Merge branch 'develop' into add-sleep-task

.github/workflows/check.yml

@@ -68,7 +68,7 @@ jobs:
       - uses: rlespinasse/github-slug-action@v4
 
       - name: Publish allure report
-        uses: andrcuns/allure-publish-action@v2.7.1
+        uses: andrcuns/allure-publish-action@v2.8.0
         if: ${{ always() && env.GOOGLE_SERVICE_ACCOUNT != 0 && (github.event.inputs.skip-test == 'false' || github.event.inputs.skip-test == '') }}
         env:
           GITHUB_AUTH_TOKEN: ${{ secrets.GITHUB_TOKEN }}

.github/workflows/e2e.yml

@@ -149,7 +149,7 @@ jobs:
       - uses: rlespinasse/github-slug-action@v4
 
       - name: Publish allure report
-        uses: andrcuns/allure-publish-action@v2.7.1
+        uses: andrcuns/allure-publish-action@v2.8.0
         if: ${{ !cancelled() && env.GOOGLE_SERVICE_ACCOUNT != 0 }}
         env:
           GITHUB_AUTH_TOKEN: ${{ secrets.GITHUB_AUTH_TOKEN }}

.github/workflows/generate_translations.yml

@@ -108,4 +108,4 @@ jobs:
             exit 0
           fi
           git commit -m "chore(translations): auto generate values for languages other than english"
-          git push origin $BRANCH_NAME
+          git push origin $BRANCH_NAME

.github/workflows/main.yml

@@ -237,7 +237,7 @@ jobs:
       - uses: rlespinasse/github-slug-action@v4
 
       - name: Publish allure report
-        uses: andrcuns/allure-publish-action@v2.7.1
+        uses: andrcuns/allure-publish-action@v2.8.0
         if: ${{ always() && env.GOOGLE_SERVICE_ACCOUNT != 0 && (github.event.inputs.skip-test == 'false' || github.event.inputs.skip-test == '') }}
         env:
           GITHUB_AUTH_TOKEN: ${{ secrets.GITHUB_TOKEN }}

.github/workflows/vulnerabilities-check.yml

@@ -72,7 +72,7 @@ jobs:
 
       # Run Trivy image scan for Docker vulnerabilities, see https://github.com/aquasecurity/trivy-action
       - name: Docker Vulnerabilities Check
-        uses: aquasecurity/trivy-action@0.27.0
+        uses: aquasecurity/trivy-action@0.28.0
         with:
           image-ref: kestra/kestra:develop
           format: table
@@ -105,7 +105,7 @@ jobs:
 
       # Run Trivy image scan for Docker vulnerabilities, see https://github.com/aquasecurity/trivy-action
       - name: Docker Vulnerabilities Check
-        uses: aquasecurity/trivy-action@0.27.0
+        uses: aquasecurity/trivy-action@0.28.0
         with:
           image-ref: kestra/kestra:latest
           format: table

.prettierignore

