debugging logs for workflow engine

Author: cathteng

src/sentry/features/temporary.py

@@ -483,6 +483,8 @@ def register_temporary_features(manager: FeatureManager):
     manager.add("organizations:workflow-engine-issue-alert-dual-write", OrganizationFeature, FeatureHandlerStrategy.FLAGPOLE, api_expose=False)
     # Enable workflow engine for issue alerts
     manager.add("organizations:workflow-engine-process-workflows", OrganizationFeature, FeatureHandlerStrategy.FLAGPOLE, api_expose=False)
+    # Enable logging to debug workflow engine process workflows
+    manager.add("organizations:workflow-engine-process-workflows-logs", OrganizationFeature, FeatureHandlerStrategy.FLAGPOLE, api_expose=False)
     # Enable firing actions for workflow engine issue alerts
     manager.add("organizations:workflow-engine-trigger-actions", OrganizationFeature, FeatureHandlerStrategy.FLAGPOLE, api_expose=False)
     # Enable dual writing for metric alert issues (see: alerts create issues)

src/sentry/rules/processing/delayed_processing.py

@@ -9,7 +9,7 @@
 from celery import Task
 from django.db.models import OuterRef, Subquery
 
-from sentry import buffer, nodestore
+from sentry import buffer, features, nodestore
 from sentry.buffer.base import BufferField
 from sentry.db import models
 from sentry.eventstore.models import Event, GroupEvent
@@ -465,6 +465,19 @@ def fire_rules(
             groupevent = group_to_groupevent[group]
             rule_fire_history = history.record(rule, group, groupevent.event_id, notification_uuid)
 
+            if features.has(
+                "organizations:workflow-engine-process-workflows-logs",
+                project.organization,
+            ):
+                logger.info(
+                    "post_process.delayed_processing.triggered_rule",
+                    extra={
+                        "rule_id": rule.id,
+                        "group_id": group.id,
+                        "event_id": groupevent.event_id,
+                    },
+                )
+
             callback_and_futures = activate_downstream_actions(
                 rule, groupevent, notification_uuid, rule_fire_history, is_post_process=False
             ).values()

src/sentry/rules/processing/processor.py

@@ -396,6 +396,21 @@ def apply_rule(self, rule: Rule, status: GroupRuleStatus) -> None:
                 organization_id=rule.project.organization.id,
                 rule_id=rule.id,
             )
