feat(core): add a Worker task to keep filesystem between different task

core/src/main/java/io/kestra/core/models/tasks/Task.java

@@ -7,6 +7,7 @@
 import io.kestra.core.models.executions.TaskRun;
 import io.kestra.core.models.tasks.retrys.AbstractRetry;
 import io.kestra.core.runners.RunContext;
+import io.kestra.core.tasks.flows.Worker;
 import io.micronaut.core.annotation.Introspected;
 import lombok.Builder;
 import lombok.Getter;
@@ -108,4 +109,9 @@ public Optional<Task> findById(String id, RunContext runContext, TaskRun taskRun
     public boolean isFlowable() {
         return this instanceof FlowableTask;
     }
+
+    @JsonIgnore
+    public boolean isSendToWorkerTask() {
+        return !(this instanceof FlowableTask) || this instanceof Worker;
+    }
 }

core/src/main/java/io/kestra/core/runners/ExecutorService.java

@@ -12,6 +12,7 @@
 import io.kestra.core.models.tasks.ResolvedTask;
 import io.kestra.core.models.tasks.Task;
 import io.kestra.core.services.ConditionService;
+import io.kestra.core.tasks.flows.Worker;
 import io.kestra.core.tasks.flows.Pause;
 import io.micronaut.context.ApplicationContext;
 import jakarta.inject.Inject;
@@ -601,6 +602,27 @@ private Executor handleFlowTask(final Executor executor) {
         return resultExecutor;
     }
 
