Rework peon.sh

Author: GWphua

distribution/docker/peon.sh

@@ -39,11 +39,12 @@
 # - DRUID_CONFIG_${service} -- full path to a file for druid 'service' properties
 
 # This script is very similar to druid.sh, used exclusively for the kubernetes-overlord-extension.
+# Specifically, it is used by K8sTaskAdapter and PodTemplateTaskAdapter to start up Peon tasks.
 
 set -e
 SERVICE="overlord"
 
-echo "$(date -Is) startup service $SERVICE"
+echo "$(date -Is) startup service peon"
 
 # We put all the config in /tmp/conf to allow for a
 # read-only root filesystem
@@ -159,15 +160,12 @@ fi
 
 # take the ${TASK_JSON} environment variable and base64 decode, unzip and throw it in ${TASK_DIR}/task.json.
 # If TASK_JSON is not set, CliPeon will pull the task.json file from deep storage.
-if [ -n "$TASK_ID" ]; then
-    # In the case of running with Kubernetes, see https://github.com/apache/druid/pull/18206 for more information.
-    mkdir -p ${TASK_DIR}/${TASK_ID}; [ -n "$TASK_JSON" ] && echo ${TASK_JSON} | base64 -d | gzip -d > ${TASK_DIR}/${TASK_ID}/task.json;
-else
-    mkdir -p ${TASK_DIR}; [ -n "$TASK_JSON" ] && echo ${TASK_JSON} | base64 -d | gzip -d > ${TASK_DIR}/task.json;
-fi
+mkdir -p ${TASK_DIR}; [ -n "$TASK_JSON" ] && echo ${TASK_JSON} | base64 -d | gzip -d > ${TASK_DIR}/task.json;
 
+# Start peon using CliPeon, with variables `Main internal peon TASK_ID ATTEMPT_ID`
 if [ -n "$TASK_ID" ]; then
-    exec bin/run-java ${JAVA_OPTS} -cp $COMMON_CONF_DIR:$SERVICE_CONF_DIR:lib/*: org.apache.druid.cli.Main internal peon --taskId "${TASK_ID}" "$@"
+    # TASK_ID is only set from PodTemplateTaskAdapter
+    exec bin/run-java ${JAVA_OPTS} -cp $COMMON_CONF_DIR:$SERVICE_CONF_DIR:lib/*: org.apache.druid.cli.Main internal peon "${TASK_DIR}" 1 --taskId "${TASK_ID}" "$@"
 else
-    exec bin/run-java ${JAVA_OPTS} -cp $COMMON_CONF_DIR:$SERVICE_CONF_DIR:lib/*: org.apache.druid.cli.Main internal peon "$@"
+    exec bin/run-java ${JAVA_OPTS} -cp $COMMON_CONF_DIR:$SERVICE_CONF_DIR:lib/*: org.apache.druid.cli.Main internal peon "${TASK_DIR}" 1 "$@"
 fi

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

@@ -124,7 +124,7 @@ public Job fromTask(Task task) throws IOException
     PeonCommandContext context = new PeonCommandContext(
         generateCommand(task),
         javaOpts(task),
-        taskConfig.getBaseTaskDir(),
+        taskConfig.getTaskDir(task.getId()),
         taskRunnerConfig.getCpuCoreInMicro(),
         node.isEnableTlsPort()
     );
@@ -408,6 +408,9 @@ private List<String> javaOpts(Task task)
       );
     }
 
+    javaOpts.add(org.apache.druid.java.util.common.StringUtils.format(
+        "-Ddruid.indexer.task.baseTaskDir=%s", taskConfig.getBaseTaskDir().getAbsolutePath()
+    ));
     javaOpts.add(org.apache.druid.java.util.common.StringUtils.format("-Ddruid.port=%d", DruidK8sConstants.PORT));
     javaOpts.add(org.apache.druid.java.util.common.StringUtils.format(
         "-Ddruid.plaintextPort=%d",
@@ -435,8 +438,6 @@ private List<String> generateCommand(Task task)
   {
     final List<String> command = new ArrayList<>();
     command.add("/peon.sh");
-    command.add(taskConfig.getBaseTaskDir().getAbsolutePath());
-    command.add("1"); // the attemptId is always 1, we never run the task twice on the same pod.
 
     String nodeType = task.getNodeType();
     if (nodeType != null) {

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

@@ -217,7 +217,7 @@ private Collection<EnvVar> getEnv(Task task) throws IOException
     List<EnvVar> envVars = Lists.newArrayList(
         new EnvVarBuilder()
             .withName(DruidK8sConstants.TASK_DIR_ENV)
-            .withValue(taskConfig.getBaseDir())
+            .withValue(taskConfig.getTaskDir(task.getId()).getAbsolutePath())
             .build(),
         new EnvVarBuilder()
             .withName(DruidK8sConstants.TASK_ID_ENV)

services/src/main/java/org/apache/druid/cli/CliPeon.java

@@ -235,12 +235,6 @@ public void configure(Binder binder)
 
             taskDirPath = taskAndStatusFile.get(0);
             attemptId = taskAndStatusFile.get(1);
-            boolean isK8sMode = properties.getProperty("druid.indexer.runner.type", "").contains("k8s");
-            if (isK8sMode) {
-              // Need to connect task directory to include taskId.
-              // This is to ensure task reports, status files will be written and read from the correct path.
-              taskDirPath = Paths.get(taskDirPath, taskId).toString();
-            }
 
             binder.bindConstant().annotatedWith(Names.named("serviceName")).to("druid/peon");
             binder.bindConstant().annotatedWith(Names.named("servicePort")).to(0);
@@ -263,11 +257,10 @@ public void configure(Binder binder)
                 .setStatusFile(Paths.get(taskDirPath, "attempt", attemptId, "status.json").toFile());
 
             binder.bind(Properties.class).toInstance(properties);
-            if (isK8sMode) {
+            if (properties.getProperty("druid.indexer.runner.type", "").contains("k8s")) {
               log.info("Running peon in k8s mode");
               executorLifecycleConfig.setParentStreamDefined(false);
             }
-            log.info("Attempt[%s] at Running peon task in taskDirPath[%s]", attemptId, taskDirPath);
 
             binder.bind(ExecutorLifecycleConfig.class).toInstance(executorLifecycleConfig);