+
+        if features.has(
+            "organizations:workflow-engine-process-workflows-logs",
+            rule.project.organization,
+        ):
+            logger.info(
+                "post_process.process_rules.triggered_rule",
+                extra={
+                    "rule_id": rule.id,
+                    "payload": state,
+                    "group_id": self.group.id,
+                    "event_id": self.event.event_id,
+                },
+            )
+
         notification_uuid = str(uuid.uuid4())
         rule_fire_history = history.record(rule, self.group, self.event.event_id, notification_uuid)
         grouped_futures = activate_downstream_actions(

src/sentry/workflow_engine/processors/delayed_workflow.py

@@ -46,6 +46,7 @@
 from sentry.workflow_engine.processors.workflow import (
     WORKFLOW_ENGINE_BUFFER_LIST_KEY,
     evaluate_workflows_action_filters,
+    log_fired_workflows,
 )
 from sentry.workflow_engine.types import DataConditionHandler, WorkflowJob
 
@@ -423,6 +424,16 @@ def fire_actions_for_groups(
                 tags={"event_type": group_event.group.type},
             )
 
+        if features.has(
+            "organizations:workflow-engine-process-workflows-logs",
+            organization,
+        ):
+            log_fired_workflows(
+                log_name="workflow_engine.delayed_workflow.fired_workflow",
+                actions=filtered_actions,
+                job=job,
+            )
+
         if features.has(
             "organizations:workflow-engine-trigger-actions",
             organization,

src/sentry/workflow_engine/processors/workflow.py

@@ -11,6 +11,7 @@
 from sentry.utils import json, metrics
 from sentry.workflow_engine.models import (
     Action,
+    AlertRuleWorkflow,
     DataCondition,
     DataConditionGroup,
     Detector,
@@ -111,6 +112,34 @@ def evaluate_workflows_action_filters(
     return filter_recently_fired_workflow_actions(filtered_action_groups, job["event"].group)
 
 
+def log_fired_workflows(log_name: str, actions: BaseQuerySet[Action], job: WorkflowJob) -> None:
+    # go from actions to workflows
+    action_ids = {action.id for action in actions}
+    action_conditions = DataConditionGroup.objects.filter(
+        dataconditiongroupaction__action_id__in=action_ids
+    ).values_list("id", flat=True)
+    workflows_to_fire = Workflow.objects.filter(
+        workflowdataconditiongroup__condition_group_id__in=action_conditions
+    )
+    workflow_to_rule = dict(
+        AlertRuleWorkflow.objects.filter(workflow__in=workflows_to_fire).values_list(
+            "workflow_id", "rule_id"
+        )
+    )
+
+    for workflow in workflows_to_fire:
+        logger.info(
+            log_name,
+            extra={
+                "workflow_id": workflow.id,
+                "rule_id": workflow_to_rule.get(workflow.id),
+                "payload": job,
+                "group_id": job["event"].group_id,
+                "event_id": job["event"].event_id,
+            },
+        )
+
+
 def process_workflows(job: WorkflowJob) -> set[Workflow]:
     """
     This method will get the detector based on the event, and then gather the associated workflows.
@@ -171,6 +200,16 @@ def process_workflows(job: WorkflowJob) -> set[Workflow]:
                 tags={"detector_type": detector.type},
             )
 
+        if features.has(
+            "organizations:workflow-engine-process-workflows-logs",
+            organization,
+        ):
+            log_fired_workflows(
+                log_name="workflow_engine.process_workflows.fired_workflow",
+                actions=actions,
+                job=job,
+            )
+
     with sentry_sdk.start_span(op="workflow_engine.process_workflows.trigger_actions"):
         if features.has(
             "organizations:workflow-engine-trigger-actions",

tests/sentry/workflow_engine/processors/test_workflow.py

@@ -6,6 +6,7 @@
 from sentry import buffer
 from sentry.eventstream.base import GroupState
 from sentry.grouping.grouptype import ErrorGroupType
+from sentry.models.rule import Rule
 from sentry.testutils.factories import Factories
 from sentry.testutils.helpers import with_feature
 from sentry.testutils.helpers.datetime import before_now, freeze_time
@@ -14,6 +15,7 @@
 from sentry.utils import json
 from sentry.workflow_engine.models import (
     Action,
+    AlertRuleWorkflow,
     DataConditionGroup,
     DataConditionGroupAction,
     Workflow,
@@ -84,6 +86,37 @@ def test_error_event(self):
         triggered_workflows = process_workflows(self.job)
         assert triggered_workflows == {self.error_workflow}
 
+    @with_feature("organizations:workflow-engine-process-workflows-logs")
+    @patch("sentry.workflow_engine.processors.workflow.logger")
+    def test_error_event__logger(self, mock_logger):
+        self.action_group, self.action = self.create_workflow_action(workflow=self.error_workflow)
+
+        rule = Rule.objects.filter(project=self.project).first()
+        AlertRuleWorkflow.objects.create(workflow=self.error_workflow, rule=rule)
+
+        triggered_workflows = process_workflows(self.job)
+        assert triggered_workflows == {self.error_workflow}
+
+        mock_logger.info.assert_called_with(
+            "workflow_engine.process_workflows.fired_workflows",
+            extra={
+                "workflow_id": self.error_workflow.id,
+                "rule_id": rule.id,
+                "payload": {
+                    "event": self.group_event,
+                    "group_state": {
+                        "id": 1,
+                        "is_new": False,
+                        "is_regression": True,
+                        "is_new_group_environment": False,
+                    },
+                    "workflow": self.error_workflow,
+                },
+                "group_id": self.group.id,
+                "event_id": self.event.event_id,
+            },
+        )
+
     def test_same_environment_only(self):
         # only processes workflows with the same env or no env specified
         self.error_workflow.update(environment=None)