Repository: kestra-io/kestra (java)

Issue:
Webhook trigger : allow to specify condition on payload to trigger execution
### Feature description

For some APIs we might want to check a condition (key exists, value is a specific one we are interested in) to decide to trigger our flow or not.

Currently this is doable within the flow definition but add some complexity, still generate flow execution and pollute history.

=> Can be nice to have an optional parameter `payloadCondition` which would contain our condition.

Flow could look like : 

```YAML
triggers:
  - id: webhook
    type: io.kestra.core.models.triggers.types.Webhook
    key: 4wjtkzwVGBM9yKnjm3yv8r
    payloadCondition: {{ trigger.body.pull_request.comments_url ?? true : false }}
```


Pull request:
feat(core,webserver): support conditions on webhooks
@anna-geller, I took the occasion to slightly improve the documentation, can you have a look?

close #2267


Patch:
diff --git a/core/src/main/java/io/kestra/core/metrics/MetricRegistry.java b/core/src/main/java/io/kestra/core/metrics/MetricRegistry.java
index 5ce504f30..d271142bb 100644
--- a/core/src/main/java/io/kestra/core/metrics/MetricRegistry.java
+++ b/core/src/main/java/io/kestra/core/metrics/MetricRegistry.java
@@ -1,6 +1,7 @@
 package io.kestra.core.metrics;
 
 import io.kestra.core.models.executions.Execution;
+import io.kestra.core.models.executions.TaskRun;
 import io.kestra.core.models.tasks.Task;
 import io.kestra.core.models.triggers.TriggerContext;
 import io.kestra.core.runners.WorkerTask;
