Making task log storage optional for Kubernetes Runner

[uds5501]

Problem Statement

In current task runner setups, we utilize a common TaskLog interface to perform all the operations like log pushes, log streams, payload and report pushes. In future, we want to provide the flexibility to the end user to be able to perform the following things simultaneously:

• Streaming task logs from an external store ( elasticsearch / datadog / deepstorage providers like s3 etc).
• Pushing task logs to separate location that is independent of streaming implementation ( could be stdout / deepstorage [currently supported] etc).

Description

The approach in general is this:

• I am using a SwitchingTaskLogs implementation that picks up appropriate options from indexer configuration set in overlord and uses a default type to push and stream reports / statuses etc while uses specific implementations for log push. / log stream if configured.

Property	Default
druid.indexer.logs.switching.defaultType	file
druid.indexer.logs.switching.logPushType	use defaultType
druid.indexer.logs.switching.logStreamType	use defaultType
druid.indexer.logs.switching.reportsType	use defaultType

configs as suggested by : @kfaraz ^

• These configs are bound at 3 areas right now:
  • CliOverlord.java
  • KubernetesOverlordModule.java
  • IndexingServiceTaskLogsModule.java

---

Key changed/added classes in this PR

• KubernetesPeonLifecycle
• CliOverlord.java
• KubernetesOverlordModule.java
• IndexingServiceTaskLogsModule.java

TODO(s) [notes for author]

• Call the binder from all the deep-storage extensions.
• Verify switching setup with K8S Runners.

[cryptoe]

Since this effort is more related to have task logs being managed externally I purpose we create a new class like

public class ExternalTaskLogs implements TaskLogs {
private TaskLogs delegate;
private ExternalLogStreamer externalLogStreamer;

@Inject
public ExternalLogStreamer(TaskLogs taskLogs, ExternalLogStreamer externalLogStreamer){
    this.delegate = taskLogs;
    this.externalLogStreamer = externalLogStreamer;
}


@Override
Optional<InputStream> streamTaskLog(String taskid, long offset) throws IOException {
    return externalLogStreamer.streamTaskLog();
};

@Override
Optional<InputStream> streamTaskReports(final String taskid) throws IOException {
    return delegate.streamTaskReports();
}

@Override
Optional<InputStream> streamTaskStatus(final String taskid) throws IOException {
    return delegate.streamTaskStatus();
}
  
@Override
void pushTaskLog(String taskid, File logFile) throws IOException {
    // do nothing. 
};

@Override
void pushTaskReports(String taskid, File reportFile) throws IOException {
    delegate.pushTaskReports();
}

@Override
void pushTaskStatus(String taskid, File reportFile) throws IOException {
    delegate.pushTaskStatus();
}
  {
  }

  @Override
  void killAll() throws IOException {
    delegate.killAll();
  };

  @Override
  void killOlderThan(long timestamp) throws IOException {
    delegate.killOlderThan();
  };

@Override
void pushTaskPayload(String taskid, File taskPayloadFile) throws IOException {
    delegate.pushTaskPayload();
}


@Override
  Optional<InputStream> streamTaskPayload(String taskid) throws IOException {
    return delegate.streamTaskPayload();
  }
  
}


public interface ExternalLogStreamer {
    Optional<InputStream> streamTaskLog(String taskid) throws IOException;
}

We can then bind the implementation of the ExternalLogStreamer as we like it.

[kfaraz]

Left some suggestions.

[kfaraz]

It can be a little confusing to have say these props set on the Overlord:

druid.indexer.logs.type=external
druid.indexer.logs.delegate.type=file

It is unclear why we need a delegate if we just wanted FileTaskLogs.

I think the intention is to direct task reports, status and payload to one place and logs to another.
So we need some kind of switching or composite task logs type, which would give us these props:

druid.indexer.logs.type=switching
druid.indexer.logs.switching.reportsType=file
druid.indexer.logs.switching.logsType=hdfs

With this, we just need a SwitchingTaskLogs implementation which takes in a reportsType and a logsType, both of which would be TaskLogs implementations.

[kfaraz]

