✨ feat(aci): add task for trigger actions (#95159)

Author: iamrajjoshi

src/sentry/conf/server.py

@@ -853,6 +853,8 @@ def SOCIAL_AUTH_DEFAULT_USERNAME() -> str:
     "sentry.integrations.vsts.tasks.kickoff_subscription_check",
     "sentry.integrations.tasks",
     "sentry.demo_mode.tasks",
+    "sentry.workflow_engine.tasks.workflows",
+    "sentry.workflow_engine.tasks.actions",
 )
 
 # Enable split queue routing
@@ -1023,6 +1025,7 @@ def SOCIAL_AUTH_DEFAULT_USERNAME() -> str:
     Queue("release_registry", routing_key="release_registry"),
     Queue("seer.seer_automation", routing_key="seer.seer_automation"),
     Queue("workflow_engine.process_workflows", routing_key="workflow_engine.process_workflows"),
+    Queue("workflow_engine.trigger_action", routing_key="workflow_engine.trigger_action"),
 ]
 
 from celery.schedules import crontab
@@ -1516,6 +1519,8 @@ def SOCIAL_AUTH_DEFAULT_USERNAME() -> str:
     "sentry.uptime.rdap.tasks",
     "sentry.uptime.subscriptions.tasks",
     "sentry.workflow_engine.processors.delayed_workflow",
+    "sentry.workflow_engine.tasks.workflows",
+    "sentry.workflow_engine.tasks.actions",
     # Used for tests
     "sentry.taskworker.tasks.examples",
 )

src/sentry/features/temporary.py

@@ -515,6 +515,8 @@ def register_temporary_features(manager: FeatureManager):
     manager.add("organizations:workflow-engine-rule-serializers", OrganizationFeature, FeatureHandlerStrategy.FLAGPOLE, api_expose=False)
     # Enable async processing of event workflows in a task rather than as part of post_process.
     manager.add("organizations:workflow-engine-post-process-async", OrganizationFeature, FeatureHandlerStrategy.FLAGPOLE, api_expose=False)
+    # Enable async processing of actions in a task rather than as part of post_process.
+    manager.add("organizations:workflow-engine-action-trigger-async", OrganizationFeature, FeatureHandlerStrategy.FLAGPOLE, api_expose=False)
     # Enable EventUniqueUserFrequencyConditionWithConditions special alert condition
     manager.add("organizations:event-unique-user-frequency-condition-with-conditions", OrganizationFeature, FeatureHandlerStrategy.FLAGPOLE, api_expose=True)
     # Use spans instead of transactions for dynamic sampling calculations. This will become the new default.

src/sentry/tasks/post_process.py

@@ -956,7 +956,7 @@ def process_workflow_engine(job: PostProcessJob) -> None:
     metrics.incr("workflow_engine.issue_platform.payload.received.occurrence")
 
     from sentry.workflow_engine.processors.workflow import process_workflows
-    from sentry.workflow_engine.tasks import process_workflows_event
+    from sentry.workflow_engine.tasks.workflows import process_workflows_event
 
     # PostProcessJob event is optional, WorkflowEventData event is required
     if "event" not in job:

src/sentry/workflow_engine/models/action.py

@@ -14,6 +14,7 @@
 from sentry.backup.scopes import RelocationScope
 from sentry.db.models import DefaultFieldsModel, region_silo_model, sane_repr
 from sentry.db.models.fields.hybrid_cloud_foreign_key import HybridCloudForeignKey
+from sentry.utils import metrics
 from sentry.workflow_engine.models.json_config import JSONConfigBase
 from sentry.workflow_engine.registry import action_handler_registry
 from sentry.workflow_engine.types import ActionHandler, WorkflowEventData
@@ -88,6 +89,11 @@ def trigger(self, event_data: WorkflowEventData, detector: Detector) -> None:
         handler = self.get_handler()
         handler.execute(event_data, self, detector)
 
