Making task log storage optional for Kubernetes Runner

extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/KubernetesOverlordModule.java

@@ -29,6 +29,7 @@
 import com.google.inject.Provides;
 import com.google.inject.multibindings.MapBinder;
 import com.google.inject.name.Named;
+import com.google.inject.name.Names;
 import io.fabric8.kubernetes.client.Config;
 import io.fabric8.kubernetes.client.ConfigBuilder;
 import org.apache.druid.discovery.NodeRole;
@@ -45,6 +46,8 @@
 import org.apache.druid.guice.annotations.Smile;
 import org.apache.druid.indexing.common.config.FileTaskLogsConfig;
 import org.apache.druid.indexing.common.config.TaskConfig;
+import org.apache.druid.indexing.common.tasklogs.ExternalLogStreamer;
+import org.apache.druid.indexing.common.tasklogs.ExternalTaskLogs;
 import org.apache.druid.indexing.common.tasklogs.FileTaskLogs;
 import org.apache.druid.indexing.overlord.RemoteTaskRunnerFactory;
 import org.apache.druid.indexing.overlord.TaskRunnerFactory;
@@ -290,13 +293,22 @@ public RunnerStrategy get()
   private void configureTaskLogs(Binder binder)
   {
     PolyBind.createChoice(binder, "druid.indexer.logs.type", Key.get(TaskLogs.class), Key.get(FileTaskLogs.class));
+    PolyBind.createChoice(binder, "druid.indexer.logs.delegate.type", Key.get(TaskLogs.class, Names.named("delegate")), Key.get(FileTaskLogs.class));
     JsonConfigProvider.bind(binder, "druid.indexer.logs", FileTaskLogsConfig.class);
 
     final MapBinder<String, TaskLogs> taskLogBinder = Binders.taskLogsBinder(binder);
     taskLogBinder.addBinding("noop").to(NoopTaskLogs.class).in(LazySingleton.class);
     taskLogBinder.addBinding("file").to(FileTaskLogs.class).in(LazySingleton.class);
+    taskLogBinder.addBinding("external").to(ExternalTaskLogs.class).in(LazySingleton.class);
+
+    final MapBinder<String, TaskLogs> delegateTaskLogBinder = PolyBind.optionBinder(binder, Key.get(TaskLogs.class, Names.named("delegate")));
+    delegateTaskLogBinder.addBinding("noop").to(NoopTaskLogs.class).in(LazySingleton.class);
+    delegateTaskLogBinder.addBinding("file").to(FileTaskLogs.class).in(LazySingleton.class);
+
     binder.bind(NoopTaskLogs.class).in(LazySingleton.class);
     binder.bind(FileTaskLogs.class).in(LazySingleton.class);
+    binder.bind(ExternalTaskLogs.class).in(LazySingleton.class);
+    binder.bind(ExternalLogStreamer.class).in(LazySingleton.class);
 
     binder.bind(TaskLogPusher.class).to(TaskLogs.class);
     binder.bind(TaskLogKiller.class).to(TaskLogs.class);

noindexing-service/src/main/java/org/apache/druid/guice/IndexingServiceTaskLogsModule.java

@@ -23,7 +23,10 @@
 import com.google.inject.Key;
 import com.google.inject.Module;
 import com.google.inject.multibindings.MapBinder;
+import com.google.inject.name.Names;
 import org.apache.druid.indexing.common.config.FileTaskLogsConfig;
+import org.apache.druid.indexing.common.tasklogs.ExternalLogStreamer;
+import org.apache.druid.indexing.common.tasklogs.ExternalTaskLogs;
 import org.apache.druid.indexing.common.tasklogs.FileTaskLogs;
 import org.apache.druid.tasklogs.NoopTaskLogs;
 import org.apache.druid.tasklogs.TaskLogKiller;
@@ -39,13 +42,22 @@ public class IndexingServiceTaskLogsModule implements Module
   public void configure(Binder binder)
   {
     PolyBind.createChoice(binder, "druid.indexer.logs.type", Key.get(TaskLogs.class), Key.get(FileTaskLogs.class));
+    PolyBind.createChoice(binder, "druid.indexer.logs.delegate.type", Key.get(TaskLogs.class, Names.named("delegate")), Key.get(FileTaskLogs.class));
     JsonConfigProvider.bind(binder, "druid.indexer.logs", FileTaskLogsConfig.class);
 
     final MapBinder<String, TaskLogs> taskLogBinder = Binders.taskLogsBinder(binder);
     taskLogBinder.addBinding("noop").to(NoopTaskLogs.class).in(LazySingleton.class);
     taskLogBinder.addBinding("file").to(FileTaskLogs.class).in(LazySingleton.class);
+    taskLogBinder.addBinding("external").to(ExternalTaskLogs.class).in(LazySingleton.class);
+
+    final MapBinder<String, TaskLogs> delegateTaskLogBinder = PolyBind.optionBinder(binder, Key.get(TaskLogs.class, Names.named("delegate")));
+    delegateTaskLogBinder.addBinding("noop").to(NoopTaskLogs.class).in(LazySingleton.class);
+    delegateTaskLogBinder.addBinding("file").to(FileTaskLogs.class).in(LazySingleton.class);
+
     binder.bind(NoopTaskLogs.class).in(LazySingleton.class);
     binder.bind(FileTaskLogs.class).in(LazySingleton.class);
+    binder.bind(ExternalTaskLogs.class).in(LazySingleton.class);
+    binder.bind(ExternalLogStreamer.class).in(LazySingleton.class);
 
     binder.bind(TaskLogPusher.class).to(TaskLogs.class);
     binder.bind(TaskLogKiller.class).to(TaskLogs.class);

noindexing-service/src/main/java/org/apache/druid/indexing/common/tasklogs/ExternalLogStreamer.java

@@ -0,0 +1,35 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.indexing.common.tasklogs;
+
+import com.google.common.base.Optional;
+import org.apache.druid.tasklogs.TaskLogStreamer;
+
+import java.io.IOException;
+import java.io.InputStream;
+
+public class ExternalLogStreamer implements TaskLogStreamer
+{
+  @Override
+  public Optional<InputStream> streamTaskLog(String taskid, long offset) throws IOException
+  {
+    return Optional.absent();
+  }
+}

noindexing-service/src/main/java/org/apache/druid/indexing/common/tasklogs/ExternalTaskLogs.java

@@ -0,0 +1,104 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.indexing.common.tasklogs;
+
+import com.google.common.base.Optional;
+import com.google.inject.Inject;
+import com.google.inject.name.Named;
+import org.apache.druid.java.util.emitter.EmittingLogger;
+import org.apache.druid.tasklogs.TaskLogs;
+
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStream;
+
+public class ExternalTaskLogs implements TaskLogs
+{
+  private static final EmittingLogger log = new EmittingLogger(ExternalTaskLogs.class);
+  private final TaskLogs delegate;
+  private final ExternalLogStreamer logStreamer;
+
+  @Inject
+  public ExternalTaskLogs(@Named("delegate") TaskLogs delegate, ExternalLogStreamer logStreamer)
+  {
+    this.delegate = delegate;
+    this.logStreamer = logStreamer;
+  }
+
+  @Override
+  public Optional<InputStream> streamTaskLog(String taskid, long offset) throws IOException
+  {
+    return logStreamer.streamTaskLog(taskid, offset);
+  }
+
+  @Override
+  public Optional<InputStream> streamTaskReports(final String taskid) throws IOException
+  {
+    return delegate.streamTaskReports(taskid);
+  }
+
+  @Override
+  public Optional<InputStream> streamTaskStatus(final String taskid) throws IOException
+  {
+    return delegate.streamTaskReports(taskid);
+  }
+
+  @Override
+  public void pushTaskLog(String taskid, File logFile)
+  {
+    log.debug("Skipping task log push for task[%s]", taskid);
+  }
+
+  @Override
+  public void pushTaskPayload(String taskid, File taskPayloadFile) throws IOException
+  {
+    delegate.pushTaskPayload(taskid, taskPayloadFile);
+  }
+
+  @Override
+  public void killAll() throws IOException
+  {
+    delegate.killAll();
+  }
+
+  @Override
+  public void killOlderThan(long timestamp) throws IOException
+  {
+    delegate.killOlderThan(timestamp);
+  }
+
+  @Override
+  public void pushTaskReports(String taskid, File reportFile) throws IOException
+  {
+    delegate.pushTaskReports(taskid, reportFile);
+  }
+
+  @Override
+  public void pushTaskStatus(String taskid, File reportFile) throws IOException
+  {
+    delegate.pushTaskStatus(taskid, reportFile);
+  }
+
+  @Override
+  public Optional<InputStream> streamTaskPayload(String taskid) throws IOException
+  {
+    return delegate.streamTaskPayload(taskid);
+  }
+}