chore(version): update to version '0.23.0-rc2-SNAPSHOT'

Closes https://github.com/kestra-io/kestra/issues/9352.

.github/workflows/auto-translate-ui-keys.yml

@@ -62,6 +62,6 @@ jobs:
               echo "No changes to commit. Exiting with success."
               exit 0
             fi
-            git commit -m "chore(translations): localize to languages other than English"
+            git commit -m "chore(core): localize to languages other than english" -m "Extended localization support by adding translations for multiple languages using English as the base. This enhances accessibility and usability for non-English-speaking users while keeping English as the source reference."
             git push -u origin $BRANCH_NAME || (git push origin --delete $BRANCH_NAME && git push -u origin $BRANCH_NAME)
             gh pr create --title "Translations from en.json" --body "This PR was created automatically by a GitHub Action." --base develop --head $BRANCH_NAME --assignee anna-geller --reviewer anna-geller

.github/workflows/codeql-analysis.yml

@@ -62,7 +62,7 @@ jobs:
 
     - name: Build with Gradle
       if: ${{ matrix.language == 'java' }}
-      run: ./gradlew testClasses -x :ui:installFrontend -x :ui:assembleFrontend
+      run: ./gradlew testClasses -x :ui:assembleFrontend
 
     # Autobuild attempts to build any compiled languages  (C/C++, C#, or Java).
     # If this step fails, then you should remove it and run the build manually (see below)

.github/workflows/docker.yml

@@ -51,7 +51,7 @@ jobs:
             python-libs: ""
           - name: ""
             plugins: ${{needs.plugins.outputs.plugins}}
-            packages: python3 python3-venv python-is-python3 python3-pip nodejs npm curl zip unzip jattach
+            packages: python3 python-is-python3 python3-pip curl jattach
             python-libs: kestra
     steps:
       - uses: actions/checkout@v4

.github/workflows/gradle-release.yml

@@ -27,7 +27,7 @@ jobs:
             echo "Invalid release version. Must match regex: ^[0-9]+(\.[0-9]+)\.0-rc[01](-SNAPSHOT)?$"
             exit 1
           fi
-            
+
           if ! [[ "$NEXT_VERSION" =~ ^[0-9]+(\.[0-9]+)\.0-SNAPSHOT$ ]]; then
             echo "Invalid next version. Must match regex: ^[0-9]+(\.[0-9]+)\.0-SNAPSHOT$"
             exit 1;
@@ -36,6 +36,7 @@ jobs:
       - uses: actions/checkout@v4
         with:
           fetch-depth: 0
+          path: kestra
 
       # Checkout GitHub Actions
       - uses: actions/checkout@v4
@@ -62,18 +63,20 @@ jobs:
       - name: Run Gradle Release
         env:
           GITHUB_PAT: ${{ secrets.GH_PERSONAL_TOKEN }}
-        run: |          
+        run: |
           # Extract the major and minor versions
           BASE_VERSION=$(echo "$RELEASE_VERSION" | sed -E 's/^([0-9]+\.[0-9]+)\..*/\1/')
           PUSH_RELEASE_BRANCH="releases/v${BASE_VERSION}.x"
-  
+
+          cd kestra
+
           # Create and push release branch
           git checkout -b "$PUSH_RELEASE_BRANCH";
           git push -u origin "$PUSH_RELEASE_BRANCH";
-      
+
           # Run gradle release
           git checkout develop;
-      
+
           if [[ "$RELEASE_VERSION" == *"-SNAPSHOT" ]]; then
             # -SNAPSHOT qualifier maybe used to test release-candidates
             ./gradlew release -Prelease.useAutomaticVersion=true \

.github/workflows/main.yml

@@ -43,6 +43,9 @@ jobs:
       SONATYPE_GPG_KEYID: ${{ secrets.SONATYPE_GPG_KEYID }}
       SONATYPE_GPG_PASSWORD: ${{ secrets.SONATYPE_GPG_PASSWORD }}
       SONATYPE_GPG_FILE: ${{ secrets.SONATYPE_GPG_FILE }}
+      GH_PERSONAL_TOKEN: ${{ secrets.GH_PERSONAL_TOKEN }}
+      SLACK_RELEASES_WEBHOOK_URL: ${{ secrets.SLACK_RELEASES_WEBHOOK_URL }}
+
 
   end:
     runs-on: ubuntu-latest

.github/workflows/vulnerabilities-check.yml

@@ -8,6 +8,9 @@ on:
 env:
   JAVA_VERSION: '21'
 
+permissions:
+  contents: read
+
 jobs:
   dependency-check:
     name: Dependency Check
@@ -57,6 +60,10 @@ jobs:
   develop-image-check:
     name: Image Check (develop)
     runs-on: ubuntu-latest
+    permissions:
+      contents: read
+      security-events: write
+      actions: read
     steps:
       # Checkout
       - uses: actions/checkout@v4
@@ -83,13 +90,25 @@ jobs:
         uses: aquasecurity/trivy-action@0.30.0
         with:
           image-ref: kestra/kestra:develop
-          format: table
+          format: 'template'
+          template: '@/contrib/sarif.tpl'
+          severity: 'CRITICAL,HIGH'
+          output: 'trivy-results.sarif'
           skip-dirs: /app/plugins
-          scanners: vuln
+
+      - name: Upload Trivy scan results to GitHub Security tab
+        uses: github/codeql-action/upload-sarif@v3
+        with:
+          sarif_file: 'trivy-results.sarif'
+          category: docker-
 
   latest-image-check:
     name: Image Check (latest)
     runs-on: ubuntu-latest
+    permissions:
+      contents: read
+      security-events: write
+      actions: read
     steps:
       # Checkout
       - uses: actions/checkout@v4
@@ -118,4 +137,11 @@ jobs:
           image-ref: kestra/kestra:latest
           format: table
           skip-dirs: /app/plugins
-          scanners: vuln
+          scanners: vuln
+          severity: 'CRITICAL,HIGH'
+          output: 'trivy-results.sarif'
+
+      - name: Upload Trivy scan results to GitHub Security tab
+        uses: github/codeql-action/upload-sarif@v3
+        with:
+          sarif_file: 'trivy-results.sarif'

.github/workflows/workflow-backend-test.yml

@@ -31,6 +31,8 @@ jobs:
     steps:
       - uses: actions/checkout@v4
         name: Checkout - Current ref
+        with:
+          fetch-depth: 0
 
       # Setup build
       - uses: kestra-io/actions/.github/actions/setup-build@main

.github/workflows/workflow-frontend-test.yml

@@ -22,8 +22,25 @@ jobs:
       - name: Checkout
         uses: actions/checkout@v4
 
+      - name: Cache Node Modules
+        id: cache-node-modules
+        uses: actions/cache@v4
+        with:
+          path: |
+            ui/node_modules
+          key: modules-${{ hashFiles('ui/package-lock.json') }}
+
+      - name: Cache Playwright Binaries
+        id: cache-playwright
+        uses: actions/cache@v4
+        with:
+          path: |
+            ~/.cache/ms-playwright
+          key: playwright-${{ hashFiles('ui/package-lock.json') }}
+
       - name: Npm - install
         shell: bash
+        if: steps.cache-node-modules.outputs.cache-hit != 'true'
         working-directory: ui
         run: npm ci
 
@@ -44,6 +61,7 @@ jobs:
       - name: Storybook - Install Playwright
         shell: bash
         working-directory: ui
+        if: steps.cache-playwright.outputs.cache-hit != 'true'
         run: npx playwright install --with-deps
 
       - name: Run front-end unit tests

.github/workflows/workflow-github-release.yml

@@ -6,19 +6,21 @@ on:
       GH_PERSONAL_TOKEN:
         description: "The Github personal token."
         required: true
+      SLACK_RELEASES_WEBHOOK_URL:
+        description: "The Slack webhook URL."
+        required: true
 
 jobs:
   publish:
     name: Github - Release
     runs-on: ubuntu-latest
     steps:
-      # Download Exec
-      - name: Artifacts - Download executable
-        uses: actions/download-artifact@v4
-        if: startsWith(github.ref, 'refs/tags/v')
+      # Check out
+      - name: Checkout - Repository
+        uses: actions/checkout@v4
         with:
-          name: exe
-          path: build/executable
+          fetch-depth: 0
+          submodules: true
 
       # Checkout GitHub Actions
       - name: Checkout - Actions
@@ -30,18 +32,27 @@ jobs:
           sparse-checkout: |
             .github/actions
 
+      # Download Exec
+      # Must be done after checkout actions
+      - name: Artifacts - Download executable
+        uses: actions/download-artifact@v4
+        if: startsWith(github.ref, 'refs/tags/v')
+        with:
+          name: exe
+          path: build/executable
+
+
       # GitHub Release
       - name: Create GitHub release
         uses: ./actions/.github/actions/github-release
         if: ${{ startsWith(github.ref, 'refs/tags/v') }}
         env:
-          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+          GITHUB_TOKEN: ${{ secrets.GH_PERSONAL_TOKEN }}
           SLACK_RELEASES_WEBHOOK_URL: ${{ secrets.SLACK_RELEASES_WEBHOOK_URL }}
 
       # Trigger gha workflow to bump helm chart version
       - name: GitHub - Trigger the Helm chart version bump
         uses: peter-evans/repository-dispatch@v3
-        if: steps.create_github_release.conclusion == 'success'
         with:
           token: ${{ secrets.GH_PERSONAL_TOKEN }}
           repository: kestra-io/helm-charts

.github/workflows/workflow-release.yml

@@ -42,6 +42,12 @@ on:
       SONATYPE_GPG_FILE:
         description: "The Sonatype GPG file."
         required: true
+      GH_PERSONAL_TOKEN:
+        description: "The Github personal token."
+        required: true
+      SLACK_RELEASES_WEBHOOK_URL:
+        description: "The Slack webhook URL."
+        required: true
 jobs:
   build-artifacts:
     name: Build - Artifacts
@@ -77,4 +83,5 @@ jobs:
     if: startsWith(github.ref, 'refs/tags/v')
     uses: ./.github/workflows/workflow-github-release.yml
     secrets:
-      GH_PERSONAL_TOKEN: ${{ secrets.GH_PERSONAL_TOKEN }}
+      GH_PERSONAL_TOKEN: ${{ secrets.GH_PERSONAL_TOKEN }}
+      SLACK_RELEASES_WEBHOOK_URL: ${{ secrets.SLACK_RELEASES_WEBHOOK_URL }}

.gitignore

@@ -59,3 +59,4 @@ core/src/main/resources/gradle.properties
 
 *storybook.log
 storybook-static
+/jmh-benchmarks/src/main/resources/gradle.properties

.plugins

@@ -33,9 +33,11 @@
 #plugin-github:io.kestra.plugin:plugin-github:LATEST
 #plugin-googleworkspace:io.kestra.plugin:plugin-googleworkspace:LATEST
 #plugin-graalvm:io.kestra.plugin:plugin-graalvm:LATEST
+#plugin-graphql:io.kestra.plugin:plugin-graphql:LATEST
 #plugin-hightouch:io.kestra.plugin:plugin-hightouch:LATEST
 #plugin-hubspot:io.kestra.plugin:plugin-hubspot:LATEST
 #plugin-huggingface:io.kestra.plugin:plugin-huggingface:LATEST
+#plugin-influxdb:io.kestra.plugin:plugin-influxdb:LATEST
 #plugin-jdbc:io.kestra.plugin:plugin-jdbc-as400:LATEST
 #plugin-jdbc:io.kestra.plugin:plugin-jdbc-clickhouse:LATEST
 #plugin-jdbc:io.kestra.plugin:plugin-jdbc-db2:LATEST
@@ -56,9 +58,11 @@
 #plugin-jdbc:io.kestra.plugin:plugin-jdbc-arrow-flight:LATEST
 #plugin-jdbc:io.kestra.plugin:plugin-jdbc-sqlite:LATEST
 #plugin-jdbc:io.kestra.plugin:plugin-jdbc-sybase:LATEST
+#plugin-jenkins:io.kestra.plugin:plugin-jenkins:LATEST
 #plugin-jira:io.kestra.plugin:plugin-jira:LATEST
 #plugin-kafka:io.kestra.plugin:plugin-kafka:LATEST
 #plugin-kubernetes:io.kestra.plugin:plugin-kubernetes:LATEST
+#plugin-langchain4j:io.kestra.plugin:plugin-langchain4j:LATEST
 #plugin-ldap:io.kestra.plugin:plugin-ldap:LATEST
 #plugin-linear:io.kestra.plugin:plugin-linear:LATEST
 #plugin-malloy:io.kestra.plugin:plugin-malloy:LATEST
@@ -70,11 +74,13 @@
 #plugin-nats:io.kestra.plugin:plugin-nats:LATEST
 #plugin-neo4j:io.kestra.plugin:plugin-neo4j:LATEST
 #plugin-notifications:io.kestra.plugin:plugin-notifications:LATEST
+#plugin-ollama:io.kestra.plugin:plugin-ollama:LATEST
 #plugin-openai:io.kestra.plugin:plugin-openai:LATEST
 #plugin-opensearch:io.kestra.plugin:plugin-opensearch:LATEST
 #plugin-powerbi:io.kestra.plugin:plugin-powerbi:LATEST
 #plugin-pulsar:io.kestra.plugin:plugin-pulsar:LATEST
 #plugin-redis:io.kestra.plugin:plugin-redis:LATEST
+#plugin-scripts:io.kestra.plugin:plugin-script-go:LATEST
 #plugin-scripts:io.kestra.plugin:plugin-script-groovy:LATEST
 #plugin-scripts:io.kestra.plugin:plugin-script-jbang:LATEST
 #plugin-scripts:io.kestra.plugin:plugin-script-julia:LATEST

Dockerfile

