Allow Bash & Python task to have some outputs & metrics
For now, Bash & Python take the whole log as output. 
This output is unusable and can crash the executor. 

We need to remove the log from output ! 
But at the same time, we will let log special string that will be let have some outputs & metrics ! 

I think mostly to issue the same concept than [github actions](https://docs.github.com/en/free-pro-team@latest/actions/reference/workflow-commands-for-github-actions).

example: 
* `::<output name="{name}" value="{value}" />::`
* `::<outputs>{json}</outputs>::`
* `::<metrics name="{name}" type="{type}" value="{value}" />::`

core/src/main/java/org/kestra/core/tasks/scripts/AbstractBash.java core/src/main/java/org/kestra/core/tasks/scripts/Node.java core/src/main/java/org/kestra/core/tasks/scripts/Python.java
core/src/main/java/org/kestra/core/tasks/scripts/AbstractBash.java core/src/main/java/org/kestra/core/tasks/scripts/Node.java core/src/main/java/org/kestra/core/tasks/scripts/Python.java core/src/main/resources/scripts/kestra.js core/src/main/resources/scripts/kestra.py
core/src/test/java/org/kestra/core/docs/JsonSchemaGeneratorTest.java core/src/test/java/org/kestra/core/tasks/BashTest.java core/src/test/java/org/kestra/core/tasks/NodeTest.java core/src/test/java/org/kestra/core/tasks/PythonTest.java webserver/src/test/java/org/kestra/webserver/controllers/PluginControllerTest.java
diff --git a/core/src/main/java/org/kestra/core/tasks/scripts/AbstractBash.java b/core/src/main/java/org/kestra/core/tasks/scripts/AbstractBash.java
index 9e8ba3d02b..68728123ed 100644
--- a/core/src/main/java/org/kestra/core/tasks/scripts/AbstractBash.java
+++ b/core/src/main/java/org/kestra/core/tasks/scripts/AbstractBash.java
@@ -1,14 +1,18 @@
 package org.kestra.core.tasks.scripts;
 
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
 import io.swagger.v3.oas.annotations.media.Schema;
 import lombok.*;
 import lombok.experimental.SuperBuilder;
 import org.apache.commons.io.FileUtils;
 import org.kestra.core.exceptions.IllegalVariableEvaluationException;
 import org.kestra.core.models.annotations.PluginProperty;
+import org.kestra.core.models.executions.AbstractMetricEntry;
 import org.kestra.core.models.tasks.RunnableTask;
 import org.kestra.core.models.tasks.Task;
 import org.kestra.core.runners.RunContext;
+import org.kestra.core.serializers.JacksonMapper;
 import org.slf4j.Logger;
 
 import java.io.*;
@@ -19,6 +23,8 @@
 import java.nio.file.Paths;
 import java.util.*;
 import java.util.function.Function;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
 import javax.validation.constraints.NotEmpty;
 import javax.validation.constraints.NotNull;
 
@@ -64,6 +70,7 @@ abstract public class AbstractBash extends Task implements RunnableTask<Abstract
         deprecated = true
     )
     @PluginProperty(dynamic = true)