To ensure that extensions register themselves correctly as a valid type for say druid.indexer.logs.switching.logsType, you would need to add a utility method in Binders, say bindTaskLogs and invoke it from all the relevant extension modules.

public static <T extends TaskLogs> void bindTaskLogs(Binder binder, String type, Class<T> clazz) {
    // bind to `druid.indexer.logs.type`
    // bind to `druid.indexer.logs.switching.reportsType`
    // bind to `druid.indexer.logs.switching.logsType`
}

[uds5501]

druid.indexer.logs.switching.logsType=hdfs

The options I am trying to provide here is different implementations of logStorage vs logStreaming, maybe I could just extend this to druid.indexer.logs.switching.logs.storage and druid.indexer.logs.switching.streaming ?

[kfaraz]

I see. Yes, the same switching concept can be extended for that use case as well.

druid.indexer.logs.switching.streaming.logsPushType=noop
druid.indexer.logs.switching.streaming.logsStreamType=file

where logsPushType would bind to a TaskLogPusher object.
and logsStreamType would bind to a TaskLogStreamer object.

Would you still need to distinguish between reports and logs though?
Because I assume reports might need to be pushed to a different place.
If yes, you would still need a

druid.indexer.logs.switching.streaming.reportsType=noop/file/hdfs

[uds5501]

This sounds good, in my head I am looking to provide the flexibility just for log types for now. The delegated / reportType could take care of other task responsibilities. So finally, imo it should look like one of these.

Option 1 [asking for explicit types for each responsibility]

druid.indexer.logs.type=switching
druid.indexer.logs.switching.reportsType=file
druid.indexer.logs.switching.payloadManagerType=file
druid.indexer.logs.switching.logsType=switching
druid.indexer.logs.switching.logsType.logsPushType=noop
druid.indexer.logs.switching.logsType.logsStreamType=file

translating to

public class SwitchingTaskLogs implements TaskLogs
{
  private final TaskLogs reportDelegate;
  private final TaskLogs payloadDelegate;
  private final TaskLogs logDelegate;

  @Inject
  public SwitchingTaskLogs(
      @Named("report") TaskLogs reportDelegate,
      @Named("payload") TaskLogs payloadDelegate,
      @Named("log") TaskLogs logDelegate
   )
  {

    this.reportDelegate = reportDelegate;
    this.payloadDelegate = payloadDelegate;
    this.logDelegate = logDelegate;
  }
}

public class TaskLogManager implements TaskLogs  // this / some flavour of switching task log will be injected as switching log delegate to handle the task related responsibilities with an appropriate factory placed in front.
{
  private final TaskLogStreamer streamer;
  private final TaskLogPusher pusher; 

  @Inject
  public TaskLogManager(
      @Named("taskLogPusher") TaskLogs pusher,
      @Named("taskLogStreamer") TaskLogs streamer,
   )
  {
    this.pusher = pusher;
    this.streamer = streamer;
  }
}

or we could create one like

Option 2 [just keeping log handler separate while keeping the rest of the responsibilities with the delegate class]

druid.indexer.logs.type=switching
druid.indexer.logs.switching.delegate=file
druid.indexer.logs.switching.logsHandlerType=switching
druid.indexer.logs.switching.logHandler.logsPushType=noop
druid.indexer.logs.switching.logHandler.logsStreamType=file

public class SwitchingTaskLogs implements TaskLogs
{
  private final TaskLogs delegate;
  private final TaskLogs switchingLogDelegate; 

  @Inject
  public SwitchingTaskLogs(
      @Named("default") TaskLogs defaultDelegate,
      @Named("log") TaskLogs logDelegate,
   )
  {
    this.defaultDelegate = defaultDelegate;
    this.logDelegate = logDelegate;
  }
}

// the TaskLogManager stays the same.

I am personally leaning towards option 2 given the usecase in hand. what do you folks think?

---

EDIT: managed to make it work with a single SwitchingTaskLogs with factory. I've added the implementation in this PR for the same.

[kfaraz]

@uds5501 , I don't think we need two levels of a switching implementation. That would really complicate the configs.

You can provide some decent defaults

When druid.indexer.logs.type=switching