+    public Execution addDynamicTaskRun(Execution execution, Flow flow, WorkerTaskResult workerTaskResult) throws InternalException {
+        // if parent, can be a Worker task that generate dynamic tasks
+        if (workerTaskResult.getTaskRun().getParentTaskRunId() != null) {
+            try {
+                execution.findTaskRunByTaskRunId(workerTaskResult.getTaskRun().getId());
+            } catch (InternalException e) {
+                TaskRun parentTaskRun = execution.findTaskRunByTaskRunId(workerTaskResult.getTaskRun().getParentTaskRunId());
+                Task parentTask = flow.findTaskByTaskId(parentTaskRun.getTaskId());
+
+                if (parentTask instanceof Worker) {
+                    ArrayList<TaskRun> taskRuns = new ArrayList<>(execution.getTaskRunList());
+                    taskRuns.add(workerTaskResult.getTaskRun());
+
+                    return execution.withTaskRunList(taskRuns);
+                }
+            }
+        }
+
+        return null;
+    }
+
     public boolean canBePurged(final Executor executor) {
         return conditionService.isTerminatedWithListeners(executor.getFlow(), executor.getExecution())
             // we don't purge pause execution in order to be able to restart automatically in case of delay

core/src/main/java/io/kestra/core/runners/RunContext.java

@@ -5,6 +5,7 @@
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.CaseFormat;
 import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.Lists;
 import io.kestra.core.exceptions.IllegalVariableEvaluationException;
 import io.kestra.core.metrics.MetricRegistry;
 import io.kestra.core.models.executions.AbstractMetricEntry;
@@ -190,11 +191,7 @@ protected Map<String, Object> variables(Flow flow, Task task, Execution executio
         }
 
         if (task != null) {
-            builder
-                .put("task", ImmutableMap.of(
-                    "id", task.getId(),
-                    "type", task.getType()
-                ));
+            builder.put("task", this.variables(task));
         }
 
         if (taskRun != null) {
@@ -287,17 +284,85 @@ private Map<String, Object> variables(TaskRun taskRun) {
         return builder.build();
     }
 
-    public RunContext forWorker(ApplicationContext applicationContext, TaskRun taskRun) {
+    private Map<String, Object> variables(Task task) {
+        return ImmutableMap.of(
+            "id", task.getId(),
+            "type", task.getType()
+        );
+    }
+
+    @SuppressWarnings("unchecked")
+    public RunContext updateVariables(WorkerTaskResult workerTaskResult, TaskRun parent) {
+        Map<String, Object> variables = new HashMap<>(this.variables);
+
+        HashMap<String, Object> outputs = this.variables.containsKey("outputs") ?
+            new HashMap<>((Map<String, Object>) this.variables.get("outputs")) :
+            new HashMap<>();
+
+
+        Map<String, Object> result = new HashMap<>();
+        Map<String, Object> current = result;
+
+        if (variables.containsKey("parents")) {
+            for (Map<String, Map<String, String>> t : Lists.reverse((List<Map<String, Map<String, String>>>) variables.get("parents"))) {
+                if (t.get("taskrun") != null && t.get("taskrun").get("value") != null) {
+                    HashMap<String, Object> item = new HashMap<>();
+                    current.put(t.get("taskrun").get("value"), item);
+                    current = item;
+                }
+            }
+        }
+
+        if (parent.getValue() != null) {
+            HashMap<String, Object> item = new HashMap<>();
+            current.put(parent.getValue(), item);
+            current = item;
+        }
+
+        if (workerTaskResult.getTaskRun().getOutputs() != null) {
+            current.putAll(workerTaskResult.getTaskRun().getOutputs());
+        }
+
+        outputs.put(workerTaskResult.getTaskRun().getTaskId(), result);
+
+        variables.remove("outputs");
+        variables.put("outputs", outputs);
+
+        return this.clone(variables);
+    }
+
+    private RunContext clone(Map<String, Object> variables) {
+        RunContext runContext = new RunContext();
+        runContext.variableRenderer = this.variableRenderer;
+        runContext.applicationContext = this.applicationContext;
+        runContext.storageInterface = this.storageInterface;
+        runContext.storageOutputPrefix = this.storageOutputPrefix;
+        runContext.storageExecutionPrefix = this.storageExecutionPrefix;
+        runContext.envPrefix = this.envPrefix;
+        runContext.variables = variables;
+        runContext.metrics = new ArrayList<>();
+        runContext.meterRegistry = this.meterRegistry;
+        runContext.runContextLogger = this.runContextLogger;
+        runContext.tempBasedPath = this.tempBasedPath;
+        runContext.temporaryDirectory = this.temporaryDirectory;
+
+        return runContext;
+    }
+
+    public RunContext forWorker(ApplicationContext applicationContext, WorkerTask workerTask) {
         this.initBean(applicationContext);
-        this.initLogger(taskRun);
+        this.initLogger(workerTask.getTaskRun());
 
-        HashMap<String, Object> clone = new HashMap<>(this.variables);
+        Map<String, Object> clone = new HashMap<>(this.variables);
 
         clone.remove("taskrun");
-        clone.put("taskrun", this.variables(taskRun));
+        clone.put("taskrun", this.variables(workerTask.getTaskRun()));
+
+        clone.remove("task");
+        clone.put("task", this.variables(workerTask.getTask()));
 
         this.variables = ImmutableMap.copyOf(clone);
-        this.storageExecutionPrefix = URI.create("/" + this.storageInterface.executionPrefix(taskRun));
+        this.storageExecutionPrefix = URI.create("/" + this.storageInterface.executionPrefix(workerTask.getTaskRun()));
 
         return this;
     }
@@ -554,17 +619,6 @@ public void cleanup() {
         }
     }
 
-    public WorkerTask cleanup(WorkerTask workerTask) {
-        try {
-            this.cleanTemporaryDirectory();
-            return MAPPER.readValue(MAPPER.writeValueAsString(workerTask), WorkerTask.class);
-        } catch (IOException ex) {
-            logger().warn("Unable to cleanup worker task", ex);
-
-            return workerTask;
-        }
-    }
-
     private void cleanTemporaryDirectory() throws IOException {
         if (temporaryDirectory != null && temporaryDirectory.toFile().exists()) {
             FileUtils.deleteDirectory(temporaryDirectory.toFile());