Add support for CEL filters to dispatcher

A dispatcher is now capable to filter out invalid events using CEL
expressions. This allows us to filter out invalid events on dispatcher
level and stop the request before reaching Tekton pipeline listeners.

JIRA: ISV-6225

Signed-off-by: Ales Raszka <[email protected]>

Author: Allda

ansible/inventory/group_vars/operator-pipeline-integration-tests-community.yml

@@ -18,6 +18,8 @@ operator_pipeline_dispatcher_config:
       max_capacity: "{{ operator_pipeline_dispatcher_hosted_capacity }}"
       namespace: "{{ oc_namespace }}"
     callback_url: "{{ operator_pipeline_community_pipeline_callback_url }}"
+    filter:
+      cel_expression: "{{ operator_pipeline_hosted_pipeline_cel_filter | replace('\n', '') }}"
 
   - name: Release pipeline for community operators
     events: "{{ operator_pipeline_dispatcher_release_pipeline_events }}"
@@ -28,3 +30,5 @@ operator_pipeline_dispatcher_config:
       max_capacity: "{{ operator_pipeline_dispatcher_release_capacity }}"
       namespace: "{{ oc_namespace }}"
     callback_url: "{{ operator_pipeline_community_pipeline_callback_url }}"
+    filter:
+      cel_expression: "{{ operator_pipeline_release_pipeline_cel_filter | replace('\n', '') }}"

ansible/inventory/group_vars/operator-pipeline-integration-tests.yml

@@ -49,6 +49,9 @@ operator_pipeline_dispatcher_config:
       max_capacity: "{{ operator_pipeline_dispatcher_hosted_capacity }}"
       namespace: "{{ oc_namespace }}"
     callback_url: "{{ operator_pipeline_callback_url }}"
+    filter:
+      cel_expression: "{{ operator_pipeline_hosted_pipeline_cel_filter | replace('\n', '') }}"
+
   - name: Release pipeline for certified operators
     events: "{{ operator_pipeline_dispatcher_release_pipeline_events }}"
     full_repository_name: "{{ integration_tests_git_upstream_repo }}"
@@ -58,3 +61,5 @@ operator_pipeline_dispatcher_config:
       max_capacity: "{{ operator_pipeline_dispatcher_release_capacity }}"
       namespace: "{{ oc_namespace }}"
     callback_url: "{{ operator_pipeline_callback_url }}"
+    filter:
+      cel_expression: "{{ operator_pipeline_release_pipeline_cel_filter | replace('\n', '') }}"

ansible/inventory/group_vars/operator-pipeline.yml

@@ -85,6 +85,34 @@ operator_pipeline_certified_operators_repository_name: "redhat-openshift-ecosyst
 operator_pipeline_marketplace_operators_repository_name: "redhat-openshift-ecosystem/redhat-marketplace-operators-preprod"
 operator_pipeline_community_operators_repository_name: "redhat-openshift-ecosystem/community-operators-pipeline-preprod"
 
+operator_pipeline_hosted_pipeline_cel_filter: >-
+  (
+    (
+      body.action in ["opened", "reopened", "synchronize", "ready_for_review"]
+      && body.pull_request.base.ref == "{{ branch }}"
+    ) ||
+    (
+      body.action == "labeled"
+      && body.label.name == "pipeline/trigger-hosted"
+      && body.pull_request.base.ref == "{{ branch }}"
+    )
+  )
+
+operator_pipeline_release_pipeline_cel_filter: >-
+  (
+    (
+      body.action == "closed"
+      && body.pull_request.base.ref == "{{ branch }}"
+      && body.pull_request.merged == true
+    ) ||
+    (
+      body.action == "labeled"
+      && body.label.name == "pipeline/trigger-release"
+      && body.pull_request.base.ref == "{{ branch }}"
+      && body.pull_request.merged == true
+    )
+  )
+
 operator_pipeline_dispatcher_config:
   - name: Hosted pipeline for certified operators
     events: "{{ operator_pipeline_dispatcher_hosted_pipeline_events }}"