@@ -60,6 +61,7 @@ public class MetricRegistry {
     public final static String TAG_STATE = "state";
     public final static String TAG_ATTEMPT_COUNT = "attempt_count";
     public final static String TAG_WORKER_GROUP = "worker_group";
+    public final static String TAG_TENANT_ID = "tenant_id";
 
     @Inject
     private MeterRegistry meterRegistry;
@@ -126,14 +128,14 @@ public class MetricRegistry {
 
     /**
      * Return tags for current {@link WorkerTask}.
-     * We don't include current state since it will breakup the values per state and it's make no sense.
+     * We don't include current state since it will break up the values per state which make no sense.
      *
      * @param workerTask the current WorkerTask
      * @param workerGroup the worker group, optional
      * @return tags to applied to metrics
      */
     public String[] tags(WorkerTask workerTask, String workerGroup, String... tags) {
-        var finalTags = ArrayUtils.addAll(
+        var baseTags = ArrayUtils.addAll(
             ArrayUtils.addAll(
                 this.tags(workerTask.getTask()),
                 tags
@@ -141,7 +143,8 @@ public class MetricRegistry {
             TAG_NAMESPACE_ID, workerTask.getTaskRun().getNamespace(),
             TAG_FLOW_ID, workerTask.getTaskRun().getFlowId()
         );
-        return workerGroup != null ? ArrayUtils.addAll(finalTags, TAG_WORKER_GROUP, workerGroup) : finalTags;
+        baseTags = workerGroup == null ? baseTags : ArrayUtils.addAll(baseTags, TAG_WORKER_GROUP, workerGroup);
+        return workerTask.getTaskRun().getTenantId() == null ? baseTags : ArrayUtils.addAll(baseTags, TAG_TENANT_ID, workerTask.getTaskRun().getTenantId());
     }
 
     /**
@@ -151,12 +154,13 @@ public class MetricRegistry {
      * @return tags to applied to metrics
      */
     public String[] tags(WorkerTaskResult workerTaskResult, String... tags) {
-        return ArrayUtils.addAll(
+        var baseTags = ArrayUtils.addAll(
             tags,
             TAG_NAMESPACE_ID, workerTaskResult.getTaskRun().getNamespace(),
             TAG_FLOW_ID, workerTaskResult.getTaskRun().getFlowId(),
             TAG_STATE, workerTaskResult.getTaskRun().getState().getCurrent().name()
         );
+        return workerTaskResult.getTaskRun().getTenantId() == null ? baseTags : ArrayUtils.addAll(baseTags, TAG_TENANT_ID, workerTaskResult.getTaskRun().getTenantId());
     }
 
     /**
@@ -178,11 +182,12 @@ public class MetricRegistry {
      * @return tags to applied to metrics
      */
     public String[] tags(Execution execution) {
-        return new String[]{
+        var baseTags = new String[]{
             TAG_FLOW_ID, execution.getFlowId(),
             TAG_NAMESPACE_ID, execution.getNamespace(),
             TAG_STATE, execution.getState().getCurrent().name(),
         };
+        return execution.getTenantId() == null ? baseTags : ArrayUtils.addAll(baseTags, TAG_TENANT_ID, execution.getTenantId());
     }
 
     /**
@@ -193,11 +198,12 @@ public class MetricRegistry {
      * @return tags to applied to metrics
      */
     public String[] tags(TriggerContext triggerContext, String workerGroup) {
-        var finalTags = new String[]{
+        var baseTags = new String[]{
             TAG_FLOW_ID, triggerContext.getFlowId(),
             TAG_NAMESPACE_ID, triggerContext.getNamespace()
         };
-        return workerGroup != null ? ArrayUtils.addAll(finalTags, TAG_WORKER_GROUP, workerGroup) : finalTags;
+        baseTags =  workerGroup == null ? baseTags : ArrayUtils.addAll(baseTags, TAG_WORKER_GROUP, workerGroup);
+        return triggerContext.getTenantId() == null ? baseTags : ArrayUtils.addAll(baseTags, TAG_TENANT_ID, triggerContext.getTenantId());
     }
 
     /**
@@ -207,10 +213,7 @@ public class MetricRegistry {
      * @return tags to applied to metrics
      */
     public String[] tags(TriggerContext triggerContext) {
-        return new String[]{
-            TAG_FLOW_ID, triggerContext.getFlowId(),
-            TAG_NAMESPACE_ID, triggerContext.getNamespace(),
-        };
+        return tags(triggerContext, null);
     }
 
     /**
diff --git a/core/src/main/java/io/kestra/core/models/triggers/types/Webhook.java b/core/src/main/java/io/kestra/core/models/triggers/types/Webhook.java
index 1c044ae58..4440bc3c6 100644
--- a/core/src/main/java/io/kestra/core/models/triggers/types/Webhook.java
+++ b/core/src/main/java/io/kestra/core/models/triggers/types/Webhook.java
@@ -31,12 +31,21 @@ import javax.validation.constraints.Size;
 @NoArgsConstructor
 @Schema(
     title = "Trigger a flow from a webhook",
-    description = "Webhook trigger allow you to trigger a flow from a webhook url. " +
-        "The trigger will generate a `key` that must be used on url : `/api/v1/executions/webhook/{namespace}/[flowId]/{key}`. " +
-        "Kestra accept `GET`, `POST` & `PUT` request on this url. " +
-        "The whole body & headers will be available as variable:\n " +
-        "- `trigger.body`\n" +
-        "- `trigger.headers`"
+    description = """
+        Webhook trigger allows you to create a unique URL that you can use to trigger a Kestra flow execution based on a presence of events in another application such as GitHub or Amazon EventBridge. In order to use that URL, you have to add a secret key that will secure your webhook URL. 
+        
+        The URL will then follow the following format: `https://{your_hostname}/api/v1/executions/webhook/{namespace}/{flowId}/{key}`. Replace the templated values accordingly to your workflow setup.
+        
+        The webhook URL accepts `GET`, `POST` and `PUT` requests.
+
+        You can access the request body and headers sent by another application using the following template variables:
+        - `{{ trigger.body }}`
+        - `{{ trigger.headers }}`.
+
+        The webhook response will be one of the following HTTP status codes:
+        - 404 if the namespace, flow or webhook key is not found
+        - 200 if the webhook triggers an execution
+        - 204 if the webhook cannot trigger an execution due to a lack of matching event conditions sent by other application."""
 )
 @Plugin(
     examples = {
@@ -49,6 +58,21 @@ import javax.validation.constraints.Size;
                 "    key: 4wjtkzwVGBM9yKnjm3yv8r"
             },
             full = true
+        ),
+        @Example(
+            title = """
+                Add a trigger matching specific webhook event condition. The flow will be executed only if the condition is met.`.
+                """,
+            code = {
+                "triggers:",
+                "  - id: webhook",
+                "    type: io.kestra.core.models.triggers.types.Webhook",
+                "    key: 4wjtkzwVGBM9yKnjm3yv8r",
+                "    conditions:",
+                "      - type: io.kestra.core.models.conditions.types.VariableCondition",
+                "        expression: \"{{ trigger.body.hello == 'world' }}\""
+            },
+            full = true
         )
     }
 )
@@ -60,7 +84,7 @@ public class Webhook extends AbstractTrigger implements TriggerOutput<Webhook.Ou
         description = "The key is used for generating the url of the webhook.\n" +
             "\n" +
             "::alert{type=\"warning\"}\n" +
-            "Take care when using manual key, the key is the only security to protect your webhook and must be considered as a secret !\n" +
+            "Make sure to keep the webhook key secure. It's the only security mechanism to protect your endpoint from bad actors, and must be considered as a secret. You can use a random key generator to create the key.\n" +
             "::\n"
     )
     @PluginProperty(dynamic = true)
diff --git a/ui/src/override/components/LeftMenu.vue b/ui/src/override/components/LeftMenu.vue
index 04823a82e..6c320e2ec 100644
--- a/ui/src/override/components/LeftMenu.vue
+++ b/ui/src/override/components/LeftMenu.vue
@@ -83,6 +83,9 @@
                         return r;
                     })
             },
+            routeStartWith(route) {
+                return this.$router.getRoutes().filter(r => r.name.startsWith(route)).map(r => r.name);
+            },
             generateMenu() {
                 return [
                     {
@@ -95,6 +98,7 @@
                     },
                     {
                         href: {name: "flows/list"},
+                        routes: this.routeStartWith("flows"),
                         title: this.$t("flows"),
                         icon: {
                             element: FileTreeOutline,
@@ -104,6 +108,7 @@
                     },
                     {
                         href: {name: "templates/list"},
+                        routes: this.routeStartWith("templates"),
                         title: this.$t("templates"),
                         icon: {
                             element: ContentCopy,
@@ -113,6 +118,7 @@
                     },
                     {
                         href: {name: "executions/list"},
+                        routes: this.routeStartWith("executions"),
                         title: this.$t("executions"),
                         icon: {
                             element: TimelineClockOutline,
@@ -121,6 +127,7 @@
                     },
                     {
                         href: {name: "taskruns/list"},
+                        routes: this.routeStartWith("taskruns"),
                         title: this.$t("taskruns"),
                         icon: {
                             element: TimelineTextOutline,
@@ -130,6 +137,7 @@
                     },
                     {
                         href: {name: "logs/list"},
+                        routes: this.routeStartWith("logs"),
                         title: this.$t("logs"),
                         icon: {
                             element: NotebookOutline,
@@ -138,7 +146,7 @@
                     },
                     {
                         href: {name: "blueprints"},
-                        routes: ["blueprints", "blueprints/view"],
+                        routes: this.routeStartWith("blueprints"),
                         title: this.$t("blueprints.title"),
                         icon: {
                             element: Ballot,
@@ -192,6 +200,7 @@
                     },
                     {
                         title: this.$t("administration"),
+                        routes: this.routeStartWith("admin"),
                         icon: {
                             element: AccountSupervisorOutline,
                             class: "menu-icon"
@@ -199,6 +208,7 @@
                         child: [
                             {
                                 href: {name: "admin/triggers"},
+                                routes: this.routeStartWith("admin/triggers"),
                                 title: this.$t("triggers"),
                                 icon: {
                                     element: TimerCogOutline,
@@ -207,6 +217,7 @@
                             },
                             {
                                 href: {name: "admin/workers"},
+                                routes: this.routeStartWith("admin/workers"),
                                 title: this.$t("workers"),
                                 icon: {
                                     element: AccountHardHatOutline,
@@ -217,6 +228,7 @@
                     },
                     {
                         href: {name: "settings"},
+                        routes: this.routeStartWith("admin/settings"),
                         title: this.$t("settings"),
                         icon: {
                             element: CogOutline,
diff --git a/webserver/src/main/java/io/kestra/webserver/controllers/ExecutionController.java b/webserver/src/main/java/io/kestra/webserver/controllers/ExecutionController.java
index 9a74902a9..be2d489b4 100644
--- a/webserver/src/main/java/io/kestra/webserver/controllers/ExecutionController.java
+++ b/webserver/src/main/java/io/kestra/webserver/controllers/ExecutionController.java
@@ -15,6 +15,7 @@ import io.kestra.core.models.hierarchies.FlowGraph;
 import io.kestra.core.models.storage.FileMetas;
 import io.kestra.core.models.tasks.Task;
 import io.kestra.core.models.triggers.AbstractTrigger;
+import io.kestra.core.models.triggers.multipleflows.MultipleConditionStorageInterface;
 import io.kestra.core.models.triggers.types.Webhook;
 import io.kestra.core.models.validations.ManualConstraintViolation;
 import io.kestra.core.queues.QueueFactoryInterface;
@@ -129,6 +130,9 @@ public class ExecutionController {
     @Inject
     private TenantService tenantService;
 
+    @Inject
+    private MultipleConditionStorageInterface multipleConditionStorageInterface;
+
     @ExecuteOn(TaskExecutors.IO)
     @Get(uri = "/search", produces = MediaType.TEXT_JSON)
     @Operation(tags = {"Executions"}, summary = "Search for executions")
@@ -343,7 +347,7 @@ public class ExecutionController {
     @ExecuteOn(TaskExecutors.IO)
     @Post(uri = "/webhook/{namespace}/{id}/{key}", produces = MediaType.TEXT_JSON)
     @Operation(tags = {"Executions"}, summary = "Trigger a new execution by POST webhook trigger")
-    public Execution webhookTriggerPost(
+    public HttpResponse<Execution> webhookTriggerPost(
         @Parameter(description = "The flow namespace") @PathVariable String namespace,
         @Parameter(description = "The flow id") @PathVariable String id,
         @Parameter(description = "The webhook trigger uid") @PathVariable String key,
@@ -355,7 +359,7 @@ public class ExecutionController {
     @ExecuteOn(TaskExecutors.IO)
     @Get(uri = "/webhook/{namespace}/{id}/{key}", produces = MediaType.TEXT_JSON)
     @Operation(tags = {"Executions"}, summary = "Trigger a new execution by GET webhook trigger")
-    public Execution webhookTriggerGet(
+    public HttpResponse<Execution> webhookTriggerGet(
         @Parameter(description = "The flow namespace") @PathVariable String namespace,
         @Parameter(description = "The flow id") @PathVariable String id,
         @Parameter(description = "The webhook trigger uid") @PathVariable String key,
@@ -367,7 +371,7 @@ public class ExecutionController {
     @ExecuteOn(TaskExecutors.IO)
     @Put(uri = "/webhook/{namespace}/{id}/{key}", produces = MediaType.TEXT_JSON)
     @Operation(tags = {"Executions"}, summary = "Trigger a new execution by PUT webhook trigger")
-    public Execution webhookTriggerPut(
+    public HttpResponse<Execution> webhookTriggerPut(
         @Parameter(description = "The flow namespace") @PathVariable String namespace,
         @Parameter(description = "The flow id") @PathVariable String id,
         @Parameter(description = "The webhook trigger uid") @PathVariable String key,
@@ -376,7 +380,7 @@ public class ExecutionController {
         return this.webhook(namespace, id, key, request);
     }
 
-    private Execution webhook(
+    private HttpResponse<Execution> webhook(
         String namespace,
         String id,
         String key,
@@ -384,7 +388,7 @@ public class ExecutionController {
     ) {
         Optional<Flow> find = flowRepository.findById(tenantService.resolveTenant(), namespace, id);
         if (find.isEmpty()) {
-            return null;
+            return HttpResponse.notFound();
         }
 
         var flow = find.get();
@@ -414,23 +418,29 @@ public class ExecutionController {
             .findFirst();
 
         if (webhook.isEmpty()) {
-            return null;
+            return HttpResponse.notFound();
         }
 
         Optional<Execution> execution = webhook.get().evaluate(request, flow);
 
         if (execution.isEmpty()) {
-            return null;
+            return HttpResponse.notFound();
         }
 
         var result = execution.get();
         if (flow.getLabels() != null) {
             result = result.withLabels(flow.getLabels());
         }
+
+        // we check conditions here as it's easier as the execution is created we have the body and headers available for the runContext
+        if (!conditionService.isValid(webhook.get(), flow, result, multipleConditionStorageInterface)) {
+            return HttpResponse.noContent();
+        }
+
         executionQueue.emit(result);
         eventPublisher.publishEvent(new CrudEvent<>(result, CrudEventType.CREATE));
 
-        return result;
+        return HttpResponse.ok(result);
     }
 
     @ExecuteOn(TaskExecutors.IO)


Test patch:
diff --git a/core/src/test/resources/flows/valids/webhook-with-condition.yaml b/core/src/test/resources/flows/valids/webhook-with-condition.yaml
new file mode 100644
index 000000000..08de8f366
--- /dev/null
+++ b/core/src/test/resources/flows/valids/webhook-with-condition.yaml
@@ -0,0 +1,15 @@
+id: webhook-with-condition
+namespace: io.kestra.tests
+
+tasks:
+  - id: out
+    type: io.kestra.core.tasks.debugs.Return
+    format: "{{ trigger | json }}"
+
+triggers:
+  - id: webhook
+    type: io.kestra.core.models.triggers.types.Webhook
+    key: webhookKey
+    conditions:
+      - type: io.kestra.core.models.conditions.types.VariableCondition
+        expression: "{{trigger.body.hello == 'world'}}"
\ No newline at end of file
diff --git a/webserver/src/test/java/io/kestra/webserver/controllers/ExecutionControllerTest.java b/webserver/src/test/java/io/kestra/webserver/controllers/ExecutionControllerTest.java
index aefdd3299..43a1af713 100644
--- a/webserver/src/test/java/io/kestra/webserver/controllers/ExecutionControllerTest.java
+++ b/webserver/src/test/java/io/kestra/webserver/controllers/ExecutionControllerTest.java
@@ -542,6 +542,34 @@ class ExecutionControllerTest extends JdbcH2ControllerTest {
         assertThat(execution.getId(), notNullValue());
     }
 
+    @Test
+    void webhookWithCondition() {
+        record Hello(String hello) {}
+
+        Execution execution = client.toBlocking().retrieve(
+            HttpRequest
+                .POST(
+                    "/api/v1/executions/webhook/" + TESTS_FLOW_NS + "/webhook-with-condition/webhookKey",
+                    new Hello("world")
+                ),
+            Execution.class
+        );
+
+        assertThat(execution, notNullValue());
+        assertThat(execution.getId(), notNullValue());
+
+        HttpResponse<Execution> response = client.toBlocking().exchange(
+            HttpRequest
+                .POST(
+                    "/api/v1/executions/webhook/" + TESTS_FLOW_NS + "/webhook-with-condition/webhookKey",
+                    new Hello("webhook")
+                ),
+            Execution.class
+        );
+        assertThat(response.getStatus(), is(HttpStatus.NO_CONTENT));
+        assertThat(response.body(), nullValue());
+    }
+
     @Test
     void resumePaused() throws TimeoutException, InterruptedException {
         // Run execution until it is paused