@@ -16,8 +16,9 @@ RUN apt-get update -y && \
     if [ -n "${APT_PACKAGES}" ]; then apt-get install -y --no-install-recommends ${APT_PACKAGES}; fi && \
     apt-get clean && \
     rm -rf /var/lib/apt/lists/* /var/tmp/* /tmp/* && \
+    curl -LsSf https://astral.sh/uv/0.6.17/install.sh | sh && mv /root/.local/bin/uv /bin && mv /root/.local/bin/uvx /bin && \
     if [ -n "${KESTRA_PLUGINS}" ]; then /app/kestra plugins install ${KESTRA_PLUGINS} && rm -rf /tmp/*; fi && \
-    if [ -n "${PYTHON_LIBRARIES}" ]; then pip install ${PYTHON_LIBRARIES}; fi && \
+    if [ -n "${PYTHON_LIBRARIES}" ]; then uv pip install --system ${PYTHON_LIBRARIES}; fi && \
     chown -R kestra:kestra /app
 
 USER kestra

build.gradle

@@ -21,7 +21,7 @@ plugins {
 
     // test
     id "com.adarshr.test-logger" version "4.0.0"
-    id "org.sonarqube" version "6.1.0.5360"
+    id "org.sonarqube" version "6.2.0.5505"
     id 'jacoco-report-aggregation'
 
     // helper
@@ -165,7 +165,7 @@ allprojects {
  * Test
  **********************************************************************************************************************/
 subprojects {
-    if (it.name != 'platform') {
+    if (it.name != 'platform' && it.name != 'jmh-benchmarks') {
         apply plugin: "com.adarshr.test-logger"
 
         java {
@@ -268,7 +268,7 @@ subprojects {
  * Allure Reports
  **********************************************************************************************************************/
 subprojects {
-    if (it.name != 'platform') {
+    if (it.name != 'platform' && it.name != 'jmh-benchmarks') {
         dependencies {
             testImplementation platform("io.qameta.allure:allure-bom")
             testImplementation "io.qameta.allure:allure-junit5"
@@ -282,7 +282,7 @@ subprojects {
         }
 
         dependencies {
-            agent "org.aspectj:aspectjweaver:1.9.23"
+            agent "org.aspectj:aspectjweaver:1.9.24"
         }
 
         test {
@@ -295,7 +295,7 @@ subprojects {
  * Jacoco
  **********************************************************************************************************************/
 subprojects {
-    if (it.name != 'platform') {
+    if (it.name != 'platform' && it.name != 'jmh-benchmarks') {
         apply plugin: 'jacoco'
 
         test {
@@ -472,6 +472,15 @@ tasks.register('runLocal', JavaExec) {
     args 'server', 'local', '--plugins', 'local/plugins'
 }
 
+tasks.register('runStandalone', JavaExec) {
+    group = "application"
+    description = "Run Kestra as server local"
+    classpath = project(":cli").sourceSets.main.runtimeClasspath
+    mainClass = mainClassName
+    environment 'MICRONAUT_ENVIRONMENTS', 'override'
+    args 'server', 'standalone', '--plugins', 'local/plugins'
+}
+
 /**********************************************************************************************************************\
  * Publish
  **********************************************************************************************************************/
@@ -487,98 +496,101 @@ nexusPublishing {
 }
 
 subprojects {
-    apply plugin: "maven-publish"
-    apply plugin: 'signing'
-    apply plugin: 'ru.vyarus.pom'
-    apply plugin: 'ru.vyarus.github-info'
 
-    javadoc {
-        options {
-            locale = 'en_US'
-            encoding = 'UTF-8'
-            addStringOption("Xdoclint:none", "-quiet")
-        }
-    }
+    if (it.name != 'jmh-benchmarks') {
+        apply plugin: "maven-publish"
+        apply plugin: 'signing'
+        apply plugin: 'ru.vyarus.pom'
+        apply plugin: 'ru.vyarus.github-info'
 
-    tasks.register('sourcesJar', Jar) {
-        dependsOn = [':core:copyGradleProperties']
-        dependsOn = [':ui:assembleFrontend']
-        archiveClassifier.set('sources')
-        from sourceSets.main.allSource
-    }
-    sourcesJar.dependsOn ':core:copyGradleProperties'
-    sourcesJar.dependsOn ':ui:assembleFrontend'
-
-    tasks.register('javadocJar', Jar) {
-        archiveClassifier.set('javadoc')
-        from javadoc
-    }
-
-    tasks.register('testsJar', Jar) {
-        group = 'build'
-        description = 'Build the tests jar'
-
-        archiveClassifier.set('tests')
-        if (sourceSets.matching { it.name == 'test'}) {
-            from sourceSets.named('test').get().output
-        }
-    }
-
-    github {
-        user 'kestra-io'
-        license 'Apache'
-        repository 'kestra'
-        site 'https://kestra.io'
-    }
-
-    maven.pom {
-        description = 'The modern, scalable orchestrator & scheduler open source platform'
-
-        developers {
-            developer {
-                id = "tchiotludo"
-                name = "Ludovic Dehon"
+        javadoc {
+            options {
+                locale = 'en_US'
+                encoding = 'UTF-8'
+                addStringOption("Xdoclint:none", "-quiet")
             }
         }
-    }
 
-    publishing {
-        publications {
-            sonatypePublication(MavenPublication) {
-                version project.version
+        tasks.register('sourcesJar', Jar) {
+            dependsOn = [':core:copyGradleProperties']
+            dependsOn = [':ui:assembleFrontend']
+            archiveClassifier.set('sources')
+            from sourceSets.main.allSource
+        }
+        sourcesJar.dependsOn ':core:copyGradleProperties'
+        sourcesJar.dependsOn ':ui:assembleFrontend'
 
-                if (project.name.contains('cli')) {
-                    groupId "io.kestra"
-                    artifactId "kestra"
+        tasks.register('javadocJar', Jar) {
+            archiveClassifier.set('javadoc')
+            from javadoc
+        }
 
-                    artifact shadowJar
-                    artifact executableJar
-                } else if (project.name.contains('platform')){
-                    groupId project.group
-                    artifactId project.name
-                } else {
-                    from components.java
+        tasks.register('testsJar', Jar) {
+            group = 'build'
+            description = 'Build the tests jar'
 
-                    groupId project.group
-                    artifactId project.name
+            archiveClassifier.set('tests')
+            if (sourceSets.matching { it.name == 'test'}) {
+                from sourceSets.named('test').get().output
+            }
+        }
 
-                    artifact sourcesJar
-                    artifact javadocJar
-                    artifact testsJar
+        github {
+            user 'kestra-io'
+            license 'Apache'
+            repository 'kestra'
+            site 'https://kestra.io'
+        }
+
+        maven.pom {
+            description = 'The modern, scalable orchestrator & scheduler open source platform'
+
+            developers {
+                developer {
+                    id = "tchiotludo"
+                    name = "Ludovic Dehon"
                 }
             }
         }
-    }
 
-    signing {
-        // only sign JARs that we publish to Sonatype
-        required { gradle.taskGraph.hasTask("publishSonatypePublicationPublicationToSonatypeRepository") }
-        sign publishing.publications.sonatypePublication
-    }
+        publishing {
+            publications {
+                sonatypePublication(MavenPublication) {
+                    version project.version
 
-    tasks.withType(GenerateModuleMetadata).configureEach {
-        // Suppression this validation error as we want to enforce the Kestra platform
-        suppressedValidationErrors.add('enforced-platform')
+                    if (project.name.contains('cli')) {
+                        groupId "io.kestra"
+                        artifactId "kestra"
+
+                        artifact shadowJar
+                        artifact executableJar
+                    } else if (project.name.contains('platform')){
+                        groupId project.group
+                        artifactId project.name
+                    } else {
+                        from components.java
+
+                        groupId project.group
+                        artifactId project.name
+
+                        artifact sourcesJar
+                        artifact javadocJar
+                        artifact testsJar
+                    }
+                }
+            }
+        }
+
+        signing {
+            // only sign JARs that we publish to Sonatype
+            required { gradle.taskGraph.hasTask("publishSonatypePublicationPublicationToSonatypeRepository") }
+            sign publishing.publications.sonatypePublication
+        }
+
+        tasks.withType(GenerateModuleMetadata).configureEach {
+            // Suppression this validation error as we want to enforce the Kestra platform
+            suppressedValidationErrors.add('enforced-platform')
+        }
     }
 }
 

cli/src/main/java/io/kestra/cli/AbstractApiCommand.java

@@ -1,5 +1,7 @@
 package io.kestra.cli;
 
+import static io.kestra.core.tenant.TenantService.MAIN_TENANT;
+
 import io.micronaut.core.annotation.Nullable;
 import io.micronaut.http.HttpHeaders;
 import io.micronaut.http.HttpRequest;
@@ -90,7 +92,7 @@ public abstract class AbstractApiCommand extends AbstractCommand {
             throw new IllegalArgumentException("'path' must be non-null and start with '/'");
         }
 
-        return tenantId == null ? "/api/v1" + path : "/api/v1/" + tenantId + path;
+        return tenantId == null ? "/api/v1/" + MAIN_TENANT + path : "/api/v1/" + tenantId + path;
     }
 
     @Builder

cli/src/main/java/io/kestra/cli/App.java

@@ -2,6 +2,7 @@ package io.kestra.cli;
 
 import io.kestra.cli.commands.configs.sys.ConfigCommand;
 import io.kestra.cli.commands.flows.FlowCommand;
+import io.kestra.cli.commands.migrations.MigrationCommand;
 import io.kestra.cli.commands.namespaces.NamespaceCommand;
 import io.kestra.cli.commands.plugins.PluginCommand;
 import io.kestra.cli.commands.servers.ServerCommand;
@@ -42,6 +43,7 @@ import java.util.concurrent.Callable;
         SysCommand.class,
         ConfigCommand.class,
         NamespaceCommand.class,
+        MigrationCommand.class,
     }
 )
 @Introspected

cli/src/main/java/io/kestra/cli/commands/migrations/MigrationCommand.java

@@ -0,0 +1,29 @@
+package io.kestra.cli.commands.migrations;
+
+import io.kestra.cli.AbstractCommand;
+import io.kestra.cli.App;
+import io.micronaut.configuration.picocli.PicocliRunner;
+import lombok.SneakyThrows;
+import lombok.extern.slf4j.Slf4j;
+import picocli.CommandLine;
+
+@CommandLine.Command(
+    name = "migrate",
+    description = "handle migrations",
+    mixinStandardHelpOptions = true,
+    subcommands = {
+        TenantMigrationCommand.class,
+    }
+)
+@Slf4j
+public class MigrationCommand extends AbstractCommand {
+    @SneakyThrows
+    @Override
+    public Integer call() throws Exception {
+        super.call();
+
+        PicocliRunner.call(App.class, "migrate",  "--help");
+
+        return 0;
+    }
+}

cli/src/main/java/io/kestra/cli/commands/migrations/TenantMigrationCommand.java

@@ -0,0 +1,49 @@
+package io.kestra.cli.commands.migrations;
+
+import io.kestra.cli.AbstractCommand;
+import io.kestra.core.repositories.TenantMigrationInterface;
+import io.micronaut.context.ApplicationContext;
+import jakarta.inject.Inject;
+import lombok.extern.slf4j.Slf4j;
+import picocli.CommandLine;
+import picocli.CommandLine.Option;
+
+@CommandLine.Command(
+    name = "default-tenant",
+    description = "migrate every elements from no tenant to the main tenant"
+)
+@Slf4j
+public class TenantMigrationCommand extends AbstractCommand {
+    @Inject
+    private ApplicationContext applicationContext;
+
+    @Option(names = "--tenant-id", description = "tenant identifier")
+    String tenantId;
+
+    @Option(names = "--tenant-name", description = "tenant name")
+    String tenantName;
+
+    @Option(names = "--dry-run", description = "Preview only, do not update")
+    boolean dryRun;
+
+    @Override
+    public Integer call() throws Exception {
+        super.call();
+
+        if (dryRun) {
+            System.out.println("🧪 Dry-run mode enabled. No changes will be applied.");
+        }
+
+        TenantMigrationService migrationService = this.applicationContext.getBean(TenantMigrationService.class);
+        try {
+            migrationService.migrateTenant(tenantId, tenantName, dryRun);
+            System.out.println("✅ Tenant migration complete.");
+        } catch (Exception e) {
+            System.err.println("❌ Tenant migration failed: " + e.getMessage());
+            e.printStackTrace();
+            return 1;
+        }
+        return 0;
+    }
+
+}

cli/src/main/java/io/kestra/cli/commands/migrations/TenantMigrationService.java

@@ -0,0 +1,56 @@
+package io.kestra.cli.commands.migrations;
+
+import static io.kestra.core.tenant.TenantService.MAIN_TENANT;
+
+import com.github.javaparser.utils.Log;
+import io.kestra.core.exceptions.KestraRuntimeException;
+import io.kestra.core.models.flows.FlowInterface;
+import io.kestra.core.queues.QueueException;
+import io.kestra.core.queues.QueueFactoryInterface;
+import io.kestra.core.queues.QueueInterface;
+import io.kestra.core.repositories.FlowRepositoryInterface;
+import io.kestra.core.repositories.TenantMigrationInterface;
+import jakarta.inject.Inject;
+import jakarta.inject.Named;
+import jakarta.inject.Singleton;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.StringUtils;
+
+@Singleton
+@Slf4j
+public class TenantMigrationService {
+
+    @Inject
+    private TenantMigrationInterface tenantMigrationInterface;
+
+    @Inject
+    private FlowRepositoryInterface flowRepository;
+
+    @Inject
+    @Named(QueueFactoryInterface.FLOW_NAMED)
+    private QueueInterface<FlowInterface> flowQueue;
+
+    public void migrateTenant(String tenantId, String tenantName, boolean dryRun) {
+        if (StringUtils.isNotBlank(tenantId) && !MAIN_TENANT.equals(tenantId)){
+            throw new KestraRuntimeException("Tenant configuration is an enterprise feature. It can only be main in OSS");
+        }
+
+        Log.info("🔁 Starting tenant migration...");
+        tenantMigrationInterface.migrateTenant(MAIN_TENANT, dryRun);
+        migrateQueue(dryRun);
+    }
+
+    protected void migrateQueue(boolean dryRun) {
+        if (!dryRun){
+            log.info("🔁 Starting restoring queue...");
+            flowRepository.findAllWithSourceForAllTenants().forEach(flow -> {
+                try {
+                    flowQueue.emit(flow);
+                } catch (QueueException e) {
+                    log.warn("Unable to send the flow {} to the queue", flow.uid(), e);
+                }
+            });
+        }
+    }
+
+}

cli/src/main/java/io/kestra/cli/commands/plugins/PluginInstallCommand.java

@@ -1,6 +1,7 @@
 package io.kestra.cli.commands.plugins;
 
 import io.kestra.core.contexts.MavenPluginRepositoryConfig;
+import io.kestra.core.exceptions.KestraRuntimeException;
 import io.kestra.core.plugins.LocalPluginManager;
 import io.kestra.core.plugins.MavenPluginDownloader;
 import io.kestra.core.plugins.PluginArtifact;
@@ -51,7 +52,7 @@ public class PluginInstallCommand extends AbstractCommand {
     Provider<MavenPluginDownloader> mavenPluginRepositoryProvider;
 
     @Inject
-    @Client("api") HttpClient httpClient;
+    Provider<PluginCatalogService> pluginCatalogService;
 
     @Override
     public Integer call() throws Exception {
@@ -85,7 +86,7 @@ public class PluginInstallCommand extends AbstractCommand {
         }
 
         if (all) {
-            PluginCatalogService service = new PluginCatalogService(httpClient, false, true);
+            PluginCatalogService service = pluginCatalogService.get();
             dependencies = service.get().stream().map(Objects::toString).toList();
         }
 
@@ -103,12 +104,21 @@ public class PluginInstallCommand extends AbstractCommand {
         }
 
         try (final PluginManager pluginManager = getPluginManager()) {
-            List<PluginArtifact> installed = pluginManager.install(
-                pluginArtifacts,
-                repositoryConfigs,
-                false,
-                pluginsPath
-            );
+
+            List<PluginArtifact> installed;
+            if (all) {
+                installed = new ArrayList<>(pluginArtifacts.size());
+                for (PluginArtifact pluginArtifact : pluginArtifacts) {
+                    try {
+                        installed.add(pluginManager.install(pluginArtifact, repositoryConfigs, false, pluginsPath));
+                    } catch (KestraRuntimeException e) {
+                        String cause = e.getCause() != null ? e.getCause().getMessage() : e.getMessage();
+                        stdErr("Failed to install plugin {0}. Cause: {1}", pluginArtifact, cause);
+                    }
+                }
+            } else {
+                installed = pluginManager.install(pluginArtifacts, repositoryConfigs, false, pluginsPath);
+            }
 
             List<URI> uris = installed.stream().map(PluginArtifact::uri).toList();
             stdOut("Successfully installed plugins {0} into {1}", dependencies, uris);

cli/src/main/java/io/kestra/cli/commands/servers/StandAloneCommand.java

@@ -98,7 +98,7 @@ public class StandAloneCommand extends AbstractServerCommand {
         if (flowPath != null) {
             try {
                 LocalFlowRepositoryLoader localFlowRepositoryLoader = applicationContext.getBean(LocalFlowRepositoryLoader.class);
-                localFlowRepositoryLoader.load(this.flowPath);
+                localFlowRepositoryLoader.load(null, this.flowPath);
             } catch (IOException e) {
                 throw new CommandLine.ParameterException(this.spec.commandLine(), "Invalid flow path", e);
             }

cli/src/main/java/io/kestra/cli/services/FileChangedEventListener.java

@@ -1,6 +1,6 @@
 package io.kestra.cli.services;
 
-import io.kestra.core.exceptions.DeserializationException;
+import io.kestra.core.exceptions.FlowProcessingException;
 import io.kestra.core.models.flows.FlowInterface;
 import io.kestra.core.models.flows.FlowWithPath;
 import io.kestra.core.models.flows.FlowWithSource;
@@ -236,7 +236,7 @@ public class FileChangedEventListener {
             FlowWithSource flow = pluginDefaultService.parseFlowWithAllDefaults(tenantId, content, false);
             modelValidator.validate(flow);
             return Optional.of(flow);
-        } catch (DeserializationException | ConstraintViolationException e) {
+        } catch (ConstraintViolationException | FlowProcessingException e) {
             log.warn("Error while parsing flow: {}", entry, e);
         }
         return Optional.empty();

cli/src/main/java/io/kestra/cli/services/LocalFlowFileWatcher.java

@@ -15,7 +15,7 @@ public class LocalFlowFileWatcher implements FlowFilesManager {
 
     @Override
     public FlowWithSource createOrUpdateFlow(final GenericFlow flow) {
-        return flowRepository.findById(null, flow.getNamespace(), flow.getId())
+        return flowRepository.findById(flow.getTenantId(), flow.getNamespace(), flow.getId())
             .map(previous -> flowRepository.update(flow, previous))
             .orElseGet(() -> flowRepository.create(flow));
     }

cli/src/main/resources/application.yml

@@ -26,11 +26,11 @@ micronaut:
     netty:
       max-chunk-size: 10MB
       max-header-size: 32768 # increased from the default of 8k
-      responses:
-        file:
-          cache-seconds: 86400
-          cache-control:
-            public: true
+    responses:
+      file:
+        cache-seconds: 86400
+        cache-control:
+          public: true
 
       # Access log configuration, see https://docs.micronaut.io/latest/guide/index.html#accessLogger
       access-logger:
@@ -138,8 +138,8 @@ kestra:
   jdbc:
     queues:
       min-poll-interval: 25ms
-      max-poll-interval: 1000ms
-      poll-switch-interval: 5s
+      max-poll-interval: 500ms
+      poll-switch-interval: 60s
 
     cleaner:
       initial-delay: 1h
@@ -195,13 +195,18 @@ kestra:
     liveness:
       enabled: true
       # The expected time between liveness probe.
-      interval: 5s
+      interval: 10s
       # The timeout used to detect service failures.
-      timeout: 45s
+      timeout: 1m
       # The time to wait before executing a liveness probe.
-      initialDelay: 45s
+      initialDelay: 1m
       # The expected time between service heartbeats.
       heartbeatInterval: 3s
+    service:
+      purge:
+        initial-delay: 1h
+        fixed-delay: 1d
+        retention: 30d
   anonymous-usage-report:
     enabled: true
     uri: https://api.kestra.io/v1/reports/usages

cli/src/test/java/io/kestra/cli/commands/configs/sys/ConfigPropertiesCommandTest.java

@@ -4,11 +4,14 @@ import io.micronaut.configuration.picocli.PicocliRunner;
 import io.micronaut.context.ApplicationContext;
 import io.micronaut.context.env.Environment;
 import org.junit.jupiter.api.Test;
+import org.yaml.snakeyaml.Yaml;
 
 import java.io.ByteArrayOutputStream;
 import java.io.PrintStream;
+import java.util.Map;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.AssertionsForClassTypes.catchThrowable;
 
 class ConfigPropertiesCommandTest {
     @Test
@@ -23,4 +26,48 @@ class ConfigPropertiesCommandTest {
             assertThat(out.toString()).contains("- test");
         }
     }
+
+    @Test
+    void shouldOutputCustomEnvironment() {
+        ByteArrayOutputStream out = new ByteArrayOutputStream();
+        System.setOut(new PrintStream(out));
+
+        try (ApplicationContext ctx = ApplicationContext.run(Environment.CLI, "custom-env")) {
+            PicocliRunner.call(ConfigPropertiesCommand.class, ctx);
+
+            assertThat(out.toString()).contains("activeEnvironments:");
+            assertThat(out.toString()).contains("- custom-env");
+        }
+    }
+
+    @Test
+    void shouldReturnZeroOnSuccess() throws Exception {
+        ByteArrayOutputStream out = new ByteArrayOutputStream();
+        System.setOut(new PrintStream(out));
+
+        try (ApplicationContext ctx = ApplicationContext.run(Environment.CLI, Environment.TEST)) {
+            ConfigPropertiesCommand cmd = ctx.createBean(ConfigPropertiesCommand.class);
+            int result = cmd.call();
+
+            assertThat(result).isZero();
+        }
+    }
+
+    @Test
+    void shouldOutputValidYaml() {
+        ByteArrayOutputStream out = new ByteArrayOutputStream();
+        System.setOut(new PrintStream(out));
+
+        try (ApplicationContext ctx = ApplicationContext.run(Environment.CLI, Environment.TEST)) {
+            PicocliRunner.call(ConfigPropertiesCommand.class, ctx);
+
+            String output = out.toString();
+            Yaml yaml = new Yaml();
+            Throwable thrown = catchThrowable(() -> {
+                Map<?, ?> parsed = yaml.load(output);
+                assertThat(parsed).isInstanceOf(Map.class);
+            });
+            assertThat(thrown).isNull();
+        }
+    }
 }

cli/src/test/java/io/kestra/cli/commands/flows/FlowCreateOrUpdateCommandTest.java

@@ -129,7 +129,7 @@ class FlowCreateOrUpdateCommandTest {
             };
             Integer call = PicocliRunner.call(FlowUpdatesCommand.class, ctx, args);
 
-            assertThat(call).isEqualTo(0);
+            assertThat(call).isZero();
             assertThat(out.toString()).contains("1 flow(s)");
         }
     }

cli/src/test/java/io/kestra/cli/commands/flows/FlowDotCommandTest.java

@@ -24,7 +24,7 @@ class FlowDotCommandTest {
             };
             Integer call = PicocliRunner.call(FlowDotCommand.class, ctx, args);
 
-            assertThat(call).isEqualTo(0);
+            assertThat(call).isZero();
             assertThat(out.toString()).contains("\"root.date\"[shape=box];");
         }
     }

cli/src/test/java/io/kestra/cli/commands/flows/FlowExpandCommandTest.java

@@ -22,7 +22,7 @@ class FlowExpandCommandTest {
             };
             Integer call = PicocliRunner.call(FlowExpandCommand.class, ctx, args);
 
-            assertThat(call).isEqualTo(0);
+            assertThat(call).isZero();
             assertThat(out.toString()).isEqualTo("id: include\n" +
                 "namespace: io.kestra.cli\n" +
                 "\n" +

cli/src/test/java/io/kestra/cli/commands/flows/FlowExportCommandTest.java

@@ -55,7 +55,7 @@ class FlowExportCommandTest {
             };
             PicocliRunner.call(FlowExportCommand.class, ctx, exportArgs);
             File file = new File("/tmp/flows.zip");
-            assertThat(file.exists()).isEqualTo(true);
+            assertThat(file.exists()).isTrue();
             ZipFile zipFile = new ZipFile(file);
 
             // When launching the test in a suite, there is 4 flows but when lauching individualy there is only 3

cli/src/test/java/io/kestra/cli/commands/flows/FlowUpdatesCommandTest.java

@@ -142,7 +142,7 @@ class FlowUpdatesCommandTest {
             };
             PicocliRunner.call(FlowUpdatesCommand.class, ctx, args);
 
-            assertThat(out.toString()).contains("Invalid entity: flow.namespace: io.kestra.outsider_quattro_-1 - flow namespace is invalid");
+            assertThat(out.toString()).contains("Invalid entity: flow.namespace: main_io.kestra.outsider_quattro_-1 - flow namespace is invalid");
         }
     }
 
@@ -169,7 +169,7 @@ class FlowUpdatesCommandTest {
             };
             Integer call = PicocliRunner.call(FlowUpdatesCommand.class, ctx, args);
 
-            assertThat(call).isEqualTo(0);
+            assertThat(call).isZero();
             assertThat(out.toString()).contains("1 flow(s)");
         }
     }

cli/src/test/java/io/kestra/cli/commands/flows/FlowValidateCommandTest.java

@@ -22,7 +22,7 @@ class FlowValidateCommandTest {
             };
             Integer call = PicocliRunner.call(FlowValidateCommand.class, ctx, args);
 
-            assertThat(call).isEqualTo(0);
+            assertThat(call).isZero();
             assertThat(out.toString()).contains("✓ - io.kestra.cli / include");
         }
     }
@@ -39,7 +39,7 @@ class FlowValidateCommandTest {
             };
             Integer call = PicocliRunner.call(FlowValidateCommand.class, ctx, args);
 
-            assertThat(call).isEqualTo(0);
+            assertThat(call).isZero();
             assertThat(out.toString()).contains("✓ - system / warning");
             assertThat(out.toString()).contains("⚠ - tasks[0] is deprecated");
             assertThat(out.toString()).contains("ℹ - io.kestra.core.tasks.log.Log is replaced by io.kestra.plugin.core.log.Log");

cli/src/test/java/io/kestra/cli/commands/flows/SingleFlowCommandsTest.java

@@ -12,12 +12,11 @@ import java.net.URL;
 
 import static org.assertj.core.api.Assertions.assertThat;
 
-public class SingleFlowCommandsTest {
-
+class SingleFlowCommandsTest {
 
     @Test
     void all() {
-        URL flow = SingleFlowCommandsTest.class.getClassLoader().getResource("flows/quattro.yml");
+        URL flow = SingleFlowCommandsTest.class.getClassLoader().getResource("crudFlow/date.yml");
         ByteArrayOutputStream out = new ByteArrayOutputStream();
         System.setOut(new PrintStream(out));
 
@@ -26,19 +25,6 @@ public class SingleFlowCommandsTest {
             EmbeddedServer embeddedServer = ctx.getBean(EmbeddedServer.class);
             embeddedServer.start();
 
-            String[] deleteArgs = {
-                "--server",
-                embeddedServer.getURL().toString(),
-                "--user",
-                "myuser:pass:word",
-                "io.kestra.outsider",
-                "quattro"
-            };
-            PicocliRunner.call(FlowDeleteCommand.class, ctx, deleteArgs);
-
-            assertThat(out.toString()).contains("Flow successfully deleted !");
-            out.reset();
-
             String[] createArgs = {
                 "--server",
                 embeddedServer.getURL().toString(),
@@ -50,21 +36,34 @@ public class SingleFlowCommandsTest {
 
             assertThat(out.toString()).contains("Flow successfully created !");
 
+            out.reset();
 
-            out.reset();String[] updateArgs = {
+            String[] updateArgs = {
                 "--server",
                 embeddedServer.getURL().toString(),
                 "--user",
                 "myuser:pass:word",
                 flow.getPath(),
-                "io.kestra.outsider",
-                "quattro"
+                "io.kestra.cli",
+                "date"
             };
             PicocliRunner.call(FlowUpdateCommand.class, ctx, updateArgs);
 
             assertThat(out.toString()).contains("Flow successfully updated !");
+
             out.reset();
+
+            String[] deleteArgs = {
+                "--server",
+                embeddedServer.getURL().toString(),
+                "--user",
+                "myuser:pass:word",
+                "io.kestra.cli",
+                "date"
+            };
+            PicocliRunner.call(FlowDeleteCommand.class, ctx, deleteArgs);
+
+            assertThat(out.toString()).contains("Flow successfully deleted !");
         }
     }
-
 }

cli/src/test/java/io/kestra/cli/commands/flows/namespaces/FlowNamespaceCommandTest.java

@@ -19,7 +19,7 @@ class FlowNamespaceCommandTest {
             String[] args = {};
             Integer call = PicocliRunner.call(FlowNamespaceCommand.class, ctx, args);
 
-            assertThat(call).isEqualTo(0);
+            assertThat(call).isZero();
             assertThat(out.toString()).contains("Usage: kestra flow namespace");
         }
     }

cli/src/test/java/io/kestra/cli/commands/flows/namespaces/FlowNamespaceUpdateCommandTest.java

@@ -162,7 +162,7 @@ class FlowNamespaceUpdateCommandTest {
             };
             Integer call = PicocliRunner.call(FlowNamespaceUpdateCommand.class, ctx, args);
 
-            assertThat(call).isEqualTo(0);
+            assertThat(call).isZero();
             assertThat(out.toString()).contains("1 flow(s)");
         }
     }

cli/src/test/java/io/kestra/cli/commands/namespaces/NamespaceCommandTest.java

@@ -19,7 +19,7 @@ class NamespaceCommandTest {
             String[] args = {};
             Integer call = PicocliRunner.call(NamespaceCommand.class, ctx, args);
 
-            assertThat(call).isEqualTo(0);
+            assertThat(call).isZero();
             assertThat(out.toString()).contains("Usage: kestra namespace");
         }
     }

cli/src/test/java/io/kestra/cli/commands/namespaces/files/NamespaceFilesCommandTest.java

@@ -19,7 +19,7 @@ class NamespaceFilesCommandTest {
             String[] args = {};
             Integer call = PicocliRunner.call(NamespaceFilesCommand.class, ctx, args);
 
-            assertThat(call).isEqualTo(0);
+            assertThat(call).isZero();
             assertThat(out.toString()).contains("Usage: kestra namespace files");
         }
     }

cli/src/test/java/io/kestra/cli/commands/namespaces/kv/KvCommandTest.java

@@ -19,7 +19,7 @@ class KvCommandTest {
             String[] args = {};
             Integer call = PicocliRunner.call(KvCommand.class, ctx, args);
 
-            assertThat(call).isEqualTo(0);
+            assertThat(call).isZero();
             assertThat(out.toString()).contains("Usage: kestra namespace kv");
         }
     }

cli/src/test/java/io/kestra/cli/commands/namespaces/kv/KvUpdateCommandTest.java

@@ -16,6 +16,7 @@ import java.io.IOException;
 import java.nio.file.Files;
 import java.util.Map;
 
+import static io.kestra.core.tenant.TenantService.MAIN_TENANT;
 import static org.assertj.core.api.Assertions.assertThat;
 
 class KvUpdateCommandTest {
@@ -40,7 +41,7 @@ class KvUpdateCommandTest {
             PicocliRunner.call(KvUpdateCommand.class, ctx, args);
 
             KVStoreService kvStoreService = ctx.getBean(KVStoreService.class);
-            KVStore kvStore = kvStoreService.get(null, "io.kestra.cli", null);
+            KVStore kvStore = kvStoreService.get(MAIN_TENANT, "io.kestra.cli", null);
 
             assertThat(kvStore.getValue("string").get()).isEqualTo(new KVValue("stringValue"));
             assertThat(((InternalKVStore) kvStore).getRawValue("string").get()).isEqualTo("\"stringValue\"");
@@ -68,7 +69,7 @@ class KvUpdateCommandTest {
             PicocliRunner.call(KvUpdateCommand.class, ctx, args);
 
             KVStoreService kvStoreService = ctx.getBean(KVStoreService.class);
-            KVStore kvStore = kvStoreService.get(null, "io.kestra.cli", null);
+            KVStore kvStore = kvStoreService.get(MAIN_TENANT, "io.kestra.cli", null);
 
             assertThat(kvStore.getValue("int").get()).isEqualTo(new KVValue(1));
             assertThat(((InternalKVStore) kvStore).getRawValue("int").get()).isEqualTo("1");
@@ -98,7 +99,7 @@ class KvUpdateCommandTest {
             PicocliRunner.call(KvUpdateCommand.class, ctx, args);
 
             KVStoreService kvStoreService = ctx.getBean(KVStoreService.class);
-            KVStore kvStore = kvStoreService.get(null, "io.kestra.cli", null);
+            KVStore kvStore = kvStoreService.get(MAIN_TENANT, "io.kestra.cli", null);
 
             assertThat(kvStore.getValue("intStr").get()).isEqualTo(new KVValue("1"));
             assertThat(((InternalKVStore) kvStore).getRawValue("intStr").get()).isEqualTo("\"1\"");
@@ -126,7 +127,7 @@ class KvUpdateCommandTest {
             PicocliRunner.call(KvUpdateCommand.class, ctx, args);
 
             KVStoreService kvStoreService = ctx.getBean(KVStoreService.class);
-            KVStore kvStore = kvStoreService.get(null, "io.kestra.cli", null);
+            KVStore kvStore = kvStoreService.get(MAIN_TENANT, "io.kestra.cli", null);
 
             assertThat(kvStore.getValue("object").get()).isEqualTo(new KVValue(Map.of("some", "json")));
             assertThat(((InternalKVStore) kvStore).getRawValue("object").get()).isEqualTo("{some:\"json\"}");
@@ -156,7 +157,7 @@ class KvUpdateCommandTest {
             PicocliRunner.call(KvUpdateCommand.class, ctx, args);
 
             KVStoreService kvStoreService = ctx.getBean(KVStoreService.class);
-            KVStore kvStore = kvStoreService.get(null, "io.kestra.cli", null);
+            KVStore kvStore = kvStoreService.get(MAIN_TENANT, "io.kestra.cli", null);
 
             assertThat(kvStore.getValue("objectStr").get()).isEqualTo(new KVValue("{\"some\":\"json\"}"));
             assertThat(((InternalKVStore) kvStore).getRawValue("objectStr").get()).isEqualTo("\"{\\\"some\\\":\\\"json\\\"}\"");
@@ -190,7 +191,7 @@ class KvUpdateCommandTest {
             PicocliRunner.call(KvUpdateCommand.class, ctx, args);
 
             KVStoreService kvStoreService = ctx.getBean(KVStoreService.class);
-            KVStore kvStore = kvStoreService.get(null, "io.kestra.cli", null);
+            KVStore kvStore = kvStoreService.get(MAIN_TENANT, "io.kestra.cli", null);
 
             assertThat(kvStore.getValue("objectFromFile").get()).isEqualTo(new KVValue(Map.of("some", "json", "from", "file")));
             assertThat(((InternalKVStore) kvStore).getRawValue("objectFromFile").get()).isEqualTo("{some:\"json\",from:\"file\"}");

cli/src/test/java/io/kestra/cli/commands/plugins/PluginCommandTest.java

@@ -23,4 +23,32 @@ class PluginCommandTest {
             assertThat(out.toString()).contains("Usage: kestra plugins");
         }
     }
-}
+    
+    
+    // Additional Coverage:
+    @Test
+    void shouldListSubcommandsInHelp() throws Exception {
+        ByteArrayOutputStream out = new ByteArrayOutputStream();
+        PrintStream originalOut = System.out;
+        System.setOut(new PrintStream(out));
+        try {
+            PluginCommand cmd = new PluginCommand();
+            cmd.call();
+            String output = out.toString();
+            assertThat(output).contains("install");
+            assertThat(output).contains("uninstall");
+            assertThat(output).contains("list");
+            assertThat(output).contains("doc");
+            assertThat(output).contains("search");
+        } finally {
+            System.setOut(originalOut);
+        }
+    }
+
+    // Passes
+    @Test
+    void shouldNotLoadExternalPlugins() {
+        PluginCommand cmd = new PluginCommand();
+        assertThat(cmd.loadExternalPlugins()).isFalse();
+    }
+}

cli/src/test/java/io/kestra/cli/commands/plugins/PluginDocCommandTest.java

@@ -46,7 +46,7 @@ class PluginDocCommandTest {
             assertThat(files.size()).isEqualTo(1);
             assertThat(files.getFirst().getFileName().toString()).isEqualTo("plugin-template-test");
             var directory = files.getFirst().toFile();
-            assertThat(directory.isDirectory()).isEqualTo(true);
+            assertThat(directory.isDirectory()).isTrue();
             assertThat(directory.listFiles().length).isEqualTo(3);
 
             var readme = directory.toPath().resolve("index.md");

cli/src/test/java/io/kestra/cli/commands/sys/ReindexCommandTest.java

@@ -42,7 +42,7 @@ class ReindexCommandTest {
                "flow",
             };
             Integer call = PicocliRunner.call(ReindexCommand.class, ctx, reindexArgs);
-            assertThat(call).isEqualTo(0);
+            assertThat(call).isZero();
             // in local it reindex 3 flows and in CI 4 for an unknown reason
             assertThat(out.toString()).contains("Successfully reindex");
         }

cli/src/test/java/io/kestra/cli/commands/sys/database/DatabaseCommandTest.java

@@ -19,7 +19,7 @@ class DatabaseCommandTest {
             String[] args = {};
             Integer call = PicocliRunner.call(DatabaseCommand.class, ctx, args);
 
-            assertThat(call).isEqualTo(0);
+            assertThat(call).isZero();
             assertThat(out.toString()).contains("Usage: kestra sys database");
         }
     }

cli/src/test/java/io/kestra/cli/commands/sys/statestore/StateStoreCommandTest.java

@@ -20,7 +20,7 @@ class StateStoreCommandTest {
             String[] args = {};
             Integer call = PicocliRunner.call(StateStoreCommand.class, ctx, args);
 
-            assertThat(call).isEqualTo(0);
+            assertThat(call).isZero();
             assertThat(out.toString()).contains("Usage: kestra sys state-store");
         }
     }

cli/src/test/java/io/kestra/cli/commands/sys/statestore/StateStoreMigrateCommandTest.java

@@ -53,7 +53,7 @@ class StateStoreMigrateCommandTest {
                 oldStateStoreUri,
                 new ByteArrayInputStream("my-value".getBytes())
             );
-            assertThat(storage.exists(tenantId, flow.getNamespace(), oldStateStoreUri)).isEqualTo(true);
+            assertThat(storage.exists(tenantId, flow.getNamespace(), oldStateStoreUri)).isTrue();
 
             RunContext runContext = ctx.getBean(RunContextFactory.class).of(flow, Map.of("flow", Map.of(
                 "tenantId", tenantId,
@@ -67,9 +67,9 @@ class StateStoreMigrateCommandTest {
             Integer call = PicocliRunner.call(StateStoreMigrateCommand.class, ctx, args);
 
             assertThat(new String(stateStore.getState(true, "my-state", "sub-name", "my-taskrun-value").readAllBytes())).isEqualTo("my-value");
-            assertThat(storage.exists(tenantId, flow.getNamespace(), oldStateStoreUri)).isEqualTo(false);
+            assertThat(storage.exists(tenantId, flow.getNamespace(), oldStateStoreUri)).isFalse();
 
-            assertThat(call).isEqualTo(0);
+            assertThat(call).isZero();
         }
     }
 }

cli/src/test/java/io/kestra/cli/commands/templates/TemplateExportCommandTest.java

@@ -54,7 +54,7 @@ class TemplateExportCommandTest {
             };
             PicocliRunner.call(TemplateExportCommand.class, ctx, exportArgs);
             File file = new File("/tmp/templates.zip");
-            assertThat(file.exists()).isEqualTo(true);
+            assertThat(file.exists()).isTrue();
             ZipFile zipFile = new ZipFile(file);
             assertThat(zipFile.stream().count()).isEqualTo(3L);
 

cli/src/test/java/io/kestra/cli/commands/templates/TemplateValidateCommandTest.java

@@ -13,7 +13,7 @@ import java.util.Map;
 
 import static org.assertj.core.api.Assertions.assertThat;
 
-public class TemplateValidateCommandTest {
+class TemplateValidateCommandTest {
     @Test
     void runLocal()  {
         URL directory = TemplateValidateCommandTest.class.getClassLoader().getResource("invalidsTemplates/template.yml");

cli/src/test/java/io/kestra/cli/commands/templates/namespaces/TemplateNamespaceCommandTest.java

@@ -19,7 +19,7 @@ class TemplateNamespaceCommandTest {
             String[] args = {};
             Integer call = PicocliRunner.call(TemplateNamespaceCommand.class, ctx, args);
 
-            assertThat(call).isEqualTo(0);
+            assertThat(call).isZero();
             assertThat(out.toString()).contains("Usage: kestra template namespace");
         }
     }

cli/src/test/java/io/kestra/cli/listeners/DeleteConfigurationApplicationListenersTest.java

@@ -27,7 +27,7 @@ class DeleteConfigurationApplicationListenersTest {
         );
 
         try (ApplicationContext ctx = ApplicationContext.run(mapPropertySource, Environment.CLI, Environment.TEST)) {
-            assertThat(tempFile.exists()).isEqualTo(false);
+            assertThat(tempFile.exists()).isFalse();
         }
     }
 }

cli/src/test/java/io/kestra/cli/services/FileChangedEventListenerTest.java

@@ -18,6 +18,7 @@ import java.util.concurrent.Executors;
 import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicBoolean;
 
+import static io.kestra.core.tenant.TenantService.MAIN_TENANT;
 import static io.kestra.core.utils.Rethrow.throwRunnable;
 import static org.assertj.core.api.Assertions.assertThat;
 
@@ -43,6 +44,7 @@ class FileChangedEventListenerTest {
     @AfterAll
     static void tearDown() throws IOException {
         if (Files.exists(Path.of(FILE_WATCH))) {
+            FileUtils.cleanDirectory(Path.of(FILE_WATCH).toFile());
             FileUtils.deleteDirectory(Path.of(FILE_WATCH).toFile());
         }
     }
@@ -57,7 +59,7 @@ class FileChangedEventListenerTest {
     @RetryingTest(5) // Flaky on CI but always pass locally
     void test() throws IOException, TimeoutException {
         // remove the flow if it already exists
-        flowRepository.findByIdWithSource(null, "io.kestra.tests.watch", "myflow").ifPresent(flow -> flowRepository.delete(flow));
+        flowRepository.findByIdWithSource(MAIN_TENANT, "io.kestra.tests.watch", "myflow").ifPresent(flow -> flowRepository.delete(flow));
 
         // create a basic flow
         String flow = """
@@ -71,11 +73,11 @@ class FileChangedEventListenerTest {
             """;
         Files.write(Path.of(FILE_WATCH + "/myflow.yaml"), flow.getBytes());
         Await.until(
-            () -> flowRepository.findById(null, "io.kestra.tests.watch", "myflow").isPresent(),
+            () -> flowRepository.findById(MAIN_TENANT, "io.kestra.tests.watch", "myflow").isPresent(),
             Duration.ofMillis(100),
             Duration.ofSeconds(10)
         );
-        Flow myflow = flowRepository.findById(null, "io.kestra.tests.watch", "myflow").orElseThrow();
+        Flow myflow = flowRepository.findById(MAIN_TENANT, "io.kestra.tests.watch", "myflow").orElseThrow();
         assertThat(myflow.getTasks()).hasSize(1);
         assertThat(myflow.getTasks().getFirst().getId()).isEqualTo("hello");
         assertThat(myflow.getTasks().getFirst().getType()).isEqualTo("io.kestra.plugin.core.log.Log");
@@ -83,7 +85,7 @@ class FileChangedEventListenerTest {
         // delete the flow
         Files.delete(Path.of(FILE_WATCH + "/myflow.yaml"));
         Await.until(
-            () -> flowRepository.findById(null, "io.kestra.tests.watch", "myflow").isEmpty(),
+            () -> flowRepository.findById(MAIN_TENANT, "io.kestra.tests.watch", "myflow").isEmpty(),
             Duration.ofMillis(100),
             Duration.ofSeconds(10)
         );
@@ -92,7 +94,7 @@ class FileChangedEventListenerTest {
     @RetryingTest(5) // Flaky on CI but always pass locally
     void testWithPluginDefault() throws IOException, TimeoutException {
         // remove the flow if it already exists
-        flowRepository.findByIdWithSource(null, "io.kestra.tests.watch", "pluginDefault").ifPresent(flow -> flowRepository.delete(flow));
+        flowRepository.findByIdWithSource(MAIN_TENANT, "io.kestra.tests.watch", "pluginDefault").ifPresent(flow -> flowRepository.delete(flow));
 
         // create a flow with plugin default
         String pluginDefault = """
@@ -110,11 +112,11 @@ class FileChangedEventListenerTest {
             """;
         Files.write(Path.of(FILE_WATCH + "/plugin-default.yaml"), pluginDefault.getBytes());
         Await.until(
-            () -> flowRepository.findById(null, "io.kestra.tests.watch", "pluginDefault").isPresent(),
+            () -> flowRepository.findById(MAIN_TENANT, "io.kestra.tests.watch", "pluginDefault").isPresent(),
             Duration.ofMillis(100),
             Duration.ofSeconds(10)
         );
-        Flow pluginDefaultFlow = flowRepository.findById(null, "io.kestra.tests.watch", "pluginDefault").orElseThrow();
+        Flow pluginDefaultFlow = flowRepository.findById(MAIN_TENANT, "io.kestra.tests.watch", "pluginDefault").orElseThrow();
         assertThat(pluginDefaultFlow.getTasks()).hasSize(1);
         assertThat(pluginDefaultFlow.getTasks().getFirst().getId()).isEqualTo("helloWithDefault");
         assertThat(pluginDefaultFlow.getTasks().getFirst().getType()).isEqualTo("io.kestra.plugin.core.log.Log");
@@ -122,7 +124,7 @@ class FileChangedEventListenerTest {
         // delete both files
         Files.delete(Path.of(FILE_WATCH + "/plugin-default.yaml"));
         Await.until(
-            () -> flowRepository.findById(null, "io.kestra.tests.watch", "pluginDefault").isEmpty(),
+            () -> flowRepository.findById(MAIN_TENANT, "io.kestra.tests.watch", "pluginDefault").isEmpty(),
             Duration.ofMillis(100),
             Duration.ofSeconds(10)
         );

cli/src/test/resources/application-file-watch.yml

@@ -2,6 +2,7 @@ micronaut:
   io:
     watch:
       enabled: true
+      tenantId: main
       paths:
         - build/file-watch
 

cli/src/test/resources/crudFlow/date.yml

@@ -0,0 +1,7 @@
+id: date
+namespace: io.kestra.cli
+
+tasks:
+  - id: date
+    type: io.kestra.plugin.core.debug.Return
+    format: "{{taskrun.startDate}}"

core/build.gradle

@@ -36,6 +36,7 @@ dependencies {
     implementation group: 'de.focus-shift', name: 'jollyday-jaxb'
     implementation 'nl.basjes.gitignore:gitignore-reader'
     implementation group: 'dev.failsafe', name: 'failsafe'
+    implementation 'com.github.ben-manes.caffeine:caffeine'
     api 'org.apache.httpcomponents.client5:httpclient5'
 
     // plugins
@@ -73,7 +74,7 @@ dependencies {
     testImplementation "io.micronaut:micronaut-http-server-netty"
     testImplementation "io.micronaut:micronaut-management"
 
-    testImplementation "org.testcontainers:testcontainers:1.20.6"
-    testImplementation "org.testcontainers:junit-jupiter:1.20.6"
+    testImplementation "org.testcontainers:testcontainers:1.21.1"
+    testImplementation "org.testcontainers:junit-jupiter:1.21.1"
     testImplementation "org.bouncycastle:bcpkix-jdk18on:1.80"
 }

core/src/main/java/io/kestra/core/cache/NoopCache.java

@@ -0,0 +1,92 @@
+package io.kestra.core.cache;
+
+import com.github.benmanes.caffeine.cache.Cache;
+import com.github.benmanes.caffeine.cache.Policy;
+import com.github.benmanes.caffeine.cache.stats.CacheStats;
+import org.jspecify.annotations.NonNull;
+import org.jspecify.annotations.Nullable;
+
+import java.util.Collections;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.function.Function;
+
+/**
+ * A No-Op implementation of a Caffeine Cache.
+ * Useful to disable caching but still use a cache to avoid if/else chains
+ */
+public class NoopCache<K, V> implements Cache<K, V> {
+    private static final ConcurrentMap<?, ?> EMPTY_MAP = new ConcurrentHashMap<>(0);
+
+    @Override
+    public @Nullable V getIfPresent(K key) {
+        return null;
+    }
+
+    @Override
+    public V get(K key, Function<? super K, ? extends V> mappingFunction) {
+        return mappingFunction.apply(key);
+    }
+
+    @Override
+    public Map<K, @NonNull V> getAllPresent(Iterable<? extends K> keys) {
+        return Collections.emptyMap();
+    }
+
+    @Override
+    public Map<K, @NonNull V> getAll(Iterable<? extends K> keys, Function<? super Set<? extends K>, ? extends Map<? extends K, ? extends @NonNull V>> mappingFunction) {
+        return Collections.emptyMap();
+    }
+
+    @Override
+    public void put(K key, @NonNull V value) {
+        // just do nothing
+    }
+
+    @Override
+    public void putAll(Map<? extends K, ? extends @NonNull V> map) {
+        // just do nothing
+    }
+
+    @Override
+    public void invalidate(K key) {
+        // just do nothing
+    }
+
+    @Override
+    public void invalidateAll(Iterable<? extends K> keys) {
+        // just do nothing
+    }
+
+    @Override
+    public void invalidateAll() {
+        // just do nothing
+    }
+
+    @Override
+    public long estimatedSize() {
+        return 0;
+    }
+
+    @Override
+    public CacheStats stats() {
+        return CacheStats.empty();
+    }
+
+    @Override
+    public ConcurrentMap<K, @NonNull V> asMap() {
+        return (ConcurrentMap<K, V>) EMPTY_MAP;
+    }
+
+    @Override
+    public void cleanUp() {
+        // just do nothing
+    }
+
+    @Override
+    public Policy<K, @NonNull V> policy() {
+        throw new UnsupportedOperationException();
+    }
+}

core/src/main/java/io/kestra/core/contexts/KestraBeansFactory.java

@@ -2,6 +2,7 @@ package io.kestra.core.contexts;
 
 import io.kestra.core.exceptions.KestraRuntimeException;
 import io.kestra.core.plugins.DefaultPluginRegistry;
+import io.kestra.core.plugins.PluginCatalogService;
 import io.kestra.core.plugins.PluginRegistry;
 import io.kestra.core.storages.StorageInterface;
 import io.kestra.core.storages.StorageInterfaceFactory;
@@ -13,6 +14,8 @@ import io.micronaut.context.annotation.Value;
 import io.micronaut.core.annotation.Nullable;
 import io.micronaut.core.convert.format.MapFormat;
 import io.micronaut.core.naming.conventions.StringConvention;
+import io.micronaut.http.client.HttpClient;
+import io.micronaut.http.client.annotation.Client;
 import jakarta.inject.Inject;
 import jakarta.inject.Singleton;
 import jakarta.validation.Validator;
@@ -35,6 +38,11 @@ public class KestraBeansFactory {
     @Value("${kestra.storage.type}")
     protected Optional<String> storageType;
 
+    @Singleton
+    public PluginCatalogService pluginCatalogService(@Client("api") HttpClient httpClient) {
+        return new PluginCatalogService(httpClient, false, true);
+    }
+
     @Requires(missingBeans = PluginRegistry.class)
     @Singleton
     public PluginRegistry pluginRegistry() {

core/src/main/java/io/kestra/core/contexts/KestraContext.java

@@ -2,11 +2,13 @@ package io.kestra.core.contexts;
 
 import io.kestra.core.models.ServerType;
 import io.kestra.core.plugins.PluginRegistry;
+import io.kestra.core.storages.StorageInterface;
 import io.kestra.core.utils.VersionProvider;
 import io.micronaut.context.ApplicationContext;
 import io.micronaut.context.annotation.Context;
 import io.micronaut.context.annotation.Requires;
 import io.micronaut.context.env.Environment;
+import io.micronaut.context.env.PropertySource;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -80,6 +82,8 @@ public abstract class KestraContext {
      */
     public abstract PluginRegistry getPluginRegistry();
 
+    public abstract StorageInterface getStorageInterface();
+
     /**
      * Shutdowns the Kestra application.
      */
@@ -146,7 +150,7 @@ public abstract class KestraContext {
                 .ifPresent(val -> configs.put(KESTRA_WORKER_GROUP_KEY, val));
 
             if (!configs.isEmpty()) {
-                environment.addPropertySource("kestra-runtime", configs);
+                environment.addPropertySource(PropertySource.of("kestra-runtime", configs));
             }
         }
 
@@ -172,5 +176,11 @@ public abstract class KestraContext {
             // Lazy init of the PluginRegistry.
             return this.applicationContext.getBean(PluginRegistry.class);
         }
+
+        @Override
+        public StorageInterface getStorageInterface() {
+            // Lazy init of the PluginRegistry.
+            return this.applicationContext.getBean(StorageInterface.class);
+        }
     }
 }

core/src/main/java/io/kestra/core/docs/AbstractClassDocumentation.java

@@ -1,9 +1,6 @@
 package io.kestra.core.docs;
 
 import com.google.common.base.CaseFormat;
-import io.kestra.core.models.Plugin;
-import io.kestra.core.models.tasks.retrys.AbstractRetry;
-import io.kestra.core.models.tasks.runners.TaskRunner;
 import lombok.AllArgsConstructor;
 import lombok.EqualsAndHashCode;
 import lombok.Getter;
@@ -59,7 +56,7 @@ public abstract class AbstractClassDocumentation<T> {
             .filter(entry -> (baseCls == null) || !entry.getKey().startsWith("io.kestra.core.models.flows.input."))
             .map(entry -> {
                 Map<String, Object> value = (Map<String, Object>) entry.getValue();
-                value.put("properties", flatten(properties(value), required(value), isTypeToKeep(entry.getKey())));
+                value.put("properties", flatten(properties(value), required(value), null));
 
                 return new AbstractMap.SimpleEntry<>(
                     entry.getKey(),
@@ -92,20 +89,13 @@ public abstract class AbstractClassDocumentation<T> {
         }
 
         if (this.propertiesSchema.containsKey("properties")) {
-            this.inputs = flatten(properties(this.propertiesSchema), required(this.propertiesSchema));
+            this.inputs = flattenWithoutType(properties(this.propertiesSchema), required(this.propertiesSchema));
         }
     }
 
-    protected static Map<String, Object> flatten(Map<String, Object> map, List<String> required) {
+    protected static Map<String, Object> flattenWithoutType(Map<String, Object> map, List<String> required) {
         map.remove("type");
-        return flatten(map, required, (String) null);
-    }
-
-    protected static Map<String, Object> flatten(Map<String, Object> map, List<String> required, Boolean keepType) {
-        if (!keepType) {
-            map.remove("type");
-        }
-        return flatten(map, required, (String) null);
+        return flatten(map, required, null);
     }
 
     @SuppressWarnings("unchecked")
@@ -141,23 +131,6 @@ public abstract class AbstractClassDocumentation<T> {
         return result;
     }
 
-    // Some task can have the `type` property but not to represent the task
-    // so we cant to keep it in the doc
-    private Boolean isTypeToKeep(String key){
-        try {
-            if (AbstractRetry.class.isAssignableFrom(Class.forName(key))) {
-                return true;
-            }
-
-            if (TaskRunner.class.isAssignableFrom(Class.forName(key))) {
-                return true;
-            }
-        } catch (ClassNotFoundException ignored) {
-            log.debug(ignored.getMessage(), ignored);
-        }
-        return false;
-    }
-
     protected static String flattenKey(String current, String parent) {
         return (parent != null ? parent + "." : "") + current;
     }

core/src/main/java/io/kestra/core/docs/ClassPluginDocumentation.java

@@ -1,9 +1,14 @@
 package io.kestra.core.docs;
 
 import io.kestra.core.plugins.PluginClassAndMetadata;
-import lombok.*;
+import lombok.AllArgsConstructor;
+import lombok.EqualsAndHashCode;
+import lombok.Getter;
+import lombok.ToString;
 
-import java.util.*;
+import java.util.List;
+import java.util.Map;
+import java.util.TreeMap;
 
 @Getter
 @EqualsAndHashCode
@@ -50,7 +55,7 @@ public class ClassPluginDocumentation<T> extends AbstractClassDocumentation<T> {
         }
 
         if (this.outputsSchema.containsKey("properties")) {
-            this.outputs = flatten(properties(this.outputsSchema), required(this.outputsSchema));
+            this.outputs = flattenWithoutType(properties(this.outputsSchema), required(this.outputsSchema));
         }
 
         // metrics

core/src/main/java/io/kestra/core/docs/DocumentationGenerator.java

@@ -7,6 +7,7 @@ import io.kestra.core.models.tasks.logs.LogExporter;
 import io.kestra.core.models.tasks.runners.TaskRunner;
 import io.kestra.core.models.tasks.Task;
 import io.kestra.core.models.triggers.AbstractTrigger;
+import io.kestra.core.plugins.AdditionalPlugin;
 import io.kestra.core.plugins.PluginClassAndMetadata;
 import io.kestra.core.plugins.RegisteredPlugin;
 import io.kestra.core.runners.pebble.Extension;
@@ -75,6 +76,7 @@ public class DocumentationGenerator {
         //noinspection unchecked
         result.addAll(this.generate(registeredPlugin, registeredPlugin.getTaskRunners(), (Class) TaskRunner.class, "task-runners"));
         result.addAll(this.generate(registeredPlugin, registeredPlugin.getLogExporters(), (Class) LogExporter.class, "log-exporters"));
+        result.addAll(this.generate(registeredPlugin, registeredPlugin.getAdditionalPlugins(), AdditionalPlugin.class, "additional-plugins"));
 
         result.addAll(guides(registeredPlugin));
 

core/src/main/java/io/kestra/core/docs/JsonSchemaGenerator.java

@@ -2,6 +2,7 @@ package io.kestra.core.docs;
 
 import com.fasterxml.classmate.ResolvedType;
 import com.fasterxml.classmate.members.HierarchicType;
+import com.fasterxml.jackson.annotation.JsonTypeInfo;
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
@@ -23,8 +24,10 @@ import io.kestra.core.models.annotations.PluginProperty;
 import io.kestra.core.models.conditions.Condition;
 import io.kestra.core.models.conditions.ScheduleCondition;
 import io.kestra.core.models.dashboards.DataFilter;
+import io.kestra.core.models.dashboards.DataFilterKPI;
 import io.kestra.core.models.dashboards.charts.Chart;
 import io.kestra.core.models.dashboards.charts.DataChart;
+import io.kestra.core.models.dashboards.charts.DataChartKPI;
 import io.kestra.core.models.property.Data;
 import io.kestra.core.models.property.Property;
 import io.kestra.core.models.tasks.Output;
@@ -33,6 +36,7 @@ import io.kestra.core.models.tasks.common.EncryptedString;
 import io.kestra.core.models.tasks.logs.LogExporter;
 import io.kestra.core.models.tasks.runners.TaskRunner;
 import io.kestra.core.models.triggers.AbstractTrigger;
+import io.kestra.core.plugins.AdditionalPlugin;
 import io.kestra.core.plugins.PluginRegistry;
 import io.kestra.core.plugins.RegisteredPlugin;
 import io.kestra.core.serializers.JacksonMapper;
@@ -54,6 +58,7 @@ import static io.kestra.core.serializers.JacksonMapper.MAP_TYPE_REFERENCE;
 @Singleton
 public class JsonSchemaGenerator {
     private static final List<Class<?>> TYPES_RESOLVED_AS_STRING = List.of(Duration.class, LocalTime.class, LocalDate.class, LocalDateTime.class, ZonedDateTime.class, OffsetDateTime.class, OffsetTime.class);
+    private static final List<Class<?>> SUBTYPE_RESOLUTION_EXCLUSION_FOR_PLUGIN_SCHEMA = List.of(Task.class, AbstractTrigger.class);
 
     private static final ObjectMapper MAPPER = JacksonMapper.ofJson().copy()
         .configure(SerializationFeature.WRITE_DURATIONS_AS_TIMESTAMPS, false);
@@ -127,6 +132,15 @@ public class JsonSchemaGenerator {
                 }
             }
         });
+
+        // do the same for all definitions
+        if (objectNode.get("definitions") instanceof ObjectNode definitions) {
+            definitions.forEach(jsonNode -> {
+                if (jsonNode instanceof ObjectNode definition) {
+                    removeRequiredOnPropsWithDefaults(definition);
+                }
+            });
+        }
     }
 
     // This hack exists because for Property we generate a anyOf for properties that are not strings.
@@ -237,15 +251,12 @@ public class JsonSchemaGenerator {
             .with(Option.DEFINITIONS_FOR_ALL_OBJECTS)
             .with(Option.DEFINITION_FOR_MAIN_SCHEMA)
             .with(Option.PLAIN_DEFINITION_KEYS)
-            .with(Option.ALLOF_CLEANUP_AT_THE_END);
+            .with(Option.ALLOF_CLEANUP_AT_THE_END);;
 
         if (!draft7) {
-            builder
-                .with(new JacksonModule(JacksonOption.IGNORE_TYPE_INFO_TRANSFORM))
-                .with(Option.MAP_VALUES_AS_ADDITIONAL_PROPERTIES);
+            builder.with(new JacksonModule(JacksonOption.IGNORE_TYPE_INFO_TRANSFORM));
         } else {
-            builder
-                .with(new JacksonModule());
+            builder.with(new JacksonModule());
         }
 
         // default value
@@ -345,6 +356,9 @@ public class JsonSchemaGenerator {
                 if (pluginPropertyAnnotation.internalStorageURI()) {
                     memberAttributes.put("$internalStorageURI", true);
                 }
+                if (!pluginPropertyAnnotation.group().isEmpty()) {
+                    memberAttributes.put("$group", pluginPropertyAnnotation.group());
+                }
             }
 
             Schema schema = member.getAnnotationConsideringFieldAndGetter(Schema.class);
@@ -436,8 +450,8 @@ public class JsonSchemaGenerator {
             return Object.class;
         });
 
-        // Subtype resolver for all plugins
         if (builder.build().getSchemaVersion() != SchemaVersion.DRAFT_2019_09) {
+            // Subtype resolver for all plugins
             builder.forTypesInGeneral()
                 .withSubtypeResolver((declaredType, context) -> {
                     TypeContext typeContext = context.getTypeContext();
@@ -510,21 +524,87 @@ public class JsonSchemaGenerator {
                     collectedTypeAttributes.remove("$examples");
                 }
             });
+        } else {
+            builder.forTypesInGeneral()
+                .withSubtypeResolver((declaredType, context) -> {
+                    TypeContext typeContext = context.getTypeContext();
 
-            // Ensure that `type` is defined as a constant in JSON Schema.
-            // The `const` property is used by editors for auto-completion based on that schema.
-            builder.forTypesInGeneral().withTypeAttributeOverride((collectedTypeAttributes, scope, context) -> {
-                final Class<?> pluginType = scope.getType().getErasedType();
-                if (pluginType.getAnnotation(Plugin.class) != null) {
-                    ObjectNode properties = (ObjectNode) collectedTypeAttributes.get("properties");
-                    if (properties != null) {
-                        properties.set("type", context.getGeneratorConfig().createObjectNode()
-                            .put("const", pluginType.getName())
-                        );
+                    if (SUBTYPE_RESOLUTION_EXCLUSION_FOR_PLUGIN_SCHEMA.contains(declaredType.getErasedType())) {
+                        return null;
                     }
-                }
-            });
+
+                    return this.subtypeResolver(declaredType, typeContext);
+                });
         }
+
+        // Ensure that `type` is defined as a constant in JSON Schema.
+        // The `const` property is used by editors for auto-completion based on that schema.
+        builder.forTypesInGeneral().withTypeAttributeOverride((collectedTypeAttributes, scope, context) -> {
+            final Class<?> pluginType = scope.getType().getErasedType();
+            if (pluginType.getAnnotation(Plugin.class) != null) {
+                ObjectNode properties = (ObjectNode) collectedTypeAttributes.get("properties");
+                if (properties != null) {
+                    properties.set("type", context.getGeneratorConfig().createObjectNode()
+                        .put("const", pluginType.getName())
+                    );
+                }
+            }
+        });
+
+        typeDefiningPropertiesToConst(builder);
+    }
+
+    /**
+     * Properties which are defining an implementation to choose among multiple ones (JsonTypeInfo.property) are simple String with default. We move them to be a "const": "defaultValue" instead
+     */
+    private void typeDefiningPropertiesToConst(SchemaGeneratorConfigBuilder builder) {
+        builder.forTypesInGeneral().withTypeAttributeOverride((collectedTypeAttributes, scope, context) -> {
+            final Class<?> targetType = scope.getType().getErasedType();
+            JsonTypeInfo jsonTypeInfo = Optional.ofNullable(targetType.getSuperclass()).map(c -> c.getAnnotation(JsonTypeInfo.class)).orElse(null);
+            if (jsonTypeInfo == null) {
+                return;
+            }
+
+            String property = jsonTypeInfo.property();
+            if (property == null) {
+                return;
+            }
+
+            ObjectNode properties = (ObjectNode) collectedTypeAttributes.get("properties");
+            if (properties == null) {
+                return;
+            }
+
+            String defaultValue = Optional.ofNullable(properties.get(property))
+                .flatMap(p -> {
+                    Optional<String> defaultOpt = p.optional("default").map(JsonNode::asText);
+                    if (defaultOpt.isPresent()) {
+                        return defaultOpt;
+                    }
+
+                    return p.optional("allOf").flatMap(node -> {
+                        if (node.isArray()) {
+                            Iterable<JsonNode> iterable = node::values;
+                            return StreamSupport.stream(
+                                    iterable.spliterator(),
+                                    false
+                                ).filter(subNode -> subNode.has("default"))
+                                .findFirst()
+                                .map(subNode -> subNode.get("default").asText());
+                        }
+
+                        return Optional.empty();
+                    });
+                })
+                .orElse(null);
+            if (defaultValue == null) {
+                return;
+            }
+
+            properties.set(property, context.getGeneratorConfig().createObjectNode()
+                .put("const", defaultValue)
+            );
+        });
     }
 
     private boolean isAssignableFromResolvedAsString(Class<?> declaredType) {
@@ -580,6 +660,16 @@ public class JsonSchemaGenerator {
                 .filter(Predicate.not(io.kestra.core.models.Plugin::isInternal))
                 .map(typeContext::resolve)
                 .toList();
+        } else if (AdditionalPlugin.class.isAssignableFrom(declaredType.getErasedType())) { // base type for addition plugin is not AdditionalPlugin but a subtype of AdditionalPlugin.
+            return getRegisteredPlugins()
+                .stream()
+                .flatMap(registeredPlugin -> registeredPlugin.getAdditionalPlugins().stream())
+                // for additional plugins, we have one subtype by type of additional plugins (for ex: embedding store for Langchain4J), so we need to filter on the correct subtype
+                .filter(cls -> declaredType.getErasedType().isAssignableFrom(cls))
+                .filter(cls -> cls != declaredType.getErasedType())
+                .filter(Predicate.not(io.kestra.core.models.Plugin::isInternal))
+                .map(typeContext::resolve)
+                .toList();
         } else if (declaredType.getErasedType() == Chart.class) {
             return getRegisteredPlugins()
                 .stream()
@@ -595,10 +685,25 @@ public class JsonSchemaGenerator {
 
                         TypeVariable<? extends Class<? extends Chart<?>>> dataFilterType = clz.getTypeParameters()[1];
                         ParameterizedType chartAwareColumnDescriptor = ((ParameterizedType) ((WildcardType) ((ParameterizedType) dataFilterType.getBounds()[0]).getActualTypeArguments()[1]).getUpperBounds()[0]);
+
                         dataFilters.forEach(dataFilter -> {
                             Type fieldsEnum = ((ParameterizedType) dataFilter.getGenericSuperclass()).getActualTypeArguments()[0];
                             consumer.accept(typeContext.resolve(clz, fieldsEnum, typeContext.resolve(dataFilter, typeContext.resolve(chartAwareColumnDescriptor, fieldsEnum))));
                         });
+                    } else if (DataChartKPI.class.isAssignableFrom(clz)) {
+                        List<Class<? extends DataFilterKPI<?, ?>>> dataFilterKPIs = getRegisteredPlugins()
+                            .stream()
+                            .flatMap(registeredPlugin -> registeredPlugin.getDataFiltersKPI().stream())
+                            .filter(Predicate.not(io.kestra.core.models.Plugin::isInternal))
+                            .toList();
+
+                        TypeVariable<? extends Class<? extends Chart<?>>> dataFilterType = clz.getTypeParameters()[1];
+                        ParameterizedType chartAwareColumnDescriptor = ((ParameterizedType) ((WildcardType) ((ParameterizedType) dataFilterType.getBounds()[0]).getActualTypeArguments()[1]).getUpperBounds()[0]);
+
+                        dataFilterKPIs.forEach(dataFilterKPI -> {
+                            Type fieldsEnum = ((ParameterizedType) dataFilterKPI.getGenericSuperclass()).getActualTypeArguments()[0];
+                            consumer.accept(typeContext.resolve(clz, fieldsEnum, typeContext.resolve(dataFilterKPI, typeContext.resolve(chartAwareColumnDescriptor, fieldsEnum))));
+                        });
                     } else {
                         consumer.accept(typeContext.resolve(clz));
                     }
@@ -642,10 +747,13 @@ public class JsonSchemaGenerator {
 
         this.build(builder, false);
 
-        // we don't return base properties unless specified with @PluginProperty
+        // we don't return base properties unless specified with @PluginProperty and hidden is false
         builder
             .forFields()
-            .withIgnoreCheck(fieldScope -> base != null && fieldScope.getAnnotation(PluginProperty.class) == null && fieldScope.getDeclaringType().getTypeName().equals(base.getName()));
+            .withIgnoreCheck(fieldScope -> base != null &&
+                (fieldScope.getAnnotation(PluginProperty.class) == null || fieldScope.getAnnotation(PluginProperty.class).hidden()) &&
+                fieldScope.getDeclaringType().getTypeName().equals(base.getName())
+            );
 
         SchemaGeneratorConfig schemaGeneratorConfig = builder.build();
 

core/src/main/java/io/kestra/core/docs/Plugin.java

@@ -37,6 +37,7 @@ public class Plugin {
     private List<String> charts;
     private List<String> dataFilters;
     private List<String> logExporters;
+    private List<String> additionalPlugins;
     private List<PluginSubGroup.PluginCategory> categories;
     private String subGroup;
 
@@ -89,17 +90,18 @@ public class Plugin {
         plugin.subGroup = subgroup;
 
         Predicate<Class<?>> packagePredicate = c -> subgroup == null || c.getPackageName().equals(subgroup);
-        plugin.tasks = filterAndGetClassName(registeredPlugin.getTasks(), includeDeprecated, packagePredicate).stream().toList();
-        plugin.triggers = filterAndGetClassName(registeredPlugin.getTriggers(), includeDeprecated, packagePredicate).stream().toList();
-        plugin.conditions = filterAndGetClassName(registeredPlugin.getConditions(), includeDeprecated, packagePredicate).stream().toList();
-        plugin.storages = filterAndGetClassName(registeredPlugin.getStorages(), includeDeprecated, packagePredicate).stream().toList();
-        plugin.secrets = filterAndGetClassName(registeredPlugin.getSecrets(), includeDeprecated, packagePredicate).stream().toList();
-        plugin.taskRunners = filterAndGetClassName(registeredPlugin.getTaskRunners(), includeDeprecated, packagePredicate).stream().toList();
-        plugin.apps = filterAndGetClassName(registeredPlugin.getApps(), includeDeprecated, packagePredicate).stream().toList();
-        plugin.appBlocks = filterAndGetClassName(registeredPlugin.getAppBlocks(), includeDeprecated, packagePredicate).stream().toList();
-        plugin.charts = filterAndGetClassName(registeredPlugin.getCharts(), includeDeprecated, packagePredicate).stream().toList();
-        plugin.dataFilters = filterAndGetClassName(registeredPlugin.getDataFilters(), includeDeprecated, packagePredicate).stream().toList();
-        plugin.logExporters = filterAndGetClassName(registeredPlugin.getLogExporters(), includeDeprecated, packagePredicate).stream().toList();
+        plugin.tasks = filterAndGetClassName(registeredPlugin.getTasks(), includeDeprecated, packagePredicate);
+        plugin.triggers = filterAndGetClassName(registeredPlugin.getTriggers(), includeDeprecated, packagePredicate);
+        plugin.conditions = filterAndGetClassName(registeredPlugin.getConditions(), includeDeprecated, packagePredicate);
+        plugin.storages = filterAndGetClassName(registeredPlugin.getStorages(), includeDeprecated, packagePredicate);
+        plugin.secrets = filterAndGetClassName(registeredPlugin.getSecrets(), includeDeprecated, packagePredicate);
+        plugin.taskRunners = filterAndGetClassName(registeredPlugin.getTaskRunners(), includeDeprecated, packagePredicate);
+        plugin.apps = filterAndGetClassName(registeredPlugin.getApps(), includeDeprecated, packagePredicate);
+        plugin.appBlocks = filterAndGetClassName(registeredPlugin.getAppBlocks(), includeDeprecated, packagePredicate);
+        plugin.charts = filterAndGetClassName(registeredPlugin.getCharts(), includeDeprecated, packagePredicate);
+        plugin.dataFilters = filterAndGetClassName(registeredPlugin.getDataFilters(), includeDeprecated, packagePredicate);
+        plugin.logExporters = filterAndGetClassName(registeredPlugin.getLogExporters(), includeDeprecated, packagePredicate);
+        plugin.additionalPlugins = filterAndGetClassName(registeredPlugin.getAdditionalPlugins(), includeDeprecated, packagePredicate);
 
         return plugin;
     }

core/src/main/java/io/kestra/core/docs/Schema.java

@@ -9,6 +9,9 @@ import java.util.Map;
 @NoArgsConstructor
 @AllArgsConstructor
 @Data
+@io.swagger.v3.oas.annotations.media.Schema(
+    name = "PluginSchema"
+)
 public class Schema {
     private Map<String, Object> properties;
     private Map<String, Object> outputs;

core/src/main/java/io/kestra/core/docs/SchemaType.java

@@ -11,6 +11,7 @@ public enum SchemaType {
     TRIGGER,
     PLUGINDEFAULT,
     APPS,
+    TESTSUITES,
     DASHBOARD;
 
     @JsonCreator

core/src/main/java/io/kestra/core/exceptions/FlowProcessingException.java

@@ -0,0 +1,24 @@
+package io.kestra.core.exceptions;
+
+import java.io.Serial;
+
+/**
+ * Exception class for all problems encountered when processing (parsing, injecting defaults, validating) a flow.
+ */
+public class FlowProcessingException extends KestraException {
+
+    @Serial
+    private static final long serialVersionUID = 1L;
+    
+    public FlowProcessingException(String message) {
+        super(message);
+    }
+
+    public FlowProcessingException(String message, Throwable cause) {
+        super(message, cause);
+    }
+
+    public FlowProcessingException(Throwable cause) {
+        super(cause);
+    }
+}

core/src/main/java/io/kestra/core/exceptions/KestraException.java

@@ -0,0 +1,27 @@
+package io.kestra.core.exceptions;
+
+import java.io.Serial;
+
+/**
+ * The top-level {@link KestraException}..
+ */
+public class KestraException extends Exception {
+    @Serial
+    private static final long serialVersionUID = 1L;
+
+    public KestraException() {
+    }
+
+    public KestraException(String message) {
+        super(message);
+    }
+
+    public KestraException(String message, Throwable cause) {
+        super(message, cause);
+    }
+
+    public KestraException(Throwable cause) {
+        super(cause);
+    }
+
+}

core/src/main/java/io/kestra/core/http/client/HttpClient.java

@@ -155,6 +155,14 @@ public class HttpClient implements Closeable {
             builder.addResponseInterceptorLast(new FailedResponseInterceptor());
         }
 
+        if (this.configuration.getAllowedResponseCodes() != null) {
+            List<Integer> list = runContext.render(this.configuration.getAllowedResponseCodes()).asList(Integer.class);
+
+            if (!list.isEmpty()) {
+                builder.addResponseInterceptorLast(new FailedResponseInterceptor(list));
+            }
+        }
+
         builder.addResponseInterceptorLast(new RunContextResponseInterceptor(this.runContext));
 
         // builder object

core/src/main/java/io/kestra/core/http/client/apache/FailedResponseInterceptor.java

@@ -3,7 +3,6 @@ package io.kestra.core.http.client.apache;
 import io.kestra.core.http.HttpResponse;
 import io.kestra.core.http.HttpService;
 import io.kestra.core.http.client.HttpClientResponseException;
-import lombok.AllArgsConstructor;
 import org.apache.hc.core5.http.EntityDetails;
 import org.apache.hc.core5.http.HttpEntityContainer;
 import org.apache.hc.core5.http.HttpException;
@@ -12,22 +11,43 @@ import org.apache.hc.core5.http.protocol.HttpContext;
 
 import java.io.IOException;
 import java.nio.charset.StandardCharsets;
+import java.util.List;
 
-@AllArgsConstructor
 public class FailedResponseInterceptor implements HttpResponseInterceptor {
+    private final boolean allErrors;
+    private List<Integer> statusCodes;
+
+    public FailedResponseInterceptor() {
+        this.allErrors = true;
+    }
+
+    public FailedResponseInterceptor(List<Integer> statusCodes) {
+        this.statusCodes = statusCodes;
+        this.allErrors = false;
+    }
+
+
     @Override
     public void process(org.apache.hc.core5.http.HttpResponse response, EntityDetails entity, HttpContext context) throws HttpException, IOException {
-        if (response.getCode() >= 400) {
-            String error = "Failed http request with response code '" + response.getCode() + "'";
+        if (this.allErrors && response.getCode() >= 400) {
+            this.raiseError(response, context);
+        }
 
-            if (response instanceof HttpEntityContainer httpEntity && httpEntity.getEntity() != null) {
-                HttpService.HttpEntityCopy copy = HttpService.copy(httpEntity.getEntity());
-                httpEntity.setEntity(copy);
-
-                error += " and body:\n" + new String(copy.getBody(), StandardCharsets.UTF_8);
-            }
-
-            throw new HttpClientResponseException(error, HttpResponse.from(response, context));
+        if (this.statusCodes != null && !this.statusCodes.contains(response.getCode())) {
+            this.raiseError(response, context);
         }
     }
+
+    private void raiseError(org.apache.hc.core5.http.HttpResponse response, HttpContext context) throws IOException, HttpClientResponseException {
+        String error = "Failed http request with response code '" + response.getCode() + "'";
+
+        if (response instanceof HttpEntityContainer httpEntity && httpEntity.getEntity() != null) {
+            HttpService.HttpEntityCopy copy = HttpService.copy(httpEntity.getEntity());
+            httpEntity.setEntity(copy);
+
+            error += " and body:\n" + new String(copy.getBody(), StandardCharsets.UTF_8);
+        }
+
+        throw new HttpClientResponseException(error, HttpResponse.from(response, context));
+    }
 }

core/src/main/java/io/kestra/core/http/client/configurations/HttpConfiguration.java

@@ -13,6 +13,7 @@ import java.net.Proxy;
 import java.nio.charset.Charset;
 import java.nio.charset.StandardCharsets;
 import java.time.Duration;
+import java.util.List;
 
 @Builder(toBuilder = true)
 @Getter
@@ -35,16 +36,20 @@ public class HttpConfiguration {
 
     @Schema(title = "Whether redirects should be followed automatically.")
     @Builder.Default
-    private Property<Boolean> followRedirects = Property.of(true);
+    private Property<Boolean> followRedirects = Property.ofValue(true);
 
     @Setter
     @Schema(title = "If true, allow a failed response code (response code >= 400)")
     @Builder.Default
-    private Property<Boolean> allowFailed = Property.of(false);
+    private Property<Boolean> allowFailed = Property.ofValue(false);
+
+    @Setter
+    @Schema(title = "List of response code allowed for this request")
+    private Property<List<Integer>> allowedResponseCodes;
 
     @Schema(title = "The default charset for the request.")
     @Builder.Default
-    private final Property<Charset> defaultCharset = Property.of(StandardCharsets.UTF_8);
+    private final Property<Charset> defaultCharset = Property.ofValue(StandardCharsets.UTF_8);
 
     @Schema(title = "The enabled log.")
     @PluginProperty
@@ -121,7 +126,7 @@ public class HttpConfiguration {
             }
 
             this.timeout = this.timeout.toBuilder()
-                .connectTimeout(Property.of(connectTimeout))
+                .connectTimeout(Property.ofValue(connectTimeout))
                 .build();
 
             return this;
@@ -135,7 +140,7 @@ public class HttpConfiguration {
             }
 
             this.timeout = this.timeout.toBuilder()
-                .readIdleTimeout(Property.of(readTimeout))
+                .readIdleTimeout(Property.ofValue(readTimeout))
                 .build();
 
             return this;
@@ -150,7 +155,7 @@ public class HttpConfiguration {
             }
 
             this.proxy = this.proxy.toBuilder()
-                .type(Property.of(proxyType))
+                .type(Property.ofValue(proxyType))
                 .build();
 
             return this;
@@ -164,7 +169,7 @@ public class HttpConfiguration {
             }
 
             this.proxy = this.proxy.toBuilder()
-                .address(Property.of(proxyAddress))
+                .address(Property.ofValue(proxyAddress))
                 .build();
 
             return this;
@@ -178,7 +183,7 @@ public class HttpConfiguration {
             }
 
             this.proxy = this.proxy.toBuilder()
-                .port(Property.of(proxyPort))
+                .port(Property.ofValue(proxyPort))
                 .build();
 
             return this;
@@ -192,7 +197,7 @@ public class HttpConfiguration {
             }
 
             this.proxy = this.proxy.toBuilder()
-                .username(Property.of(proxyUsername))
+                .username(Property.ofValue(proxyUsername))
                 .build();
 
             return this;
@@ -206,7 +211,7 @@ public class HttpConfiguration {
             }
 
             this.proxy = this.proxy.toBuilder()
-                .password(Property.of(proxyPassword))
+                .password(Property.ofValue(proxyPassword))
                 .build();
 
             return this;
@@ -222,7 +227,7 @@ public class HttpConfiguration {
             }
 
             this.auth = ((BasicAuthConfiguration) this.auth).toBuilder()
-                .username(Property.of(basicAuthUser))
+                .username(Property.ofValue(basicAuthUser))
                 .build();
 
             return this;
@@ -237,7 +242,7 @@ public class HttpConfiguration {
             }
 
             this.auth = ((BasicAuthConfiguration) this.auth).toBuilder()
-                .password(Property.of(basicAuthPassword))
+                .password(Property.ofValue(basicAuthPassword))
                 .build();
 
             return this;

core/src/main/java/io/kestra/core/http/client/configurations/ProxyConfiguration.java

@@ -14,7 +14,7 @@ import java.net.Proxy;
 public class ProxyConfiguration {
     @Schema(title = "The type of proxy to use.")
     @Builder.Default
-    private final Property<java.net.Proxy.Type> type = Property.of(Proxy.Type.DIRECT);
+    private final Property<java.net.Proxy.Type> type = Property.ofValue(Proxy.Type.DIRECT);
 
     @Schema(title = "The address of the proxy server.")
     private final Property<String> address;

core/src/main/java/io/kestra/core/http/client/configurations/TimeoutConfiguration.java

@@ -15,5 +15,5 @@ public class TimeoutConfiguration {
 
     @Schema(title = "The time allowed for a read connection to remain idle before closing it.")
     @Builder.Default
-    Property<Duration> readIdleTimeout = Property.of(Duration.ofMinutes(5));
+    Property<Duration> readIdleTimeout = Property.ofValue(Duration.ofMinutes(5));
 }

core/src/main/java/io/kestra/core/log/KestraLogFilter.java

@@ -12,6 +12,7 @@ public class KestraLogFilter extends EventEvaluatorBase<ILoggingEvent> {
         // we use startWith and do all checks successfully instead of using a more elegant construct like Stream...
         return message.startsWith("outOfOrder mode is active. Migration of schema") ||
             message.startsWith("Version mismatch         : Database version is older than what dialect POSTGRES supports") ||
-            message.startsWith("Failed to bind as java.util.concurrent.Executors$AutoShutdownDelegatedExecutorService is unsupported.");
+            message.startsWith("Failed to bind as java.util.concurrent.Executors$AutoShutdownDelegatedExecutorService is unsupported.") ||
+            message.startsWith("The cache 'default' is not recording statistics.");
     }
 }

core/src/main/java/io/kestra/core/metrics/MetricRegistry.java

@@ -1,16 +1,15 @@
 package io.kestra.core.metrics;
 
 import io.kestra.core.models.executions.Execution;
+import io.kestra.core.models.executions.ExecutionKilled;
 import io.kestra.core.models.tasks.Task;
 import io.kestra.core.models.triggers.AbstractTrigger;
 import io.kestra.core.models.triggers.TriggerContext;
-import io.kestra.core.runners.SubflowExecutionResult;
-import io.kestra.core.runners.WorkerTask;
-import io.kestra.core.runners.WorkerTaskResult;
-import io.kestra.core.runners.WorkerTrigger;
+import io.kestra.core.runners.*;
 import io.kestra.core.schedulers.SchedulerExecutionWithTrigger;
 import io.micrometer.core.instrument.*;
 import io.micrometer.core.instrument.binder.MeterBinder;
+import io.micrometer.core.instrument.search.Search;
 import jakarta.inject.Inject;
 import jakarta.inject.Singleton;
 import lombok.extern.slf4j.Slf4j;
@@ -20,49 +19,119 @@ import org.apache.commons.lang3.ArrayUtils;
 @Slf4j
 public class MetricRegistry {
     public static final String METRIC_WORKER_JOB_PENDING_COUNT = "worker.job.pending";
+    public static final String METRIC_WORKER_JOB_PENDING_COUNT_DESCRIPTION = "The number of jobs (tasks or triggers) pending to be run by the Worker";
     public static final String METRIC_WORKER_JOB_RUNNING_COUNT = "worker.job.running";
+    public static final String METRIC_WORKER_JOB_RUNNING_COUNT_DESCRIPTION = "The number of jobs (tasks or triggers) currently running inside the Worker";
     public static final String METRIC_WORKER_JOB_THREAD_COUNT = "worker.job.thread";
+    public static final String METRIC_WORKER_JOB_THREAD_COUNT_DESCRIPTION = "The number of worker threads";
     public static final String METRIC_WORKER_RUNNING_COUNT = "worker.running.count";
+    public static final String METRIC_WORKER_RUNNING_COUNT_DESCRIPTION = "The number of tasks currently running inside the Worker";
     public static final String METRIC_WORKER_QUEUED_DURATION = "worker.queued.duration";
+    public static final String METRIC_WORKER_QUEUED_DURATION_DESCRIPTION = "Task queued duration inside the Worker";
     public static final String METRIC_WORKER_STARTED_COUNT = "worker.started.count";
+    public static final String METRIC_WORKER_STARTED_COUNT_DESCRIPTION = "The total number of tasks started by the Worker";
     public static final String METRIC_WORKER_TIMEOUT_COUNT = "worker.timeout.count";
+    public static final String METRIC_WORKER_TIMEOUT_COUNT_DESCRIPTION = "The total number of tasks that timeout inside the Worker";
     public static final String METRIC_WORKER_ENDED_COUNT = "worker.ended.count";
+    public static final String METRIC_WORKER_ENDED_COUNT_DESCRIPTION = "The total number of tasks ended by the Worker";
     public static final String METRIC_WORKER_ENDED_DURATION = "worker.ended.duration";
+    public static final String METRIC_WORKER_ENDED_DURATION_DESCRIPTION = "Task run duration inside the Worker";
     public static final String METRIC_WORKER_TRIGGER_DURATION = "worker.trigger.duration";
+    public static final String METRIC_WORKER_TRIGGER_DURATION_DESCRIPTION = "Trigger evaluation duration inside the Worker";
     public static final String METRIC_WORKER_TRIGGER_RUNNING_COUNT = "worker.trigger.running.count";
+    public static final String METRIC_WORKER_TRIGGER_RUNNING_COUNT_DESCRIPTION = "The number of triggers currently evaluating inside the Worker";
     public static final String METRIC_WORKER_TRIGGER_STARTED_COUNT = "worker.trigger.started.count";
+    public static final String METRIC_WORKER_TRIGGER_STARTED_COUNT_DESCRIPTION = "The total number of trigger evaluations started by the Worker";
     public static final String METRIC_WORKER_TRIGGER_ENDED_COUNT = "worker.trigger.ended.count";
+    public static final String METRIC_WORKER_TRIGGER_ENDED_COUNT_DESCRIPTION = "The total number of trigger evaluations ended by the Worker";
     public static final String METRIC_WORKER_TRIGGER_ERROR_COUNT = "worker.trigger.error.count";
+    public static final String METRIC_WORKER_TRIGGER_ERROR_COUNT_DESCRIPTION = "The total number of trigger evaluations that failed inside the Worker";
     public static final String METRIC_WORKER_TRIGGER_EXECUTION_COUNT = "worker.trigger.execution.count";
+    public static final String METRIC_WORKER_TRIGGER_EXECUTION_COUNT_DESCRIPTION = "The total number of triggers evaluated by the Worker";
+    public static final String METRIC_WORKER_KILLED_COUNT = "worker.killed.count";
+    public static final String METRIC_WORKER_KILLED_COUNT_DESCRIPTION = "The total number of executions killed events received the Executor";
 
-    public static final String EXECUTOR_TASKRUN_NEXT_COUNT = "executor.taskrun.next.count";
-    public static final String EXECUTOR_TASKRUN_ENDED_COUNT = "executor.taskrun.ended.count";
-    public static final String EXECUTOR_TASKRUN_ENDED_DURATION = "executor.taskrun.ended.duration";
-    public static final String EXECUTOR_WORKERTASKRESULT_COUNT = "executor.workertaskresult.count";
-    public static final String EXECUTOR_EXECUTION_STARTED_COUNT = "executor.execution.started.count";
-    public static final String EXECUTOR_EXECUTION_END_COUNT = "executor.execution.end.count";
-    public static final String EXECUTOR_EXECUTION_DURATION = "executor.execution.duration";
+    public static final String METRIC_EXECUTOR_THREAD_COUNT = "executor.thread.count";
+    public static final String METRIC_EXECUTOR_THREAD_COUNT_DESCRIPTION = "The number of executor threads";
+    public static final String METRIC_EXECUTOR_TASKRUN_CREATED_COUNT = "executor.taskrun.created.count";
+    public static final String METRIC_EXECUTOR_TASKRUN_CREATED_COUNT_DESCRIPTION = "The total number of tasks created by the Executor";
+    public static final String METRIC_EXECUTOR_TASKRUN_ENDED_COUNT = "executor.taskrun.ended.count";
+    public static final String METRIC_EXECUTOR_TASKRUN_ENDED_COUNT_DESCRIPTION = "The total number of tasks ended by the Executor";
+    public static final String METRIC_EXECUTOR_TASKRUN_ENDED_DURATION = "executor.taskrun.ended.duration";
+    public static final String METRIC_EXECUTOR_TASKRUN_ENDED_DURATION_DESCRIPTION = "Task duration inside the Executor";
+    public static final String METRIC_EXECUTOR_FLOWABLE_EXECUTION_COUNT = "executor.flowable.execution.count";
+    public static final String METRIC_EXECUTOR_FLOWABLE_EXECUTION_COUNT_DESCRIPTION = "The total number of flowable tasks executed by the Executor";
+    public static final String METRIC_EXECUTOR_EXECUTION_STARTED_COUNT = "executor.execution.started.count";
+    public static final String METRIC_EXECUTOR_EXECUTION_STARTED_COUNT_DESCRIPTION = "The total number of executions started by the Executor";
+    public static final String METRIC_EXECUTOR_EXECUTION_END_COUNT = "executor.execution.end.count";
+    public static final String METRIC_EXECUTOR_EXECUTION_END_COUNT_DESCRIPTION = "The total number of executions ended by the Executor";
+    public static final String METRIC_EXECUTOR_EXECUTION_DURATION = "executor.execution.duration";
+    public static final String METRIC_EXECUTOR_EXECUTION_DURATION_DESCRIPTION = "Execution duration inside the Executor";
+    public static final String METRIC_EXECUTOR_EXECUTION_MESSAGE_PROCESS_DURATION = "executor.execution.message.process";
+    public static final String METRIC_EXECUTOR_EXECUTION_MESSAGE_PROCESS_DURATION_DESCRIPTION = "Duration of a single execution message processed by the Executor";
+    public static final String METRIC_EXECUTOR_KILLED_COUNT = "executor.killed.count";
+    public static final String METRIC_EXECUTOR_KILLED_COUNT_DESCRIPTION = "The total number of executions killed events received the Executor";
+    public static final String METRIC_EXECUTOR_SLA_EXPIRED_COUNT = "executor.sla.expired.count";
+    public static final String METRIC_EXECUTOR_SLA_EXPIRED_COUNT_DESCRIPTION = "The total number of expired SLA (i.e. executions with SLA of type MAX_DURATION that took longer than the SLA) evaluated by the Executor";
+    public static final String METRIC_EXECUTOR_SLA_VIOLATION_COUNT = "executor.sla.violation.count";
+    public static final String METRIC_EXECUTOR_SLA_VIOLATION_COUNT_DESCRIPTION = "The total number of expired SLA (i.e. executions with SLA of type MAX_DURATION that took longer than the SLA) evaluated by the Executor";
+    public static final String METRIC_EXECUTOR_EXECUTION_DELAY_CREATED_COUNT = "executor.execution.delay.created.count";
+    public static final String METRIC_EXECUTOR_EXECUTION_DELAY_CREATED_COUNT_DESCRIPTION = "The total number of execution delays created by the Executor";
+    public static final String METRIC_EXECUTOR_EXECUTION_DELAY_ENDED_COUNT = "executor.execution.delay.ended.count";
+    public static final String METRIC_EXECUTOR_EXECUTION_DELAY_ENDED_COUNT_DESCRIPTION = "The total number of execution delays ended (resumed) by the Executor";
+    public static final String METRIC_EXECUTOR_WORKER_JOB_RESUBMIT_COUNT = "executor.worker.job.resubmit.count";
+    public static final String METRIC_EXECUTOR_WORKER_JOB_RESUBMIT_COUNT_DESCRIPTION = "The total number of worker jobs resubmitted to the Worker by the Executor";
+    public static final String METRIC_EXECUTOR_EXECUTION_QUEUED_COUNT = "executor.execution.queued.count";
+    public static final String METRIC_EXECUTOR_EXECUTION_QUEUED_COUNT_DESCRIPTION = "The total number of executions queued by the Executor";
+    public static final String METRIC_EXECUTOR_EXECUTION_POPPED_COUNT = "executor.execution.popped.count";
+    public static final String METRIC_EXECUTOR_EXECUTION_POPPED_COUNT_DESCRIPTION = "The total number of executions popped by the Executor";
 
     public static final String METRIC_INDEXER_REQUEST_COUNT = "indexer.request.count";
+    public static final String METRIC_INDEXER_REQUEST_COUNT_DESCRIPTION = "Total number of batches of records received by the Indexer";
     public static final String METRIC_INDEXER_REQUEST_DURATION = "indexer.request.duration";
+    public static final String METRIC_INDEXER_REQUEST_DURATION_DESCRIPTION = "Batch of records duration inside the Indexer";
     public static final String METRIC_INDEXER_REQUEST_RETRY_COUNT = "indexer.request.retry.count";
+    public static final String METRIC_INDEXER_REQUEST_RETRY_COUNT_DESCRIPTION = "Total number of batches of records retried by the Indexer";
     public static final String METRIC_INDEXER_SERVER_DURATION = "indexer.server.duration";
+    public static final String METRIC_INDEXER_SERVER_DURATION_DESCRIPTION = "Batch of records indexation duration";
     public static final String METRIC_INDEXER_MESSAGE_FAILED_COUNT = "indexer.message.failed.count";
+    public static final String METRIC_INDEXER_MESSAGE_FAILED_COUNT_DESCRIPTION = "Total number of records which failed to be indexed by the Indexer";
     public static final String METRIC_INDEXER_MESSAGE_IN_COUNT = "indexer.message.in.count";
+    public static final String METRIC_INDEXER_MESSAGE_IN_COUNT_DESCRIPTION = "Total number of records received by the Indexer";
     public static final String METRIC_INDEXER_MESSAGE_OUT_COUNT = "indexer.message.out.count";
+    public static final String METRIC_INDEXER_MESSAGE_OUT_COUNT_DESCRIPTION = "Total number of records indexed by the Indexer";
 
-    public static final String SCHEDULER_LOOP_COUNT = "scheduler.loop.count";
-    public static final String SCHEDULER_TRIGGER_COUNT = "scheduler.trigger.count";
-    public static final String SCHEDULER_TRIGGER_DELAY_DURATION = "scheduler.trigger.delay.duration";
-    public static final String SCHEDULER_EVALUATE_COUNT = "scheduler.evaluate.count";
-    public static final String SCHEDULER_EXECUTION_RUNNING_DURATION = "scheduler.execution.running.duration";
-    public static final String SCHEDULER_EXECUTION_MISSING_DURATION = "scheduler.execution.missing.duration";
+    public static final String METRIC_SCHEDULER_LOOP_COUNT = "scheduler.loop.count";
+    public static final String METRIC_SCHEDULER_LOOP_COUNT_DESCRIPTION = "Total number of evaluation loops executed by the Scheduler";
+    public static final String METRIC_SCHEDULER_TRIGGER_EVALUATION_DURATION = "scheduler.trigger.evaluation.duration";
+    public static final String METRIC_SCHEDULER_TRIGGER_EVALUATION_DURATION_DESCRIPTION = "Trigger evaluation duration for trigger executed inside the Scheduler (Schedulable triggers)";
+    public static final String METRIC_SCHEDULER_TRIGGER_COUNT = "scheduler.trigger.count";
+    public static final String METRIC_SCHEDULER_TRIGGER_COUNT_DESCRIPTION = "Total number of executions triggered by the Scheduler";
+    public static final String METRIC_SCHEDULER_TRIGGER_DELAY_DURATION = "scheduler.trigger.delay.duration";
+    public static final String METRIC_SCHEDULER_TRIGGER_DELAY_DURATION_DESCRIPTION = "Trigger delay duration inside the Scheduler";
+    public static final String METRIC_SCHEDULER_EVALUATE_COUNT = "scheduler.evaluate.count";
+    public static final String METRIC_SCHEDULER_EVALUATE_COUNT_DESCRIPTION = "Total number of triggers evaluated by the Scheduler";
+    public static final String METRIC_SCHEDULER_EXECUTION_LOCK_DURATION = "scheduler.execution.lock.duration";
+    public static final String METRIC_SCHEDULER_EXECUTION_LOCK_DURATION_DESCRIPTION = "Trigger lock duration waiting for an execution to be terminated";
+    public static final String METRIC_SCHEDULER_EXECUTION_MISSING_DURATION = "scheduler.execution.missing.duration";
+    public static final String METRIC_SCHEDULER_EXECUTION_MISSING_DURATION_DESCRIPTION = "Missing execution duration inside the Scheduler. A missing execution is an execution that was triggered by the Scheduler but not yet started by the Executor";
+    public static final String METRIC_SCHEDULER_EVALUATION_LOOP_DURATION = "scheduler.evaluation.loop.duration";
+    public static final String METRIC_SCHEDULER_EVALUATION_LOOP_DURATION_DESCRIPTION = "Trigger evaluation loop duration inside the Scheduler";
 
-    public static final String STREAMS_STATE_COUNT = "stream.state.count";
+    public static final String METRIC_STREAMS_STATE_COUNT = "stream.state.count";
+    public static final String METRIC_STREAMS_STATE_COUNT_DESCRIPTION = "Number of Kafka Stream applications by state";
 
-    public static final String JDBC_QUERY_DURATION = "jdbc.query.duration";
+    public static final String METRIC_JDBC_QUERY_DURATION = "jdbc.query.duration";
+    public static final String METRIC_JDBC_QUERY_DURATION_DESCRIPTION = "Duration of database queries";
 
-    public static final String QUEUE_BIG_MESSAGE_COUNT = "queue.big_message.count";
+    public static final String METRIC_QUEUE_BIG_MESSAGE_COUNT = "queue.big_message.count";
+    public static final String METRIC_QUEUE_BIG_MESSAGE_COUNT_DESCRIPTION = "Total number of big messages";
+    public static final String METRIC_QUEUE_PRODUCE_COUNT = "queue.produce.count";
+    public static final String METRIC_QUEUE_PRODUCE_COUNT_DESCRIPTION = "Total number of produced messages";
+    public static final String METRIC_QUEUE_RECEIVE_DURATION = "queue.receive.duration";
+    public static final String METRIC_QUEUE_RECEIVE_DURATION_DESCRIPTION = "Queue duration to receive and consume a batch of messages";
+    public static final String METRIC_QUEUE_POLL_SIZE = "queue.poll.size";
+    public static final String METRIC_QUEUE_POLL_SIZE_DESCRIPTION = "Size of a poll to the queue (message batch size)";
 
     public static final String TAG_TASK_TYPE = "task_type";
     public static final String TAG_TRIGGER_TYPE = "trigger_type";
@@ -73,6 +142,10 @@ public class MetricRegistry {
     public static final String TAG_WORKER_GROUP = "worker_group";
     public static final String TAG_TENANT_ID = "tenant_id";
     public static final String TAG_CLASS_NAME = "class_name";
+    public static final String TAG_EXECUTION_KILLED_TYPE = "execution_killed_type";
+    public static final String TAG_QUEUE_CONSUMER = "consumer";
+    public static final String TAG_QUEUE_CONSUMER_GROUP = "consumer_group";
+    public static final String TAG_QUEUE_TYPE = "queue_type";
 
     @Inject
     private MeterRegistry meterRegistry;
@@ -84,47 +157,72 @@ public class MetricRegistry {
      * Tracks a monotonically increasing value.
      *
      * @param name The base metric name
+     * @param description The metric description
      * @param tags MUST be an even number of arguments representing key/value pairs of tags.
      * @return A new or existing counter.
      */
-    public Counter counter(String name, String... tags) {
-        return this.meterRegistry.counter(metricName(name), tags);
+    public Counter counter(String name, String description, String... tags) {
+        return Counter.builder(metricName(name))
+            .description(description)
+            .tags(tags)
+            .register(this.meterRegistry);
     }
 
     /**
      * Register a gauge that reports the value of the {@link Number}.
      *
      * @param name   Name of the gauge being registered.
+     * @param description The metric description
      * @param number Thread-safe implementation of {@link Number} used to access the value.
      * @param tags   Sequence of dimensions for breaking down the name.
      * @param <T>    The type of the number from which the gauge value is extracted.
      * @return The number that was passed in so the registration can be done as part of an assignment
      * statement.
      */
-    public <T extends Number> T gauge(String name, T number, String... tags) {
-        return this.meterRegistry.gauge(metricName(name), Tags.of(tags), number);
+    public <T extends Number> T gauge(String name, String description, T number, String... tags) {
+        Gauge.builder(metricName(name), () -> number)
+            .description(description)
+            .tags(tags)
+            .register(this.meterRegistry);
+        return number;
     }
 
     /**
      * Measures the time taken for short tasks and the count of these tasks.
      *
      * @param name The base metric name
+     * @param description The metric description
      * @param tags MUST be an even number of arguments representing key/value pairs of tags.
      * @return A new or existing timer.
      */
-    public Timer timer(String name, String... tags) {
-        return this.meterRegistry.timer(metricName(name), tags);
+    public Timer timer(String name, String description, String... tags) {
+        return Timer.builder(metricName(name))
+            .description(description)
+            .tags(tags)
+            .register(this.meterRegistry);
     }
 
     /**
      * Measures the distribution of samples.
      *
      * @param name The base metric name
+     * @param description The metric description
      * @param tags MUST be an even number of arguments representing key/value pairs of tags.
      * @return A new or existing distribution summary.
      */
-    public DistributionSummary summary(String name, String... tags) {
-        return this.meterRegistry.summary(metricName(name), tags);
+    public DistributionSummary summary(String name, String description, String... tags) {
+        return DistributionSummary.builder(metricName(name))
+            .description(description)
+            .tags(tags)
+            .register(this.meterRegistry);
+    }
+
+    /**
+     * Search for an existing Meter in the meter registry
+     * @param name The base metric name
+     */
+    public Search find(String name) {
+        return this.meterRegistry.find(metricName(name));
     }
 
     /**
@@ -260,7 +358,7 @@ public class MetricRegistry {
      * Return tags for current {@link AbstractTrigger}
      *
      * @param trigger the current Trigger
-     * @return tags to applied to metrics
+     * @return tags to apply to metrics
      */
     public String[] tags(AbstractTrigger trigger) {
         return new String[]{
@@ -310,6 +408,19 @@ public class MetricRegistry {
         );
     }
 
+    /**
+     * Return tags for current {@link ExecutionKilled}
+     *
+     * @param executionKilled the current Trigger
+     * @return tags to apply to metrics
+     */
+    public String[] tags(ExecutionKilled executionKilled) {
+        var baseTags = new String[]{
+            TAG_EXECUTION_KILLED_TYPE, executionKilled.getType(),
+        };
+        return executionKilled.getTenantId() == null ? baseTags : ArrayUtils.addAll(baseTags, TAG_TENANT_ID, executionKilled.getTenantId());
+    }
+
 
     /**
      * Return globals tags

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

@@ -20,6 +20,7 @@ public record Label(@NotNull String key, @NotNull String value) {
     public static final String REPLAY = SYSTEM_PREFIX + "replay";
     public static final String REPLAYED = SYSTEM_PREFIX + "replayed";
     public static final String SIMULATED_EXECUTION = SYSTEM_PREFIX + "simulatedExecution";
+    public static final String TEST = SYSTEM_PREFIX + "test";
 
     /**
      * Static helper method for converting a list of labels to a nested map.

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

@@ -43,9 +43,13 @@ public record QueryFilter(
         STARTS_WITH,
         ENDS_WITH,
         CONTAINS,
-        REGEX;
+        REGEX,
+        PREFIX
     }
 
+    private List<Object> asValues(Object value) {
+        return value instanceof String valueStr ? Arrays.asList(valueStr.split(",")) : (List<Object>) value;
+    }
 
     @SuppressWarnings("unchecked")
     public <T extends Enum<T>> AbstractFilter<T> toDashboardFilterBuilder(T field, Object value) {
@@ -63,9 +67,9 @@ public record QueryFilter(
             case LESS_THAN_OR_EQUAL_TO:
                 return LessThanOrEqualTo.<T>builder().field(field).value(value).build();
             case IN:
-                return In.<T>builder().field(field).values((List<Object>) value).build();
+                return In.<T>builder().field(field).values(asValues(value)).build();
             case NOT_IN:
-                return NotIn.<T>builder().field(field).values((List<Object>) value).build();
+                return NotIn.<T>builder().field(field).values(asValues(value)).build();
             case STARTS_WITH:
                 return StartsWith.<T>builder().field(field).value(value.toString()).build();
             case ENDS_WITH:
@@ -74,6 +78,8 @@ public record QueryFilter(
                 return Contains.<T>builder().field(field).value(value.toString()).build();
             case REGEX:
                 return Regex.<T>builder().field(field).value(value.toString()).build();
+            case PREFIX:
+                return Regex.<T>builder().field(field).value("^" + value.toString().replace(".", "\\.") + "(?:\\..+)?$").build();
             default:
                 throw new IllegalArgumentException("Unsupported operation: " + this.operation);
         }
@@ -83,7 +89,7 @@ public record QueryFilter(
         QUERY("q") {
             @Override
             public List<Op> supportedOp() {
-                return List.of(Op.EQUALS, Op.NOT_EQUALS, Op.REGEX);
+                return List.of(Op.EQUALS, Op.NOT_EQUALS);
             }
         },
         SCOPE("scope") {
@@ -95,7 +101,7 @@ public record QueryFilter(
         NAMESPACE("namespace") {
             @Override
             public List<Op> supportedOp() {
-                return List.of(Op.EQUALS, Op.NOT_EQUALS, Op.CONTAINS, Op.STARTS_WITH, Op.ENDS_WITH, Op.REGEX, Op.IN);
+                return List.of(Op.EQUALS, Op.NOT_EQUALS, Op.CONTAINS, Op.STARTS_WITH, Op.ENDS_WITH, Op.REGEX, Op.IN, Op.PREFIX);
             }
         },
         LABELS("labels") {
@@ -113,7 +119,7 @@ public record QueryFilter(
         START_DATE("startDate") {
             @Override
             public List<Op> supportedOp() {
-                return List.of(Op.GREATER_THAN, Op.LESS_THAN, Op.EQUALS, Op.NOT_EQUALS);
+                return List.of(Op.GREATER_THAN_OR_EQUAL_TO, Op.GREATER_THAN, Op.LESS_THAN_OR_EQUAL_TO, Op.LESS_THAN, Op.EQUALS, Op.NOT_EQUALS);
             }
         },
         END_DATE("endDate") {

core/src/main/java/io/kestra/core/models/collectors/FlowUsage.java

@@ -22,6 +22,10 @@ import java.util.stream.Stream;
 @Jacksonized
 @Introspected
 public class FlowUsage {
+
+    // Namespace used for 'Getting Started' flows.
+    private static final String TUTORIAL_NAMESPACE = "tutorial";
+
     private final Integer count;
     private final Long namespacesCount;
     private final Map<String, Long> taskTypeCount;
@@ -37,12 +41,13 @@ public class FlowUsage {
     }
 
     public static FlowUsage of(List<Flow> flows) {
+        List<Flow> filtered = flows.stream().filter(flow -> !TUTORIAL_NAMESPACE.equals(flow.getNamespace())).toList();
         return FlowUsage.builder()
-            .count(count(flows))
-            .namespacesCount(namespacesCount(flows))
-            .taskTypeCount(taskTypeCount(flows))
-            .triggerTypeCount(triggerTypeCount(flows))
-            .taskRunnerTypeCount(taskRunnerTypeCount(flows))
+            .count(count(filtered))
+            .namespacesCount(namespacesCount(filtered))
+            .taskTypeCount(taskTypeCount(filtered))
+            .triggerTypeCount(triggerTypeCount(filtered))
+            .taskRunnerTypeCount(taskRunnerTypeCount(filtered))
             .build();
     }
 

core/src/main/java/io/kestra/core/models/collectors/PluginMetric.java

@@ -0,0 +1,4 @@
+package io.kestra.core.models.collectors;
+
+public record PluginMetric(String type, double count, double totalTime, double meanTime){
+}

core/src/main/java/io/kestra/core/models/collectors/Usage.java

@@ -2,7 +2,7 @@ package io.kestra.core.models.collectors;
 
 import io.kestra.core.models.ServerType;
 import io.micronaut.core.annotation.Introspected;
-import io.micronaut.core.annotation.Nullable;
+import jakarta.annotation.Nullable;
 import jakarta.validation.Valid;
 import jakarta.validation.constraints.NotNull;
 import lombok.AllArgsConstructor;
@@ -13,6 +13,7 @@ import lombok.extern.jackson.Jacksonized;
 import java.time.Instant;
 import java.time.ZoneId;
 import java.util.List;
+import java.util.Map;
 import java.util.Set;
 
 @SuperBuilder(toBuilder = true)
@@ -66,4 +67,8 @@ public class Usage {
     @Valid
     @Nullable
     private ServiceUsage services;
+
+    @Valid
+    @Nullable
+    private List<PluginMetric> pluginMetrics;
 }

core/src/main/java/io/kestra/core/models/dashboards/ChartOption.java

@@ -1,7 +1,10 @@
 package io.kestra.core.models.dashboards;
 
+import jakarta.validation.constraints.Max;
+import jakarta.validation.constraints.Min;
 import jakarta.validation.constraints.NotBlank;
 import jakarta.validation.constraints.NotNull;
+import lombok.Builder;
 import lombok.EqualsAndHashCode;
 import lombok.Getter;
 import lombok.NoArgsConstructor;
@@ -21,6 +24,11 @@ public class ChartOption {
 
     private String description;
 
+    @Builder.Default
+    @Min(1)
+    @Max(12)
+    private int width = 6;
+
     public List<String> neededColumns() {
         return Collections.emptyList();
     }

core/src/main/java/io/kestra/core/models/dashboards/DataFilter.java

@@ -4,6 +4,7 @@ import io.kestra.core.models.QueryFilter;
 import io.kestra.core.models.annotations.Plugin;
 import io.kestra.core.models.dashboards.filters.AbstractFilter;
 import io.kestra.core.repositories.QueryBuilderInterface;
+import io.kestra.plugin.core.dashboard.data.IData;
 import jakarta.validation.constraints.NotBlank;
 import jakarta.validation.constraints.NotNull;
 import jakarta.validation.constraints.Pattern;
@@ -24,13 +25,12 @@ import java.util.Set;
 @NoArgsConstructor
 @Plugin
 @EqualsAndHashCode
-public abstract class DataFilter<F extends Enum<F>, C extends ColumnDescriptor<F>> implements io.kestra.core.models.Plugin {
+public abstract class DataFilter<F extends Enum<F>, C extends ColumnDescriptor<F>> implements io.kestra.core.models.Plugin, IData<F> {
     @NotNull
     @NotBlank
     @Pattern(regexp = "\\p{javaJavaIdentifierStart}\\p{javaJavaIdentifierPart}*(\\.\\p{javaJavaIdentifierStart}\\p{javaJavaIdentifierPart}*)*")
     private String type;
 
-
     private Map<String, C> columns;
 
     @Setter
@@ -42,8 +42,10 @@ public abstract class DataFilter<F extends Enum<F>, C extends ColumnDescriptor<F
         return Collections.emptySet();
     }
 
+    public void updateWhereWithGlobalFilters(List<QueryFilter> queryFilterList, ZonedDateTime startDate, ZonedDateTime endDate) {
+        this.where = whereWithGlobalFilters(queryFilterList, startDate, endDate, this.where);
+    }
+
     public abstract Class<? extends QueryBuilderInterface<F>> repositoryClass();
 
-    public abstract void setGlobalFilter(List<QueryFilter> queryFilterList, ZonedDateTime startDate, ZonedDateTime endDate);
-
 }

core/src/main/java/io/kestra/core/models/dashboards/DataFilterKPI.java

@@ -0,0 +1,56 @@
+package io.kestra.core.models.dashboards;
+
+import io.kestra.core.models.QueryFilter;
+import io.kestra.core.models.annotations.Plugin;
+import io.kestra.core.models.dashboards.filters.AbstractFilter;
+import io.kestra.core.repositories.QueryBuilderInterface;
+import io.kestra.plugin.core.dashboard.data.IData;
+import jakarta.validation.constraints.NotBlank;
+import jakarta.validation.constraints.NotNull;
+import jakarta.validation.constraints.Pattern;
+import lombok.EqualsAndHashCode;
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+import lombok.Setter;
+import lombok.experimental.SuperBuilder;
+
+import java.time.ZonedDateTime;
+import java.util.Collections;
+import java.util.List;
+import java.util.Set;
+
+@SuperBuilder(toBuilder = true)
+@Getter
+@NoArgsConstructor
+@Plugin
+@EqualsAndHashCode
+public abstract class DataFilterKPI<F extends Enum<F>, C extends ColumnDescriptor<F>> implements io.kestra.core.models.Plugin, IData<F> {
+    @NotNull
+    @NotBlank
+    @Pattern(regexp = "\\p{javaJavaIdentifierStart}\\p{javaJavaIdentifierPart}*(\\.\\p{javaJavaIdentifierStart}\\p{javaJavaIdentifierPart}*)*")
+    private String type;
+
+    private C columns;
+
+    @Setter
+    private List<AbstractFilter<F>> numerator;
+
+    private List<AbstractFilter<F>> where;
+
+    public Set<F> aggregationForbiddenFields() {
+        return Collections.emptySet();
+    }
+
+    public DataFilterKPI<F, C> clearFilters() {
+        this.numerator = Collections.emptyList();
+
+        return this;
+    }
+
+    public void updateWhereWithGlobalFilters(List<QueryFilter> queryFilterList, ZonedDateTime startDate, ZonedDateTime endDate) {
+        this.where = whereWithGlobalFilters(queryFilterList, startDate, endDate, this.where);
+    }
+
+    public abstract Class<? extends QueryBuilderInterface<F>> repositoryClass();
+
+}

core/src/main/java/io/kestra/core/models/dashboards/charts/DataChartKPI.java

@@ -0,0 +1,32 @@
+package io.kestra.core.models.dashboards.charts;
+
+import com.fasterxml.jackson.annotation.JsonInclude;
+import io.kestra.core.models.annotations.Plugin;
+import io.kestra.core.models.dashboards.DataFilterKPI;
+import io.kestra.core.validations.DataChartKPIValidation;
+import io.kestra.plugin.core.dashboard.chart.kpis.KpiOption;
+import jakarta.validation.constraints.NotNull;
+import lombok.EqualsAndHashCode;
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+import lombok.experimental.SuperBuilder;
+
+@SuperBuilder(toBuilder = true)
+@Getter
+@NoArgsConstructor
+@Plugin
+@JsonInclude(JsonInclude.Include.NON_DEFAULT)
+@EqualsAndHashCode
+@DataChartKPIValidation
+public abstract class DataChartKPI<P extends KpiOption, D extends DataFilterKPI<?, ?>> extends Chart<P> implements io.kestra.core.models.Plugin {
+    @NotNull
+    private D data;
+
+    public Integer minNumberOfAggregations() {
+        return null;
+    }
+
+    public Integer maxNumberOfAggregations() {
+        return null;
+    }
+}

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

@@ -7,6 +7,7 @@ import io.kestra.core.metrics.MetricRegistry;
 import io.kestra.core.models.executions.metrics.Counter;
 import io.kestra.core.models.executions.metrics.Timer;
 import io.micronaut.core.annotation.Introspected;
+import jakarta.annotation.Nullable;
 import lombok.EqualsAndHashCode;
 import lombok.Getter;
 import lombok.NoArgsConstructor;
@@ -36,12 +37,15 @@ abstract public class AbstractMetricEntry<T> {
     @NotNull
     protected String name;
 
+    protected String description;
+
     protected Map<String, String> tags;
 
     protected Instant timestamp = Instant.now();
 
-    protected AbstractMetricEntry(@NotNull String name, String[] tags) {
+    protected AbstractMetricEntry(@NotNull String name, @Nullable String description, String[] tags) {
         this.name = name;
+        this.description = description;
         this.tags = tagsAsMap(tags);
     }
 
@@ -79,7 +83,7 @@ abstract public class AbstractMetricEntry<T> {
 
     abstract public T getValue();
 
-    abstract public void register(MetricRegistry meterRegistry, String prefix, Map<String, String> tags);
+    abstract public void register(MetricRegistry meterRegistry, String name, @Nullable String description, Map<String, String> tags);
 
     abstract public void increment(T value);
 }

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

@@ -22,10 +22,11 @@ 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.test.flow.TaskFixture;
 import io.kestra.core.utils.IdUtils;
 import io.kestra.core.utils.MapUtils;
-import io.micronaut.core.annotation.Nullable;
 import io.swagger.v3.oas.annotations.Hidden;
+import jakarta.annotation.Nullable;
 import jakarta.validation.constraints.NotNull;
 import jakarta.validation.constraints.Pattern;
 import lombok.*;
@@ -112,6 +113,13 @@ public class Execution implements DeletedInterface, TenantInterface {
     @Setter
     String traceParent;
 
+    @With
+    @Nullable
+    List<TaskFixture> fixtures;
+
+    @Nullable
+    ExecutionKind kind;
+
     /**
      * Factory method for constructing a new {@link Execution} object for the given {@link Flow}.
      *
@@ -210,7 +218,9 @@ public class Execution implements DeletedInterface, TenantInterface {
             this.deleted,
             this.metadata,
             this.scheduleDate,
-            this.traceParent
+            this.traceParent,
+            this.fixtures,
+            this.kind
         );
     }
 
@@ -234,7 +244,9 @@ public class Execution implements DeletedInterface, TenantInterface {
             this.deleted,
             this.metadata,
             this.scheduleDate,
-            this.traceParent
+            this.traceParent,
+            this.fixtures,
+            this.kind
         );
     }
 
@@ -271,7 +283,9 @@ public class Execution implements DeletedInterface, TenantInterface {
             this.deleted,
             this.metadata,
             this.scheduleDate,
-            this.traceParent
+            this.traceParent,
+            this.fixtures,
+            this.kind
         );
     }
 
@@ -295,7 +309,9 @@ public class Execution implements DeletedInterface, TenantInterface {
             this.deleted,
             this.metadata,
             this.scheduleDate,
-            this.traceParent
+            this.traceParent,
+            this.fixtures,
+            this.kind
         );
     }
 
@@ -728,6 +744,16 @@ public class Execution implements DeletedInterface, TenantInterface {
             );
     }
 
+    public Optional<TaskFixture> getFixtureForTaskRun(TaskRun taskRun) {
+        if (this.fixtures == null) {
+            return Optional.empty();
+        }
+
+        return this.fixtures.stream()
+            .filter(fixture -> Objects.equals(fixture.getId(), taskRun.getTaskId()) && Objects.equals(fixture.getValue(), taskRun.getValue()))
+            .findFirst();
+    }
+
     /**
      * Create a new attempt for failed worker execution
      *
@@ -735,7 +761,7 @@ public class Execution implements DeletedInterface, TenantInterface {
      * @param e the exception raise
      * @return new taskRun with added attempt
      */
-    private static FailedTaskRunWithLog newAttemptsTaskRunForFailedExecution(TaskRun taskRun,
+    private FailedTaskRunWithLog newAttemptsTaskRunForFailedExecution(TaskRun taskRun,
         Exception e) {
         return new FailedTaskRunWithLog(
             taskRun
@@ -746,7 +772,7 @@ public class Execution implements DeletedInterface, TenantInterface {
                         .withState(State.Type.FAILED))
                 )
                 .withState(State.Type.FAILED),
-            RunContextLogger.logEntries(loggingEventFromException(e), LogEntry.of(taskRun))
+            RunContextLogger.logEntries(loggingEventFromException(e), LogEntry.of(taskRun, kind))
         );
     }
 
@@ -758,7 +784,7 @@ public class Execution implements DeletedInterface, TenantInterface {
      * @param e the exception raise
      * @return new taskRun with updated attempt with logs
      */
-    private static FailedTaskRunWithLog lastAttemptsTaskRunForFailedExecution(TaskRun taskRun,
+    private FailedTaskRunWithLog lastAttemptsTaskRunForFailedExecution(TaskRun taskRun,
         TaskRunAttempt lastAttempt, Exception e) {
         return new FailedTaskRunWithLog(
             taskRun
@@ -772,7 +798,7 @@ public class Execution implements DeletedInterface, TenantInterface {
                         .toList()
                 )
                 .withState(State.Type.FAILED),
-            RunContextLogger.logEntries(loggingEventFromException(e), LogEntry.of(taskRun))
+            RunContextLogger.logEntries(loggingEventFromException(e), LogEntry.of(taskRun, kind))
         );
     }
 
@@ -794,7 +820,7 @@ public class Execution implements DeletedInterface, TenantInterface {
     /**
      * Transform an exception to {@link ILoggingEvent}
      *
-     * @param e the current execption
+     * @param e the current exception
      * @return the {@link ILoggingEvent} waited to generate {@link LogEntry}
      */
     public static ILoggingEvent loggingEventFromException(Exception e) {
@@ -827,10 +853,12 @@ public class Execution implements DeletedInterface, TenantInterface {
             .forEach((taskId, taskRuns) -> {
                 Map<String, Object> taskOutputs = new HashMap<>();
                 for (TaskRun current : taskRuns) {
-                    if (current.getIteration() != null) {
-                        taskOutputs = MapUtils.merge(taskOutputs, outputs(current, byIds));
-                    } else {
-                        taskOutputs.putAll(outputs(current, byIds));
+                    if (!MapUtils.isEmpty(current.getOutputs())) {
+                        if (current.getIteration() != null) {
+                            taskOutputs = MapUtils.merge(taskOutputs, outputs(current, byIds));
+                        } else {
+                            taskOutputs.putAll(outputs(current, byIds));
+                        }
                     }
                 }
                 result.put(taskId, taskOutputs);

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

@@ -2,7 +2,6 @@ package io.kestra.core.models.executions;
 
 import com.fasterxml.jackson.annotation.JsonInclude;
 import io.kestra.core.models.TenantInterface;
-import io.kestra.core.models.triggers.TriggerContext;
 import io.kestra.core.runners.WorkerTask;
 import jakarta.validation.constraints.NotNull;
 import lombok.*;
@@ -25,6 +24,11 @@ public class ExecutionKilledExecution extends ExecutionKilled implements TenantI
     @NotNull
     String executionId;
 
+    /**
+     * The state to move the execution to after kill.
+     */
+    io.kestra.core.models.flows.State.Type executionState;
+
     /**
      * Specifies whether killing the execution, also kill all sub-flow executions.
      */

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

@@ -0,0 +1,10 @@
+package io.kestra.core.models.executions;
+
+/**
+ * Describe the kind of execution:
+ * - TEST: created by a test
+ * - NORMAL: anything else, for backward compatibility NORMAL is not persisted but null is used instead
+ */
+public enum ExecutionKind {
+    NORMAL, TEST
+}

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

@@ -6,8 +6,8 @@ import io.kestra.core.models.TenantInterface;
 import io.kestra.core.models.flows.Flow;
 import io.kestra.core.models.triggers.AbstractTrigger;
 import io.kestra.core.models.triggers.TriggerContext;
-import io.micronaut.core.annotation.Nullable;
 import io.swagger.v3.oas.annotations.Hidden;
+import jakarta.annotation.Nullable;
 import lombok.Builder;
 import lombok.Value;
 import org.slf4j.event.Level;
@@ -60,6 +60,9 @@ public class LogEntry implements DeletedInterface, TenantInterface {
     @Builder.Default
     boolean deleted = false;
 
+    @Nullable
+    ExecutionKind executionKind;
+
     public static List<Level> findLevelsByMin(Level minLevel) {
         if (minLevel == null) {
             return Arrays.asList(Level.values());
@@ -76,10 +79,11 @@ public class LogEntry implements DeletedInterface, TenantInterface {
             .namespace(execution.getNamespace())
             .flowId(execution.getFlowId())
             .executionId(execution.getId())
+            .executionKind(execution.getKind())
             .build();
     }
 
-    public static LogEntry of(TaskRun taskRun) {
+    public static LogEntry of(TaskRun taskRun, ExecutionKind executionKind) {
         return LogEntry.builder()
             .tenantId(taskRun.getTenantId())
             .namespace(taskRun.getNamespace())
@@ -88,24 +92,27 @@ public class LogEntry implements DeletedInterface, TenantInterface {
             .executionId(taskRun.getExecutionId())
             .taskRunId(taskRun.getId())
             .attemptNumber(taskRun.attemptNumber())
+            .executionKind(executionKind)
             .build();
     }
 
-    public static LogEntry of(Flow flow, AbstractTrigger abstractTrigger) {
+    public static LogEntry of(Flow flow, AbstractTrigger abstractTrigger, ExecutionKind executionKind) {
         return LogEntry.builder()
             .tenantId(flow.getTenantId())
             .namespace(flow.getNamespace())
             .flowId(flow.getId())
             .triggerId(abstractTrigger.getId())
+            .executionId(abstractTrigger.getId())
             .build();
     }
 
-    public static LogEntry of(TriggerContext triggerContext, AbstractTrigger abstractTrigger) {
+    public static LogEntry of(TriggerContext triggerContext, AbstractTrigger abstractTrigger, ExecutionKind executionKind) {
         return LogEntry.builder()
             .tenantId(triggerContext.getTenantId())
             .namespace(triggerContext.getNamespace())
             .flowId(triggerContext.getFlowId())
             .triggerId(abstractTrigger.getId())
+            .executionId(abstractTrigger.getId())
             .build();
     }
 
@@ -122,7 +129,8 @@ public class LogEntry implements DeletedInterface, TenantInterface {
                 new AbstractMap.SimpleEntry<>("taskId", this.taskId),
                 new AbstractMap.SimpleEntry<>("executionId", this.executionId),
                 new AbstractMap.SimpleEntry<>("taskRunId", this.taskRunId),
-                new AbstractMap.SimpleEntry<>("triggerId", this.triggerId)
+                new AbstractMap.SimpleEntry<>("triggerId", this.triggerId),
+                new AbstractMap.SimpleEntry<>("executionKind", Optional.ofNullable(this.executionKind).map(executionKind -> executionKind.name()).orElse(null)  )
             )
             .filter(e -> e.getValue() != null)
             .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));

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

@@ -5,8 +5,8 @@ import io.kestra.core.models.DeletedInterface;
 import io.kestra.core.models.TenantInterface;
 import io.kestra.core.models.executions.metrics.Counter;
 import io.kestra.core.models.executions.metrics.Timer;
-import io.micronaut.core.annotation.Nullable;
 import io.swagger.v3.oas.annotations.Hidden;
+import jakarta.annotation.Nullable;
 import lombok.Builder;
 import lombok.Value;
 
@@ -57,7 +57,10 @@ public class MetricEntry implements DeletedInterface, TenantInterface {
     @Builder.Default
     boolean deleted = false;
 
-    public static MetricEntry of(TaskRun taskRun, AbstractMetricEntry<?> metricEntry) {
+    @Nullable
+    ExecutionKind executionKind;
+
+    public static MetricEntry of(TaskRun taskRun, AbstractMetricEntry<?> metricEntry, ExecutionKind executionKind) {
         return MetricEntry.builder()
             .tenantId(taskRun.getTenantId())
             .namespace(taskRun.getNamespace())
@@ -66,10 +69,11 @@ public class MetricEntry implements DeletedInterface, TenantInterface {
             .taskId(taskRun.getTaskId())
             .taskRunId(taskRun.getId())
             .type(metricEntry.getType())
-            .name(metricEntry.name)
+            .name(metricEntry.getName())
             .tags(metricEntry.getTags())
             .value(computeValue(metricEntry))
             .timestamp(metricEntry.getTimestamp())
+            .executionKind(executionKind)
             .build();
     }
 

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

@@ -50,8 +50,8 @@ public class TaskRun implements TenantInterface {
     List<TaskRunAttempt> attempts;
 
     @With
-    @JsonInclude(JsonInclude.Include.ALWAYS) // always include outputs so it's easier to reason about in expressions
-    Map<String, Object> outputs;
+    @JsonInclude(JsonInclude.Include.ALWAYS)
+    Variables outputs;
 
     @NotNull
     State state;

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

@@ -1,15 +1,15 @@
 package io.kestra.core.models.executions;
 
+import io.kestra.core.models.flows.State;
+import jakarta.annotation.Nullable;
+import jakarta.validation.constraints.NotNull;
 import lombok.Builder;
 import lombok.Value;
-import io.kestra.core.models.flows.State;
+import lombok.With;
 
 import java.net.URI;
 import java.util.List;
 
-import jakarta.validation.constraints.NotNull;
-import lombok.With;
-
 @Value
 @Builder
 public class TaskRunAttempt {
@@ -18,18 +18,22 @@ public class TaskRunAttempt {
      */
     @Deprecated
     public void setMetrics(List<AbstractMetricEntry<?>> metrics) {
-
     }
 
     @NotNull
     State state;
 
+    @Nullable
+    String workerId;
+
     @With
+    @Nullable
     URI logFile;
 
     public TaskRunAttempt withState(State.Type state) {
         return new TaskRunAttempt(
             this.state.withState(state),
+            this.workerId,
             this.logFile
         );
     }

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

@@ -0,0 +1,271 @@
+package io.kestra.core.models.executions;
+
+import com.fasterxml.jackson.core.JsonGenerator;
+import com.fasterxml.jackson.core.JsonParser;
+import com.fasterxml.jackson.core.JsonToken;
+import com.fasterxml.jackson.databind.DeserializationContext;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.SerializerProvider;
+import com.fasterxml.jackson.databind.annotation.JsonDeserialize;
+import com.fasterxml.jackson.databind.annotation.JsonSerialize;
+import com.fasterxml.jackson.databind.deser.std.StdDeserializer;
+import com.fasterxml.jackson.databind.ser.std.StdSerializer;
+import io.kestra.core.contexts.KestraContext;
+import io.kestra.core.serializers.JacksonMapper;
+import io.kestra.core.storages.Storage;
+import io.kestra.core.storages.StorageInterface;
+import io.kestra.core.utils.MapUtils;
+import io.kestra.core.utils.ReadOnlyDelegatingMap;
+
+import java.io.*;
+import java.net.URI;
+import java.nio.file.Files;
+import java.util.*;
+import java.util.stream.Collectors;
+
+import static io.kestra.core.utils.Rethrow.throwBiConsumer;
+
+/**
+ * A <code>Variables</code> represent a set of output variables.
+ * Variables can be stored in-memory or inside the internal storage.
+ * <p>
+ * The easiest way to construct a <code>Variables</code> object is to use the {@link io.kestra.core.services.VariablesService}.
+ *
+ * @see io.kestra.core.services.VariablesService
+ */
+@JsonSerialize(using = Variables.Serializer.class)
+@JsonDeserialize(using = Variables.Deserializer.class)
+public sealed interface Variables extends Map<String, Object> {
+    String TYPE = "io.kestra.datatype:outputs";
+    Variables EMPTY = new InMemoryVariables(Collections.emptyMap());
+
+    /**
+     * Returns an empty Variables.
+     */
+    static Variables empty() {
+        return EMPTY;
+    }
+
+    /**
+     * Creates an InMemoryVariables with an output map.
+     * This is safer to use {@link io.kestra.core.services.VariablesService#of(io.kestra.core.storages.StorageContext, Map)} instead.
+     *
+     * @see InMemoryVariables
+     * @see io.kestra.core.services.VariablesService
+     */
+    static Variables inMemory(Map<String, Object> outputs) {
+        if (MapUtils.isEmpty(outputs)) {
+            return empty();
+        }
+        return new InMemoryVariables(outputs);
+    }
+
+    /**
+     * Creates an InStorageVariables with a {@link Storage} and an output map.
+     * The output map will be immediately stored inside the internal storage.
+     *
+     * @see InStorageVariables
+     * @see io.kestra.core.services.VariablesService
+     */
+    static Variables inStorage(Storage storage, Map<String, Object> outputs) {
+        if (MapUtils.isEmpty(outputs)) {
+            return empty();
+        }
+        return new InStorageVariables(storage, outputs);
+    }
+
+    /**
+     * Creates an InStorageVariables with an internal storage URI.
+     * The output map will be read lazily from the internal storage URI at access time.
+     *
+     * @see InStorageVariables
+     * @see io.kestra.core.services.VariablesService
+     */
+    static Variables inStorage(StorageContext storageContext, URI uri) {
+        return new InStorageVariables(storageContext, uri);
+    }
+
+    record StorageContext(String tenantId, String namespace) {}
+
+    final class InMemoryVariables extends ReadOnlyDelegatingMap<String, Object> implements Variables {
+        private final Map<String, Object> delegate;
+
+        InMemoryVariables(Map<String, Object> outputs) {
+            this.delegate = outputs;
+        }
+
+        @Override
+        protected Map<String, Object> getDelegate() {
+            return MapUtils.emptyOnNull(delegate);
+        }
+    }
+
+    final class InStorageVariables extends ReadOnlyDelegatingMap<String, Object> implements Variables {
+        private static final ObjectMapper ION_MAPPER = JacksonMapper.ofIon();
+
+        private final URI storageUri;
+        private final StorageContext storageContext;
+
+        private Map<String, Object> delegate;
+        private State state;
+
+        // we need to store the tenantId and namespace for loading the file from the storage
+
+        InStorageVariables(Storage storage, Map<String, Object> outputs) {
+            // expand the map in case it already contains variable in it
+            this.delegate = expand(outputs);
+            this.state = State.DEFLATED;
+            this.storageContext = new StorageContext(storage.namespace().tenantId(), storage.namespace().namespace());
+
+            if (!MapUtils.isEmpty(outputs)) {
+                try {
+                    File file = Files.createTempFile("output-", ".ion").toFile();
+                    ION_MAPPER.writeValue(file, outputs);
+                    this.storageUri = storage.putFile(file);
+                } catch (IOException e) {
+                    // FIXME check if we should not declare it
+                    throw new UncheckedIOException(e);
+                }
+            } else {
+                this.storageUri = null;
+            }
+        }
+
+        InStorageVariables(StorageContext storageContext, URI storageUri) {
+            this.storageUri = storageUri;
+            this.state = State.INIT;
+            this.storageContext = storageContext;
+        }
+
+        URI getStorageUri() {
+            return storageUri;
+        }
+
+        StorageContext getStorageContext() {
+            return storageContext;
+        }
+
+        @Override
+        protected Map<String, Object> getDelegate() {
+            return loadFromStorage();
+        }
+
+        private Map<String, Object> loadFromStorage() {
+            if (this.state == State.INIT) {
+                if (storageUri == null) {
+                    return Collections.emptyMap();
+                }
+
+                StorageInterface storage = KestraContext.getContext().getStorageInterface();
+                try (InputStream file = storage.get(storageContext.tenantId(), storageContext.namespace(), storageUri)) {
+                    delegate = ION_MAPPER.readValue(file, JacksonMapper.MAP_TYPE_REFERENCE);
+                    state = State.DEFLATED;
+                } catch (IOException e) {
+                    throw new UncheckedIOException(e);
+                }
+
+                // check all entries to possibly deflate them also
+                return MapUtils.emptyOnNull(expand(delegate));
+            }
+
+            return MapUtils.emptyOnNull(delegate);
+        }
+
+        @SuppressWarnings("unchecked")
+        private static Map<String, Object> expand(Map<String, Object> variables) {
+            if (MapUtils.isEmpty(variables)) {
+                return variables;
+            }
+
+            return variables.entrySet()
+                .stream()
+                .map(entry -> {
+                    if (entry.getValue() instanceof InStorageVariables var) {
+                        return Map.entry(entry.getKey(), (Object) expand(var.loadFromStorage()));
+                    } else if (entry.getValue() instanceof Map<?, ?> map) {
+                        if (TYPE.equals(map.get("type"))) {
+                            String uriString = (String) map.get("storageUri");
+                            if (uriString != null) {
+                                Map<String, String> storageContextMap = (Map<String, String>) map.get("storageContext");
+                                StorageContext storageContext = new StorageContext(storageContextMap.get("tenantId"), storageContextMap.get("namespace"));
+                                URI storageUri = URI.create(uriString);
+                                InStorageVariables inStorage = new InStorageVariables(storageContext, storageUri);
+                                return Map.entry(entry.getKey(), (Object) expand(inStorage.loadFromStorage()));
+                            }
+                            InStorageVariables inStorage = new InStorageVariables((StorageContext) null, null);
+                            return Map.entry(entry.getKey(), (Object) inStorage.loadFromStorage());
+                        }
+                        else {
+                            return Map.entry(entry.getKey(), (Object) expand((Map<String, Object>) map));
+                        }
+                    } else {
+                        return entry;
+                    }
+                })
+                .collect(Collectors.toMap(entry -> entry.getKey(), entry -> entry.getValue()));
+        }
+
+        enum State { INIT, DEFLATED }
+    }
+
+    class Serializer extends StdSerializer<Variables> {
+        protected Serializer() {
+            super(Variables.class);
+        }
+
+        @Override
+        public void serialize(Variables value, JsonGenerator gen, SerializerProvider serializers) throws IOException {
+            if (value == null) {
+                gen.writeNull();
+            } else {
+                switch (value) {
+                    case InMemoryVariables inMemory -> {
+                        // we must write entry by entry otherwise nulls are not included
+                        gen.writeStartObject();
+                        inMemory.getDelegate().forEach(throwBiConsumer((k, v) -> gen.writeObjectField(k, v)));
+                        gen.writeEndObject();
+                    }
+                    case InStorageVariables inStorage -> {
+                        gen.writeStartObject();
+                        gen.writeStringField("type", TYPE); // marker to be sure at deserialization time it's a Variables not some random Map
+                        gen.writeStringField("storageUri", inStorage.getStorageUri() != null ? inStorage.getStorageUri().toString() : null);
+                        gen.writeObjectField("storageContext", inStorage.getStorageContext());
+                        gen.writeEndObject();
+                    }
+                }
+            }
+        }
+    }
+
+    class Deserializer extends StdDeserializer<Variables> {
+        public Deserializer() {
+            super(Variables.class);
+        }
+
+        @Override
+        @SuppressWarnings("unchecked")
+        public Variables deserialize(JsonParser parser, DeserializationContext ctx) throws IOException {
+            if (parser.hasToken(JsonToken.VALUE_NULL)) {
+                return null;
+            } else if (parser.hasToken(JsonToken.START_OBJECT)) {
+                // deserialize as map
+                Map<String, Object> ret = ctx.readValue(parser, Map.class);
+                if (TYPE.equals(ret.get("type"))) {
+                    String uriString = (String) ret.get("storageUri");
+                    if (uriString != null) {
+                        Map<String, String> storageContextMap = (Map<String, String>) ret.get("storageContext");
+                        StorageContext storageContext = new StorageContext(storageContextMap.get("tenantId"), storageContextMap.get("namespace"));
+                        URI storageUri = URI.create(uriString);
+                        return new InStorageVariables(storageContext, storageUri);
+                    }
+                    return new InStorageVariables((StorageContext) null, null);
+                }
+
+                // If the type is not TYPE, a real map has been serialized so we build a Variables with it.
+                return new InMemoryVariables(ret);
+            }
+            throw new IllegalArgumentException("Unable to deserialize value as it's not an object");
+        }
+    }
+}
+

core/src/main/java/io/kestra/core/models/executions/metrics/Counter.java

@@ -1,6 +1,7 @@
 package io.kestra.core.models.executions.metrics;
 
 import com.fasterxml.jackson.annotation.JsonInclude;
+import jakarta.annotation.Nullable;
 import lombok.EqualsAndHashCode;
 import lombok.Getter;
 import lombok.NoArgsConstructor;
@@ -17,6 +18,7 @@ import java.util.Map;
 @NoArgsConstructor
 public final class Counter extends AbstractMetricEntry<Double> {
     public static final String TYPE = "counter";
+
     @NotNull
     @JsonInclude
     private final String type = TYPE;
@@ -25,32 +27,48 @@ public final class Counter extends AbstractMetricEntry<Double> {
     @EqualsAndHashCode.Exclude
     private Double value;
 
-    private Counter(@NotNull String name, @NotNull Double value, String... tags) {
-        super(name, tags);
+    private Counter(@NotNull String name, @Nullable String description, @NotNull Double value, String... tags) {
+        super(name, description, tags);
 
         this.value = value;
     }
 
     public static Counter of(@NotNull String name, @NotNull Double value, String... tags) {
-        return new Counter(name, value, tags);
+        return new Counter(name, null, value, tags);
+    }
+
+    public static Counter of(@NotNull String name, @Nullable String description, @NotNull Double value, String... tags) {
+        return new Counter(name, description, value, tags);
     }
 
     public static Counter of(@NotNull String name, @NotNull Integer value, String... tags) {
-        return new Counter(name, (double) value, tags);
+        return new Counter(name, null, (double) value, tags);
+    }
+
+    public static Counter of(@NotNull String name, @Nullable String description, @NotNull Integer value, String... tags) {
+        return new Counter(name, description, (double) value, tags);
     }
 
     public static Counter of(@NotNull String name, @NotNull Long value, String... tags) {
-        return new Counter(name, (double) value, tags);
+        return new Counter(name, null, (double) value, tags);
+    }
+
+    public static Counter of(@NotNull String name, @Nullable String description, @NotNull Long value, String... tags) {
+        return new Counter(name, description, (double) value, tags);
     }
 
     public static Counter of(@NotNull String name, @NotNull Float value, String... tags) {
-        return new Counter(name, (double) value, tags);
+        return new Counter(name, null, (double) value, tags);
+    }
+
+    public static Counter of(@NotNull String name, @Nullable String description, @NotNull Float value, String... tags) {
+        return new Counter(name, description, (double) value, tags);
     }
 
     @Override
-    public void register(MetricRegistry meterRegistry, String prefix, Map<String, String> tags) {
+    public void register(MetricRegistry meterRegistry, String name, String description, Map<String, String> tags) {
         meterRegistry
-            .counter(this.metricName(prefix), this.tagsAsArray(tags))
+            .counter(this.metricName(name), description, this.tagsAsArray(tags))
             .increment(this.value);
     }
 

core/src/main/java/io/kestra/core/models/executions/metrics/Timer.java

@@ -1,6 +1,7 @@
 package io.kestra.core.models.executions.metrics;
 
 import com.fasterxml.jackson.annotation.JsonInclude;
+import jakarta.annotation.Nullable;
 import lombok.EqualsAndHashCode;
 import lombok.Getter;
 import lombok.NoArgsConstructor;
@@ -27,20 +28,24 @@ public class Timer extends AbstractMetricEntry<Duration> {
     @EqualsAndHashCode.Exclude
     private Duration value;
 
-    private Timer(@NotNull String name, @NotNull Duration value, String... tags) {
-        super(name, tags);
+    private Timer(@NotNull String name, @Nullable String description, @NotNull Duration value, String... tags) {
+        super(name, description, tags);
 
         this.value = value;
     }
 
     public static Timer of(@NotNull String name, @NotNull Duration value, String... tags) {
-        return new Timer(name, value, tags);
+        return new Timer(name, null, value, tags);
+    }
+
+    public static Timer of(@NotNull String name, @Nullable String description, @NotNull Duration value, String... tags) {
+        return new Timer(name, description, value, tags);
     }
 
     @Override
-    public void register(MetricRegistry meterRegistry, String prefix, Map<String, String> tags) {
+    public void register(MetricRegistry meterRegistry, String name, String description, Map<String, String> tags) {
         meterRegistry
-            .timer(this.metricName(prefix), this.tagsAsArray(tags))
+            .timer(this.metricName(name), description, this.tagsAsArray(tags))
             .record(this.value);
     }
 

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

@@ -27,6 +27,7 @@ public class FlowWithPath {
 
     public static FlowWithPath of(FlowInterface flow, String path) {
         return FlowWithPath.builder()
+            .tenantId(flow.getTenantId())
             .id(flow.getId())
             .namespace(flow.getNamespace())
             .path(path)

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

@@ -39,30 +39,6 @@ import java.util.Optional;
 @NoArgsConstructor
 @JsonDeserialize
 public class GenericFlow extends AbstractFlow implements HasUID {
-
-    private String id;
-
-    private String namespace;
-
-    private Integer revision;
-
-    private List<Input<?>> inputs;
-
-    private Map<String, Object> variables;
-
-    @Builder.Default
-    private boolean disabled = false;
-
-    @Builder.Default
-    private boolean deleted = false;
-
-    @JsonSerialize(using = ListOrMapOfLabelSerializer.class)
-    @JsonDeserialize(using = ListOrMapOfLabelDeserializer.class)
-    @Schema(implementation = Object.class, oneOf = {List.class, Map.class})
-    private List<Label> labels;
-
-    private String tenantId;
-
     private String source;
 
     private List<SLA> sla;
@@ -84,7 +60,6 @@ public class GenericFlow extends AbstractFlow implements HasUID {
      * @return a new {@link GenericFlow}
      * @throws DeserializationException if source cannot be deserialized.
      */
-    @VisibleForTesting
     public static GenericFlow of(final FlowInterface flow) throws DeserializationException {
         return fromYaml(flow.getTenantId(), flow.sourceOrGenerateIfNull());
     }

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

@@ -69,7 +69,7 @@ public class State {
 
     public State withState(Type state) {
         if (this.current == state) {
-            log.warn("Can't change state, already " + current);
+            log.warn("Can't change state, already {}", current);
             return this;
         }
 

core/src/main/java/io/kestra/core/models/hierarchies/SubflowGraphTask.java

@@ -7,7 +7,7 @@ import io.kestra.core.models.executions.TaskRun;
 import io.kestra.core.models.flows.Flow;
 import io.kestra.core.models.flows.FlowInterface;
 import io.kestra.core.models.tasks.*;
-import io.kestra.core.runners.FlowExecutorInterface;
+import io.kestra.core.runners.FlowMetaStoreInterface;
 import io.kestra.core.runners.RunContext;
 import io.kestra.core.runners.SubflowExecution;
 import io.kestra.core.runners.SubflowExecutionResult;
@@ -48,7 +48,7 @@ public class SubflowGraphTask extends AbstractGraphTask {
 
     public record SubflowTaskWrapper<T extends Output>(RunContext runContext, ExecutableTask<T> subflowTask) implements TaskInterface, ExecutableTask<T> {
         @Override
-        public List<SubflowExecution<?>> createSubflowExecutions(RunContext runContext, FlowExecutorInterface flowExecutorInterface, Flow currentFlow, Execution currentExecution, TaskRun currentTaskRun) throws InternalException {
+        public List<SubflowExecution<?>> createSubflowExecutions(RunContext runContext, FlowMetaStoreInterface flowExecutorInterface, Flow currentFlow, Execution currentExecution, TaskRun currentTaskRun) throws InternalException {
             return subflowTask.createSubflowExecutions(runContext, flowExecutorInterface, currentFlow, currentExecution, currentTaskRun);
         }
 

core/src/main/java/io/kestra/core/models/namespaces/Namespace.java

@@ -1,5 +1,6 @@
 package io.kestra.core.models.namespaces;
 
+import io.swagger.v3.oas.annotations.media.Schema;
 import jakarta.validation.constraints.NotNull;
 import jakarta.validation.constraints.Pattern;
 import lombok.*;
@@ -11,6 +12,7 @@ import lombok.experimental.SuperBuilder;
 @NoArgsConstructor
 @ToString
 @EqualsAndHashCode
+@Schema(name = "NamespaceLight")
 public class Namespace implements NamespaceInterface {
     @NotNull
     @Pattern(regexp="^[a-z0-9][a-z0-9._-]*")

core/src/main/java/io/kestra/core/models/property/Data.java

@@ -44,7 +44,7 @@ public class Data<T> {
      * @see #ofList(List)
      */
     public static Data<?> ofURI(URI uri) {
-        return Data.builder().fromURI(Property.of(uri)).build();
+        return Data.builder().fromURI(Property.ofValue(uri)).build();
     }
 
     /**
@@ -54,7 +54,7 @@ public class Data<T> {
      * @see #ofList(List)
      */
     public static Data<?> ofMap(Map<String, Object> map) {
-        return Data.builder().fromMap(Property.of(map)).build();
+        return Data.builder().fromMap(Property.ofValue(map)).build();
     }
 
     /**
@@ -64,7 +64,7 @@ public class Data<T> {
      * @see #ofMap(Map)
      */
     public static Data<?> ofList(List<Map<String, Object>> list) {
-        return Data.builder().fromList(Property.of(list)).build();
+        return Data.builder().fromList(Property.ofValue(list)).build();
     }
 
     /**

core/src/main/java/io/kestra/core/models/property/Property.java

@@ -8,9 +8,11 @@ import com.fasterxml.jackson.databind.annotation.JsonDeserialize;
 import com.fasterxml.jackson.databind.annotation.JsonSerialize;
 import com.fasterxml.jackson.databind.deser.std.StdDeserializer;
 import com.fasterxml.jackson.databind.ser.std.StdSerializer;
+import com.google.common.annotations.VisibleForTesting;
 import io.kestra.core.exceptions.IllegalVariableEvaluationException;
 import io.kestra.core.runners.RunContext;
 import io.kestra.core.serializers.JacksonMapper;
+import jakarta.validation.constraints.NotNull;
 import lombok.AccessLevel;
 import lombok.AllArgsConstructor;
 import lombok.Builder;
@@ -45,10 +47,16 @@ public class Property<T> {
     private String expression;
     private T value;
 
+    /**
+     * @deprecated use {@link #ofExpression(String)} instead.
+     */
+    @Deprecated
+    // Note: when not used, this constructor would not be deleted but made private so it can only be used by ofExpression(String) and the deserializer
     public Property(String expression) {
         this.expression = expression;
     }
 
+    @VisibleForTesting
     public Property(Map<?, ?> map) {
         try {
             expression = MAPPER.writeValueAsString(map);
@@ -65,8 +73,10 @@ public class Property<T> {
      * Build a new Property object with a value already set.<br>
      *
      * A property build with this method will always return the value passed at build time, no rendering will be done.
+     *
+     * Use {@link #ofExpression(String)} to build a property with a Pebble expression instead.
      */
-    public static <V> Property<V> of(V value) {
+    public static <V> Property<V> ofValue(V value) {
         // trick the serializer so the property would not be null at deserialization time
         String expression;
         if (value instanceof Map<?, ?> || value instanceof List<?>) {
@@ -93,6 +103,28 @@ public class Property<T> {
         return p;
     }
 
+    /**
+     * @deprecated use {@link #ofValue(Object)} instead.
+     */
+    @Deprecated
+    public static <V> Property<V> of(V value) {
+        return ofValue(value);
+    }
+
+    /**
+     * Build a new Property object with a Pebble expression.<br>
+     *
+     * Use {@link #ofValue(Object)} to build a property with a value instead.
+     */
+    public static <V> Property<V> ofExpression(@NotNull String expression) {
+        Objects.requireNonNull(expression, "'expression' is required");
+        if(!expression.contains("{")) {
+            throw new IllegalArgumentException("'expression' must be a valid Pebble expression");
+        }
+
+        return new Property<>(expression);
+    }
+
     /**
      * Render a property then convert it to its target type.<br>
      *

core/src/main/java/io/kestra/core/models/tasks/ExecutableTask.java

@@ -6,7 +6,7 @@ import io.kestra.core.models.executions.TaskRun;
 import io.kestra.core.models.flows.Flow;
 import io.kestra.core.models.flows.FlowId;
 import io.kestra.core.models.flows.FlowInterface;
-import io.kestra.core.runners.FlowExecutorInterface;
+import io.kestra.core.runners.FlowMetaStoreInterface;
 import io.kestra.core.runners.RunContext;
 import io.kestra.core.runners.SubflowExecution;
 import io.kestra.core.runners.SubflowExecutionResult;
@@ -23,7 +23,7 @@ public interface ExecutableTask<T extends Output>{
      * Each SubflowExecution will generate a subflow execution.
      */
     List<SubflowExecution<?>> createSubflowExecutions(RunContext runContext,
-                                                      FlowExecutorInterface flowExecutorInterface,
+                                                      FlowMetaStoreInterface flowExecutorInterface,
                                                       Flow currentFlow, Execution currentExecution,
                                                       TaskRun currentTaskRun) throws InternalException;
 

core/src/main/java/io/kestra/core/models/tasks/NamespaceFiles.java

@@ -23,7 +23,7 @@ public class NamespaceFiles {
         title = "Whether to enable namespace files to be loaded into the working directory. If explicitly set to `true` in a task, it will load all [Namespace Files](https://kestra.io/docs/developer-guide/namespace-files) into the task's working directory. Note that this property is by default set to `true` so that you can specify only the `include` and `exclude` properties to filter the files to load without having to explicitly set `enabled` to `true`."
     )
     @Builder.Default
-    private Property<Boolean> enabled = Property.of(true);
+    private Property<Boolean> enabled = Property.ofValue(true);
 
     @Schema(
         title = "A list of filters to include only matching glob patterns. This allows you to only load a subset of the [Namespace Files](https://kestra.io/docs/developer-guide/namespace-files) into the working directory."
@@ -41,12 +41,12 @@ public class NamespaceFiles {
         title = "A list of namespaces in which searching files. The files are loaded in the namespace order, and only the latest version of a file is kept. Meaning if a file is present in the first and second namespace, only the file present on the second namespace will be loaded."
     )
     @Builder.Default
-    private Property<List<String>> namespaces = new Property<>("""
+    private Property<List<String>> namespaces = Property.ofExpression("""
         ["{{flow.namespace}}"]""");
 
     @Schema(
         title = "Comportment of the task if a file already exist in the working directory."
     )
     @Builder.Default
-    private Property<FileExistComportment> ifExists = Property.of(FileExistComportment.OVERWRITE);
+    private Property<FileExistComportment> ifExists = Property.ofValue(FileExistComportment.OVERWRITE);
 }