+        metrics.incr(
+            "workflow_engine.action.trigger",
+            tags={"action_type": self.type, "detector_type": detector.type},
+        )
+
         logger.info(
             "workflow_engine.action.trigger",
             extra={

src/sentry/workflow_engine/processors/delayed_workflow.py

@@ -61,6 +61,7 @@
     evaluate_workflows_action_filters,
 )
 from sentry.workflow_engine.processors.workflow_fire_history import create_workflow_fire_histories
+from sentry.workflow_engine.tasks.actions import build_trigger_action_task_params, trigger_action
 from sentry.workflow_engine.types import DataConditionHandler, WorkflowEventData
 from sentry.workflow_engine.utils import log_context
 
@@ -646,7 +647,16 @@ def fire_actions_for_groups(
                     organization,
                 ):
                     for action in filtered_actions:
-                        action.trigger(workflow_event_data, detector)
+                        if features.has(
+                            "organizations:workflow-engine-action-trigger-async",
+                            organization,
+                        ):
+                            task_params = build_trigger_action_task_params(
+                                action, detector, workflow_event_data
+                            )
+                            trigger_action.delay(**task_params)
+                        else:
+                            action.trigger(workflow_event_data, detector)
 
     logger.info(
         "workflow_engine.delayed_workflow.triggered_actions_summary",

src/sentry/workflow_engine/processors/workflow.py

@@ -29,6 +29,7 @@
 from sentry.workflow_engine.processors.data_condition_group import process_data_condition_group
 from sentry.workflow_engine.processors.detector import get_detector_by_event
 from sentry.workflow_engine.processors.workflow_fire_history import create_workflow_fire_histories
+from sentry.workflow_engine.tasks.actions import build_trigger_action_task_params, trigger_action
 from sentry.workflow_engine.types import WorkflowEventData
 from sentry.workflow_engine.utils import log_context
 from sentry.workflow_engine.utils.metrics import metrics_incr
@@ -414,19 +415,14 @@ def process_workflows(
             organization,
         ):
             for action in actions:
-                action.trigger(event_data, detector)
-                metrics_incr(
-                    "action.trigger",
-                    tags={"action_type": action.type},
-                )
-
-                logger.info(
-                    "workflow_engine.action.trigger",
-                    extra={
-                        "action_id": action.id,
-                        "event_data": asdict(event_data),
-                    },
-                )
+                if features.has(
+                    "organizations:workflow-engine-action-trigger-async",
+                    organization,
+                ):
+                    task_params = build_trigger_action_task_params(action, detector, event_data)
+                    trigger_action.delay(**task_params)
+                else:
+                    action.trigger(event_data, detector)
         else:
             logger.info(
                 "workflow_engine.triggered_actions",

src/sentry/workflow_engine/tasks/__init__.py

@@ -0,0 +1,123 @@
+from django.db.models import Value
+
+from sentry.eventstore.models import GroupEvent
+from sentry.eventstream.base import GroupState
+from sentry.models.activity import Activity
+from sentry.silo.base import SiloMode
+from sentry.tasks.base import instrumented_task
+from sentry.taskworker import config, namespaces
+from sentry.taskworker.retry import Retry
+from sentry.utils import metrics
+from sentry.workflow_engine.models import Action, Detector
+from sentry.workflow_engine.tasks.utils import build_workflow_event_data_from_event
+from sentry.workflow_engine.types import WorkflowEventData
+from sentry.workflow_engine.utils import log_context
+
+logger = log_context.get_logger(__name__)
+
+
+def build_trigger_action_task_params(action, detector, event_data: WorkflowEventData):
+    """Build parameters for trigger_action.delay() call."""
+    event_id = None
+    activity_id = None
+    occurrence_id = None
+
+    if isinstance(event_data.event, GroupEvent):
+        event_id = event_data.event.event_id
+        occurrence_id = event_data.event.occurrence_id
+    elif isinstance(event_data.event, Activity):
+        activity_id = event_data.event.id
+
+    return {
+        "action_id": action.id,
+        "detector_id": detector.id,
+        "workflow_id": getattr(action, "workflow_id", None),
+        "event_id": event_id,
+        "activity_id": activity_id,
+        "group_id": event_data.event.group_id,
+        "occurrence_id": occurrence_id,
+        "group_state": event_data.group_state,
+        "has_reappeared": event_data.has_reappeared,
+        "has_escalated": event_data.has_escalated,
+        "workflow_env_id": event_data.workflow_env.id if event_data.workflow_env else None,
+    }
+
+
+@instrumented_task(
+    name="sentry.workflow_engine.tasks.trigger_action",
+    queue="workflow_engine.trigger_action",
+    acks_late=True,
+    default_retry_delay=5,
+    max_retries=3,
+    soft_time_limit=25,
+    time_limit=30,
+    silo_mode=SiloMode.REGION,
+    taskworker_config=config.TaskworkerConfig(
+        namespace=namespaces.workflow_engine_tasks,
+        processing_deadline_duration=30,
+        retry=Retry(
+            times=3,
+            delay=5,
+        ),
+    ),
+)
+def trigger_action(
+    action_id: int,
+    detector_id: int,
+    workflow_id: int,
+    event_id: str | None,
+    activity_id: int | None,
+    group_id: int,
+    occurrence_id: str | None,
+    group_state: GroupState,
+    has_reappeared: bool,
+    has_escalated: bool,
+    workflow_env_id: int | None,
+) -> None:
+
+    # XOR check to ensure exactly one of event_id or activity_id is provided
+    if (event_id is not None) == (activity_id is not None):
+        logger.error(
+            "Exactly one of event_id or activity_id must be provided",
+            extra={"event_id": event_id, "activity_id": activity_id},
+        )
+        raise ValueError("Exactly one of event_id or activity_id must be provided")
+
+    action = Action.objects.annotate(workflow_id=Value(workflow_id)).get(id=action_id)
+    detector = Detector.objects.get(id=detector_id)
+
+    metrics.incr(
+        "workflow_engine.tasks.trigger_action_task_started",
+        tags={"action_type": action.type, "detector_type": detector.type},
+        sample_rate=1.0,
+    )
+
+    project_id = detector.project_id
+
+    if event_id is not None:
+        event_data = build_workflow_event_data_from_event(
+            project_id=project_id,
+            event_id=event_id,
+            group_id=group_id,
+            occurrence_id=occurrence_id,
+            group_state=group_state,
+            has_reappeared=has_reappeared,
+            has_escalated=has_escalated,
+            workflow_env_id=workflow_env_id,
+        )
+
+        # TODO(iamrajjoshi): remove this once we are sure everything is working as expected
+        logger.info(
+            "workflow_engine.tasks.trigger_action.build_workflow_event_data_from_event",
+            extra={
+                "action_id": action_id,
+                "detector_id": detector_id,
+                "workflow_id": workflow_id,
+            },
+        )
+
+    else:
+        # Here, we probably build the event data from the activity
+        raise NotImplementedError("Activity ID is not supported yet")
+
+    action.trigger(event_data, detector)

src/sentry/workflow_engine/tasks/actions.py

@@ -0,0 +1,84 @@
+from google.api_core.exceptions import DeadlineExceeded, RetryError, ServiceUnavailable
+
+from sentry import nodestore
+from sentry.eventstore.models import Event, GroupEvent
+from sentry.eventstream.base import GroupState
+from sentry.issues.issue_occurrence import IssueOccurrence
+from sentry.models.environment import Environment
+from sentry.models.group import Group
+from sentry.types.activity import ActivityType
+from sentry.utils.retries import ConditionalRetryPolicy, exponential_delay
+from sentry.workflow_engine.types import WorkflowEventData
+from sentry.workflow_engine.utils import log_context
+
+SUPPORTED_ACTIVITIES = [ActivityType.SET_RESOLVED.value]
+
+
+logger = log_context.get_logger(__name__)
+
+
+def _should_retry_nodestore_fetch(attempt: int, e: Exception) -> bool:
+    return not attempt > 3 and (
+        # ServiceUnavailable and DeadlineExceeded are generally retriable;
+        # we also include RetryError because the nodestore interface doesn't let
+        # us specify a timeout to BigTable and the default is 5s; see c5e2b40.
+        isinstance(e, (ServiceUnavailable, RetryError, DeadlineExceeded))
+    )
+
+
+def fetch_event(event_id: str, project_id: int) -> Event | None:
+    """
+    Fetch a single Event, with retries.
+    """
+    node_id = Event.generate_node_id(project_id, event_id)
+    fetch_retry_policy = ConditionalRetryPolicy(
+        _should_retry_nodestore_fetch, exponential_delay(1.00)
+    )
+    data = fetch_retry_policy(lambda: nodestore.backend.get(node_id))
+    if data is None:
+        return None
+    return Event(
+        event_id=event_id,
+        project_id=project_id,
+        data=data,
+    )
+
+
+def build_workflow_event_data_from_event(
+    project_id: int,
+    event_id: str,
+    group_id: int,
+    occurrence_id: str | None = None,
+    group_state: GroupState | None = None,
+    has_reappeared: bool = False,
+    has_escalated: bool = False,
+    workflow_env_id: int | None = None,
+) -> WorkflowEventData:
+    """
+    Build a WorkflowEventData object from individual parameters.
+    This method handles all the database fetching and object construction logic.
+    """
+
+    event = fetch_event(event_id, project_id)
+    if event is None:
+        raise ValueError(f"Event not found: event_id={event_id}, project_id={project_id}")
+
+    occurrence = IssueOccurrence.fetch(occurrence_id, project_id) if occurrence_id else None
+    # TODO(iamrajjoshi): Should we use get_from_cache here?
+    group = Group.objects.get(id=group_id)
+    group_event = GroupEvent.from_event(event, group)
+    group_event.occurrence = occurrence
+
+    # Fetch environment if provided
+    workflow_env = None
+    if workflow_env_id:
+        workflow_env = Environment.objects.get(id=workflow_env_id)
+
+    return WorkflowEventData(
+        event=group_event,
+        group=group,
+        group_state=group_state,
+        has_reappeared=has_reappeared,
+        has_escalated=has_escalated,
+        workflow_env=workflow_env,
+    )