Repository: kestra-io/kestra (java)

Issue:
Allow a subflow to inherit parent flow's execution labels 
### Feature description

When triggering a subflow, the user currently can define custom execution labels using the `labels` property. 

It would be great to allow a subflow to inherit labels from a parent flow's execution. 

## Possible implementation

Add an optional boolean property `inheritLabels` to the `Flow` task which should be by default set to `false` https://kestra.io/plugins/core/tasks/flows/io.kestra.core.tasks.flows.flow#labels 

Pull request description:
Allow a subflow to inherit calling flow's labels
The label inheritance is disabled by default.

Please, validate the usage of the exec repo.

close #2114


Patch:
diff --git a/core/src/main/java/io/kestra/core/runners/ExecutorService.java b/core/src/main/java/io/kestra/core/runners/ExecutorService.java
index de656bbf8..f869302b9 100644
--- a/core/src/main/java/io/kestra/core/runners/ExecutorService.java
+++ b/core/src/main/java/io/kestra/core/runners/ExecutorService.java
@@ -579,7 +579,7 @@ public class ExecutorService {
                 );
 
                 try {
-                    Execution execution = flowTask.createExecution(runContext, flowExecutorInterface());
+                    Execution execution = flowTask.createExecution(runContext, flowExecutorInterface(), executor.getExecution());
 
                     WorkerTaskExecution workerTaskExecution = WorkerTaskExecution.builder()
                         .task(flowTask)
diff --git a/core/src/main/java/io/kestra/core/tasks/flows/Flow.java b/core/src/main/java/io/kestra/core/tasks/flows/Flow.java
index cedce2ebb..29b503beb 100644
--- a/core/src/main/java/io/kestra/core/tasks/flows/Flow.java
+++ b/core/src/main/java/io/kestra/core/tasks/flows/Flow.java
@@ -95,6 +95,14 @@ public class Flow extends Task implements RunnableTask<Flow.Output> {
     @PluginProperty
     private final Boolean transmitFailed = false;
 
+    @Builder.Default
+    @Schema(
+        title = "Inherit labels from the calling execution",
+        description = "By default, we don't inherit any labels of the calling execution"
+    )
+    @PluginProperty
+    private final Boolean inheritLabels = false;
+
     @Schema(
         title = "Extract outputs from triggered executions.",
         description = "Allow to specify key value (with value rendered), in order to extract any outputs from " +
@@ -117,7 +125,7 @@ public class Flow extends Task implements RunnableTask<Flow.Output> {
     }
 
     @SuppressWarnings("unchecked")
-    public Execution createExecution(RunContext runContext, FlowExecutorInterface flowExecutorInterface) throws Exception {
+    public Execution createExecution(RunContext runContext, FlowExecutorInterface flowExecutorInterface, Execution currentExecution) throws Exception {
         RunnerUtils runnerUtils = runContext.getApplicationContext().getBean(RunnerUtils.class);
 
         Map<String, String> inputs = new HashMap<>();
@@ -128,6 +136,9 @@ public class Flow extends Task implements RunnableTask<Flow.Output> {
         }
 
         List<Label> labels = new ArrayList<>();
+        if (this.inheritLabels) {
+            labels.addAll(currentExecution.getLabels());
+        }
         if (this.labels != null) {
             for (Map.Entry<String, String> entry: this.labels.entrySet()) {
                 labels.add(new Label(entry.getKey(), runContext.render(entry.getValue())));
diff --git a/ui/src/components/logs/LogList.vue b/ui/src/components/logs/LogList.vue
index 3e0fd8a5f..bfd453518 100644
--- a/ui/src/components/logs/LogList.vue
+++ b/ui/src/components/logs/LogList.vue
@@ -10,8 +10,8 @@
                             class="task-icon me-1"
                         >
                             <task-icon
-                                :cls="taskIcon(currentTaskRun.taskId)"
-                                v-if="taskIcon(currentTaskRun.taskId)"
+                                :cls="taskIcon(currentTaskRun)"
+                                v-if="taskIcon(currentTaskRun)"
                                 only-icon
                             />
                         </div>
@@ -550,9 +550,13 @@
             uniqueTaskRunDisplayFilter(currentTaskRun) {
                 return !(this.taskRunId && this.taskRunId !== currentTaskRun.id);
             },
-            taskIcon(taskId) {
-                let findTaskById = FlowUtils.findTaskById(this.flow, taskId);
-                return findTaskById ? findTaskById.type : undefined;
+            taskIcon(taskRun) {
+                const task = FlowUtils.findTaskById(this.flow, taskRun.taskId);
+                const parentTaskRunId = taskRun.parentTaskRunId;
+                if(task === undefined && parentTaskRunId) {
+                    return this.taskIcon(this.taskRunById[parentTaskRunId])
+                }
+                return task ? task.type : undefined;
             },
             loadLogs(executionId) {
                 this.$store.dispatch("execution/loadLogsNoCommit", {


Test patch:
diff --git a/core/src/test/java/io/kestra/core/tasks/flows/FlowCaseTest.java b/core/src/test/java/io/kestra/core/tasks/flows/FlowCaseTest.java
index 79deee803..4e04d4c64 100644
--- a/core/src/test/java/io/kestra/core/tasks/flows/FlowCaseTest.java
+++ b/core/src/test/java/io/kestra/core/tasks/flows/FlowCaseTest.java
@@ -1,6 +1,7 @@
 package io.kestra.core.tasks.flows;
 
 import com.google.common.collect.ImmutableMap;
+import io.kestra.core.models.Label;
 import io.kestra.core.models.executions.Execution;
 import io.kestra.core.models.flows.State;
 import io.kestra.core.queues.QueueFactoryInterface;
@@ -8,10 +9,12 @@ import io.kestra.core.queues.QueueInterface;
 import io.kestra.core.runners.RunnerUtils;
 
 import java.time.Duration;
+import java.util.List;
 import java.util.Map;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicReference;
+
 import jakarta.inject.Inject;
 import jakarta.inject.Named;
 import jakarta.inject.Singleton;
@@ -41,7 +44,7 @@ public class FlowCaseTest {
     }
 
     @SuppressWarnings({"ResultOfMethodCallIgnored", "unchecked"})
-    void run(String input, State.Type fromState, State.Type triggerState,  int count, String outputs) throws Exception {
+    void run(String input, State.Type fromState, State.Type triggerState, int count, String outputs) throws Exception {
         CountDownLatch countDownLatch = new CountDownLatch(1);
         AtomicReference<Execution> triggered = new AtomicReference<>();
 
@@ -58,7 +61,8 @@ public class FlowCaseTest {
             "task-flow",
             null,
             (f, e) -> ImmutableMap.of("string", input),
-            Duration.ofMinutes(1)
+            Duration.ofMinutes(1),
+            List.of(new Label("mainFlowExecutionLabel", "execFoo"))
         );
 
         countDownLatch.await(1, TimeUnit.MINUTES);
@@ -85,5 +89,12 @@ public class FlowCaseTest {
 
         assertThat(triggered.get().getTaskRunList(), hasSize(count));
         assertThat(triggered.get().getState().getCurrent(), is(triggerState));
+
+        assertThat(triggered.get().getLabels(), hasItems(
+            new Label("mainFlowExecutionLabel", "execFoo"),
+            new Label("mainFlowLabel", "flowFoo"),
+            new Label("launchTaskLabel", "launchFoo"),
+            new Label("switchFlowLabel", "switchFoo")
+        ));
     }
 }
diff --git a/core/src/test/resources/flows/valids/switch.yaml b/core/src/test/resources/flows/valids/switch.yaml
index b2c88be51..f911bcbac 100644
--- a/core/src/test/resources/flows/valids/switch.yaml
+++ b/core/src/test/resources/flows/valids/switch.yaml
@@ -8,6 +8,9 @@ inputs:
     type: STRING
     defaults: amazing
 
+labels:
+  switchFlowLabel: switchFoo
+
 tasks:
   - id: parent-seq
     type: io.kestra.core.tasks.flows.Switch
diff --git a/core/src/test/resources/flows/valids/task-flow.yaml b/core/src/test/resources/flows/valids/task-flow.yaml
index 409725981..54a0a7457 100644
--- a/core/src/test/resources/flows/valids/task-flow.yaml
+++ b/core/src/test/resources/flows/valids/task-flow.yaml
@@ -5,6 +5,9 @@ inputs:
   - name: string
     type: STRING
 
+labels:
+  mainFlowLabel: flowFoo
+
 tasks:
   - id: launch
     type: io.kestra.core.tasks.flows.Flow
@@ -14,5 +17,8 @@ tasks:
       string: "{{ inputs.string }}"
     wait: true
     transmitFailed: true
+    inheritLabels: true
+    labels:
+      launchTaskLabel: launchFoo
     outputs:
       extracted: "{{ outputs.default.value ?? outputs['error-t1'].value }}"
\ No newline at end of file