Repository: kestra-io/kestra
Issue: Labels are not created correctly when deployed by Terraform
Labels: bug

### Expected Behavior

When deploying a flow using Terraform, I want the labels to be created correctly.

If I wrote labels in a flow, using the key/value representation, such as
```yaml
labels:
  env: prd
  country: FR
```

I want them to be created as following, in the target Kestra server:
```yaml
labels:
  env: prd
  country: FR
```

### Actual Behaviour

Instead of the expected behavior, the flows created by Terraform don't contain labels. Instead, we get a list with an empty object.

```yaml
labels:
- {}
```

We get this unexpected behavior for any flow.

### Steps To Reproduce

_No response_

### Environment Information

- Kestra Version: 0.11.0
- Operating System (OS / Docker / Kubernetes): Kubernetes
- Kestra TF Operator Version: 0.7.0

### Example flow

_No response_

Issue comments:
can you try to update terraform provider version please to [0.10.0](https://registry.terraform.io/providers/kestra-io/kestra/latest) please ? 
Hello!

Indeed, upgrading the TF provider to 0.10.0+ solved our issue. The problem seems to originate from the JSON resource on the web server, which was used to deploy flows in previous versions.

This also involved updating the flows, and adding/uncommenting the id/namespace pairs: they were previously not allowed with the JSON resource, but are now required with the YAML resource. If a flow is created without them, the web server returns an HTTP 401 error without clear explanation.

Thanks for your help,
Yoann

Pull request review comments:

Patch that resolved the issue:
diff --git a/core/src/main/java/io/kestra/core/models/flows/Flow.java b/core/src/main/java/io/kestra/core/models/flows/Flow.java
index e29e7d1df6..a12b51a664 100644
--- a/core/src/main/java/io/kestra/core/models/flows/Flow.java
+++ b/core/src/main/java/io/kestra/core/models/flows/Flow.java
@@ -10,7 +10,6 @@
 import io.kestra.core.exceptions.InternalException;
 import io.kestra.core.models.DeletedInterface;
 import io.kestra.core.models.Label;
-import io.kestra.core.models.annotations.PluginProperty;
 import io.kestra.core.models.executions.Execution;
 import io.kestra.core.models.listeners.Listener;
 import io.kestra.core.models.tasks.FlowableTask;
@@ -29,17 +28,16 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.util.*;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
 import javax.validation.ConstraintViolation;
 import javax.validation.ConstraintViolationException;
 import javax.validation.Valid;
 import javax.validation.constraints.*;
-import java.util.*;
-import java.util.stream.Collectors;
-import java.util.stream.Stream;
 
 @SuperBuilder(toBuilder = true)
 @Getter
-@AllArgsConstructor
 @NoArgsConstructor
 @Introspected
 @ToString
@@ -64,7 +62,6 @@ public boolean hasIgnoreMarker(final AnnotatedMember m) {
     @Pattern(regexp = "[a-z0-9._-]+")
     String namespace;
 
-    @With
     @Min(value = 1)
     Integer revision;
 
@@ -75,7 +72,6 @@ public boolean hasIgnoreMarker(final AnnotatedMember m) {
     @Schema(implementation = Object.class, anyOf = {List.class, Map.class})
     List<Label> labels;
 
-
     @Valid
     List<Input<?>> inputs;
 
@@ -322,21 +318,9 @@ public String generateSource() {
     }
 
     public Flow toDeleted() {
-        return new Flow(
-            this.id,
-            this.namespace,
-            this.revision + 1,
-            this.description,
-            this.labels,
-            this.inputs,
-            this.variables,
-            this.tasks,
-            this.errors,
-            this.listeners,
-            this.triggers,
-            this.taskDefaults,
-            this.disabled,
-            true
-        );
+        return this.toBuilder()
+            .revision(this.revision + 1)
+            .deleted(true)
+            .build();
     }
 }
diff --git a/core/src/main/java/io/kestra/core/models/flows/FlowWithException.java b/core/src/main/java/io/kestra/core/models/flows/FlowWithException.java
index abf1c5a5d2..4e09794ba3 100644
--- a/core/src/main/java/io/kestra/core/models/flows/FlowWithException.java
+++ b/core/src/main/java/io/kestra/core/models/flows/FlowWithException.java
@@ -1,12 +1,14 @@
 package io.kestra.core.models.flows;
 
 import io.micronaut.core.annotation.Introspected;
-import lombok.*;
+import lombok.EqualsAndHashCode;
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+import lombok.ToString;
 import lombok.experimental.SuperBuilder;
 
 @SuperBuilder(toBuilder = true)
 @Getter
-@AllArgsConstructor
 @NoArgsConstructor
 @Introspected
 @ToString
diff --git a/core/src/main/java/io/kestra/core/models/flows/FlowWithSource.java b/core/src/main/java/io/kestra/core/models/flows/FlowWithSource.java
index 6c950f01c4..be9049b12a 100644
--- a/core/src/main/java/io/kestra/core/models/flows/FlowWithSource.java
+++ b/core/src/main/java/io/kestra/core/models/flows/FlowWithSource.java
@@ -2,20 +2,16 @@
 
 import io.kestra.core.services.FlowService;
 import io.micronaut.core.annotation.Introspected;
-import lombok.AllArgsConstructor;
 import lombok.Getter;
 import lombok.NoArgsConstructor;
 import lombok.ToString;
 import lombok.experimental.SuperBuilder;
-import lombok.extern.slf4j.Slf4j;
 
 @SuperBuilder(toBuilder = true)
 @Getter
-@AllArgsConstructor
 @NoArgsConstructor
 @Introspected
 @ToString
-@Slf4j
 public class FlowWithSource extends Flow {
     String source;
 
diff --git a/jdbc/src/main/java/io/kestra/jdbc/repository/AbstractJdbcFlowRepository.java b/jdbc/src/main/java/io/kestra/jdbc/repository/AbstractJdbcFlowRepository.java
index 8f10e1f2ff..13defaef92 100644
--- a/jdbc/src/main/java/io/kestra/jdbc/repository/AbstractJdbcFlowRepository.java
+++ b/jdbc/src/main/java/io/kestra/jdbc/repository/AbstractJdbcFlowRepository.java
@@ -348,6 +348,10 @@ public FlowWithSource update(Flow flow, Flow previous, String flowSource, Flow f
 
     @SneakyThrows
     private FlowWithSource save(Flow flow, CrudEventType crudEventType, String flowSource) throws ConstraintViolationException {
+        if (flow instanceof FlowWithSource) {
+            flow = ((FlowWithSource) flow).toFlow();
+        }
+
         // flow exists, return it
         Optional<FlowWithSource> exists = this.findByIdWithSource(flow.getNamespace(), flow.getId());
         if (exists.isPresent() && exists.get().isUpdatable(flow, flowSource)) {
@@ -357,9 +361,9 @@ private FlowWithSource save(Flow flow, CrudEventType crudEventType, String flowS
         List<FlowWithSource> revisions = this.findRevisions(flow.getNamespace(), flow.getId());
 
         if (revisions.size() > 0) {
-            flow = flow.withRevision(revisions.get(revisions.size() - 1).getRevision() + 1);
+            flow = flow.toBuilder().revision(revisions.get(revisions.size() - 1).getRevision() + 1).build();
         } else {
-            flow = flow.withRevision(1);
+            flow = flow.toBuilder().revision(1).build();
         }
 
         Map<Field<Object>, Object> fields = this.jdbcRepository.persistFields(flow);
@@ -376,6 +380,10 @@ private FlowWithSource save(Flow flow, CrudEventType crudEventType, String flowS
     @SneakyThrows
     @Override
     public Flow delete(Flow flow) {
+        if (flow instanceof FlowWithSource) {
+            flow = ((FlowWithSource) flow).toFlow();
+        }
+
         Optional<Flow> revision = this.findById(flow.getNamespace(), flow.getId(), Optional.of(flow.getRevision()));
         if (revision.isEmpty()) {
             throw new IllegalStateException("Flow " + flow.getId() + " doesn't exists");
diff --git a/repository-memory/src/main/java/io/kestra/repository/memory/MemoryFlowRepository.java b/repository-memory/src/main/java/io/kestra/repository/memory/MemoryFlowRepository.java
index 31abf66595..9e3e49e8dc 100644
--- a/repository-memory/src/main/java/io/kestra/repository/memory/MemoryFlowRepository.java
+++ b/repository-memory/src/main/java/io/kestra/repository/memory/MemoryFlowRepository.java
@@ -195,6 +195,10 @@ public FlowWithSource update(Flow flow, Flow previous, String flowSource, Flow f
     }
 
     private FlowWithSource save(Flow flow, CrudEventType crudEventType, String flowSource) throws ConstraintViolationException {
+        if (flow instanceof FlowWithSource) {
+            flow = ((FlowWithSource) flow).toFlow();
+        }
+
         // flow exists, return it
         Optional<Flow> exists = this.findById(flow.getNamespace(), flow.getId());
         Optional<String> existsSource = this.findSourceById(flow.getNamespace(), flow.getId());
@@ -205,9 +209,9 @@ private FlowWithSource save(Flow flow, CrudEventType crudEventType, String flowS
         List<FlowWithSource> revisions = this.findRevisions(flow.getNamespace(), flow.getId());
 
         if (revisions.size() > 0) {
-            flow = flow.withRevision(revisions.get(revisions.size() - 1).getRevision() + 1);
+            flow = flow.toBuilder().revision(revisions.get(revisions.size() - 1).getRevision() + 1).build();
         } else {
-            flow = flow.withRevision(1);
+            flow = flow.toBuilder().revision(1).build();
         }
 
         this.flows.put(flowId(flow), flow);
@@ -222,6 +226,10 @@ private FlowWithSource save(Flow flow, CrudEventType crudEventType, String flowS
 
     @Override
     public Flow delete(Flow flow) {
+        if (flow instanceof FlowWithSource) {
+            flow = ((FlowWithSource) flow).toFlow();
+        }
+
         if (this.findById(flow.getNamespace(), flow.getId(), Optional.of(flow.getRevision())).isEmpty()) {
             throw new IllegalStateException("Flow " + flow.getId() + " doesn't exists");
         }
@@ -232,8 +240,9 @@ public Flow delete(Flow flow) {
         this.flows.remove(flowId(deleted));
         this.revisions.put(deleted.uid(), deleted);
 
+        Flow finalFlow = flow;
         ListUtils.emptyOnNull(flow.getTriggers())
-            .forEach(abstractTrigger -> triggerQueue.delete(Trigger.of(flow, abstractTrigger)));
+            .forEach(abstractTrigger -> triggerQueue.delete(Trigger.of(finalFlow, abstractTrigger)));
 
         eventPublisher.publishEvent(new CrudEvent<>(flow, CrudEventType.DELETE));
 


Test patch:
diff --git a/core/src/test/java/io/kestra/core/services/FlowTopologyServiceTest.java b/core/src/test/java/io/kestra/core/services/FlowTopologyServiceTest.java
index 70e7042aec..553c383dab 100644
--- a/core/src/test/java/io/kestra/core/services/FlowTopologyServiceTest.java
+++ b/core/src/test/java/io/kestra/core/services/FlowTopologyServiceTest.java
@@ -173,14 +173,14 @@ public void multipleCondition() {
 
     @Test
     public void self1() {
-        Flow flow = parse("flows/valids/trigger-multiplecondition-listener.yaml").withRevision(1);
+        Flow flow = parse("flows/valids/trigger-multiplecondition-listener.yaml").toBuilder().revision(1).build();
 
         assertThat(flowTopologyService.isChild(flow, flow), nullValue());
     }
 
     @Test
     public void self() {
-        Flow flow = parse("flows/valids/trigger-flow-listener.yaml").withRevision(1);
+        Flow flow = parse("flows/valids/trigger-flow-listener.yaml").toBuilder().revision(1).build();
 
         assertThat(flowTopologyService.isChild(flow, flow), nullValue());
     }
diff --git a/webserver/src/test/java/io/kestra/webserver/controllers/FlowControllerTest.java b/webserver/src/test/java/io/kestra/webserver/controllers/FlowControllerTest.java
index 40b9b336f7..e2f8c18722 100644
--- a/webserver/src/test/java/io/kestra/webserver/controllers/FlowControllerTest.java
+++ b/webserver/src/test/java/io/kestra/webserver/controllers/FlowControllerTest.java
@@ -10,6 +10,7 @@
 import io.kestra.core.models.hierarchies.FlowGraph;
 import io.kestra.core.models.tasks.Task;
 import io.kestra.core.models.validations.ValidateConstraintViolation;
+import io.kestra.core.serializers.JacksonMapper;
 import io.kestra.core.serializers.YamlFlowParser;
 import io.kestra.core.tasks.debugs.Return;
 import io.kestra.core.tasks.flows.Sequential;
@@ -251,7 +252,18 @@ void createFlow() {
         Flow get = parseFlow(client.toBlocking().retrieve(HttpRequest.GET("/api/v1/flows/" + flow.getNamespace() + "/" + flow.getId()), String.class));
         assertThat(get.getId(), is(flow.getId()));
         assertThat(get.getInputs().get(0).getName(), is("a"));
+    }
+
+    @Test
+    void createFlowWithJsonLabels() {
+        Map<String, Object> flow = JacksonMapper.toMap(generateFlow("io.kestra.unittest", "a"));
+        flow.put("labels", Map.of("a", "b"));
+
+        Flow result = parseFlow(client.toBlocking().retrieve(POST("/api/v1/flows", flow), String.class));
 
+        assertThat(result.getId(), is(flow.get("id")));
+        assertThat(result.getLabels().get(0).key(), is("a"));
+        assertThat(result.getLabels().get(0).value(), is("b"));
     }
 
     @Test