+    @Deprecated
     protected List<String> files;
 
     @Schema(
@@ -91,6 +98,7 @@ abstract public class AbstractBash extends Task implements RunnableTask<Abstract
     @Builder.Default
     protected transient List<File> cleanupDirectory = new ArrayList<>();
     protected transient Path workingDirectory;
+    protected static transient ObjectMapper mapper = JacksonMapper.ofJson();
 
     protected Map<String, String> handleOutputFiles(Map<String, Object> additionalVars) throws IOException {
         List<String> outputs = new ArrayList<>();
@@ -195,8 +203,6 @@ protected AbstractBash.Output run(RunContext runContext, Function<Map<String, Ob
 
         int exitCode = process.waitFor();
 
-        this.cleanup();
-
         if (exitCode != 0) {
             stdOut.join();
             stdErr.join();
@@ -224,11 +230,18 @@ protected AbstractBash.Output run(RunContext runContext, Function<Map<String, Ob
             bashTempFiles.delete();
         }
 
+        this.cleanup();
+
+        Map<String, Object> outputs = new HashMap<>();
+        outputs.putAll(parseOut(runContext, stdOut.getLogs()));
+        outputs.putAll(parseOut(runContext, stdErr.getLogs()));
+
         // output
         return Output.builder()
             .exitCode(exitCode)
-            .stdOut(stdOut.getLogs())
-            .stdErr(stdErr.getLogs())
+            .stdOutLineCount(stdOut.getLogs().size())
+            .stdErrLineCount(stdErr.getLogs().size())
+            .vars(outputs)
             .files(uploaded)
             .build();
     }
@@ -249,6 +262,41 @@ protected Path tmpWorkingDirectory(Map<String, Object> additionalVars) throws IO
         return this.workingDirectory;
     }
 
+    protected Map<String, Object> parseOut(RunContext runContext, List<String> outs) throws JsonProcessingException {
+        Map<String, Object> outputs = new HashMap<>();
+
+        for (String out : outs) {
+            // captures output per line
+            String pattern = "^::(\\{.*\\})::$";
+
+            Pattern r = Pattern.compile(pattern);
+            Matcher m = r.matcher(out);
+
+            if (m.find()) {
+                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);
+                }
+            }
+        }
+
+        return outputs;
+    }
+
+    @NoArgsConstructor
+    @Data
+    public static class BashCommand <T> {
+        private Map<String, Object> outputs;
+        private List<AbstractMetricEntry<T>> metrics;
+
+    }
+
     protected LogThread readInput(Logger logger, InputStream inputStream, boolean isStdErr) {
         LogThread thread = new LogThread(logger, inputStream, isStdErr);
         thread.setName("bash-log");
@@ -302,14 +350,19 @@ public List<String> getLogs() {
     @Getter
     public static class Output implements org.kestra.core.models.tasks.Output {
         @Schema(
-            title = "The standard output of the commands"
+            title = "The value extract from output of the commands"
         )
-        private final List<String> stdOut;
+        private final Map<String, Object> vars;
 
         @Schema(
-            title = "The standard error of the commands"
+            title = "The standard output line count"
         )
-        private final List<String> stdErr;
+        private final int stdOutLineCount;
+
+        @Schema(
+            title = "The standard error line count"
+        )
+        private final int stdErrLineCount;
 
         @Schema(
             title = "The exit code of the whole execution"
diff --git a/core/src/main/java/org/kestra/core/tasks/scripts/Node.java b/core/src/main/java/org/kestra/core/tasks/scripts/Node.java
index 5f732712b3..71e02c0c26 100644
--- a/core/src/main/java/org/kestra/core/tasks/scripts/Node.java
+++ b/core/src/main/java/org/kestra/core/tasks/scripts/Node.java
@@ -1,8 +1,10 @@
 package org.kestra.core.tasks.scripts;
 
+import com.google.common.base.Charsets;
 import io.swagger.v3.oas.annotations.media.Schema;
 import lombok.*;
 import lombok.experimental.SuperBuilder;
+import org.apache.commons.io.IOUtils;
 import org.kestra.core.models.annotations.Example;
 import org.kestra.core.models.annotations.Plugin;
 import org.kestra.core.models.annotations.PluginProperty;
@@ -14,6 +16,7 @@
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
+import java.util.Objects;
 
 import static org.kestra.core.utils.Rethrow.throwFunction;
 
@@ -28,7 +31,17 @@
         "The task will create a temprorary folder for every tasks and allows you to install some npm packages defined in an optional `package.json` file.\n" +
         "\n" +
         "By convention, you need to define at least a `main.js` files in `inputFiles` that will be the script used.\n" +
-        "You can also  add as many javascript files as you need in `inputFiles`."
+        "You can also  add as many javascript files as you need in `inputFiles`.\n" +
+        "\n" +
+        "You can send outputs & metrics from your node script that can be used by others tasks. In order to help, we inject a node package directly on the working dir." +
+        "Here is an example usage:\n" +
+        "```javascript\n" +
+        "const Kestra = require(\"./kestra\");\n" +
+        "Kestra.outputs({test: 'value', int: 2, bool: true, float: 3.65});\n" +
+        "Kestra.counter('count', 1, {tag1: 'i', tag2: 'win'});\n" +
+        "Kestra.timer('timer1', (callback) => { setTimeout(callback, 1000) }, {tag1: 'i', tag2: 'lost'});\n" +
+        "Kestra.timer('timer2', 2.12, {tag1: 'i', tag2: 'destroy'});\n" +
+        "```"
 )
 @Plugin(
     examples = {
@@ -37,11 +50,12 @@
             code = {
                 "inputFiles:",
                 "  main.js: |",
+                "    const Kestra = require(\"./kestra\");",
                 "    const fs = require('fs')",
                 "    const result = fs.readFileSync(process.argv[2], \"utf-8\")",
                 "    console.log(JSON.parse(result).status)",
                 "    const axios = require('axios')",
-                "    axios.get('http://google.fr').then(d => console.log(d.status))",
+                "    axios.get('http://google.fr').then(d => console.log(d.status); Kestra.outputs({'status': d.status, 'text': d.data}))",
                 "    console.log(require('./mymodule').value)",
                 "  data.json: |",
                 "    {\"status\": \"OK\"}",
@@ -98,6 +112,11 @@ public Bash.Output run(RunContext runContext) throws Exception {
             throw new Exception("Invalid input files structure, expecting inputFiles property to contain at least a main.js key with javascript code value.");
         }
 
+        this.inputFiles.put("kestra.js", IOUtils.toString(
+            Objects.requireNonNull(this.getClass().getClassLoader().getResourceAsStream("scripts/kestra.js")),
+            Charsets.UTF_8
+        ));
+
         return run(runContext, throwFunction((additionalVars) -> {
             Path workingDirectory = this.tmpWorkingDirectory(additionalVars);
 
diff --git a/core/src/main/java/org/kestra/core/tasks/scripts/Python.java b/core/src/main/java/org/kestra/core/tasks/scripts/Python.java
index 9bf4252312..d1770bcc52 100644
--- a/core/src/main/java/org/kestra/core/tasks/scripts/Python.java
+++ b/core/src/main/java/org/kestra/core/tasks/scripts/Python.java
@@ -1,8 +1,10 @@
 package org.kestra.core.tasks.scripts;
 
+import com.google.common.base.Charsets;
 import io.swagger.v3.oas.annotations.media.Schema;
 import lombok.*;
 import lombok.experimental.SuperBuilder;
+import org.apache.commons.io.IOUtils;
 import org.kestra.core.models.annotations.Example;
 import org.kestra.core.models.annotations.Plugin;
 import org.kestra.core.models.annotations.PluginProperty;
@@ -13,6 +15,7 @@
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
+import java.util.Objects;
 import javax.validation.constraints.NotEmpty;
 import javax.validation.constraints.NotNull;
 
@@ -29,9 +32,19 @@
         "The task will create a fresh `virtualenv` for every tasks and allow you to install some python package define in `requirements` property.\n" +
         "\n" +
         "By convention, you need to define at least a `main.py` files in `inputFiles` that will be the script used.\n" +
-        "But you are also to add as many script as you need in `inputFiles`.\n" +
+        "But you are also able to add as many script as you need in `inputFiles`.\n" +
         "\n" +
-        "You can also add a `pip.conf` in `inputFiles` to customize the pip download of dependencies (like a private registry)."
+        "You can also add a `pip.conf` in `inputFiles` to customize the pip download of dependencies (like a private registry).\n" +
+        "\n" +
+        "You can send outputs & metrics from your python script that can be used by others tasks. In order to help, we inject a python package directly on the working dir." +
+        "Here is an example usage:\n" +
+        "```python\n" +
+        "from kestra import Kestra\n" +
+        "Kestra.outputs({'test': 'value', 'int': 2, 'bool': True, 'float': 3.65})\n" +
+        "Kestra.counter('count', 1, {'tag1': 'i', 'tag2': 'win'})\n" +
+        "Kestra.timer('timer1', lambda: time.sleep(1), {'tag1': 'i', 'tag2': 'lost'})\n" +
+        "Kestra.timer('timer2', 2.12, {'tag1': 'i', 'tag2': 'destroy'})\n" +
+        "```"
 )
 @Plugin(
     examples = {
@@ -42,12 +55,15 @@
                 "  data.json: |",
                 "          {\"status\": \"OK\"}",
                 "  main.py: |",
+                "    from kestra import Kestra",
                 "    import json",
                 "    import requests",
                 "    import sys",
                 "    result = json.loads(open(sys.argv[1]).read())",
                 "    print(f\"python script {result['status']}\")",
-                "    print(requests.get('http://google.com').status_code)",
+                "    response = requests.get('http://google.com')",
+                "    print(response.status_code)",
+                "    Kestra.outputs({'status': response.status_code, 'text': response.text})",
                 "  pip.conf: |",
                 "    # some specific pip repository configuration",
                 "args:",
@@ -89,6 +105,11 @@ public Bash.Output run(RunContext runContext) throws Exception {
             throw new Exception("Invalid input files structure, expecting inputFiles property to contain at least a main.py key with python code value.");
         }
 
+        this.inputFiles.put("kestra.py", IOUtils.toString(
+            Objects.requireNonNull(this.getClass().getClassLoader().getResourceAsStream("scripts/kestra.py")),
+            Charsets.UTF_8
+        ));
+
         return run(runContext, throwFunction((additionalVars) -> {
             Path workingDirectory = this.tmpWorkingDirectory(additionalVars);
 
diff --git a/core/src/main/resources/scripts/kestra.js b/core/src/main/resources/scripts/kestra.js
new file mode 100644
index 0000000000..c287080218
--- /dev/null
+++ b/core/src/main/resources/scripts/kestra.js
@@ -0,0 +1,43 @@
+function Kestra() {
+
+}
+
+Kestra._send = (map) => {
+    console.log("::" + JSON.stringify(map) + "::");
+}
+
+Kestra._metrics = (name, type, value, tags) => {
+    Kestra._send({
+        "metrics": [
+            {
+                "name": name,
+                "type": type,
+                "value": value,
+                tags: tags || {}
+            }
+        ]
+    });
+}
+
+Kestra.outputs = (outputs) => {
+    Kestra._send({
+        "outputs": outputs
+    });
+}
+
+Kestra.counter = (name, value, tags) => {
+    Kestra._metrics(name, "counter", value, tags);
+}
+
+Kestra.timer = (name, duration, tags) => {
+    if (typeof duration === "function") {
+        const start = new Date();
+        duration(() => {
+            Kestra._metrics(name, "timer", (new Date() - start) / 1000, tags);
+        });
+    } else {
+        Kestra._metrics(name, "timer", duration, tags)
+    }
+}
+
+module.exports = Kestra;
\ No newline at end of file
diff --git a/core/src/main/resources/scripts/kestra.py b/core/src/main/resources/scripts/kestra.py
new file mode 100644
index 0000000000..d153ff492d
--- /dev/null
+++ b/core/src/main/resources/scripts/kestra.py
@@ -0,0 +1,42 @@
+import json
+from datetime import datetime
+
+class Kestra:
+    def __init__(self):
+        pass
+
+    @staticmethod
+    def _send(map):
+        print("::" + json.dumps(map) + "::")
+
+    @staticmethod
+    def _metrics(name, type, value, tags):
+        Kestra._send({
+            "metrics": [
+                {
+                    "name": name,
+                    "type": type,
+                    "value": value,
+                    "tags": tags or {}
+                }
+            ]
+        })
+
+    @staticmethod
+    def outputs(map):
+        Kestra._send({
+            "outputs": map
+        })
+
+    @staticmethod
+    def counter(name, value, tags):
+        Kestra._metrics(name, "counter", value, tags);
+
+    @staticmethod
+    def timer(name, duration, tags):
+        if callable(duration):
+            start = datetime.now()
+            duration()
+            Kestra._metrics(name, "timer", (datetime.now().microsecond - start.microsecond) / 1000, tags);
+        else:
+            Kestra._metrics(name, "timer", duration, tags);

diff --git a/core/src/test/java/org/kestra/core/docs/JsonSchemaGeneratorTest.java b/core/src/test/java/org/kestra/core/docs/JsonSchemaGeneratorTest.java
index b45d64c140..0c88524d3a 100644
--- a/core/src/test/java/org/kestra/core/docs/JsonSchemaGeneratorTest.java
+++ b/core/src/test/java/org/kestra/core/docs/JsonSchemaGeneratorTest.java
@@ -53,7 +53,7 @@ void bash() {
         generate = jsonSchemaGenerator.outputs(Task.class, Bash.class);
 
         assertThat(generate, is(not(nullValue())));
-        assertThat(((Map<String, Map<String, Object>>) generate.get("properties")).size(), is(4));
+        assertThat(((Map<String, Map<String, Object>>) generate.get("properties")).size(), is(5));
     }
 
     @SuppressWarnings("unchecked")
diff --git a/core/src/test/java/org/kestra/core/tasks/BashTest.java b/core/src/test/java/org/kestra/core/tasks/BashTest.java
index 01e3ff4197..c174f88cc1 100644
--- a/core/src/test/java/org/kestra/core/tasks/BashTest.java
+++ b/core/src/test/java/org/kestra/core/tasks/BashTest.java
@@ -5,10 +5,12 @@
 import io.micronaut.test.annotation.MicronautTest;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.condition.DisabledIfEnvironmentVariable;
+import org.kestra.core.models.executions.AbstractMetricEntry;
 import org.kestra.core.runners.RunContext;
 import org.kestra.core.runners.RunContextFactory;
 import org.kestra.core.storages.StorageInterface;
 import org.kestra.core.tasks.scripts.Bash;
+import org.kestra.core.tasks.scripts.Node;
 
 import javax.inject.Inject;
 import java.io.File;
@@ -19,9 +21,11 @@
 import java.net.URL;
 import java.nio.charset.StandardCharsets;
 import java.nio.file.Files;
+import java.time.Duration;
 import java.util.*;
 
 import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.greaterThan;
 import static org.hamcrest.Matchers.is;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 
@@ -47,8 +51,8 @@ void run() throws Exception {
         Bash.Output run = bash.run(runContext);
 
         assertThat(run.getExitCode(), is(0));
-        assertThat(run.getStdOut().size(), is(2));
-        assertThat(run.getStdErr().size() > 0, is(true));
+        assertThat(run.getStdOutLineCount(), is(2));
+        assertThat(run.getStdErrLineCount() > 0, is(true));
     }
 
     @Test
@@ -56,18 +60,23 @@ void files() throws Exception {
         RunContext runContext = runContextFactory.of();
 
         Bash bash = Bash.builder()
-            .files(Arrays.asList("xml", "csv"))
+            .outputsFiles(Arrays.asList("xml", "csv"))
             .inputFiles(ImmutableMap.of("files/in/in.txt", "I'm here"))
-            .commands(new String[]{"cat files/in/in.txt", "echo 1 >> {{ outputFiles.xml }}", "echo 2 >> {{ outputFiles.csv }}", "echo 3 >> {{ outputFiles.xml }}"})
+            .commands(new String[]{
+                "echo '::{\"outputs\": {\"extract\":\"'$(cat files/in/in.txt)'\"}}::'",
+                "echo 1 >> {{ outputFiles.xml }}",
+                "echo 2 >> {{ outputFiles.csv }}",
+                "echo 3 >> {{ outputFiles.xml }}"
+            })
             .build();
 
         Bash.Output run = bash.run(runContext);
 
         assertThat(run.getExitCode(), is(0));
-        assertThat(run.getStdErr().size(), is(0));
+        assertThat(run.getStdErrLineCount(), is(0));
 
-        assertThat(run.getStdOut().size(), is(1));
-        assertThat(run.getStdOut().get(0), is("I'm here"));
+        assertThat(run.getStdOutLineCount(), is(1));
+        assertThat(run.getVars().get("extract"), is("I'm here"));
 
         InputStream get = storageInterface.get(run.getFiles().get("xml"));
 
@@ -134,8 +143,8 @@ void dontStopOnFirstFailed() throws Exception {
         Bash.Output run = bash.run(runContext);
 
         assertThat(run.getExitCode(), is(0));
-        assertThat(run.getStdOut().size(), is(1));
-        assertThat(run.getStdErr().size(), is(1));
+        assertThat(run.getStdOutLineCount(), is(1));
+        assertThat(run.getStdErrLineCount(), is(1));
     }
 
     @Test
@@ -154,8 +163,8 @@ void longBashCreateTempFiles() throws Exception {
         Bash.Output run = bash.run(runContext);
 
         assertThat(run.getExitCode(), is(0));
-        assertThat(run.getStdOut().size(), is(2));
-        assertThat(run.getStdErr().size() > 0, is(false));
+        assertThat(run.getStdOutLineCount(), is(2));
+        assertThat(run.getStdErrLineCount() > 0, is(false));
     }
 
     @Test
@@ -163,7 +172,7 @@ void useInputFiles() throws Exception {
         RunContext runContext = runContextFactory.of();
 
         Map<String, String> files = new HashMap<>();
-        files.put("test.sh", "tst() { echo 'testbash' ; echo '{{workingDir}}'; }");
+        files.put("test.sh", "tst() { echo '::{\"outputs\": {\"extract\":\"testbash\"}}::' ; echo '{{workingDir}}'; }");
 
         List<String> commands = new ArrayList<>();
         commands.add("source {{workingDir}}/test.sh && tst");
@@ -177,7 +186,7 @@ void useInputFiles() throws Exception {
         Bash.Output run = bash.run(runContext);
 
         assertThat(run.getExitCode(), is(0));
-        assertThat(run.getStdOut().get(0), is("testbash"));
+        assertThat(run.getVars().get("extract"), is("testbash"));
     }
 
     @Test
@@ -196,19 +205,53 @@ void useInputFilesFromKestraFs() throws Exception {
         files.put("fscontent.txt", put.toString());
 
         List<String> commands = new ArrayList<>();
-        commands.add("cat fscontent.txt");
+        commands.add("cat fscontent.txt > {{ outputFiles.out }} ");
 
         Bash bash = Bash.builder()
             .interpreter("/bin/bash")
             .commands(commands.toArray(String[]::new))
             .inputFiles(files)
+            .outputsFiles(Collections.singletonList("out"))
             .build();
 
         Bash.Output run = bash.run(runContext);
 
         assertThat(run.getExitCode(), is(0));
-        String outputContent = String.join("\n", run.getStdOut());
+        InputStream get = storageInterface.get(run.getFiles().get("out"));
+        String outputContent = CharStreams.toString(new InputStreamReader(get));
         String fileContent = String.join("\n", Files.readAllLines(new File(resource.getPath()).toPath(), StandardCharsets.UTF_8));
         assertThat(outputContent, is(fileContent));
     }
+
+    static void controlOutputs(RunContext runContext, Bash.Output run) {
+        assertThat(run.getVars().get("test"), is("value"));
+        assertThat(run.getVars().get("int"), is(2));
+        assertThat(run.getVars().get("bool"), is(true));
+        assertThat(run.getVars().get("float"), is(3.65));
+
+        assertThat(BashTest.getMetrics(runContext, "count").getValue(), is(1D));
+        assertThat(BashTest.getMetrics(runContext, "count").getTags().size(), is(2));
+        assertThat(BashTest.getMetrics(runContext, "count").getTags().get("tag1"), is("i"));
+        assertThat(BashTest.getMetrics(runContext, "count").getTags().get("tag2"), is("win"));
+
+        assertThat(BashTest.<Duration>getMetrics(runContext, "timer1").getValue().getNano(), greaterThan(0));
+        assertThat(BashTest.<Duration>getMetrics(runContext, "timer1").getTags().size(), is(2));
+        assertThat(BashTest.<Duration>getMetrics(runContext, "timer1").getTags().get("tag1"), is("i"));
+        assertThat(BashTest.<Duration>getMetrics(runContext, "timer1").getTags().get("tag2"), is("lost"));
+
+        assertThat(BashTest.<Duration>getMetrics(runContext, "timer2").getValue().getNano(), greaterThan(100000000));
+        assertThat(BashTest.<Duration>getMetrics(runContext, "timer2").getTags().size(), is(2));
+        assertThat(BashTest.<Duration>getMetrics(runContext, "timer2").getTags().get("tag1"), is("i"));
+        assertThat(BashTest.<Duration>getMetrics(runContext, "timer2").getTags().get("tag2"), is("destroy"));
+    }
+
+    @SuppressWarnings("unchecked")
+    static <T> AbstractMetricEntry<T> getMetrics(RunContext runContext, String name) {
+        return (AbstractMetricEntry<T>) runContext.metrics()
+            .stream()
+            .filter(abstractMetricEntry -> abstractMetricEntry.getName().equals(name))
+            .findFirst()
+            .orElseThrow();
+
+    }
 }
diff --git a/core/src/test/java/org/kestra/core/tasks/NodeTest.java b/core/src/test/java/org/kestra/core/tasks/NodeTest.java
index d9289799fa..52dad0719a 100644
--- a/core/src/test/java/org/kestra/core/tasks/NodeTest.java
+++ b/core/src/test/java/org/kestra/core/tasks/NodeTest.java
@@ -9,10 +9,10 @@
 import org.kestra.core.tasks.scripts.Bash;
 import org.kestra.core.tasks.scripts.Node;
 
-import javax.inject.Inject;
 import java.util.Arrays;
 import java.util.HashMap;
 import java.util.Map;
+import javax.inject.Inject;
 
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.hamcrest.Matchers.equalTo;
@@ -31,7 +31,7 @@ class NodeTest {
     void run() throws Exception {
         RunContext runContext = runContextFactory.of();
         Map<String, String> files = new HashMap<>();
-        files.put("main.js", "console.log('hello world')");
+        files.put("main.js", "console.log('::{\"outputs\": {\"extract\":\"hello world\"}}::')");
 
         Node node = Node.builder()
             .id("test-node-task")
@@ -42,9 +42,9 @@ void run() throws Exception {
         Bash.Output run = node.run(runContext);
 
         assertThat(run.getExitCode(), is(0));
-        assertThat(run.getStdOut().size(), is(1));
-        assertThat(run.getStdOut().get(0), is("hello world"));
-        assertThat(run.getStdErr().size(), equalTo(0));
+        assertThat(run.getStdOutLineCount(), is(1));
+        assertThat(run.getVars().get("extract"), is("hello world"));
+        assertThat(run.getStdErrLineCount(), equalTo(0));
     }
 
     @Test
@@ -72,7 +72,7 @@ void failed() throws Exception {
     void requirements() throws Exception {
         RunContext runContext = runContextFactory.of();
         Map<String, String> files = new HashMap<>();
-        files.put("main.js", "require('axios').get('http://google.com').then(r => { console.log( r.status ) })");
+        files.put("main.js", "require('axios').get('http://google.com').then(r => { console.log('::{\"outputs\": {\"extract\":\"' + r.status + '\"}}::') })");
         files.put("package.json", "{\"dependencies\":{\"axios\":\"^0.20.0\"}}");
 
         Node node = Node.builder()
@@ -85,14 +85,14 @@ void requirements() throws Exception {
         Bash.Output run = node.run(runContext);
 
         assertThat(run.getExitCode(), is(0));
-        assertThat(run.getStdOut().get(0), is("200"));
+        assertThat(run.getVars().get("extract"), is("200"));
     }
 
     @Test
     void manyFiles() throws Exception {
         RunContext runContext = runContextFactory.of();
         Map<String, String> files = new HashMap<>();
-        files.put("main.js", "console.log(require('./otherfile').value)");
+        files.put("main.js", "console.log('::{\"outputs\": {\"extract\":\"' + (require('./otherfile').value) + '\"}}::')");
         files.put("otherfile.js", "module.exports.value = 'success'");
 
         Node node = Node.builder()
@@ -104,14 +104,14 @@ void manyFiles() throws Exception {
         Bash.Output run = node.run(runContext);
 
         assertThat(run.getExitCode(), is(0));
-        assertThat(run.getStdOut().get(0), is("success"));
+        assertThat(run.getVars().get("extract"), is("success"));
     }
 
     @Test
     void fileInSubFolders() throws Exception {
         RunContext runContext = runContextFactory.of();
         Map<String, String> files = new HashMap<>();
-        files.put("main.js", "console.log(require('fs').readFileSync('./sub/folder/file/test.txt', 'utf-8'))");
+        files.put("main.js", "console.log('::{\"outputs\": {\"extract\":\"' + (require('fs').readFileSync('./sub/folder/file/test.txt', 'utf-8')) + '\"}}::')");
         files.put("sub/folder/file/test.txt", "OK");
         files.put("sub/folder/file/test1.txt", "OK");
 
@@ -124,14 +124,14 @@ void fileInSubFolders() throws Exception {
         Bash.Output run = node.run(runContext);
 
         assertThat(run.getExitCode(), is(0));
-        assertThat(run.getStdOut().get(0), is("OK"));
+        assertThat(run.getVars().get("extract"), is("OK"));
     }
 
     @Test
     void args() throws Exception {
         RunContext runContext = runContextFactory.of(ImmutableMap.of("test", "value"));
         Map<String, String> files = new HashMap<>();
-        files.put("main.js", "console.log(process.argv.slice(2).join(' '))");
+        files.put("main.js", "console.log('::{\"outputs\": {\"extract\":\"' + (process.argv.slice(2).join(' ')) + '\"}}::')");
 
         Node node = Node.builder()
             .id("test-node-task")
@@ -142,6 +142,28 @@ void args() throws Exception {
 
         Bash.Output run = node.run(runContext);
 
-        assertThat(run.getStdOut().get(0), is("test param value"));
+        assertThat(run.getVars().get("extract"), is("test param value"));
+    }
+
+    @Test
+    void outputs() throws Exception {
+        RunContext runContext = runContextFactory.of(ImmutableMap.of("test", "value"));
+        Map<String, String> files = new HashMap<>();
+        files.put("main.js", "const Kestra = require(\"./kestra\");" +
+            "Kestra.outputs({test: 'value', int: 2, bool: true, float: 3.65});" +
+            "Kestra.counter('count', 1, {tag1: 'i', tag2: 'win'});" +
+            "Kestra.timer('timer1', (callback) => { setTimeout(callback, 1000) }, {tag1: 'i', tag2: 'lost'});" +
+            "Kestra.timer('timer2', 2.12, {tag1: 'i', tag2: 'destroy'});"
+        );
+
+        Node node = Node.builder()
+            .id("test-node-task")
+            .nodePath("node")
+            .inputFiles(files)
+            .build();
+
+        Bash.Output run = node.run(runContext);
+
+        BashTest.controlOutputs(runContext, run);
     }
 }
diff --git a/core/src/test/java/org/kestra/core/tasks/PythonTest.java b/core/src/test/java/org/kestra/core/tasks/PythonTest.java
index d1bda3afe3..cb8a34f3a5 100644
--- a/core/src/test/java/org/kestra/core/tasks/PythonTest.java
+++ b/core/src/test/java/org/kestra/core/tasks/PythonTest.java
@@ -9,10 +9,10 @@
 import org.kestra.core.tasks.scripts.Bash;
 import org.kestra.core.tasks.scripts.Python;
 
-import javax.inject.Inject;
 import java.util.Arrays;
 import java.util.HashMap;
 import java.util.Map;
+import javax.inject.Inject;
 
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.hamcrest.Matchers.equalTo;
@@ -31,7 +31,7 @@ class PythonTest {
     void run() throws Exception {
         RunContext runContext = runContextFactory.of();
         Map<String, String> files = new HashMap<>();
-        files.put("main.py", "print('hello world')");
+        files.put("main.py", "print('::{\"outputs\": {\"extract\":\"hello world\"}}::')");
 
         Python python = Python.builder()
             .id("test-python-task")
@@ -42,9 +42,9 @@ void run() throws Exception {
         Bash.Output run = python.run(runContext);
 
         assertThat(run.getExitCode(), is(0));
-        assertThat(run.getStdOut().size(), is(1));
-        assertThat(run.getStdOut().get(0), is("hello world"));
-        assertThat(run.getStdErr().size(), equalTo(0));
+        assertThat(run.getStdOutLineCount(), is(1));
+        assertThat(run.getVars().get("extract"), is("hello world"));
+        assertThat(run.getStdErrLineCount(), equalTo(0));
     }
 
     @Test
@@ -72,7 +72,7 @@ void failed() throws Exception {
     void requirements() throws Exception {
         RunContext runContext = runContextFactory.of();
         Map<String, String> files = new HashMap<>();
-        files.put("main.py", "import requests; print(requests.get('http://google.com').status_code)");
+        files.put("main.py", "import requests; print('::{\"outputs\": {\"extract\":\"' + str(requests.get('http://google.com').status_code) + '\"}}::')");
 
         Python python = Python.builder()
             .id("test-python-task")
@@ -84,7 +84,7 @@ void requirements() throws Exception {
         Bash.Output run = python.run(runContext);
 
         assertThat(run.getExitCode(), is(0));
-        assertThat(run.getStdOut().get(0), is("200"));
+        assertThat(run.getVars().get("extract"), is("200"));
     }
 
     @Test
@@ -92,7 +92,7 @@ void manyFiles() throws Exception {
         RunContext runContext = runContextFactory.of();
         Map<String, String> files = new HashMap<>();
         files.put("main.py", "import otherfile; otherfile.test()");
-        files.put("otherfile.py", "def test(): print('success')");
+        files.put("otherfile.py", "def test(): print('::{\"outputs\": {\"extract\":\"success\"}}::')");
 
         Python python = Python.builder()
             .id("test-python-task")
@@ -103,14 +103,14 @@ void manyFiles() throws Exception {
         Bash.Output run = python.run(runContext);
 
         assertThat(run.getExitCode(), is(0));
-        assertThat(run.getStdOut().get(0), is("success"));
+        assertThat(run.getVars().get("extract"), is("success"));
     }
 
     @Test
     void pipConf() throws Exception {
         RunContext runContext = runContextFactory.of();
         Map<String, String> files = new HashMap<>();
-        files.put("main.py", "print(open('pip.conf').read())");
+        files.put("main.py", "print('::{\"outputs\": {\"extract\":\"' + str('#it worked !' in open('pip.conf').read()) + '\"}}::')");
         files.put("pip.conf", "[global]\nno-cache-dir = false\n#it worked !");
 
         Python python = Python.builder()
@@ -122,14 +122,14 @@ void pipConf() throws Exception {
         Bash.Output run = python.run(runContext);
 
         assertThat(run.getExitCode(), is(0));
-        assertThat(run.getStdOut().get(2), is("#it worked !"));
+        assertThat(run.getVars().get("extract"), is("True"));
     }
 
     @Test
     void fileInSubFolders() throws Exception {
         RunContext runContext = runContextFactory.of();
         Map<String, String> files = new HashMap<>();
-        files.put("main.py", "print(open('sub/folder/file/test.txt').read())");
+        files.put("main.py", "print('::{\"outputs\": {\"extract\":\"' + open('sub/folder/file/test.txt').read() + '\"}}::')");
         files.put("sub/folder/file/test.txt", "OK");
         files.put("sub/folder/file/test1.txt", "OK");
 
@@ -142,14 +142,14 @@ void fileInSubFolders() throws Exception {
         Bash.Output run = python.run(runContext);
 
         assertThat(run.getExitCode(), is(0));
-        assertThat(run.getStdOut().get(0), is("OK"));
+        assertThat(run.getVars().get("extract"), is("OK"));
     }
 
     @Test
     void args() throws Exception {
         RunContext runContext = runContextFactory.of(ImmutableMap.of("test", "value"));
         Map<String, String> files = new HashMap<>();
-        files.put("main.py", "import sys; print(' '.join(sys.argv))");
+        files.put("main.py", "import sys; print('::{\"outputs\": {\"extract\":\"' + ' '.join(sys.argv) + '\"}}::')");
 
         Python python = Python.builder()
             .id("test-python-task")
@@ -160,6 +160,34 @@ void args() throws Exception {
 
         Bash.Output run = python.run(runContext);
 
-        assertThat(run.getStdOut().get(0), is("main.py test param value"));
+        assertThat(run.getVars().get("extract"), is("main.py test param value"));
+    }
+
+    @Test
+    void outputs() throws Exception {
+        RunContext runContext = runContextFactory.of(ImmutableMap.of("test", "value"));
+        Map<String, String> files = new HashMap<>();
+        files.put("main.py", "from kestra import Kestra\n" +
+            "import time\n" +
+            "Kestra.outputs({'test': 'value', 'int': 2, 'bool': True, 'float': 3.65})\n" +
+            "Kestra.counter('count', 1, {'tag1': 'i', 'tag2': 'win'})\n" +
+            "Kestra.timer('timer1', lambda: time.sleep(1), {'tag1': 'i', 'tag2': 'lost'})\n" +
+            "Kestra.timer('timer2', 2.12, {'tag1': 'i', 'tag2': 'destroy'})\n"
+        );
+
+        Python node = Python.builder()
+            .id("test-node-task")
+            .pythonPath("python3")
+            .inputFiles(files)
+            .build();
+
+        Bash.Output run = node.run(runContext);
+
+        assertThat(run.getVars().get("test"), is("value"));
+        assertThat(run.getVars().get("int"), is(2));
+        assertThat(run.getVars().get("bool"), is(true));
+        assertThat(run.getVars().get("float"), is(3.65));
+
+        BashTest.controlOutputs(runContext, run);
     }
 }
diff --git a/webserver/src/test/java/org/kestra/webserver/controllers/PluginControllerTest.java b/webserver/src/test/java/org/kestra/webserver/controllers/PluginControllerTest.java
index dcce47fbf7..a5ad222ba6 100644
--- a/webserver/src/test/java/org/kestra/webserver/controllers/PluginControllerTest.java
+++ b/webserver/src/test/java/org/kestra/webserver/controllers/PluginControllerTest.java
@@ -58,9 +58,9 @@ void bash() throws URISyntaxException {
 
             assertThat(doc.getMarkdown(), containsString("org.kestra.core.tasks.scripts.Bash"));
             assertThat(doc.getMarkdown(), containsString("Exit if any non true return value"));
-            assertThat(doc.getMarkdown(), containsString("The standard error of the commands"));
+            assertThat(doc.getMarkdown(), containsString("The standard output line count"));
             assertThat(((Map<String, Object>) doc.getSchema().getProperties().get("properties")).size(), is(7));
-            assertThat(((Map<String, Object>) doc.getSchema().getOutputs().get("properties")).size(), is(4));
+            assertThat(((Map<String, Object>) doc.getSchema().getOutputs().get("properties")).size(), is(5));
         });
     }
 
