set workflow in job when evaluating action filters

Author: cathteng

src/sentry/workflow_engine/processors/workflow.py

@@ -16,6 +16,7 @@
     DataConditionGroup,
     Detector,
     Workflow,
+    WorkflowDataConditionGroup,
 )
 from sentry.workflow_engine.processors.action import filter_recently_fired_workflow_actions
 from sentry.workflow_engine.processors.data_condition_group import process_data_condition_group
@@ -73,6 +74,8 @@ def evaluate_workflow_triggers(workflows: set[Workflow], job: WorkflowJob) -> se
             if evaluation:
                 triggered_workflows.add(workflow)
 
+        job.pop("workflow", None)
+
     return triggered_workflows
 
 
@@ -82,14 +85,25 @@ def evaluate_workflows_action_filters(
 ) -> BaseQuerySet[Action]:
     filtered_action_groups: set[DataConditionGroup] = set()
 
-    # gets the list of the workflow ids, and then get the workflow_data_condition_groups for those workflows
-    workflow_ids = {workflow.id for workflow in workflows}
+    # Gets the list of the workflow ids, and then get the workflow_data_condition_groups for those workflows
+    workflow_ids_to_workflows = {workflow.id: workflow for workflow in workflows}
 
     action_conditions = DataConditionGroup.objects.filter(
-        workflowdataconditiongroup__workflow_id__in=workflow_ids
+        workflowdataconditiongroup__workflow_id__in=list(workflow_ids_to_workflows.keys())
     ).distinct()
 
+    workflow_to_dcg = dict(
+        WorkflowDataConditionGroup.objects.filter(
+            condition_group_id__in=action_conditions
+        ).values_list("condition_group_id", "workflow")
+    )
+
     for action_condition in action_conditions:
+        # Populate the workflow in the job for the action_condition evaluation
+        workflow_id = workflow_to_dcg.get(action_condition.id)
+        if workflow_id:
+            job["workflow"] = workflow_ids_to_workflows[workflow_id]
+
         (evaluation, result), remaining_conditions = process_data_condition_group(
             action_condition.id, job
         )
@@ -109,6 +123,8 @@ def evaluate_workflows_action_filters(
             if evaluation:
                 filtered_action_groups.add(action_condition)
 
+        job.pop("workflow", None)
+
     return filter_recently_fired_workflow_actions(filtered_action_groups, job["event"].group)
 
 

tests/sentry/workflow_engine/processors/test_workflow.py

@@ -117,6 +117,54 @@ def test_error_event__logger(self, mock_logger):
             },
         )
 
+    @patch("sentry.workflow_engine.processors.workflow.filter_recently_fired_workflow_actions")
+    def test_populate_workflow_for_filters(self, mock_filter):
+        # this should not pass because the environment is not None
+        self.error_workflow.update(environment=self.group_event.get_environment())
+        error_workflow_filters = self.create_data_condition_group(
+            logic_type=DataConditionGroup.Type.ANY_SHORT_CIRCUIT
+        )
+        self.create_data_condition(
+            condition_group=error_workflow_filters,
+            type=Condition.FIRST_SEEN_EVENT,
+            comparison=True,
+            condition_result=True,
+        )
+        self.create_workflow_data_condition_group(
+            workflow=self.error_workflow, condition_group=error_workflow_filters
+        )
+
+        workflow_triggers = self.create_data_condition_group(
+            logic_type=DataConditionGroup.Type.ANY_SHORT_CIRCUIT
+        )
+        workflow_filters = self.create_data_condition_group(
+            logic_type=DataConditionGroup.Type.ANY_SHORT_CIRCUIT
+        )
+        # this should pass because the environment is None
+        self.create_data_condition(
+            condition_group=workflow_filters,
+            type=Condition.FIRST_SEEN_EVENT,
+            comparison=True,
+            condition_result=True,
+        )
+        workflow = self.create_workflow(
+            name="testy",
+            when_condition_group=workflow_triggers,
+        )
+        self.create_detector_workflow(
+            detector=self.error_detector,
+            workflow=workflow,
+        )
+        self.create_workflow_data_condition_group(
+            workflow=workflow, condition_group=workflow_filters
+        )
+
+        self.job["group_state"]["is_new"] = True
+
+        process_workflows(self.job)
+
+        mock_filter.assert_called_with({workflow_filters}, self.group)
+
     def test_same_environment_only(self):
         # only processes workflows with the same env or no env specified
         self.error_workflow.update(environment=None)