@@ -95,6 +123,9 @@ operator_pipeline_dispatcher_config:
       max_capacity: "{{ operator_pipeline_dispatcher_hosted_capacity }}"
       namespace: "{{ oc_namespace }}"
     callback_url: "{{ operator_pipeline_callback_url }}"
+    filter:
+      cel_expression: "{{ operator_pipeline_hosted_pipeline_cel_filter | replace('\n', '') }}"
+
   - name: Release pipeline for certified operators
     events: "{{ operator_pipeline_dispatcher_release_pipeline_events }}"
     full_repository_name: "{{ operator_pipeline_certified_operators_repository_name }}"
@@ -104,6 +135,8 @@ operator_pipeline_dispatcher_config:
       max_capacity: "{{ operator_pipeline_dispatcher_release_capacity }}"
       namespace: "{{ oc_namespace }}"
     callback_url: "{{ operator_pipeline_callback_url }}"
+    filter:
+      cel_expression: "{{ operator_pipeline_release_pipeline_cel_filter | replace('\n', '') }}"
 
   - name: Hosted pipeline for marketplace operators
     events: "{{ operator_pipeline_dispatcher_hosted_pipeline_events }}"
@@ -114,6 +147,8 @@ operator_pipeline_dispatcher_config:
       max_capacity: "{{ operator_pipeline_dispatcher_hosted_capacity }}"
       namespace: "{{ oc_namespace }}"
     callback_url: "{{ operator_pipeline_callback_url }}"
+    filter:
+      cel_expression: "{{ operator_pipeline_hosted_pipeline_cel_filter | replace('\n', '') }}"
 
   - name: Release pipeline for marketplace operators
     events: "{{ operator_pipeline_dispatcher_release_pipeline_events }}"
@@ -124,6 +159,8 @@ operator_pipeline_dispatcher_config:
       max_capacity: "{{ operator_pipeline_dispatcher_release_capacity }}"
       namespace: "{{ oc_namespace }}"
     callback_url: "{{ operator_pipeline_callback_url }}"
+    filter:
+      cel_expression: "{{ operator_pipeline_release_pipeline_cel_filter | replace('\n', '') }}"
 
   - name: Hosted pipeline for community operators
     events: "{{ operator_pipeline_dispatcher_hosted_pipeline_events }}"
@@ -134,6 +171,8 @@ operator_pipeline_dispatcher_config:
       max_capacity: "{{ operator_pipeline_dispatcher_hosted_capacity }}"
       namespace: "{{ oc_namespace }}"
     callback_url: "{{ operator_pipeline_community_pipeline_callback_url }}"
+    filter:
+      cel_expression: "{{ operator_pipeline_hosted_pipeline_cel_filter | replace('\n', '') }}"
 
   - name: Release pipeline for community operators
     events: "{{ operator_pipeline_dispatcher_release_pipeline_events }}"
@@ -144,3 +183,5 @@ operator_pipeline_dispatcher_config:
       max_capacity: "{{ operator_pipeline_dispatcher_release_capacity }}"
       namespace: "{{ oc_namespace }}"
     callback_url: "{{ operator_pipeline_community_pipeline_callback_url }}"
+    filter:
+      cel_expression: "{{ operator_pipeline_release_pipeline_cel_filter | replace('\n', '') }}"

operator-pipeline-images/operatorcert/webhook_dispatcher/api.py

@@ -8,6 +8,7 @@
 
 import flask
 from flask import jsonify, request
+
 from operatorcert.webhook_dispatcher.config import load_config
 from operatorcert.webhook_dispatcher.database import get_database, get_db_session
 from operatorcert.webhook_dispatcher.models import (
@@ -43,6 +44,7 @@ def event_to_dict(event: WebhookEvent) -> dict[str, Any]:
     return {
         "id": event.id,
         "delivery_id": event.delivery_id,
+        "action": event.action,
         "repository_full_name": event.repository_full_name,
         "pull_request_number": event.pull_request_number,
         "processed": event.processed,

operator-pipeline-images/operatorcert/webhook_dispatcher/config.py

@@ -1,11 +1,13 @@
 """A module for configuration models."""
 
 import os
-from typing import List, Optional
+from typing import Any, List, Optional
 from urllib.parse import quote_plus
 
+import celpy
+import celpy.celparser
 import yaml
-from pydantic import BaseModel, Field
+from pydantic import BaseModel, ConfigDict, Field, field_validator
 
 
 class DatabaseConfig(BaseModel):
@@ -60,6 +62,42 @@ class CapacityConfig(BaseModel):
     namespace: str
 
 
+class Filter(BaseModel):
+    """Filter configuration for webhook events."""
+
+    model_config = ConfigDict(arbitrary_types_allowed=True)
+
+    cel_expression: celpy.Expression[Any] | None = Field(
+        default=None,
+        description="CEL expression to filter events. "
+        "If empty, no filtering is applied.",
+    )
+
+    @field_validator("cel_expression", mode="before")
+    @classmethod
+    def validate_cel_expression_and_compile(
+        cls, value: str
+    ) -> celpy.Expression[Any] | None:
+        """
+        Validate and parse the CEL expression syntax.
+
+        Raise ValueError if the expression is invalid.
+
+        Args:
+            value (celpy.Expression[Any] | None): A parsed cel expression if
+            available or None.
+        """
+        if not value:
+            return None
+
+        try:
+            env = celpy.Environment()
+            ast = env.compile(value)
+            return ast
+        except celpy.celparser.CELParseError as e:
+            raise ValueError(f"Invalid CEL expression: {e}") from e
+
+
 class DispatcherConfigItem(BaseModel):
     """Configuration for a webhook dispatcher item."""
 
@@ -68,6 +106,7 @@ class DispatcherConfigItem(BaseModel):
     full_repository_name: str
     callback_url: str
     capacity: CapacityConfig
+    filter: Optional[Filter] = None
 
 
 class DispatcherConfig(BaseModel):

operator-pipeline-images/operatorcert/webhook_dispatcher/dispatcher.py

@@ -3,10 +3,14 @@
 and dispatches them to the appropriate pipelines based on the configuration.
 """
 
+import asyncio
 import logging
-import time
 from datetime import datetime
+from typing import Any
 
+import celpy
+import celpy.adapter
+import celpy.evaluation
 from operatorcert.webhook_dispatcher.config import (
     DispatcherConfig,
     DispatcherConfigItem,
@@ -59,7 +63,7 @@ async def run(self) -> None:
 
             except Exception:  # pylint: disable=broad-except
                 LOGGER.exception("Error in event dispatcher")
-            time.sleep(5)
+            await asyncio.sleep(5)
 
     async def process_repository_events(
         self, repository_events: dict[int, list[WebhookEvent]]
@@ -154,9 +158,50 @@ def assign_configs_to_event(
                 continue
             if event.action not in item.events:
                 continue
+            if item.filter and item.filter.cel_expression:
+                if not self.match_cel_expression(
+                    event,
+                    item.filter.cel_expression,
+                ):
+                    LOGGER.debug(
+                        "Event %s did not match the CEL expression for config %s",
+                        event.id,
+                        item.name,
+                    )
+                    continue
             configs.append(item)
         return configs
 
+    def match_cel_expression(
+        self,
+        event: WebhookEvent,
+        cel_expression: celpy.Expression[Any],
+    ) -> bool:
+        """
+        Match the event against a CEL expression and return whether it matches.
+
+        Args:
+            event (WebhookEvent): A WebhookEvent object containing event data.
+            cel_expression (celpy.Expression): A CEL expression to match against the event.
+
+        Returns:
+            bool: A True if the event matches the CEL expression, False otherwise.
+        """
+        try:
+            env = celpy.Environment()
+            program = env.program(cel_expression)
+
+            context = {
+                "body": celpy.adapter.json_to_cel(event.payload),
+                "headers": celpy.adapter.json_to_cel(event.request_headers),
+            }
+
+            result = program.evaluate(context)
+            return bool(result)
+        except (celpy.evaluation.CELEvalError, celpy.evaluation.CELUnsupportedError):
+            LOGGER.exception("Failed to evaluate CEL.")
+            return False
+
     def convert_to_pipeline_events(self, event: WebhookEvent) -> list[PipelineEvent]:
         """
         Convert a WebhookEvent to a PipelineEvents with the assigned configuration.
@@ -198,6 +243,10 @@ async def process_pipeline_event(self, pipeline_event: PipelineEvent) -> None:
             # Don't mark the event as processed, so it will be retried in the next loop
             return
 
+        # Give some time for the pipeline to start so it can be tracked
+        # by the capacity manager in the next loop
+        await asyncio.sleep(10)
+
         event.processed = True
         event.processing_error = None
         event.processed_at = datetime.now()

operator-pipeline-images/tests/webhook_dispatcher/test_api.py

@@ -8,6 +8,7 @@
     CapacityConfig,
     DispatcherConfig,
     DispatcherConfigItem,
+    Filter,
     SecurityConfig,
     WebhookDispatcherConfig,
 )
@@ -30,6 +31,7 @@ def test_get_config(mock_load_config: MagicMock) -> None:
                         max_capacity=10,
                         namespace="test",
                     ),
+                    filter=Filter(cel_expression="body.action == 'pull_request'"),  # type: ignore[arg-type]
                 )
             ],
         ),
@@ -48,6 +50,7 @@ def test_event_to_dict() -> None:
     event = WebhookEvent(
         id=1,
         delivery_id="123",
+        action="opened",
         repository_full_name="test/test",
         pull_request_number=123,
         processed=False,
@@ -59,6 +62,7 @@ def test_event_to_dict() -> None:
     assert api.event_to_dict(event) == {
         "id": 1,
         "delivery_id": "123",
+        "action": "opened",
         "repository_full_name": "test/test",
         "pull_request_number": 123,
         "processed": False,
@@ -97,6 +101,7 @@ def test_github_pipeline_webhook(
                         max_capacity=10,
                         namespace="test",
                     ),
+                    filter=None,
                 )
             ],
         ),

operator-pipeline-images/tests/webhook_dispatcher/test_config.py

@@ -1,6 +1,9 @@
 from unittest.mock import MagicMock, patch
 
-from operatorcert.webhook_dispatcher.config import load_config
+import lark
+from operatorcert.webhook_dispatcher.config import Filter, load_config
+import pytest
+from pydantic import ValidationError
 
 
 @patch("operatorcert.webhook_dispatcher.config.yaml.safe_load")
@@ -21,6 +24,9 @@ def test_load_config(mock_open: MagicMock, mock_yaml_load: MagicMock) -> None:
                         "pipeline_name": "test",
                         "namespace": "test",
                     },
+                    "filter": {
+                        "cel_expression": "body.action == 'push'",
+                    },
                 }
             ]
         },
@@ -34,3 +40,17 @@ def test_load_config(mock_open: MagicMock, mock_yaml_load: MagicMock) -> None:
     assert config.dispatcher.items[0].name == "test"
     assert config.dispatcher.items[0].events == ["push"]
     assert config.dispatcher.items[0].full_repository_name == "test/test"
+
+
+def test_cel_expression_compilation() -> None:
+    # Valid expression
+    filter_config = Filter(cel_expression='body.action == "push"')  # type: ignore[arg-type]
+    assert isinstance(filter_config.cel_expression, lark.Tree)
+
+    # Empty expression
+    filter_config_no_expr = Filter(cel_expression="")  # type: ignore[arg-type]
+    assert filter_config_no_expr.cel_expression is None
+
+    # Invalid expression
+    with pytest.raises(ValidationError):
+        Filter(cel_expression="body.action = push")  # type: ignore[arg-type]