chore(version): update to version 'v0.5.1'.

kestra-io · Sep 26, 2022 · 638695e · 638695e
2 parents af5584b + 17b8887
commit 638695e
Show file tree

Hide file tree

Showing 75 changed files with 1,734 additions and 253 deletions.
diff --git a/.github/workflows/main.yml b/.github/workflows/main.yml
@@ -122,6 +122,7 @@ jobs:
         if: startsWith(github.ref, 'refs/tags/v')
         with:
           repo_token: "${{ secrets.GITHUB_TOKEN }}"
+          prerelease: false
           files: |
             build/executable/*
 
@@ -137,7 +138,7 @@ jobs:
             plugins: ""
             packages: ""
           - name: "-full"
-            plugins: io.kestra.storage:storage-azure:LATEST io.kestra.storage:storage-gcs:LATEST io.kestra.storage:storage-minio:LATEST io.kestra.plugin:plugin-aws:LATEST io.kestra.plugin:plugin-azure:LATEST io.kestra.plugin:plugin-powerbi:LATEST io.kestra.plugin:plugin-pulsar:LATEST io.kestra.plugin:plugin-cassandra:LATEST io.kestra.plugin:plugin-compress:LATEST io.kestra.plugin:plugin-crypto:LATEST io.kestra.plugin:plugin-dbt:LATEST io.kestra.plugin:plugin-debezium-mysql:LATEST io.kestra.plugin:plugin-debezium-postgres:LATEST io.kestra.plugin:plugin-debezium-sqlserver:LATEST io.kestra.plugin:plugin-elasticsearch:LATEST io.kestra.plugin:plugin-fs:LATEST io.kestra.plugin:plugin-gcp:LATEST io.kestra.plugin:plugin-googleworkspace:LATEST io.kestra.plugin:plugin-jdbc-clickhouse:LATEST io.kestra.plugin:plugin-jdbc-mysql:LATEST io.kestra.plugin:plugin-jdbc-oracle:LATEST io.kestra.plugin:plugin-jdbc-pinot:LATEST io.kestra.plugin:plugin-jdbc-postgres:LATEST io.kestra.plugin:plugin-jdbc-redshift:LATEST io.kestra.plugin:plugin-jdbc-rockset:LATEST io.kestra.plugin:plugin-jdbc-snowflake:LATEST io.kestra.plugin:plugin-jdbc-sqlserver:LATEST io.kestra.plugin:plugin-jdbc-trino:LATEST io.kestra.plugin:plugin-jdbc-vertica:LATEST io.kestra.plugin:plugin-jdbc-vectorwise:LATEST io.kestra.plugin:plugin-kafka:LATEST io.kestra.plugin:plugin-kubernetes:LATEST io.kestra.plugin:plugin-mongodb:LATEST io.kestra.plugin:plugin-mqtt:LATEST io.kestra.plugin:plugin-notifications:LATEST io.kestra.plugin:plugin-script-groovy:LATEST io.kestra.plugin:plugin-script-jython:LATEST io.kestra.plugin:plugin-script-nashorn:LATEST io.kestra.plugin:plugin-serdes:LATEST io.kestra.plugin:plugin-servicenow:LATEST io.kestra.plugin:plugin-singer:LATEST io.kestra.plugin:plugin-soda:LATEST io.kestra.plugin:plugin-spark:LATEST io.kestra.plugin:plugin-tika:LATEST
+            plugins: io.kestra.storage:storage-azure:LATEST io.kestra.storage:storage-gcs:LATEST io.kestra.storage:storage-minio:LATEST io.kestra.plugin:plugin-aws:LATEST io.kestra.plugin:plugin-azure:LATEST io.kestra.plugin:plugin-powerbi:LATEST io.kestra.plugin:plugin-pulsar:LATEST io.kestra.plugin:plugin-cassandra:LATEST io.kestra.plugin:plugin-compress:LATEST io.kestra.plugin:plugin-crypto:LATEST io.kestra.plugin:plugin-dbt:LATEST io.kestra.plugin:plugin-debezium-mysql:LATEST io.kestra.plugin:plugin-debezium-postgres:LATEST io.kestra.plugin:plugin-debezium-sqlserver:LATEST io.kestra.plugin:plugin-elasticsearch:LATEST io.kestra.plugin:plugin-fs:LATEST io.kestra.plugin:plugin-gcp:LATEST io.kestra.plugin:plugin-googleworkspace:LATEST io.kestra.plugin:plugin-jdbc-clickhouse:LATEST io.kestra.plugin:plugin-jdbc-duckdb:LATEST io.kestra.plugin:plugin-jdbc-mysql:LATEST io.kestra.plugin:plugin-jdbc-oracle:LATEST io.kestra.plugin:plugin-jdbc-pinot:LATEST io.kestra.plugin:plugin-jdbc-postgres:LATEST io.kestra.plugin:plugin-jdbc-redshift:LATEST io.kestra.plugin:plugin-jdbc-rockset:LATEST io.kestra.plugin:plugin-jdbc-snowflake:LATEST io.kestra.plugin:plugin-jdbc-sqlserver:LATEST io.kestra.plugin:plugin-jdbc-trino:LATEST io.kestra.plugin:plugin-jdbc-vertica:LATEST io.kestra.plugin:plugin-jdbc-vectorwise:LATEST io.kestra.plugin:plugin-kafka:LATEST io.kestra.plugin:plugin-kubernetes:LATEST io.kestra.plugin:plugin-mongodb:LATEST io.kestra.plugin:plugin-mqtt:LATEST io.kestra.plugin:plugin-notifications:LATEST io.kestra.plugin:plugin-script-groovy:LATEST io.kestra.plugin:plugin-script-jython:LATEST io.kestra.plugin:plugin-script-nashorn:LATEST io.kestra.plugin:plugin-serdes:LATEST io.kestra.plugin:plugin-servicenow:LATEST io.kestra.plugin:plugin-singer:LATEST io.kestra.plugin:plugin-soda:LATEST io.kestra.plugin:plugin-spark:LATEST io.kestra.plugin:plugin-tika:LATEST
             packages: python3 python3-venv python-is-python3 nodejs curl zip unzip
     steps:
       - uses: actions/checkout@v2

diff --git a/build.gradle b/build.gradle
@@ -156,7 +156,6 @@ subprojects {
 
         retry {
             maxRetries = 5
-            maxFailures = 15
         }
     }
 

diff --git a/cli/src/main/java/io/kestra/cli/AbstractApiCommand.java b/cli/src/main/java/io/kestra/cli/AbstractApiCommand.java
@@ -12,7 +12,7 @@
 import java.util.Map;
 
 public abstract class AbstractApiCommand extends AbstractCommand {
-    @CommandLine.Option(names = {"--server"}, description = "Also write core tasks plugins")
+    @CommandLine.Option(names = {"--server"}, description = "Also write core tasks plugins", defaultValue = "http://localhost:8080")
     protected URL server;
 
     @CommandLine.Option(names = {"--headers"}, description = "Also write core tasks plugins")

diff --git a/core/src/main/java/io/kestra/core/models/executions/ExecutionTrigger.java b/core/src/main/java/io/kestra/core/models/executions/ExecutionTrigger.java
@@ -28,4 +28,12 @@ public static ExecutionTrigger of(AbstractTrigger abstractTrigger, Output output
             .variables(output.toMap())
             .build();
     }
+
+    public static ExecutionTrigger of(AbstractTrigger abstractTrigger, Map<String, Object> variables) {
+        return ExecutionTrigger.builder()
+            .id(abstractTrigger.getId())
+            .type(abstractTrigger.getType())
+            .variables(variables)
+            .build();
+    }
 }
diff --git a/core/src/main/java/io/kestra/core/models/tasks/Output.java b/core/src/main/java/io/kestra/core/models/tasks/Output.java
@@ -3,6 +3,7 @@
 import io.kestra.core.models.flows.State;
 import io.kestra.core.serializers.JacksonMapper;
 
+import java.time.ZoneId;
 import java.util.Map;
 import java.util.Optional;
 
@@ -14,4 +15,8 @@ default Optional<State.Type> finalState() {
     default Map<String, Object> toMap() {
         return JacksonMapper.toMap(this);
     }
+
+    default Map<String, Object> toMap(ZoneId zoneId) {
+        return JacksonMapper.toMap(this, zoneId);
+    }
 }
diff --git a/core/src/main/java/io/kestra/core/models/templates/Template.java b/core/src/main/java/io/kestra/core/models/templates/Template.java
@@ -34,6 +34,8 @@ public class Template implements DeletedInterface {
     @Pattern(regexp="[a-z0-9._-]+")
     private String namespace;
 
+    String description;
+
     @Valid
     @NotEmpty
     private List<Task> tasks;
@@ -95,6 +97,7 @@ public Template toDeleted() {
         return new Template(
             this.id,
             this.namespace,
+            this.description,
             this.tasks,
             this.errors,
             true

diff --git a/core/src/main/java/io/kestra/core/models/triggers/types/Schedule.java b/core/src/main/java/io/kestra/core/models/triggers/types/Schedule.java
@@ -1,7 +1,6 @@
 package io.kestra.core.models.triggers.types;
 
 import com.cronutils.model.Cron;
-import com.cronutils.model.CronType;
 import com.cronutils.model.definition.CronDefinitionBuilder;
 import com.cronutils.model.time.ExecutionTime;
 import com.cronutils.parser.CronParser;
@@ -27,6 +26,7 @@
 import lombok.experimental.SuperBuilder;
 
 import java.time.Duration;
+import java.time.ZoneId;
 import java.time.ZonedDateTime;
 import java.time.temporal.ChronoUnit;
 import java.util.HashMap;
@@ -126,6 +126,12 @@ public class Schedule extends AbstractTrigger implements PollingTriggerInterface
     )
     private String cron;
 
+    @Schema(
+        title = "The time zone id to use for evaluate cron. Default value is the server default zone id."
+    )
+    @PluginProperty(dynamic = true)
+    private String timezone = ZoneId.systemDefault().toString();
+
     @Schema(
         title = "Backfill options in order to fill missing previous past date",
         description = "Kestra will handle optionally a backfill. The concept of backfill is the replay the missing schedule because we create the flow later.\n" +
@@ -162,25 +168,28 @@ public class Schedule extends AbstractTrigger implements PollingTriggerInterface
     public ZonedDateTime nextEvaluationDate(ConditionContext conditionContext, Optional<? extends TriggerContext> last) {
         ExecutionTime executionTime = this.executionTime();
 
-        // previous present & scheduleConditions
-        if (last.isPresent() && this.scheduleConditions != null) {
-            Optional<ZonedDateTime> next = this.truePreviousNextDateWithCondition(
-                executionTime,
-                conditionContext,
-                last.get().getDate(),
-                true
-            );
-
-            if (next.isPresent()) {
-                return next.get().truncatedTo(ChronoUnit.SECONDS);
+        if (last.isPresent()) {
+            ZonedDateTime lastDate = convertDateTime(last.get().getDate());
+
+            // previous present & scheduleConditions
+            if (this.scheduleConditions != null) {
+                Optional<ZonedDateTime> next = this.truePreviousNextDateWithCondition(
+                    executionTime,
+                    conditionContext,
+                    lastDate,
+                    true
+                );
+
+                if (next.isPresent()) {
+                    return next.get().truncatedTo(ChronoUnit.SECONDS);
+                }
             }
-        }
 
-        // previous present but no scheduleConditions
-        if (last.isPresent()) {
-            return computeNextEvaluationDate(executionTime, last.get().getDate()).orElse(null);
+            // previous present but no scheduleConditions
+            return computeNextEvaluationDate(executionTime, lastDate).orElse(null);
         }
 
+
         // no previous present but backfill
         if (backfill != null && backfill.getStart() != null) {
             return backfill.getStart();
@@ -193,7 +202,7 @@ public ZonedDateTime nextEvaluationDate(ConditionContext conditionContext, Optio
     public Optional<Execution> evaluate(ConditionContext conditionContext, TriggerContext context) throws Exception {
         RunContext runContext = conditionContext.getRunContext();
         ExecutionTime executionTime = this.executionTime();
-        ZonedDateTime previousDate = context.getDate();
+        ZonedDateTime previousDate = convertDateTime(context.getDate());
 
         Output output = this.output(executionTime, previousDate).orElse(null);
 
@@ -242,7 +251,14 @@ public Optional<Execution> evaluate(ConditionContext conditionContext, TriggerCo
             }
         }
 
-        ExecutionTrigger executionTrigger = ExecutionTrigger.of(this, output);
+        Map<String, Object> variables;
+        if (this.timezone != null) {
+            variables = output.toMap(ZoneId.of(this.timezone));
+        } else {
+            variables = output.toMap();
+        }
+
+        ExecutionTrigger executionTrigger = ExecutionTrigger.of(this, variables);
 
         Execution execution = Execution.builder()
             .id(IdUtils.create())
@@ -269,12 +285,14 @@ private Optional<Output> output(ExecutionTime executionTime, ZonedDateTime date)
         }
 
         Output.OutputBuilder<?, ?> outputBuilder = Output.builder()
-            .date(next.get());
+            .date(convertDateTime(next.get()));
 
         computeNextEvaluationDate(executionTime, next.get())
+            .map(this::convertDateTime)
             .ifPresent(outputBuilder::next);
 
         executionTime.lastExecution(date)
+            .map(this::convertDateTime)
             .ifPresent(outputBuilder::previous);
 
         Output output = outputBuilder.build();
@@ -299,6 +317,14 @@ private synchronized ExecutionTime executionTime() {
         return this.executionTime;
     }
 
+    private ZonedDateTime convertDateTime(ZonedDateTime date) {
+        if (this.timezone == null) {
+            return date;
+        }
+
+        return date.withZoneSameInstant(ZoneId.of(this.timezone));
+    }
+
     private Optional<ZonedDateTime> computeNextEvaluationDate(ExecutionTime executionTime, ZonedDateTime date) {
         return executionTime.nextExecution(date).map(zonedDateTime -> zonedDateTime.truncatedTo(ChronoUnit.SECONDS));
     }

diff --git a/core/src/main/java/io/kestra/core/repositories/ExecutionRepositoryInterface.java b/core/src/main/java/io/kestra/core/repositories/ExecutionRepositoryInterface.java
@@ -7,13 +7,18 @@
 import io.kestra.core.models.executions.statistics.Flow;
 import io.kestra.core.models.flows.State;
 import io.micronaut.data.model.Pageable;
+import io.reactivex.Flowable;
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+import lombok.experimental.SuperBuilder;
 
 import java.time.ZonedDateTime;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 import java.util.function.Function;
 import javax.annotation.Nullable;
+import javax.validation.constraints.NotNull;
 
 public interface ExecutionRepositoryInterface extends SaveRepositoryInterface<Execution> {
     Boolean isTaskRunEnabled();
@@ -32,6 +37,15 @@ ArrayListTotal<Execution> find(
         @Nullable List<State.Type> state
     );
 
+    Flowable<Execution> find(
+        @Nullable String query,
+        @Nullable String namespace,
+        @Nullable String flowId,
+        @Nullable ZonedDateTime startDate,
+        @Nullable ZonedDateTime endDate,
+        @Nullable List<State.Type> state
+    );
+
     ArrayListTotal<TaskRun> findTaskRun(
         Pageable pageable,
         @Nullable String query,
@@ -44,6 +58,8 @@ ArrayListTotal<TaskRun> findTaskRun(
 
     Execution delete(Execution execution);
 
+    Integer purge(Execution execution);
+
     Integer maxTaskRunSetting();
 
     List<DailyExecutionStatistics> dailyStatistics(
@@ -59,11 +75,22 @@ Map<String, Map<String, List<DailyExecutionStatistics>>> dailyGroupByFlowStatist
         @Nullable String query,
         @Nullable String namespace,
         @Nullable String flowId,
+        @Nullable List<FlowFilter> flows,
         @Nullable ZonedDateTime startDate,
         @Nullable ZonedDateTime endDate,
         boolean groupByNamespaceOnly
     );
 
+    @Getter
+    @SuperBuilder
+    @NoArgsConstructor
+    class FlowFilter {
+        @NotNull
+        private String namespace;
+        @NotNull
+        private String id;
+    }
+
     List<ExecutionCount> executionCounts(
         List<Flow> flows,
         @Nullable List<State.Type> states,

diff --git a/core/src/main/java/io/kestra/core/repositories/LogRepositoryInterface.java b/core/src/main/java/io/kestra/core/repositories/LogRepositoryInterface.java
@@ -1,7 +1,8 @@
 package io.kestra.core.repositories;
 
-import io.micronaut.data.model.Pageable;
+import io.kestra.core.models.executions.Execution;
 import io.kestra.core.models.executions.LogEntry;
+import io.micronaut.data.model.Pageable;
 import org.slf4j.event.Level;
 
 import java.time.ZonedDateTime;
@@ -26,4 +27,6 @@ ArrayListTotal<LogEntry> find(
     );
 
     LogEntry save(LogEntry log);
+
+    Integer purge(Execution execution);
 }
diff --git a/core/src/main/java/io/kestra/core/runners/DefaultFlowExecutor.java b/core/src/main/java/io/kestra/core/runners/DefaultFlowExecutor.java
@@ -42,4 +42,9 @@ public Optional<Flow> findById(String namespace, String id, Optional<Integer> re
             return flowRepositoryInterface.findById(namespace, id, revision);
         }
     }
+
+    @Override
+    public Boolean isReady() {
+        return true;
+    }
 }
diff --git a/core/src/main/java/io/kestra/core/runners/Executor.java b/core/src/main/java/io/kestra/core/runners/Executor.java
@@ -36,6 +36,10 @@ public Executor(WorkerTaskResult workerTaskResult) {
         this.joined = workerTaskResult;
     }
 
+    public Boolean canBeProcessed() {
+        return !(this.getException() != null || this.getFlow() == null || this.getExecution().isDeleted());
+    }
+
     public Executor withFlow(Flow flow) {
         this.flow = flow;
 

diff --git a/core/src/main/java/io/kestra/core/runners/ExecutorService.java b/core/src/main/java/io/kestra/core/runners/ExecutorService.java
@@ -56,7 +56,7 @@ protected FlowExecutorInterface flowExecutorInterface() {
 
     public Executor process(Executor executor) {
         // previous failed (flow join can fail), just forward
-        if (executor.getException() != null || executor.getExecution().isDeleted()) {
+        if (!executor.canBeProcessed()) {
             return executor;
         }
 
@@ -672,12 +672,13 @@ public void log(Logger log, Boolean in, Execution value) {
 
     public void log(Logger log, Boolean in, Executor value) {
         log.debug(
-            "{} {} [key='{}', from='{}', offset='{}']\n{}",
+            "{} {} [key='{}', from='{}', offset='{}', crc32='{}']\n{}",
             in ? "<< IN " : ">> OUT",
             value.getClass().getSimpleName(),
             value.getExecution().getId(),
             value.getFrom(),
             value.getOffset(),
+            value.getExecution().toCrc32State(),
             value.getExecution().toStringState()
         );
     }

diff --git a/core/src/main/java/io/kestra/core/runners/FlowExecutorInterface.java b/core/src/main/java/io/kestra/core/runners/FlowExecutorInterface.java
@@ -11,6 +11,8 @@ public interface FlowExecutorInterface {
 
     Optional<Flow> findById(String namespace, String id, Optional<Integer> revision);
 
+    Boolean isReady();
+
     default Optional<Flow> findByIdFromFlowTask(String namespace, String id, Optional<Integer> revision, String fromNamespace, String fromId) {
         return this.findById(
             namespace,