Repository: kestra-io/kestra (java)

Issue:
Add labels to the Flow trigger context
### Feature description

It seems to be a quick win to add labels to the `{{trigger}}` runContext object, let's add that.

Context:

![Image](https://github.com/user-attachments/assets/48732327-9b31-4ce1-a6f1-d84314d8b3d8)

Pull request description:
feat(core): forward execution labels in Flow Trigger
- fixes #8421

Patch:
diff --git a/core/src/main/java/io/kestra/core/utils/NamespaceFilesUtils.java b/core/src/main/java/io/kestra/core/utils/NamespaceFilesUtils.java
index 533c356f4..649df1362 100644
--- a/core/src/main/java/io/kestra/core/utils/NamespaceFilesUtils.java
+++ b/core/src/main/java/io/kestra/core/utils/NamespaceFilesUtils.java
@@ -52,19 +52,15 @@ public class NamespaceFilesUtils {
             .as(FileExistComportment.class).orElse(FileExistComportment.OVERWRITE);
         List<String> namespaces = runContext.render(namespaceFiles.getNamespaces()).asList(String.class);
 
-        Map<String, NamespaceFile> namespaceFileMap = new HashMap<>();
+        List<NamespaceFile> matchedNamespaceFiles = new ArrayList<>();
         for (String namespace : namespaces) {
             List<NamespaceFile> files = runContext.storage()
                 .namespace(namespace)
                 .findAllFilesMatching(include, exclude);
 
-            for (NamespaceFile file : files) {
-                namespaceFileMap.put(file.storagePath().toFile().getName(), file);
-            }
+          matchedNamespaceFiles.addAll(files);
         }
 
-        List<NamespaceFile> matchedNamespaceFiles = new ArrayList<>(namespaceFileMap.values());
-
         Flux.fromIterable(matchedNamespaceFiles)
             .doOnNext(throwConsumer(namespaceFile -> {
                 InputStream content = runContext.storage().getFile(namespaceFile.uri());
diff --git a/core/src/main/java/io/kestra/plugin/core/trigger/Flow.java b/core/src/main/java/io/kestra/plugin/core/trigger/Flow.java
index 1060dca10..647460696 100644
--- a/core/src/main/java/io/kestra/plugin/core/trigger/Flow.java
+++ b/core/src/main/java/io/kestra/plugin/core/trigger/Flow.java
@@ -277,6 +277,7 @@ public class Flow extends AbstractTrigger implements TriggerOutput<Flow.Output>
                 this,
                 Output.builder()
                     .executionId(current.getId())
+                    .executionLabels(Label.toNestedMap(current.getLabels().stream().filter(label -> !label.key().equals(Label.CORRELATION_ID)).collect(Collectors.toList())))
                     .namespace(current.getNamespace())
                     .flowId(current.getFlowId())
                     .flowRevision(current.getFlowRevision())
@@ -579,6 +580,10 @@ public class Flow extends AbstractTrigger implements TriggerOutput<Flow.Output>
         @NotNull
         private String executionId;
 
+        @Schema(title = "The execution labels that triggered the current flow.")
+        @NotNull
+        private Map<String, Object> executionLabels;
+
         @Schema(title = "The execution state.")
         @NotNull
         private State.Type state;


Test patch:
diff --git a/core/src/test/java/io/kestra/core/utils/NamespaceFilesUtilsTest.java b/core/src/test/java/io/kestra/core/utils/NamespaceFilesUtilsTest.java
index 486ddfd49..6cbb39bbc 100644
--- a/core/src/test/java/io/kestra/core/utils/NamespaceFilesUtilsTest.java
+++ b/core/src/test/java/io/kestra/core/utils/NamespaceFilesUtilsTest.java
@@ -91,6 +91,30 @@ class NamespaceFilesUtilsTest {
         assertThat((Duration) runContext.metrics().stream().filter(m -> m.getName().equals("namespacefiles.duration")).findFirst().orElseThrow().getValue()).isInstanceOf(Duration.class);
     }
 
+    @Test
+    void multiple_folder_ns() throws Exception {
+        List<LogEntry> logs = new CopyOnWriteArrayList<>();
+        Flux<LogEntry> receive = TestsUtils.receive(workerTaskLogQueue, either -> logs.add(either.getLeft()));
+
+        Log task = Log.builder().id(IdUtils.create()).type(Log.class.getName()).message("Yo!").build();
+        var runContext = TestsUtils.mockRunContext(runContextFactory, task, ImmutableMap.of());
+        String namespace = IdUtils.create();
+
+        ByteArrayInputStream data = new ByteArrayInputStream("a".repeat(1024).getBytes(StandardCharsets.UTF_8));
+        storageInterface.put(null, namespace, toNamespacedStorageUri(namespace, URI.create("/folder1/test.txt")), data);
+        storageInterface.put(null, namespace, toNamespacedStorageUri(namespace, URI.create("/folder2/test.txt")), data);
+        storageInterface.put(null, namespace, toNamespacedStorageUri(namespace, URI.create("/test.txt")), data);
+
+        namespaceFilesUtils.loadNamespaceFiles(runContext, NamespaceFiles.builder().namespaces(Property.of(List.of(namespace))).build());
+
+        List<LogEntry> logEntry = TestsUtils.awaitLogs(logs, 1);
+        receive.blockLast();
+
+        assertThat(logEntry.getFirst().getMessage()).contains("Loaded 3 namespace files");
+        assertThat(runContext.metrics().stream().filter(m -> m.getName().equals("namespacefiles.count")).findFirst().orElseThrow().getValue()).isEqualTo(3D);
+        assertThat((Duration) runContext.metrics().stream().filter(m -> m.getName().equals("namespacefiles.duration")).findFirst().orElseThrow().getValue()).isInstanceOf(Duration.class);
+    }
+
     private URI toNamespacedStorageUri(String namespace, @Nullable URI relativePath) {
         return NamespaceFile.of(namespace, relativePath).storagePath().toUri();
     }
diff --git a/core/src/test/java/io/kestra/plugin/core/trigger/FlowTest.java b/core/src/test/java/io/kestra/plugin/core/trigger/FlowTest.java
index 11681e860..bf9c5ed9e 100644
--- a/core/src/test/java/io/kestra/plugin/core/trigger/FlowTest.java
+++ b/core/src/test/java/io/kestra/plugin/core/trigger/FlowTest.java
@@ -2,6 +2,7 @@ package io.kestra.plugin.core.trigger;
 
 import io.kestra.core.models.Label;
 import io.kestra.core.models.executions.Execution;
+import io.kestra.core.models.executions.ExecutionTrigger;
 import io.kestra.core.models.flows.State;
 import io.kestra.core.models.property.Property;
 import io.kestra.core.models.triggers.multipleflows.MultipleConditionStorageInterface;
@@ -14,6 +15,7 @@ import org.junit.jupiter.api.Test;
 
 import java.util.Collections;
 import java.util.List;
+import java.util.Map;
 import java.util.Optional;
 
 import static org.assertj.core.api.Assertions.assertThat;
@@ -173,5 +175,7 @@ class FlowTest {
         assertThat(evaluate.get().getLabels()).contains(new Label("trigger-label-2", "trigger-label-2"));
         assertThat(evaluate.get().getLabels()).contains(new Label("trigger-label-3", ""));
         assertThat(evaluate.get().getLabels()).contains(new Label(Label.CORRELATION_ID, "correlationId"));
+        assertThat(evaluate.get().getTrigger()).extracting(ExecutionTrigger::getVariables).hasFieldOrProperty("executionLabels");
+        assertThat(evaluate.get().getTrigger().getVariables().get("executionLabels")).isEqualTo(Map.of("execution-label", "execution"));
     }
 }