From 791ced866baf13f067237e647240dc99ae806c2e Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Lo=C3=AFc=20Mathieu?= <loikeseke@gmail.com>
Date: Fri, 25 Oct 2024 11:10:27 +0200
Subject: [PATCH] feat(core): refactor labels handling

Fixes https://github.com/kestra-io/kestra-ee/issues/2001
---
 .../core/models/executions/Execution.java     | 62 ++---------------
 .../kestra/core/runners/ExecutableUtils.java  | 14 ++--
 .../java/io/kestra/core/runners/Worker.java   |  7 +-
 .../core/schedulers/AbstractScheduler.java    |  2 +-
 .../io/kestra/core/services/LabelService.java | 56 +++++++++++++++
 .../kestra/plugin/core/execution/Labels.java  | 14 +++-
 .../io/kestra/plugin/core/trigger/Flow.java   | 35 ++--------
 .../kestra/plugin/core/trigger/Schedule.java  | 16 +----
 .../plugin/core/trigger/ScheduleOnDates.java  | 22 +-----
 .../core/serializers/YamlFlowParserTest.java  |  2 +-
 .../core/services/LabelServiceTest.java       | 68 +++++++++++++++++++
 .../controllers/api/ExecutionController.java  | 37 ++++++----
 .../api/ExecutionControllerTest.java          | 17 ++++-
 13 files changed, 200 insertions(+), 152 deletions(-)
 create mode 100644 core/src/main/java/io/kestra/core/services/LabelService.java
 create mode 100644 core/src/test/java/io/kestra/core/services/LabelServiceTest.java

diff --git a/core/src/main/java/io/kestra/core/models/executions/Execution.java b/core/src/main/java/io/kestra/core/models/executions/Execution.java
index 89407f5933..18ddade573 100644
--- a/core/src/main/java/io/kestra/core/models/executions/Execution.java
+++ b/core/src/main/java/io/kestra/core/models/executions/Execution.java
@@ -20,6 +20,7 @@
 import io.kestra.core.runners.RunContextLogger;
 import io.kestra.core.serializers.ListOrMapOfLabelDeserializer;
 import io.kestra.core.serializers.ListOrMapOfLabelSerializer;
+import io.kestra.core.services.LabelService;
 import io.kestra.core.utils.IdUtils;
 import io.kestra.core.utils.MapUtils;
 import io.micronaut.core.annotation.Nullable;
@@ -39,8 +40,6 @@
 import java.util.stream.Stream;
 import java.util.zip.CRC32;
 
-import static io.kestra.core.models.Label.SYSTEM_PREFIX;
-
 @Builder(toBuilder = true)
 @Slf4j
 @Getter