@@ -0,0 +1 @@
+**/*.*

core/src/main/java/io/kestra/core/models/Label.java

@@ -2,4 +2,11 @@
 
 import jakarta.validation.constraints.NotNull;
 
-public record Label(@NotNull String key, @NotNull String value) {}
+public record Label(@NotNull String key, @NotNull String value) {
+    public static final String SYSTEM_PREFIX = "system_";
+
+    // system labels
+    public static final String CORRELATION_ID = SYSTEM_PREFIX + "correlationId";
+    public static final String USERNAME = SYSTEM_PREFIX + "username";
+    public static final String APP = SYSTEM_PREFIX + "app";
+}

core/src/main/java/io/kestra/core/models/executions/Execution.java

@@ -39,6 +39,8 @@
 import java.util.stream.Stream;
 import java.util.zip.CRC32;
 
+import static io.kestra.core.models.Label.SYSTEM_PREFIX;
+
 @Builder(toBuilder = true)
 @Slf4j
 @Getter
@@ -76,7 +78,6 @@ public class Execution implements DeletedInterface, TenantInterface {
     @JsonInclude(JsonInclude.Include.NON_EMPTY)
     Map<String, Object> outputs;
 
-    @With
     @JsonSerialize(using = ListOrMapOfLabelSerializer.class)
     @JsonDeserialize(using = ListOrMapOfLabelDeserializer.class)
     List<Label> labels;
@@ -180,6 +181,12 @@ 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) {
@@ -205,6 +212,75 @@ 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,
+            this.namespace,
+            this.flowId,
+            this.flowRevision,
+            this.taskRunList,
+            this.inputs,
+            this.outputs,
+            newLabels,
+            this.variables,
+            this.state,
+            this.parentId,
+            this.originalId,
+            this.trigger,
+            this.deleted,
+            this.metadata,
+            this.scheduleDate,
+            this.error
+        );
+    }
+
     public Execution withTaskRun(TaskRun taskRun) throws InternalException {
         ArrayList<TaskRun> newTaskRunList = new ArrayList<>(this.taskRunList);
 

core/src/main/java/io/kestra/core/models/flows/Input.java

@@ -1,6 +1,5 @@
 package io.kestra.core.models.flows;
 
-import com.fasterxml.jackson.annotation.JsonInclude;
 import com.fasterxml.jackson.annotation.JsonSetter;
 import com.fasterxml.jackson.annotation.JsonSubTypes;
 import com.fasterxml.jackson.annotation.JsonTypeInfo;
@@ -44,7 +43,6 @@
     @JsonSubTypes.Type(value = YamlInput.class, name = "YAML"),
     @JsonSubTypes.Type(value = EmailInput.class, name = "EMAIL"),
 })
-@JsonInclude(JsonInclude.Include.NON_DEFAULT)
 public abstract class Input<T> implements Data {
     @Schema(
         title = "The ID of the input."

core/src/main/java/io/kestra/core/models/tasks/retrys/Constant.java

@@ -2,6 +2,7 @@
 
 import com.fasterxml.jackson.annotation.JsonInclude;
 import dev.failsafe.RetryPolicyBuilder;
+import io.kestra.core.validations.ConstantRetryValidation;
 import jakarta.validation.constraints.NotNull;
 import lombok.Builder;
 import lombok.Getter;
@@ -14,6 +15,7 @@
 @SuperBuilder
 @Getter
 @NoArgsConstructor
+@ConstantRetryValidation
 public class Constant extends AbstractRetry {
     @NotNull
     @JsonInclude

core/src/main/java/io/kestra/core/models/tasks/retrys/Exponential.java

@@ -2,6 +2,7 @@
 
 import com.fasterxml.jackson.annotation.JsonInclude;
 import dev.failsafe.RetryPolicyBuilder;
+import io.kestra.core.validations.ExponentialRetryValidation;
 import jakarta.validation.constraints.NotNull;
 import lombok.Builder;
 import lombok.Getter;
@@ -15,6 +16,7 @@
 @SuperBuilder
 @Getter
 @NoArgsConstructor
+@ExponentialRetryValidation
 public class Exponential extends AbstractRetry {
     @NotNull
     @JsonInclude

core/src/main/java/io/kestra/core/models/tasks/retrys/Random.java

@@ -2,6 +2,7 @@
 
 import com.fasterxml.jackson.annotation.JsonInclude;
 import dev.failsafe.RetryPolicyBuilder;
+import io.kestra.core.validations.RandomRetryValidation;
 import jakarta.validation.constraints.NotNull;
 import lombok.Builder;
 import lombok.Getter;
@@ -15,6 +16,7 @@
 @SuperBuilder
 @Getter
 @NoArgsConstructor
+@RandomRetryValidation
 public class Random extends AbstractRetry {
     @NotNull
     @JsonInclude

core/src/main/java/io/kestra/core/runners/ExecutableUtils.java

@@ -13,10 +13,12 @@
 import io.kestra.core.models.tasks.Task;
 import io.kestra.core.storages.Storage;
 import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.stream.Streams;
 
 import java.time.Instant;
 import java.time.ZonedDateTime;
 import java.util.*;
+import java.util.stream.Collectors;
 
 @Slf4j
 public final class ExecutableUtils {
@@ -91,6 +93,14 @@ public static <T extends Task & ExecutableTask<?>> SubflowExecution<?> subflowEx
             "flowRevision", currentFlow.getRevision()
         );
 
+        // propagate system labels and compute correlation ID if not already existing
+        List<Label> systemLabels = 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()));
+        }
+
         FlowInputOutput flowInputOutput = ((DefaultRunContext)runContext).getApplicationContext().getBean(FlowInputOutput.class);
         Instant scheduleOnDate = scheduleDate != null ? scheduleDate.as(runContext, ZonedDateTime.class).toInstant() : null;
         Execution execution = Execution
@@ -105,7 +115,8 @@ public static <T extends Task & ExecutableTask<?>> SubflowExecution<?> subflowEx
                 .variables(variables)
                 .build()
             )
-            .withScheduleDate(scheduleOnDate);
+            .withScheduleDate(scheduleOnDate)
+            .withSystemLabels(systemLabels);
         return SubflowExecution.builder()
             .parentTask(currentTask)
             .parentTaskRun(currentTaskRun.withState(State.Type.RUNNING))

core/src/main/java/io/kestra/core/services/FlowService.java

@@ -320,6 +320,24 @@ public static String generateSource(Flow flow, @Nullable String source) {
         return source;
     }
 
+    // Used in Git plugin
+    public List<Flow> findByNamespacePrefix(String tenantId, String namespacePrefix) {
+        if (flowRepository.isEmpty()) {
+            throw noRepositoryException();
+        }
+
+        return flowRepository.get().findByNamespacePrefix(tenantId, namespacePrefix);
+    }
+
+    // Used in Git plugin
+    public FlowWithSource delete(FlowWithSource flow) {
+        if (flowRepository.isEmpty()) {
+            throw noRepositoryException();
+        }
+
+        return flowRepository.get().delete(flow);
+    }
+
     @SneakyThrows
     private static String toYamlWithoutDefault(Object object) throws JsonProcessingException {
         String json = NON_DEFAULT_OBJECT_MAPPER.writeValueAsString(object);

core/src/main/java/io/kestra/core/validations/ConstantRetryValidation.java

@@ -0,0 +1,16 @@
+package io.kestra.core.validations;
+
+import io.kestra.core.validations.validator.ConstantRetryValidator;
+import jakarta.validation.Constraint;
+import jakarta.validation.Payload;
+
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+
+@Retention(RetentionPolicy.RUNTIME)
+@Constraint(validatedBy = ConstantRetryValidator.class)
+public @interface ConstantRetryValidation {
+    String message() default "invalid constant retry";
+    Class<?>[] groups() default {};
+    Class<? extends Payload>[] payload() default {};
+}

core/src/main/java/io/kestra/core/validations/ExponentialRetryValidation.java

@@ -0,0 +1,16 @@
+package io.kestra.core.validations;
+
+import io.kestra.core.validations.validator.ExponentialRetryValidator;
+import jakarta.validation.Constraint;
+import jakarta.validation.Payload;
+
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+
+@Retention(RetentionPolicy.RUNTIME)
+@Constraint(validatedBy = ExponentialRetryValidator.class)
+public @interface ExponentialRetryValidation {
+    String message() default "invalid exponential retry";
+    Class<?>[] groups() default {};
+    Class<? extends Payload>[] payload() default {};
+}

core/src/main/java/io/kestra/core/validations/RandomRetryValidation.java

@@ -0,0 +1,16 @@
+package io.kestra.core.validations;
+
+import io.kestra.core.validations.validator.RandomRetryValidator;
+import jakarta.validation.Constraint;
+import jakarta.validation.Payload;
+
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+
+@Retention(RetentionPolicy.RUNTIME)
+@Constraint(validatedBy = RandomRetryValidator.class)
+public @interface RandomRetryValidation {
+    String message() default "invalid random retry";
+    Class<?>[] groups() default {};
+    Class<? extends Payload>[] payload() default {};
+}

core/src/main/java/io/kestra/core/validations/validator/ConstantRetryValidator.java

@@ -0,0 +1,28 @@
+package io.kestra.core.validations.validator;
+
+import io.kestra.core.models.tasks.retrys.Constant;
+import io.kestra.core.validations.ConstantRetryValidation;
+import io.micronaut.core.annotation.AnnotationValue;
+import io.micronaut.core.annotation.NonNull;
+import io.micronaut.core.annotation.Nullable;
+import io.micronaut.validation.validator.constraints.ConstraintValidator;
+import io.micronaut.validation.validator.constraints.ConstraintValidatorContext;
+import jakarta.inject.Singleton;
+
+@Singleton
+public class ConstantRetryValidator implements ConstraintValidator<ConstantRetryValidation, Constant> {
+    @Override
+    public boolean isValid(@Nullable Constant value, @NonNull AnnotationValue<ConstantRetryValidation> annotationMetadata, @NonNull ConstraintValidatorContext context) {
+        if (value == null) {
+            return true;
+        }
+
+        if (value.getMaxDuration() != null && value.getInterval() != null && value.getMaxDuration().compareTo(value.getInterval()) <= 0) {
+            context.disableDefaultConstraintViolation();
+            context.buildConstraintViolationWithTemplate( "'interval' must be less than 'maxDuration' but is " + value.getInterval())
+                .addConstraintViolation();
+            return false;
+        }
+        return true;
+    }
+}