feat(core): Add displayName to flow level outputs(backend)

kestra-io · Oct 23, 2024 · 663c819 · 663c819
1 parent 48da2a7
commit 663c819
Show file tree

Hide file tree

Showing 9 changed files with 44 additions and 10 deletions.
diff --git a/core/src/main/java/io/kestra/core/models/flows/Output.java b/core/src/main/java/io/kestra/core/models/flows/Output.java
@@ -41,4 +41,9 @@ public class Output implements Data {
     @NotNull
     @Valid
     Type type;
+    /**
+     * The display name of the output.
+     */
+    @NotNull
+    String displayName;
 }
diff --git a/core/src/main/java/io/kestra/core/runners/ExecutorService.java b/core/src/main/java/io/kestra/core/runners/ExecutorService.java
@@ -1,5 +1,7 @@
 package io.kestra.core.runners;
 
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
 import com.google.common.collect.ImmutableMap;
 import io.kestra.core.exceptions.InternalException;
 import io.kestra.core.metrics.MetricRegistry;
@@ -383,7 +385,17 @@ private Executor onEnd(Executor executor) {
             try {
                 Map<String, Object> outputs = flow.getOutputs()
                     .stream()
-                    .collect(HashMap::new, (map, entry) -> map.put(entry.getId(), entry.getValue()), Map::putAll);
+                    .collect(HashMap::new, (map, entry) -> {
+                        final ObjectMapper mapper = new ObjectMapper();
+                        final HashMap<String, Object> entryInfo = new HashMap<>();
+                        entryInfo.put("value", entry.getValue());
+                        entryInfo.put("displayName", entry.getDisplayName());
+                        try {
+                            map.put(entry.getId(), mapper.writeValueAsString(entryInfo));
+                        } catch (JsonProcessingException e) {
+                            throw new RuntimeException(e);
+                        }
+                    }, Map::putAll);
                 outputs = runContext.render(outputs);
                 outputs = flowInputOutput.typedOutputs(flow, executor.getExecution(), outputs);
                 newExecution = newExecution.withOutputs(outputs);

diff --git a/core/src/main/java/io/kestra/core/runners/FlowInputOutput.java b/core/src/main/java/io/kestra/core/runners/FlowInputOutput.java
@@ -1,5 +1,7 @@
 package io.kestra.core.runners;
 
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.core.type.TypeReference;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.collect.ImmutableMap;
@@ -31,6 +33,8 @@
 import jakarta.inject.Singleton;
 import jakarta.validation.ConstraintViolationException;
 import jakarta.validation.constraints.NotNull;
+
+import org.jcodings.util.Hash;
 import org.reactivestreams.Publisher;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -46,6 +50,7 @@
 import java.time.LocalDate;
 import java.time.LocalTime;
 import java.util.AbstractMap;
+import java.util.AbstractMap.SimpleEntry;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.HashMap;
@@ -345,24 +350,31 @@ public Map<String, Object> typedOutputs(
         if (flow.getOutputs() == null) {
             return ImmutableMap.of();
         }
+        final ObjectMapper mapper = new ObjectMapper();
         Map<String, Object> results = flow
             .getOutputs()
             .stream()
             .map(output -> {
-                Object current = in == null ? null : in.get(output.getId());
+                String current = in == null ? null : in.get(output.getId()).toString();
+                Object currentValue;
+                try {
+                    currentValue = mapper.readValue(current, new TypeReference<HashMap<String, Object>>() {}).get("value");
+                } catch (JsonProcessingException e) {
+                    throw new RuntimeException(e);
+                }
                 try {
-                    return parseData(execution, output, current)
+                    return parseData(execution, output, currentValue)
                         .map(entry -> {
                             if (output.getType().equals(Type.SECRET)) {
-                                return new AbstractMap.SimpleEntry<>(
+                                return new SimpleEntry<>(
                                     entry.getKey(),
                                     EncryptedString.from(entry.getValue().toString())
                                 );
                             }
                             return entry;
                         });
                 } catch (Exception e) {
-                    throw output.toConstraintViolationException(e.getMessage(), current);
+                    throw output.toConstraintViolationException(e.getMessage(), currentValue);
                 }
             })
             .filter(Optional::isPresent)

diff --git a/core/src/test/java/io/kestra/plugin/core/flow/FlowOutputTest.java b/core/src/test/java/io/kestra/plugin/core/flow/FlowOutputTest.java
@@ -20,7 +20,7 @@ class FlowOutputTest extends AbstractMemoryRunnerTest {
     void shouldGetSuccessExecutionForFlowWithOutputs() throws TimeoutException, QueueException {
         Execution execution = runnerUtils.runOne(null, NAMESPACE, "flow-with-outputs", null, null);
         assertThat(execution.getOutputs(), aMapWithSize(1));
-        assertThat(execution.getOutputs().get("key"), is("{\"value\":\"flow-with-outputs\"}"));
+        assertThat(execution.getOutputs().get("key"), is("{\"displayName\":\"Final returned value\",\"value\":\"{\"value\":\"flow-with-outputs\"}\"}"));
         assertThat(execution.getState().getCurrent(), is(State.Type.SUCCESS));
     }
 

diff --git a/core/src/test/resources/flows/valids/flow-with-array-outputs.yml b/core/src/test/resources/flows/valids/flow-with-array-outputs.yml
@@ -13,4 +13,5 @@ outputs:
     type: ARRAY
     value:
       - "{{ outputs.return.values.one }}"
-      - "{{ outputs.return.values.two }}"
+      - "{{ outputs.return.values.two }}"
+    displayName: Final returned value
diff --git a/core/src/test/resources/flows/valids/flow-with-outputs-failed.yml b/core/src/test/resources/flows/valids/flow-with-outputs-failed.yml
@@ -9,4 +9,5 @@ tasks:
 outputs:
 - id: "key"
   value: "{{ invalid }}"
-  type: STRING
+  type: STRING
+  displayName: Final returned value
diff --git a/core/src/test/resources/flows/valids/flow-with-outputs.yml b/core/src/test/resources/flows/valids/flow-with-outputs.yml
@@ -9,4 +9,5 @@ tasks:
 outputs:
 - id: "key"
   value: "{{ outputs.return }}"
-  type: STRING
+  type: STRING
+  displayName: Final returned value
diff --git a/core/src/test/resources/flows/valids/for-each-item-outputs-subflow.yaml b/core/src/test/resources/flows/valids/for-each-item-outputs-subflow.yaml
@@ -13,4 +13,5 @@ tasks:
 outputs:
   - id: value
     value: "{{ outputs.return.value }}"
-    type: STRING
+    type: STRING
+    displayName: Final returned value
diff --git a/core/src/test/resources/flows/valids/trigger-flow-listener-with-pause.yaml b/core/src/test/resources/flows/valids/trigger-flow-listener-with-pause.yaml
@@ -5,6 +5,7 @@ outputs:
   - id: status
     type: STRING
     value: "{{trigger.state}}"
+    displayName: Final returned value
 
 tasks:
   - id: hello