@@ -143,10 +142,7 @@ public static Execution newExecution(final Flow flow,
             .scheduleDate(scheduleDate.map(ChronoZonedDateTime::toInstant).orElse(null))
             .build();
 
-        List<Label> executionLabels = new ArrayList<>();
-        if (flow.getLabels() != null) {
-            executionLabels.addAll(flow.getLabels());
-        }
+        List<Label> executionLabels = new ArrayList<>(LabelService.labelsExcludingSystem(flow));
 
         if (labels != null) {
             executionLabels.addAll(labels);
@@ -162,6 +158,8 @@ public static Execution newExecution(final Flow flow,
         return execution;
     }
 
+
+
     public static class ExecutionBuilder {
         void prebuild() {
             this.originalId = this.id;
@@ -181,12 +179,6 @@ public Execution build() {
             this.prebuild();
             return super.build();
         }
-
-        @Override
-        public ExecutionBuilder labels(List<Label> labels) {
-            checkForSystemLabels(labels);
-            return super.labels(labels);
-        }
     }
 
     public Execution withState(State.Type state) {
@@ -212,53 +204,9 @@ public Execution withState(State.Type state) {
         );
     }
 
-    /**
-     * This method replaces labels with new ones.
-     * It refuses system labels as they must be passed via the withSystemLabels method.
-     */
     public Execution withLabels(List<Label> labels) {
-        checkForSystemLabels(labels);
-
-        return new Execution(
-            this.tenantId,
-            this.id,
-            this.namespace,
-            this.flowId,
-            this.flowRevision,
-            this.taskRunList,
-            this.inputs,
-            this.outputs,
-            labels,
-            this.variables,
-            this.state,
-            this.parentId,
-            this.originalId,
-            this.trigger,
-            this.deleted,
-            this.metadata,
-            this.scheduleDate,
-            this.error
-        );
-    }
 
-    private static void checkForSystemLabels(List<Label> labels) {
-        if (labels != null) {
-            Optional<Label> first = labels.stream().filter(label -> label.key() != null && label.key().startsWith(SYSTEM_PREFIX)).findFirst();
-            if (first.isPresent()) {
-                throw new IllegalArgumentException("System labels can only be set by Kestra itself, offending label: " + first.get().key() + "=" + first.get().value());
-            }
-        }
-    }
 
-    /**
-     * This method in <b>only to be used</b> to add system labels to an execution.
-     * It will not replace exisiting labels but add new one (possibly duplicating).
-     */
-    public Execution withSystemLabels(List<Label> labels) {
-        List<Label> newLabels = this.labels == null ? new ArrayList<>() : this.labels;
-        if (labels != null) {
-            newLabels.addAll(labels);
-        }
         return new Execution(
             this.tenantId,
             this.id,
@@ -268,7 +216,7 @@ public Execution withSystemLabels(List<Label> labels) {
             this.taskRunList,
             this.inputs,
             this.outputs,
-            newLabels,
+            labels,
             this.variables,
             this.state,
             this.parentId,
diff --git a/core/src/main/java/io/kestra/core/runners/ExecutableUtils.java b/core/src/main/java/io/kestra/core/runners/ExecutableUtils.java
index af5aaac76c..46d086ab4b 100644
--- a/core/src/main/java/io/kestra/core/runners/ExecutableUtils.java
+++ b/core/src/main/java/io/kestra/core/runners/ExecutableUtils.java
@@ -94,11 +94,14 @@ public static <T extends Task & ExecutableTask<?>> SubflowExecution<?> subflowEx
         );
 
         // propagate system labels and compute correlation ID if not already existing
-        List<Label> systemLabels = Streams.of(currentExecution.getLabels())
+        List<Label> newLabels = Streams.of(currentExecution.getLabels())
             .filter(label -> label.key().startsWith(Label.SYSTEM_PREFIX))
             .collect(Collectors.toList());
-        if (systemLabels.stream().noneMatch(label -> label.key().equals(Label.CORRELATION_ID))) {
-            systemLabels.add(new Label(Label.CORRELATION_ID, currentExecution.getId()));
+        if (newLabels.stream().noneMatch(label -> label.key().equals(Label.CORRELATION_ID))) {
+            newLabels.add(new Label(Label.CORRELATION_ID, currentExecution.getId()));
+        }
+        if (labels != null) {
+            newLabels.addAll(labels);
         }
 
         FlowInputOutput flowInputOutput = ((DefaultRunContext)runContext).getApplicationContext().getBean(FlowInputOutput.class);
@@ -107,7 +110,7 @@ public static <T extends Task & ExecutableTask<?>> SubflowExecution<?> subflowEx
             .newExecution(
                 flow,
                 (f, e) -> flowInputOutput.readExecutionInputs(f, e, inputs),
-                labels,
+                newLabels,
                 Optional.empty())
             .withTrigger(ExecutionTrigger.builder()
                 .id(currentTask.getId())
@@ -115,8 +118,7 @@ public static <T extends Task & ExecutableTask<?>> SubflowExecution<?> subflowEx
                 .variables(variables)
                 .build()
             )
-            .withScheduleDate(scheduleOnDate)
-            .withSystemLabels(systemLabels);
+            .withScheduleDate(scheduleOnDate);
         return SubflowExecution.builder()
             .parentTask(currentTask)
             .parentTaskRun(currentTaskRun.withState(State.Type.RUNNING))
diff --git a/core/src/main/java/io/kestra/core/runners/Worker.java b/core/src/main/java/io/kestra/core/runners/Worker.java
index 03c1308ab1..c3e1bb3dd7 100644
--- a/core/src/main/java/io/kestra/core/runners/Worker.java
+++ b/core/src/main/java/io/kestra/core/runners/Worker.java
@@ -19,6 +19,7 @@
 import io.kestra.core.server.ServerConfig;
 import io.kestra.core.server.Service;
 import io.kestra.core.server.ServiceStateChangeEvent;
+import io.kestra.core.services.LabelService;
 import io.kestra.core.services.LogService;
 import io.kestra.core.services.WorkerGroupService;
 import io.kestra.core.utils.Await;
@@ -361,11 +362,11 @@ private void publishTriggerExecution(WorkerTrigger workerTrigger, Optional<Execu
             );
         }
 
-        var flowLabels = workerTrigger.getConditionContext().getFlow().getLabels();
-        if (flowLabels != null) {
+        var flow = workerTrigger.getConditionContext().getFlow();
+        if (flow.getLabels() != null) {
             evaluate = evaluate.map(execution -> {
                     List<Label> executionLabels = execution.getLabels() != null ? execution.getLabels() : new ArrayList<>();
-                    executionLabels.addAll(flowLabels);
+                    executionLabels.addAll(LabelService.labelsExcludingSystem(flow));
                     return execution.withLabels(executionLabels);
                 }
             );
diff --git a/core/src/main/java/io/kestra/core/schedulers/AbstractScheduler.java b/core/src/main/java/io/kestra/core/schedulers/AbstractScheduler.java
index 73e16b2ff9..c5106e2826 100644
--- a/core/src/main/java/io/kestra/core/schedulers/AbstractScheduler.java
+++ b/core/src/main/java/io/kestra/core/schedulers/AbstractScheduler.java
@@ -564,7 +564,7 @@ private void handle() {
                             .namespace(f.getTriggerContext().getNamespace())
                             .flowId(f.getTriggerContext().getFlowId())
                             .flowRevision(f.getFlow().getRevision())
-                            .labels(f.getFlow().getLabels())
+                            .labels(LabelService.labelsExcludingSystem(f.getFlow()))
                             .state(new State().withState(State.Type.FAILED))
                             .error(ExecutionError.from(ie))
                             .build();
diff --git a/core/src/main/java/io/kestra/core/services/LabelService.java b/core/src/main/java/io/kestra/core/services/LabelService.java
new file mode 100644
index 0000000000..f47456df84
--- /dev/null
+++ b/core/src/main/java/io/kestra/core/services/LabelService.java
@@ -0,0 +1,56 @@
+package io.kestra.core.services;
+
+import io.kestra.core.exceptions.IllegalVariableEvaluationException;
+import io.kestra.core.models.Label;
+import io.kestra.core.models.flows.Flow;
+import io.kestra.core.models.triggers.AbstractTrigger;
+import io.kestra.core.runners.RunContext;
+import io.kestra.core.utils.ListUtils;
+
+import java.util.ArrayList;
+import java.util.List;
+
+public final class LabelService {
+    private LabelService() {}
+
+    /**
+     * Return flow labels excluding system labels.
+     */
+    public static List<Label> labelsExcludingSystem(Flow flow) {
+        return ListUtils.emptyOnNull(flow.getLabels()).stream().filter(label -> !label.key().startsWith(Label.SYSTEM_PREFIX)).toList();
+    }
+
+    /**
+     * Return flow labels excluding system labels concatenated with trigger labels.
+     *
+     * Trigger labels will be rendered via the run context but not flow labels.
+     * In case rendering is not possible, the label will be omitted.
+     */
+    public static List<Label> fromTrigger(RunContext runContext, Flow flow, AbstractTrigger trigger) {
+        final List<Label> labels = new ArrayList<>();
+
+        if (flow.getLabels() != null) {
+            labels.addAll(LabelService.labelsExcludingSystem(flow)); // no need for rendering
+        }
+
+        if (trigger.getLabels() != null) {
+            for (Label label : trigger.getLabels()) {
+                final var value = renderLabelValue(runContext, label);
+                if (value != null) {
+                    labels.add(new Label(label.key(), value));
+                }
+            }
+        }
+
+        return labels;
+    }
+
+    private static String renderLabelValue(RunContext runContext, Label label) {
+        try {
+            return runContext.render(label.value());
+        } catch (IllegalVariableEvaluationException e) {
+            runContext.logger().warn("Failed to render label '{}', it will be omitted", label.key(), e);
+            return null;
+        }
+    }
+}
diff --git a/core/src/main/java/io/kestra/plugin/core/execution/Labels.java b/core/src/main/java/io/kestra/plugin/core/execution/Labels.java
index aa0b1d50e7..7b8b866997 100644
--- a/core/src/main/java/io/kestra/plugin/core/execution/Labels.java
+++ b/core/src/main/java/io/kestra/plugin/core/execution/Labels.java
@@ -26,6 +26,7 @@
 import java.util.Optional;
 import java.util.stream.Collectors;
 
+import static io.kestra.core.models.Label.SYSTEM_PREFIX;
 import static io.kestra.core.utils.Rethrow.throwBiConsumer;
 import static io.kestra.core.utils.Rethrow.throwFunction;
 
@@ -35,7 +36,8 @@
 @Getter
 @NoArgsConstructor
 @Schema(
-    title = "Allow to add or overwrite labels for the current execution at runtime."
+    title = "Allow to add or overwrite labels for the current execution at runtime.",
+    description = "Trying to pass a system label (a label starting with `system_`) will fail the task."
 )
 @Plugin(
     examples = {
@@ -131,11 +133,17 @@ public Execution update(Execution execution, RunContext runContext) throws Excep
                 );
             }));
 
+            // check for system labels: none can be passed at runtime
+            Optional<Map.Entry<String, String>> first = newLabels.entrySet().stream().filter(entry -> entry.getKey().startsWith(SYSTEM_PREFIX)).findFirst();
+            if (first.isPresent()) {
+                throw new IllegalArgumentException("System labels can only be set by Kestra itself, offending label: " + first.get().getKey() + "=" + first.get().getValue());
+            }
+
             return execution.withLabels(newLabels.entrySet().stream()
-                    .map(throwFunction(entry -> new Label(
+                    .map(entry -> new Label(
                         entry.getKey(),
                         entry.getValue()
-                    )))
+                    ))
                     .toList());
         } else {
             throw new IllegalVariableEvaluationException("Unknown value type: " + labels.getClass());
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 13c1e37ad4..63ddd0f62f 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
@@ -4,6 +4,7 @@
 import io.kestra.core.models.Label;
 import io.kestra.core.models.annotations.PluginProperty;
 import io.kestra.core.models.property.Property;
+import io.kestra.core.services.LabelService;
 import io.swagger.v3.oas.annotations.media.Schema;
 import java.util.ArrayList;
 import java.util.List;
@@ -60,7 +61,7 @@
                 outputs:
                   - id: last_ingested_date
                     type: STRING
-                    value: "{{ outputs.final_date.value }}"            
+                    value: "{{ outputs.final_date.value }}"
                 ```
                 Below is the `transform` flow triggered in response to the `extract` flow's successful completion.""",
             code = """
@@ -74,7 +75,7 @@
 
                 variables:
                   result: |
-                    Ingestion done in {{ trigger.executionId }}. 
+                    Ingestion done in {{ trigger.executionId }}.
                     Now transforming data up to {{ inputs.last_ingested_date }}
 
                 tasks:
@@ -144,7 +145,7 @@ public Optional<Execution> evaluate(RunContext runContext, io.kestra.core.models
             .namespace(flow.getNamespace())
             .flowId(flow.getId())
             .flowRevision(flow.getRevision())
-            .labels(generateLabels(runContext, flow))
+            .labels(LabelService.fromTrigger(runContext, flow, this))
             .state(new State())
             .trigger(ExecutionTrigger.of(
                 this,
@@ -181,34 +182,6 @@ public Optional<Execution> evaluate(RunContext runContext, io.kestra.core.models
         }
     }
 
-    private List<Label> generateLabels(RunContext runContext, io.kestra.core.models.flows.Flow flow) {
-        final List<Label> labels = new ArrayList<>();
-
-        if (flow.getLabels() != null) {
-            labels.addAll(flow.getLabels()); // no need for rendering
-        }
-
-        if (this.getLabels() != null) {
-            for (Label label : this.getLabels()) {
-                final var value = renderLabelValue(runContext, label);
-                if (value != null) {
-                    labels.add(new Label(label.key(), value));
-                }
-            }
-        }
-
-        return labels;
-    }
-
-    private String renderLabelValue(RunContext runContext, Label label) {
-        try {
-            return runContext.render(label.value());
-        } catch (IllegalVariableEvaluationException e) {
-            runContext.logger().warn("Failed to render label '{}', it will be omitted", label.key(), e);
-            return null;
-        }
-    }
-
     @Builder
     @ToString
     @EqualsAndHashCode
diff --git a/core/src/main/java/io/kestra/plugin/core/trigger/Schedule.java b/core/src/main/java/io/kestra/plugin/core/trigger/Schedule.java
index cb125ef39a..1a348e03ae 100644
--- a/core/src/main/java/io/kestra/plugin/core/trigger/Schedule.java
+++ b/core/src/main/java/io/kestra/plugin/core/trigger/Schedule.java
@@ -20,6 +20,7 @@
 import io.kestra.core.runners.DefaultRunContext;
 import io.kestra.core.runners.RunContext;
 import io.kestra.core.services.ConditionService;
+import io.kestra.core.services.LabelService;
 import io.kestra.core.utils.ListUtils;
 import io.kestra.core.validations.ScheduleValidation;
 import io.kestra.core.validations.TimezoneId;
@@ -425,11 +426,7 @@ public Cron parseCron() {
     }
 
     private List<Label> generateLabels(RunContext runContext, ConditionContext conditionContext, Backfill backfill) throws IllegalVariableEvaluationException {
-        List<Label> labels = new ArrayList<>();
-
-        if (conditionContext.getFlow().getLabels() != null) {
-            labels.addAll(conditionContext.getFlow().getLabels()); // no need for rendering
-        }
+        List<Label> labels = LabelService.fromTrigger(runContext, conditionContext.getFlow(), this);
 
         if (backfill != null && backfill.getLabels() != null) {
             for (Label label : backfill.getLabels()) {
@@ -440,15 +437,6 @@ private List<Label> generateLabels(RunContext runContext, ConditionContext condi
             }
         }
 
-        if (this.getLabels() != null) {
-            for (Label label : this.getLabels()) {
-                final var value = runContext.render(label.value());
-                if (value != null) {
-                    labels.add(new Label(label.key(), value));
-                }
-            }
-        }
-
         return labels;
     }
 
diff --git a/core/src/main/java/io/kestra/plugin/core/trigger/ScheduleOnDates.java b/core/src/main/java/io/kestra/plugin/core/trigger/ScheduleOnDates.java
index 46ae73d66f..675aa9d5ff 100644
--- a/core/src/main/java/io/kestra/plugin/core/trigger/ScheduleOnDates.java
+++ b/core/src/main/java/io/kestra/plugin/core/trigger/ScheduleOnDates.java
@@ -10,6 +10,7 @@
 import io.kestra.core.models.tasks.VoidOutput;
 import io.kestra.core.models.triggers.*;
 import io.kestra.core.runners.RunContext;
+import io.kestra.core.services.LabelService;
 import io.kestra.core.validations.TimezoneId;
 import io.swagger.v3.oas.annotations.media.Schema;
 import jakarta.validation.constraints.NotNull;
@@ -83,7 +84,7 @@ public Optional<Execution> evaluate(ConditionContext conditionContext, TriggerCo
                 this,
                 conditionContext,
                 triggerContext,
-                generateLabels(runContext, conditionContext),
+                LabelService.fromTrigger(runContext, conditionContext.getFlow(), this),
                 this.inputs != null ? runContext.render(this.inputs) : Collections.emptyMap(),
                 Collections.emptyMap(),
                 nextDate
@@ -132,23 +133,4 @@ private Optional<ZonedDateTime> nextDate(RunContext runContext, Predicate<ZonedD
             .findFirst()
             .map(date -> date.truncatedTo(ChronoUnit.SECONDS));
     }
-
-    private List<Label> generateLabels(RunContext runContext, ConditionContext conditionContext) throws IllegalVariableEvaluationException {
-        List<Label> labels = new ArrayList<>();
-
-        if (conditionContext.getFlow().getLabels() != null) {
-            labels.addAll(conditionContext.getFlow().getLabels()); // no need for rendering
-        }
-
-        if (this.getLabels() != null) {
-            for (Label label : this.getLabels()) {
-                final var value = runContext.render(label.value());
-                if (value != null) {
-                    labels.add(new Label(label.key(), value));
-                }
-            }
-        }
-
-        return labels;
-    }
 }
diff --git a/core/src/test/java/io/kestra/core/serializers/YamlFlowParserTest.java b/core/src/test/java/io/kestra/core/serializers/YamlFlowParserTest.java
index 216ed2b519..94ee3db6a2 100644
--- a/core/src/test/java/io/kestra/core/serializers/YamlFlowParserTest.java
+++ b/core/src/test/java/io/kestra/core/serializers/YamlFlowParserTest.java
@@ -167,7 +167,7 @@ void serialization() throws IOException {
         Flow flow = this.parse("flows/valids/minimal.yaml");
 
         String s = mapper.writeValueAsString(flow);
-        assertThat(s, is("{\"id\":\"minimal\",\"namespace\":\"io.kestra.tests\",\"revision\":2,\"disabled\":false,\"deleted\":false,\"tasks\":[{\"id\":\"date\",\"type\":\"io.kestra.plugin.core.debug.Return\",\"format\":\"{{taskrun.startDate}}\"}]}"));
+        assertThat(s, is("{\"id\":\"minimal\",\"namespace\":\"io.kestra.tests\",\"revision\":2,\"disabled\":false,\"deleted\":false,\"labels\":[{\"key\":\"system_readOnly\",\"value\":\"true\"}],\"tasks\":[{\"id\":\"date\",\"type\":\"io.kestra.plugin.core.debug.Return\",\"format\":\"{{taskrun.startDate}}\"}]}"));
     }
 
     @Test
diff --git a/core/src/test/java/io/kestra/core/services/LabelServiceTest.java b/core/src/test/java/io/kestra/core/services/LabelServiceTest.java
new file mode 100644
index 0000000000..cb28162e87
--- /dev/null
+++ b/core/src/test/java/io/kestra/core/services/LabelServiceTest.java
@@ -0,0 +1,68 @@
+package io.kestra.core.services;
+
+import io.kestra.core.junit.annotations.KestraTest;
+import io.kestra.core.models.Label;
+import io.kestra.core.models.flows.Flow;
+import io.kestra.core.models.triggers.AbstractTrigger;
+import io.kestra.core.runners.RunContext;
+import io.kestra.core.runners.RunContextFactory;
+import io.kestra.plugin.core.trigger.Schedule;
+import jakarta.inject.Inject;
+import org.junit.jupiter.api.Test;
+
+import java.util.List;
+import java.util.Map;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.*;
+
+@KestraTest
+class LabelServiceTest {
+
+    @Inject
+    private RunContextFactory runContextFactory;
+
+    @Test
+    void shouldFilterSystemLabels() {
+        Flow flow = Flow.builder()
+            .labels(List.of(new Label("key", "value"), new Label("system_Label", "systemValue")))
+            .build();
+
+        List<Label> labels = LabelService.labelsExcludingSystem(flow);
+
+        assertThat(labels, hasSize(1));
+        assertThat(labels.getFirst(), is(new Label("key", "value")));
+    }
+
+    @Test
+    void shouldReturnLabelsFromFlowAndTrigger() {
+        RunContext runContext = runContextFactory.of(Map.of("variable", "variableValue"));
+        Flow flow = Flow.builder()
+            .labels(List.of(new Label("key", "value"), new Label("system_Label", "systemValue")))
+            .build();
+        AbstractTrigger trigger = Schedule.builder()
+            .labels(List.of(new Label("scheduleLabel", "scheduleValue"), new Label("variable", "{{variable}}")))
+            .build();
+
+        List<Label> labels = LabelService.fromTrigger(runContext, flow, trigger);
+
+        assertThat(labels, hasSize(3));
+        assertThat(labels, hasItems(new Label("key", "value"), new Label("scheduleLabel", "scheduleValue"), new Label("variable", "variableValue")));
+    }
+
+    @Test
+    void shouldFilterNonRenderableLabels() {
+        RunContext runContext = runContextFactory.of();
+        Flow flow = Flow.builder()
+            .labels(List.of(new Label("key", "value"), new Label("system_Label", "systemValue")))
+            .build();
+        AbstractTrigger trigger = Schedule.builder()
+            .labels(List.of(new Label("scheduleLabel", "scheduleValue"), new Label("variable", "{{variable}}")))
+            .build();
+
+        List<Label> labels = LabelService.fromTrigger(runContext, flow, trigger);
+
+        assertThat(labels, hasSize(2));
+        assertThat(labels, hasItems(new Label("key", "value"), new Label("scheduleLabel", "scheduleValue")));
+    }
+}
\ No newline at end of file
diff --git a/webserver/src/main/java/io/kestra/webserver/controllers/api/ExecutionController.java b/webserver/src/main/java/io/kestra/webserver/controllers/api/ExecutionController.java
index 5f682b0a4e..3aa7ca76e1 100644
--- a/webserver/src/main/java/io/kestra/webserver/controllers/api/ExecutionController.java
+++ b/webserver/src/main/java/io/kestra/webserver/controllers/api/ExecutionController.java
@@ -29,10 +29,7 @@
 import io.kestra.core.runners.FlowInputOutput;
 import io.kestra.core.runners.RunContext;
 import io.kestra.core.runners.RunContextFactory;
-import io.kestra.core.services.ConditionService;
-import io.kestra.core.services.ExecutionService;
-import io.kestra.core.services.FlowService;
-import io.kestra.core.services.GraphService;
+import io.kestra.core.services.*;
 import io.kestra.core.storages.StorageContext;
 import io.kestra.core.storages.StorageInterface;
 import io.kestra.core.tenant.TenantService;
@@ -89,7 +86,6 @@
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.io.FilenameUtils;
 import org.apache.commons.lang3.exception.ExceptionUtils;
-import org.apache.commons.lang3.tuple.Pair;
 import org.reactivestreams.Publisher;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.FluxSink;
@@ -112,6 +108,8 @@
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.stream.Collectors;
 
+import static io.kestra.core.models.Label.CORRELATION_ID;
+import static io.kestra.core.models.Label.SYSTEM_PREFIX;
 import static io.kestra.core.utils.DateUtils.validateTimeline;
 import static io.kestra.core.utils.Rethrow.throwConsumer;
 import static io.kestra.core.utils.Rethrow.throwFunction;
@@ -533,7 +531,7 @@ protected HttpResponse<Execution> webhook(
 
         var result = execution.get();
         if (flow.getLabels() != null) {
-            result = result.withLabels(flow.getLabels());
+            result = result.withLabels(LabelService.labelsExcludingSystem(flow));
         }
 
         // we check conditions here as it's easier as the execution is created we have the body and headers available for the runContext
@@ -583,8 +581,8 @@ public Publisher<ApiValidateExecutionInputsResponse> validateInputsOnCreate(
         @Parameter(description = "The flow revision or latest if null") @QueryValue Optional<Integer> revision
     ) {
         Flow flow = flowService.getFlowIfExecutableOrThrow(tenantService.resolveTenant(), namespace, id, revision);
-        Pair<List<Label>, List<Label>> parsedLabels = parseLabels(labels);
-        Execution execution = Execution.newExecution(flow, parsedLabels.getLeft()).withSystemLabels(parsedLabels.getRight());
+        List<Label> parsedLabels = parseLabels(labels);
+        Execution execution = Execution.newExecution(flow, parsedLabels);
         return flowInputOutput
             .validateExecutionInputs(flow.getInputs(), execution, inputs)
             .map(values -> ApiValidateExecutionInputsResponse.of(id, namespace, values));
@@ -605,8 +603,8 @@ public Publisher<ExecutionResponse> create(
         @Parameter(description = "Schedule the flow on a specific date") @QueryValue Optional<ZonedDateTime> scheduleDate
     ) throws IOException {
         Flow flow = flowService.getFlowIfExecutableOrThrow(tenantService.resolveTenant(), namespace, id, revision);
-        Pair<List<Label>, List<Label>> parsedLabels = parseLabels(labels);
-        Execution current = Execution.newExecution(flow, null, parsedLabels.getLeft(), scheduleDate).withSystemLabels(parsedLabels.getRight());
+        List<Label> parsedLabels = parseLabels(labels);
+        Execution current = Execution.newExecution(flow, null, parsedLabels, scheduleDate);
         Mono<CompletableFuture<ExecutionResponse>> handle = flowInputOutput.readExecutionInputs(flow, current, inputs)
             .handle((executionInputs, sink) -> {
                 Execution executionWithInputs = current.withInputs(executionInputs);
@@ -685,14 +683,17 @@ public static ExecutionResponse fromExecution(Execution execution, URI url) {
         }
     }
 
-    protected Pair<List<Label>, List<Label>> parseLabels(List<String> labels) {
-        // We allow passing the correlation id from the API but only this one, other system labels will go through and fail at execution creation.
+    protected List<Label> parseLabels(List<String> labels) {
         List<Label> parsedLabels =  labels == null ? Collections.emptyList() : RequestUtils.toMap(labels).entrySet().stream()
             .map(entry -> new Label(entry.getKey(), entry.getValue()))
             .toList();
-        List<Label> standardLabels = parsedLabels.stream().filter(label -> !label.key().equals(Label.CORRELATION_ID)).toList();
-        List<Label> systemLabels = parsedLabels.stream().filter(label -> label.key().equals(Label.CORRELATION_ID)).toList();
-        return Pair.of(standardLabels, systemLabels);
+
+        // check for system labels: none can be passed at execution creation time except system_correlationId
+        Optional<Label> first = parsedLabels.stream().filter(label -> !label.key().equals(CORRELATION_ID) && label.key().startsWith(SYSTEM_PREFIX)).findFirst();
+        if (first.isPresent()) {
+            throw new IllegalArgumentException("System labels can only be set by Kestra itself, offending label: " + first.get().key() + "=" + first.get().value());
+        }
+        return parsedLabels;
     }
 
     protected <T> HttpResponse<T> validateFile(String executionId, URI path, String redirect) {
@@ -1699,6 +1700,12 @@ public HttpResponse<?> setLabels(
     }
 
     private Execution setLabels(Execution execution, List<Label> labels) {
+        // check for system labels: none can be passed at runtime
+        Optional<Label> first = labels.stream().filter(label -> label.key().startsWith(SYSTEM_PREFIX)).findFirst();
+        if (first.isPresent()) {
+            throw new IllegalArgumentException("System labels can only be set by Kestra itself, offending label: " + first.get().key() + "=" + first.get().value());
+        }
+
         Map<String, String> newLabels = labels.stream().collect(Collectors.toMap(Label::key, Label::value));
         if (execution.getLabels() != null) {
             execution.getLabels().forEach(
diff --git a/webserver/src/test/java/io/kestra/webserver/controllers/api/ExecutionControllerTest.java b/webserver/src/test/java/io/kestra/webserver/controllers/api/ExecutionControllerTest.java
index df7226217b..b2f8979111 100644
--- a/webserver/src/test/java/io/kestra/webserver/controllers/api/ExecutionControllerTest.java
+++ b/webserver/src/test/java/io/kestra/webserver/controllers/api/ExecutionControllerTest.java
@@ -1594,7 +1594,7 @@ void shouldPauseByQueryRunningFlows() throws TimeoutException, QueueException {
     }
 
     @Test
-    void shouldRefuseSystemLabels() {
+    void shouldRefuseSystemLabelsWhenCreatingAnExecution() {
         var error = assertThrows(HttpClientResponseException.class, () -> client.toBlocking().retrieve(
             HttpRequest
                 .POST("/api/v1/executions/io.kestra.tests/minimal?labels=system_label:system", null)
@@ -1604,4 +1604,19 @@ void shouldRefuseSystemLabels() {
 
         assertThat(error.getStatus(), is(HttpStatus.UNPROCESSABLE_ENTITY));
     }
+
+    @Test
+    void shouldRefuseSystemLabelsWhenUpdatingLabels() throws QueueException, TimeoutException {
+        // update label on a terminated execution
+        Execution result = runnerUtils.runOne(null, "io.kestra.tests", "minimal");
+        assertThat(result.getState().getCurrent(), is(State.Type.SUCCESS));
+
+        var error = assertThrows(HttpClientResponseException.class, () -> client.toBlocking().retrieve(
+                HttpRequest.POST("/api/v1/executions/" + result.getId() + "/labels", List.of(new Label("system_label", "value"))),
+                Execution.class
+            )
+        );
+
+        assertThat(error.getStatus(), is(HttpStatus.UNPROCESSABLE_ENTITY));
+    }
 }