Repository: kestra-io/kestra
Issue: Allow a subflow to inherit parent flow's execution labels 
Labels: enhancement

### 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 

Issue comments:

Pull request review comments:
Please also add labels to switch flow and within the "launch" subflow task to ensure you're not overriding subflow default labels
Please don't use the ExecutionRepository, you can instead change the method signature to include the parent execution and retrieve the information you want.
The only place where this method is called is [here](https://github.com/kestra-io/kestra/blob/develop/core/src/main/java/io/kestra/core/runners/ExecutorService.java#L582) so you can just add it there :+1: 

Patch that resolved the issue:
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 de656bbf80..f869302b94 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 @@ private Executor handleFlowTask(final Executor executor) {
                 );
 
                 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 cedce2ebb7..29b503beb0 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 String flowUidWithoutRevision() {
     }
 
     @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 Execution createExecution(RunContext runContext, FlowExecutorInterface fl
         }
 
         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())));


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 79deee8031..4e04d4c647 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.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 void invalidOutputs() throws Exception {
     }
 
     @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 @@ void run(String input, State.Type fromState, State.Type triggerState,  int count
             "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 @@ void run(String input, State.Type fromState, State.Type triggerState,  int count
 
         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 b2c88be514..f911bcbacf 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 4097259810..54a0a74579 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