Property	Default
druid.indexer.logs.switching.defaultType	file
druid.indexer.logs.switching.logPushType	use defaultType
druid.indexer.logs.switching.logStreamType	use defaultType
druid.indexer.logs.switching.reportsType	use defaultType

I don't think report and payload will ever need to be different.
If needed, we can add it in the future.

[uds5501]

Hmm, I see, will use this.

[uds5501]

Using this combination noew.

[uds5501]

I have verified the noop task streaming options in CLI Overlord.

Used the following configuration in embedded test setup to check the same -

overlord.addProperty("druid.indexer.task.default.context", "{\"useConcurrentLocks\": true}")
            .addProperty("druid.manager.segments.useIncrementalCache", "ifSynced")
            .addProperty("druid.manager.segments.pollDuration", "PT0.1s")
            .addProperty("druid.manager.segments.killUnused.enabled", "true")
            .addProperty("druid.manager.segments.killUnused.bufferPeriod", "PT0.1s")
            .addProperty("druid.manager.segments.killUnused.dutyPeriod", "PT1s")
            .addProperty("druid.indexer.logs.type", "switching")
            .addProperty("druid.indexer.logs.switching.defaultType", "file")
            .addProperty("druid.indexer.logs.switching.streamType", "noop");

Next test: Using this in conjunction with Kubernetes Task Runners.

[kfaraz]

Thanks for the update, @uds5501 ! I will try to take a look at the changes today.

[uds5501]

Tested locally, the logs streaming is done via s3 as expected, the log pushes + temporary log savings have been shut down (noop).

Task status -

{
  "id": "noindex_parallel_kttm1_mbhlngbb_2025-08-04T09:26:54.347Z",
  "groupId": "noindex_parallel_kttm1_mbhlngbb_2025-08-04T09:26:54.347Z",
  "type": "noindex_parallel",
  "createdTime": "2025-08-04T09:26:54.359Z",
  "queueInsertionTime": "1970-01-01T00:00:00.000Z",
  "statusCode": "SUCCESS",
  "status": "SUCCESS",
  "runnerStatusCode": "WAITING",
  "duration": 51000,
  "location": {
    "host": "10.244.0.153",
    "port": 8100,
    "tlsPort": -1,
    "k8sPodName": "noindexparallelkttm1mbhlngbb2025-d8c190f944e0b8893f24ce3edfdb7q9r"
  },
  "dataSource": "kttm1",
  "errorMsg": null
}

report -

{
  "ingestionState": "COMPLETED",
  "unparseableEvents": {},
  "rowStats": {
    "determinePartitions": {
      "processed": 465346,
      "processedBytes": 360464067,
      "processedWithError": 0,
      "thrownAway": 0,
      "unparseable": 0
    },
    "buildSegments": {
      "processed": 465346,
      "processedBytes": 360464067,
      "processedWithError": 0,
      "thrownAway": 0,
      "unparseable": 0
    }
  },
  "errorMsg": null,
  "segmentAvailabilityConfirmed": false,
  "segmentAvailabilityWaitTimeMs": 0,
  "recordsProcessed": {},
  "segmentsPublished": 1
}

[kfaraz]

Thanks for the PR, @uds5501 !
I have left some thoughts.

[kfaraz]

It seems weird to have a flag to disable the one thing that this interface is meant to do.
Given that we already have a noop impl, where do we need this?

[uds5501]

The TaskLogPusher has varied responsibilities internally (reports, logs, statuses etc). This boolean helps in the scenario of K8S runners where it creates a temp file before pushing the logs to deep storage. This method helps skip the entire step

[kfaraz]

This will help you retain the zero arg constructor.

Suggested change

	public IndexingServiceTaskLogsModule(Properties props)
	public setProperties(Properties props)

[uds5501]

This didn't workout. It seems that configure() happens first and then the @inject is triggered for Modules.
Keeping it as is for the time being.

[kfaraz]

Minor nitpicks.

[kfaraz]

Thanks for fixing up the docker test job, @uds5501 !

As a follow up to this PR, could you please take up the following:

• Add docs for the new configs added here
• Add a Release note section to the description of this PR
• Add embedded tests for task logs