Repository: kestra-io/kestra
Issue: Worker temporary files cleanup 

Handle globally worker cleanup that use temporary files, for example [here](https://github.com/kestra-io/kestra/blob/develop/core/src/main/java/io/kestra/core/tasks/scripts/AbstractBash.java#L330).
It will be better to have a global function to handle this to be sure that plugins will not forgot to delete temporary files

Issue comments:

Pull request review comments:

Patch that resolved the issue:
diff --git a/core/src/main/java/io/kestra/core/models/tasks/RunnableTask.java b/core/src/main/java/io/kestra/core/models/tasks/RunnableTask.java
index e08a37a9c9..2b2cf3df68 100644
--- a/core/src/main/java/io/kestra/core/models/tasks/RunnableTask.java
+++ b/core/src/main/java/io/kestra/core/models/tasks/RunnableTask.java
@@ -2,6 +2,10 @@
 
 import io.kestra.core.runners.RunContext;
 
+import java.io.IOException;
+
 public interface RunnableTask <T extends Output> {
     T run(RunContext runContext) throws Exception;
+
+    void cleanup() throws IOException;
 }
diff --git a/core/src/main/java/io/kestra/core/models/tasks/Task.java b/core/src/main/java/io/kestra/core/models/tasks/Task.java
index 8a7fedaa53..4d514c7fed 100644
--- a/core/src/main/java/io/kestra/core/models/tasks/Task.java
+++ b/core/src/main/java/io/kestra/core/models/tasks/Task.java
@@ -8,12 +8,18 @@
 import io.kestra.core.models.tasks.retrys.AbstractRetry;
 import io.kestra.core.runners.RunContext;
 import io.micronaut.core.annotation.Introspected;
-import lombok.Builder;
-import lombok.Getter;
-import lombok.NoArgsConstructor;
+import lombok.*;
 import lombok.experimental.SuperBuilder;
+import org.apache.commons.io.FileUtils;
 
+import java.io.File;
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
 import java.time.Duration;
+import java.util.ArrayList;
+import java.util.List;
 import java.util.Optional;
 import javax.validation.Valid;
 import javax.validation.constraints.NotBlank;
@@ -49,6 +55,38 @@ abstract public class Task {
     @Builder.Default
     protected Boolean disabled = false;
 
+    @Getter(AccessLevel.NONE)
+    protected transient Path temporaryDirectory;
+
+    protected Path tempDir() throws IOException {
+        if (this.temporaryDirectory == null) {
+            this.temporaryDirectory = Files.createTempDirectory("task-temp-dir");
+        }
+
+        return this.temporaryDirectory;
+    }
+
+    protected Path tempFile() throws IOException {
+        return this.tempFile(null);
+    }
+
+    protected Path tempFile(byte[] content) throws IOException {
+        Path tempFile = Files.createTempFile(this.tempDir(), null, null);
+
+        if (content != null) {
+            Files.write(tempFile, content);
+        }
+
+        return tempFile;
+    }
+
+    public void cleanup() throws IOException {
+        if (temporaryDirectory != null) {
+            FileUtils.deleteDirectory(temporaryDirectory.toFile());
+            this.temporaryDirectory = null;
+        }
+    }
+
     public Optional<Task> findById(String id) {
         if (this.getId().equals(id)) {
             return Optional.of(this);
diff --git a/core/src/main/java/io/kestra/core/runners/Worker.java b/core/src/main/java/io/kestra/core/runners/Worker.java
index 1dfd6459f5..2b4d15d74c 100644
--- a/core/src/main/java/io/kestra/core/runners/Worker.java
+++ b/core/src/main/java/io/kestra/core/runners/Worker.java
@@ -426,6 +426,12 @@ public void run() {
                 this.exceptionHandler(this, new TimeoutExceededException(workerTask.getTask().getTimeout(), e));
             } catch (Exception e) {
                 this.exceptionHandler(this, e);
+            } finally {
+                try {
+                    task.cleanup();
+                } catch (IOException e) {
+                    log.warn("Unable to cleanup worker task", e);
+                }
             }
         }
 
diff --git a/core/src/main/java/io/kestra/core/tasks/scripts/AbstractBash.java b/core/src/main/java/io/kestra/core/tasks/scripts/AbstractBash.java
index c6dc65822b..ae5e0303b8 100644
--- a/core/src/main/java/io/kestra/core/tasks/scripts/AbstractBash.java
+++ b/core/src/main/java/io/kestra/core/tasks/scripts/AbstractBash.java
@@ -1,36 +1,30 @@
 package io.kestra.core.tasks.scripts;
 
-import com.fasterxml.jackson.core.JsonProcessingException;
-import com.fasterxml.jackson.databind.ObjectMapper;
-import io.kestra.core.utils.ExecutorsUtils;
-import io.swagger.v3.oas.annotations.media.Schema;
-import lombok.*;
-import lombok.experimental.SuperBuilder;
-import org.apache.commons.io.FileUtils;
-import io.kestra.core.exceptions.IllegalVariableEvaluationException;
 import io.kestra.core.models.annotations.PluginProperty;
 import io.kestra.core.models.executions.AbstractMetricEntry;
 import io.kestra.core.models.tasks.Task;
 import io.kestra.core.runners.RunContext;
-import io.kestra.core.serializers.JacksonMapper;
+import io.swagger.v3.oas.annotations.media.Schema;
+import lombok.*;
+import lombok.experimental.SuperBuilder;
 import org.slf4j.Logger;
 
-import java.io.*;
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStream;
 import java.net.URI;
-import java.net.URISyntaxException;
-import java.nio.file.Files;
 import java.nio.file.Path;
-import java.nio.file.Paths;
-import java.util.*;
-import java.util.function.Function;
+import java.util.AbstractMap;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
 import java.util.function.Supplier;
-import java.util.regex.Matcher;
-import java.util.regex.Pattern;
 import java.util.stream.Collectors;
 import javax.validation.constraints.NotEmpty;
 import javax.validation.constraints.NotNull;
 
-import static io.kestra.core.utils.Rethrow.*;
+import static io.kestra.core.utils.Rethrow.throwBiConsumer;
+import static io.kestra.core.utils.Rethrow.throwFunction;
 
 @SuperBuilder
 @ToString
@@ -96,8 +90,8 @@ abstract public class AbstractBash extends Task {
     @Schema(
         title = "Input files are extra files supplied by user that make it simpler organize code.",
         description = "Describe a files map that will be written and usable in execution context. In python execution " +
-            "context is in a temp folder, for bash scripts, you can reach files using a inputsDirectory variable " +
-            "like 'source {{inputsDirectory}}/myfile.sh' "
+            "context is in a temp folder, for bash scripts, you can reach files using a workingDir variable " +
+            "like 'source {{workingDir}}/myfile.sh' "
     )
     @PluginProperty(
         additionalProperties = String.class,
@@ -114,121 +108,40 @@ abstract public class AbstractBash extends Task {
     )
     protected Map<String, String> env;
 
-    @Builder.Default
-    @Getter(AccessLevel.NONE)
-    protected transient List<File> cleanupDirectory = new ArrayList<>();
-
-    @Getter(AccessLevel.NONE)
-    protected transient Path workingDirectory;
-
     @Builder.Default
     @Getter(AccessLevel.NONE)
     protected transient Map<String, Object> additionalVars = new HashMap<>();
 
-    protected Map<String, String> handleOutputFiles() {
-        List<String> outputs = new ArrayList<>();
-
-        if (this.outputFiles != null && this.outputFiles.size() > 0) {
-            outputs.addAll(this.outputFiles);
-        }
-
-        if (this.outputsFiles != null && this.outputsFiles.size() > 0) {
-            outputs.addAll(this.outputsFiles);
-        }
-
-        if (files != null && files.size() > 0) {
-            outputs.addAll(files);
-        }
-
-        Map<String, String> outputFiles = new HashMap<>();
-        if (outputs.size() > 0) {
-            outputs
-                .forEach(throwConsumer(s -> {
-                    File tempFile = File.createTempFile(s + "_", ".tmp");
-
-                    outputFiles.put(s, tempFile.getAbsolutePath());
-                }));
-
-            additionalVars.put("temp", outputFiles);
-            additionalVars.put("outputFiles", outputFiles);
-        }
-
-        return outputFiles;
-    }
-
     protected Map<String, String> finalInputFiles() throws IOException {
         return this.inputFiles;
     }
 
-    protected void handleInputFiles(RunContext runContext) throws IOException, IllegalVariableEvaluationException, URISyntaxException {
-        Map<String, String> finalInputFiles = this.finalInputFiles();
-
-        if (finalInputFiles != null && finalInputFiles.size() > 0) {
-            Path workingDirectory = tmpWorkingDirectory();
-
-            for (String fileName : finalInputFiles.keySet()) {
-                File file = new File(fileName);
-
-                // path with "/", create the subfolders
-                if (file.getParent() != null) {
-                    Path subFolder = Paths.get(
-                        workingDirectory.toAbsolutePath().toString(),
-                        new File(fileName).getParent()
-                    );
-
-                    if (!subFolder.toFile().exists()) {
-                        Files.createDirectories(subFolder);
-                    }
-                }
-
-                String filePath = workingDirectory + "/" + fileName;
-                String render = runContext.render(finalInputFiles.get(fileName), additionalVars);
-
-                if (render.startsWith("kestra://")) {
-                    try (
-                        InputStream inputStream = runContext.uriToInputStream(new URI(render));
-                        OutputStream outputStream = new BufferedOutputStream(new FileOutputStream(filePath))
-                    ) {
-                        int byteRead;
-                        while ((byteRead = inputStream.read()) != -1) {
-                            outputStream.write(byteRead);
-                        }
-                        outputStream.flush();
-                    }
-                } else {
-                    try (BufferedWriter writer = new BufferedWriter(new FileWriter(filePath))) {
-                        writer.write(render);
-                    }
-                }
-            }
-        }
-    }
-
     protected List<String> finalCommandsWithInterpreter(String commandAsString) throws IOException {
-        // build the final commands
-        List<String> commandsWithInterpreter = new ArrayList<>(Collections.singletonList(this.interpreter));
-
-        File bashTempFiles = null;
-        // https://www.in-ulm.de/~mascheck/various/argmax/ MAX_ARG_STRLEN (131072)
-        if (commandAsString.length() > 131072) {
-            bashTempFiles = File.createTempFile("bash", ".sh", this.tmpWorkingDirectory().toFile());
-            Files.write(bashTempFiles.toPath(), commandAsString.getBytes());
-
-            commandAsString = bashTempFiles.getAbsolutePath();
-        } else {
-            commandsWithInterpreter.addAll(Arrays.asList(this.interpreterArgs));
-        }
-
-        commandsWithInterpreter.add(commandAsString);
-
-        return commandsWithInterpreter;
+        return BashService.finalCommandsWithInterpreter(
+            this.tempDir(),
+            this.interpreter,
+            this.interpreterArgs,
+            commandAsString
+        );
     }
 
     protected ScriptOutput run(RunContext runContext, Supplier<String> supplier) throws Exception {
         Logger logger = runContext.logger();
 
-        Map<String, String> outputFiles = this.handleOutputFiles();
-        this.handleInputFiles(runContext);
+        Path workingDirectory = this.tempDir();
+
+        Map<String, String> outputFiles = BashService.createOutputFiles(
+            workingDirectory,
+            this.outputFiles,
+            additionalVars
+        );
+
+        BashService.createInputFiles(
+            runContext,
+            workingDirectory,
+            this.finalInputFiles(),
+            additionalVars
+        );
 
         String commandAsString = supplier.get();
 
@@ -252,9 +165,7 @@ protected ScriptOutput run(RunContext runContext, Supplier<String> supplier) thr
         Map<String, URI> uploaded = new HashMap<>();
 
         outputFiles.
-            forEach(throwBiConsumer((k, v) -> {
-                uploaded.put(k, runContext.putTempFile(new File(v)));
-            }));
+            forEach(throwBiConsumer((k, v) -> uploaded.put(k, runContext.putTempFile(new File(runContext.render(v, additionalVars))))));
 
         this.cleanup();
 
@@ -327,20 +238,11 @@ protected RunResult run(RunContext runContext, Logger logger, Path workingDirect
         }
     }
 
-    protected void cleanup() throws IOException {
-        for (File folder : cleanupDirectory) {
-            FileUtils.deleteDirectory(folder);
-        }
-    }
-
-    protected Path tmpWorkingDirectory() throws IOException {
-        if (this.workingDirectory == null) {
-            this.workingDirectory = Files.createTempDirectory("working-dir");
-            this.cleanupDirectory.add(workingDirectory.toFile());
-            additionalVars.put("workingDir", workingDirectory.toAbsolutePath().toString());
-        }
+    protected Path tempDir() throws IOException {
+        Path path = super.tempDir();
+        additionalVars.put("workingDir", path.toAbsolutePath().toString());
 
-        return this.workingDirectory;
+        return path;
     }
 
     @NoArgsConstructor
@@ -348,7 +250,6 @@ protected Path tmpWorkingDirectory() throws IOException {
     public static class BashCommand <T> {
         private Map<String, Object> outputs;
         private List<AbstractMetricEntry<T>> metrics;
-
     }
 
     @FunctionalInterface
@@ -357,9 +258,6 @@ public interface LogSupplier {
     }
 
     public static class LogThread extends AbstractLogThread {
-        protected static final ObjectMapper MAPPER = JacksonMapper.ofJson();
-        private static final Pattern PATTERN = Pattern.compile("^::(\\{.*\\})::$");
-
         private final Logger logger;
         private final boolean isStdErr;
         private final RunContext runContext;
@@ -373,7 +271,7 @@ public LogThread(Logger logger, InputStream inputStream, boolean isStdErr, RunCo
         }
 
         protected void call(String line) {
-            this.parseOut(line, logger, runContext);
+            outputs.putAll(BashService.parseOut(line, logger, runContext));
 
             if (isStdErr) {
                 logger.warn(line);
@@ -381,27 +279,6 @@ protected void call(String line) {
                 logger.info(line);
             }
         }
-
-        protected void parseOut(String line, Logger logger, RunContext runContext)  {
-            Matcher m = PATTERN.matcher(line);
-
-            if (m.find()) {
-                try {
-                    BashCommand<?> bashCommand = MAPPER.readValue(m.group(1), BashCommand.class);
-
-                    if (bashCommand.outputs != null) {
-                        outputs.putAll(bashCommand.outputs);
-                    }
-
-                    if (bashCommand.metrics != null) {
-                        bashCommand.metrics.forEach(runContext::metric);
-                    }
-                }
-                catch (JsonProcessingException e) {
-                    logger.warn("Invalid outputs '{}'", e.getMessage(), e);
-                }
-            }
-        }
     }
 
     @Getter
@@ -419,6 +296,5 @@ public BashException(int exitCode, int stdOutSize, int stdErrSize) {
             this.stdOutSize = stdOutSize;
             this.stdErrSize = stdErrSize;
         }
-
     }
 }
diff --git a/core/src/main/java/io/kestra/core/tasks/scripts/AbstractPython.java b/core/src/main/java/io/kestra/core/tasks/scripts/AbstractPython.java
index 38b553dd1b..2223485d9c 100644
--- a/core/src/main/java/io/kestra/core/tasks/scripts/AbstractPython.java
+++ b/core/src/main/java/io/kestra/core/tasks/scripts/AbstractPython.java
@@ -99,7 +99,7 @@ public class AbstractPython extends AbstractBash {
 
     protected String virtualEnvCommand(RunContext runContext, List<String> requirements) throws IllegalVariableEvaluationException, IOException {
         List<String> renderer = new ArrayList<>();
-        Path workingDirectory = this.tmpWorkingDirectory();
+        Path workingDirectory = this.tempDir();
 
         if (this.exitOnFailed) {
             renderer.add("set -o errexit");
diff --git a/core/src/main/java/io/kestra/core/tasks/scripts/BashService.java b/core/src/main/java/io/kestra/core/tasks/scripts/BashService.java
new file mode 100644
index 0000000000..24340b434e
--- /dev/null
+++ b/core/src/main/java/io/kestra/core/tasks/scripts/BashService.java
@@ -0,0 +1,146 @@
+package io.kestra.core.tasks.scripts;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import io.kestra.core.exceptions.IllegalVariableEvaluationException;
+import io.kestra.core.runners.RunContext;
+import io.kestra.core.serializers.JacksonMapper;
+import org.slf4j.Logger;
+
+import java.io.*;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.*;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+import static io.kestra.core.utils.Rethrow.throwConsumer;
+
+abstract public class BashService {
+    protected static final ObjectMapper MAPPER = JacksonMapper.ofJson();
+    private static final Pattern PATTERN = Pattern.compile("^::(\\{.*\\})::$");
+
+    public static List<String> finalCommandsWithInterpreter(
+        Path workingDirectory,
+        String interpreter,
+        String[] interpreterArgs,
+        String commandAsString
+    ) throws IOException {
+        // build the final commands
+        List<String> commandsWithInterpreter = new ArrayList<>(Collections.singletonList(interpreter));
+
+        File bashTempFiles = null;
+        // https://www.in-ulm.de/~mascheck/various/argmax/ MAX_ARG_STRLEN (131072)
+        if (commandAsString.length() > 131072) {
+            bashTempFiles = File.createTempFile("bash", ".sh", workingDirectory.toFile());
+            Files.write(bashTempFiles.toPath(), commandAsString.getBytes());
+
+            commandAsString = bashTempFiles.getAbsolutePath();
+        } else {
+            commandsWithInterpreter.addAll(Arrays.asList(interpreterArgs));
+        }
+
+        commandsWithInterpreter.add(commandAsString);
+
+        return commandsWithInterpreter;
+    }
+
+    public static Map<String, String> createOutputFiles(
+        Path workingDirectory,
+        List<String> outputFiles,
+        Map<String, Object> additionalVars
+    ) {
+        List<String> outputs = new ArrayList<>();
+
+        if (outputFiles != null && outputFiles.size() > 0) {
+            outputs.addAll(outputFiles);
+        }
+
+        Map<String, String> result = new HashMap<>();
+        if (outputs.size() > 0) {
+            outputs
+                .forEach(throwConsumer(s -> {
+                    File tempFile = File.createTempFile(s + "_", ".tmp", workingDirectory.toFile());
+
+                    result.put(s, "{{workingDir}}/" + tempFile.getName());
+                }));
+
+            additionalVars.put("temp", result);
+            additionalVars.put("outputFiles", result);
+        }
+
+        return result;
+    }
+
+    public static void createInputFiles(
+        RunContext runContext,
+        Path workingDirectory,
+        Map<String, String> inputFiles,
+        Map<String, Object> additionalVars
+    ) throws IOException, IllegalVariableEvaluationException, URISyntaxException {
+        if (inputFiles != null && inputFiles.size() > 0) {
+            for (String fileName : inputFiles.keySet()) {
+                File file = new File(fileName);
+
+                // path with "/", create the subfolders
+                if (file.getParent() != null) {
+                    Path subFolder = Paths.get(
+                        workingDirectory.toAbsolutePath().toString(),
+                        new File(fileName).getParent()
+                    );
+
+                    if (!subFolder.toFile().exists()) {
+                        Files.createDirectories(subFolder);
+                    }
+                }
+
+                String filePath = workingDirectory + "/" + fileName;
+                String render = runContext.render(inputFiles.get(fileName), additionalVars);
+
+                if (render.startsWith("kestra://")) {
+                    try (
+                        InputStream inputStream = runContext.uriToInputStream(new URI(render));
+                        OutputStream outputStream = new BufferedOutputStream(new FileOutputStream(filePath))
+                    ) {
+                        int byteRead;
+                        while ((byteRead = inputStream.read()) != -1) {
+                            outputStream.write(byteRead);
+                        }
+                        outputStream.flush();
+                    }
+                } else {
+                    try (BufferedWriter writer = new BufferedWriter(new FileWriter(filePath))) {
+                        writer.write(render);
+                    }
+                }
+            }
+        }
+    }
+
+    public static Map<String, Object> parseOut(String line, Logger logger, RunContext runContext)  {
+        Matcher m = PATTERN.matcher(line);
+        Map<String, Object> outputs = new HashMap<>();
+
+        if (m.find()) {
+            try {
+                AbstractBash.BashCommand<?> bashCommand = MAPPER.readValue(m.group(1), AbstractBash.BashCommand.class);
+
+                if (bashCommand.getOutputs() != null) {
+                    outputs.putAll(bashCommand.getOutputs());
+                }
+
+                if (bashCommand.getMetrics() != null) {
+                    bashCommand.getMetrics().forEach(runContext::metric);
+                }
+            }
+            catch (JsonProcessingException e) {
+                logger.warn("Invalid outputs '{}'", e.getMessage(), e);
+            }
+        }
+
+        return outputs;
+    }
+}


Test patch: