diff --git a/src/sentry/workflow_engine/docs/execution.md b/src/sentry/workflow_engine/docs/execution.md index d6dd52af6997..e2dfbe56f8e1 100644 --- a/src/sentry/workflow_engine/docs/execution.md +++ b/src/sentry/workflow_engine/docs/execution.md @@ -362,6 +362,9 @@ before event-level deduplication and task execution. A tuple is suppressed when passed within `Workflow.config.frequency` minutes, even if a later equivalent action was deduplicated or an external handler failed. +Only alert workflows use group-frequency suppression; Inbox workflows bypass it, +including when both match the same issue Activity. + ### Event-level deduplication `get_unique_active_actions` first removes actions whose status is not active, then keeps @@ -373,11 +376,12 @@ imply Workflow execution order. ### History and task dispatch The processor updates frequency statuses and creates -[`WorkflowFireHistory`](../models/workflow_fire_history.py) rows before `fire_actions` +[`WorkflowFireHistory`](../models/workflow_fire_history.py) rows for eligible workflows before `fire_actions` removes inactive actions and deduplicates by key. It then schedules [`trigger_action`](../tasks/actions.py) for retained actions. Fire-history rows describe scheduling, not successful external delivery, and can exist for actions that are later -filtered or deduplicated. +filtered or deduplicated. Sharing an action with an eligible workflow does not create +history for a suppressed workflow. The action task: diff --git a/src/sentry/workflow_engine/processors/action.py b/src/sentry/workflow_engine/processors/action.py index 95fd71cc4a3f..32eb75ce8992 100644 --- a/src/sentry/workflow_engine/processors/action.py +++ b/src/sentry/workflow_engine/processors/action.py @@ -295,34 +295,14 @@ def fire_actions( trigger_action.apply_async(kwargs=task_params, headers={"sentry-propagate-traces": False}) -def filter_recently_fired_workflow_actions( - filtered_action_groups: set[DataConditionGroup], event_data: WorkflowEventData -) -> tuple[BaseQuerySet[Action], dict[ActionId, WorkflowId]]: - """ - Returns actions associated with the provided DataConditionGroups, excluding - those that have been recently fired, along with a mapping of action_id to - the workflow_id each action is firing for. Also updates associated - WorkflowActionGroupStatus objects. - """ - - data_condition_group_actions = DataConditionGroupAction.objects.filter( - condition_group__in=filtered_action_groups - ).values_list("action_id", "condition_group__workflowdataconditiongroup__workflow_id") - - action_to_workflows_ids: dict[int, set[int]] = defaultdict(set) - workflow_ids: set[int] = set() - - for action_id, workflow_id in data_condition_group_actions: - # If we are mid-deletion, the workflow_id can be none. - if workflow_id is not None: - action_to_workflows_ids[action_id].add(workflow_id) - workflow_ids.add(workflow_id) - +def _filter_recently_fired_alert_actions( + action_to_workflows_ids: dict[int, set[int]], group: Group +) -> dict[int, set[int]]: + workflow_ids = {workflow_id for ids in action_to_workflows_ids.values() for workflow_id in ids} workflows = Workflow.objects.filter(id__in=workflow_ids) - action_to_statuses = get_workflow_action_group_statuses( action_to_workflows_ids=action_to_workflows_ids, - group=event_data.require_group(), + group=group, workflow_ids=workflow_ids, ) now = timezone.now() @@ -331,7 +311,7 @@ def filter_recently_fired_workflow_actions( action_to_workflows_ids=action_to_workflows_ids, action_to_statuses=action_to_statuses, workflows=workflows, - group=event_data.require_group(), + group=group, now=now, ) ) @@ -343,14 +323,48 @@ def filter_recently_fired_workflow_actions( if not action_to_workflows_ids[action_id]: action_to_workflows_ids.pop(action_id) - actions_queryset = Action.objects.filter(id__in=list(action_to_workflows_ids.keys())) + return action_to_workflows_ids - # Build a mapping of action_id -> workflow_id (pick one arbitrarily but deterministically) - action_to_workflow_id: dict[ActionId, WorkflowId] = { - action_id: min(wf_ids) for action_id, wf_ids in action_to_workflows_ids.items() - } - return actions_queryset, action_to_workflow_id +def filter_recently_fired_workflow_actions( + filtered_action_groups: set[DataConditionGroup], event_data: WorkflowEventData +) -> tuple[BaseQuerySet[Action], dict[ActionId, WorkflowId], set[WorkflowId]]: + """Select actions and their eligible workflows before equivalent-action deduplication. + + Inbox workflows bypass group-frequency state. Alert workflows retain their + existing per-group suppression. The complete eligible workflow set is returned + for history, while each action is attributed to one workflow for task dispatch. + """ + organization = event_data.get_organization() + associations = DataConditionGroupAction.objects.filter( + condition_group__in=filtered_action_groups, + condition_group__workflowdataconditiongroup__workflow__organization_id=organization.id, + ).values_list( + "action_id", + "condition_group__workflowdataconditiongroup__workflow_id", + "condition_group__workflowdataconditiongroup__workflow__type", + ) + inbox_actions: dict[int, set[int]] = defaultdict(set) + alert_actions: dict[int, set[int]] = defaultdict(set) + for action_id, workflow_id, workflow_type in associations: + if workflow_type == Workflow.Type.INBOX: + inbox_actions[action_id].add(workflow_id) + elif event_data.group is not None: + alert_actions[action_id].add(workflow_id) + + eligible_actions = inbox_actions + if alert_actions and event_data.group is not None: + for action_id, workflow_ids in _filter_recently_fired_alert_actions( + alert_actions, event_data.group + ).items(): + eligible_actions[action_id].update(workflow_ids) + + actions = get_supported_actions(Action.objects.filter(id__in=eligible_actions), event_data) + action_to_workflow_id = {action.id: min(eligible_actions[action.id]) for action in actions} + eligible_workflow_ids = { + workflow_id for action in actions for workflow_id in eligible_actions[action.id] + } + return actions, action_to_workflow_id, eligible_workflow_ids def get_available_action_integrations_for_org(organization: Organization) -> list[RpcIntegration]: diff --git a/src/sentry/workflow_engine/processors/delayed_workflow.py b/src/sentry/workflow_engine/processors/delayed_workflow.py index 40790b3980f4..b17fb946607d 100644 --- a/src/sentry/workflow_engine/processors/delayed_workflow.py +++ b/src/sentry/workflow_engine/processors/delayed_workflow.py @@ -848,8 +848,8 @@ def fire_actions_for_groups( ) dcgs_for_group = groups_to_fire.get(group.id, set()) - filtered_actions, action_to_workflow_id = filter_recently_fired_workflow_actions( - dcgs_for_group, workflow_event_data + filtered_actions, action_to_workflow_id, eligible_workflow_ids = ( + filter_recently_fired_workflow_actions(dcgs_for_group, workflow_event_data) ) # TODO: trigger service hooks from here @@ -864,6 +864,7 @@ def fire_actions_for_groups( workflow_event_data, is_delayed=True, start_timestamp=start_timestamp, + eligible_workflow_ids=eligible_workflow_ids, ) # Create mapping: workflow_id -> notification_uuid for propagation diff --git a/src/sentry/workflow_engine/processors/workflow.py b/src/sentry/workflow_engine/processors/workflow.py index c640bba14210..4bfd946d4220 100644 --- a/src/sentry/workflow_engine/processors/workflow.py +++ b/src/sentry/workflow_engine/processors/workflow.py @@ -643,7 +643,7 @@ def process_workflows( return result enqueue_workflows(batch_client, delayed_items) - actions, action_to_workflow_id = filter_recently_fired_workflow_actions( + actions, action_to_workflow_id, eligible_workflow_ids = filter_recently_fired_workflow_actions( action_groups, event_data ) @@ -657,6 +657,7 @@ def process_workflows( event_data, is_delayed=False, start_timestamp=event_start_time, + eligible_workflow_ids=eligible_workflow_ids, ) # Create mapping: workflow_id -> notification_uuid for propagation diff --git a/src/sentry/workflow_engine/processors/workflow_fire_history.py b/src/sentry/workflow_engine/processors/workflow_fire_history.py index e603e2d53e1a..c3f464ea9244 100644 --- a/src/sentry/workflow_engine/processors/workflow_fire_history.py +++ b/src/sentry/workflow_engine/processors/workflow_fire_history.py @@ -1,7 +1,7 @@ from __future__ import annotations import logging -from collections.abc import Sequence +from collections.abc import Collection, Sequence from datetime import datetime, timezone from django.db.models import OuterRef, Subquery @@ -30,6 +30,8 @@ def create_workflow_fire_histories( event_data: WorkflowEventData, is_delayed: bool = False, start_timestamp: datetime | None = None, + *, + eligible_workflow_ids: Collection[int] | None = None, ) -> list[WorkflowFireHistory]: """ Record that the workflows associated with these actions were fired for this @@ -46,6 +48,9 @@ def create_workflow_fire_histories( ).values_list("workflow_id", flat=True) ) + if eligible_workflow_ids is not None: + workflow_ids.intersection_update(eligible_workflow_ids) + event_id = event_data.event_id if start_timestamp: diff --git a/tests/sentry/workflow_engine/processors/test_action.py b/tests/sentry/workflow_engine/processors/test_action.py index 4e9d67f2999e..7600048417de 100644 --- a/tests/sentry/workflow_engine/processors/test_action.py +++ b/tests/sentry/workflow_engine/processors/test_action.py @@ -1,27 +1,36 @@ from datetime import timedelta from unittest.mock import MagicMock, patch +from uuid import uuid4 from django.db import connection from django.utils import timezone from sentry.integrations.base import IntegrationFeatures +from sentry.models.activity import Activity +from sentry.notifications.models.notificationaction import ActionTarget from sentry.testutils.helpers.datetime import freeze_time +from sentry.types.activity import ActivityType +from sentry.workflow_engine.buffer.batch_client import DelayedWorkflowClient from sentry.workflow_engine.models import ( Action, DataConditionGroup, Workflow, WorkflowActionGroupStatus, + WorkflowFireHistory, ) from sentry.workflow_engine.processors.action import ( StatusUpdateResult, filter_recently_fired_workflow_actions, + fire_actions, get_workflow_action_group_statuses, is_action_permitted, process_workflow_action_group_statuses, update_workflow_action_group_statuses, ) +from sentry.workflow_engine.processors.workflow import process_workflows +from sentry.workflow_engine.processors.workflow_fire_history import create_workflow_fire_histories from sentry.workflow_engine.types import ActionHandler, FeatureGate, WorkflowEventData -from tests.sentry.workflow_engine.test_base import BaseWorkflowTest +from tests.sentry.workflow_engine.test_base import BaseWorkflowTest, OrganizationEvent @freeze_time("2024-01-09") @@ -52,7 +61,7 @@ def test(self) -> None: workflow=self.workflow, action=action, group=self.group ) - triggered_actions, action_to_workflow_id = filter_recently_fired_workflow_actions( + triggered_actions, action_to_workflow_id, _ = filter_recently_fired_workflow_actions( set(DataConditionGroup.objects.all()), self.event_data ) assert set(triggered_actions) == {self.action} @@ -83,7 +92,7 @@ def test_multiple_workflows(self) -> None: ) status_3.update(date_updated=timezone.now() - timedelta(days=2)) - triggered_actions, action_to_workflow_id = filter_recently_fired_workflow_actions( + triggered_actions, action_to_workflow_id, _ = filter_recently_fired_workflow_actions( set(DataConditionGroup.objects.all()), self.event_data ) assert set(triggered_actions) == {self.action, action_3} @@ -101,7 +110,7 @@ def test_multiple_workflows_single_action__first_fire(self) -> None: ) # shared action self.create_workflow_data_condition_group(workflow, action_group) - triggered_actions, action_to_workflow_id = filter_recently_fired_workflow_actions( + triggered_actions, action_to_workflow_id, _ = filter_recently_fired_workflow_actions( set(DataConditionGroup.objects.all()), self.event_data ) # dedupes action if both workflows will fire it @@ -125,7 +134,7 @@ def test_multiple_workflows_single_action__later_fire(self) -> None: ) status.update(date_updated=timezone.now() - timedelta(hours=1)) - triggered_actions, action_to_workflow_id = filter_recently_fired_workflow_actions( + triggered_actions, action_to_workflow_id, _ = filter_recently_fired_workflow_actions( set(DataConditionGroup.objects.all()), self.event_data ) # fires one action for the workflow that can fire it @@ -276,7 +285,7 @@ def test_does_not_fire_for_uncreated_statuses(self, mock_update: MagicMock) -> N updated=0, created=0, not_created=[(self.workflow.id, self.action.id)] ) - triggered_actions, action_to_workflow_id = filter_recently_fired_workflow_actions( + triggered_actions, action_to_workflow_id, _ = filter_recently_fired_workflow_actions( set(DataConditionGroup.objects.all()), self.event_data ) @@ -296,7 +305,7 @@ def test_fires_for_non_conflicting_workflow(self, mock_update: MagicMock) -> Non updated=0, created=0, not_created=[(self.workflow.id, self.action.id)] ) - triggered_actions, action_to_workflow_id = filter_recently_fired_workflow_actions( + triggered_actions, action_to_workflow_id, _ = filter_recently_fired_workflow_actions( set(DataConditionGroup.objects.all()), self.event_data ) @@ -312,7 +321,7 @@ def test_skips_action_with_no_workflow(self) -> None: ) # No WorkflowDataConditionGroup links orphan_group to any workflow - triggered_actions, action_to_workflow_id = filter_recently_fired_workflow_actions( + triggered_actions, action_to_workflow_id, _ = filter_recently_fired_workflow_actions( {self.action_group, orphan_group}, self.event_data ) @@ -410,3 +419,169 @@ def test_basic(self, mock_get_features: MagicMock) -> None: } ): assert is_action_permitted(Action.Type.JIRA, org) + + +@freeze_time("2024-01-09") +class TestInboxActionFrequency(BaseWorkflowTest): + def setUp(self) -> None: + self.inbox = self.create_workflow(type=Workflow.Type.INBOX, config={"frequency": 1440}) + self.inbox_group, self.inbox_action = self.create_workflow_action( + self.inbox, + action=self.create_action( + type=Action.Type.EMAIL, + config={"target_type": ActionTarget.USER, "target_identifier": str(self.user.id)}, + ), + ) + self.alert = self.create_workflow(config={"frequency": 30}) + self.alert_group, self.alert_action = self.create_workflow_action(self.alert) + self.activity = Activity( + project=self.project, + group=self.group, + type=ActivityType.ASSIGNED.value, + data={"assignee": str(self.user.id)}, + ) + self.activity.save() + self.event_data = WorkflowEventData(event=self.activity, group=self.group) + self.groups = {self.inbox_group, self.alert_group} + + def test_inbox_only_does_not_access_frequency_state(self) -> None: + with ( + patch( + "sentry.workflow_engine.processors.action.get_workflow_action_group_statuses" + ) as read, + patch( + "sentry.workflow_engine.processors.action.update_workflow_action_group_statuses" + ) as write, + ): + actions, _, workflows = filter_recently_fired_workflow_actions( + {self.inbox_group}, self.event_data + ) + assert list(actions) == [self.inbox_action] + assert workflows == {self.inbox.id} + read.assert_not_called() + write.assert_not_called() + + def test_group_less_event_excludes_alerts_and_unsupported_actions(self) -> None: + event_data = WorkflowEventData( + event=OrganizationEvent(organization_id=self.organization.id, event_id=uuid4()), + organization=self.organization, + ) + unsupported_group, _ = self.create_workflow_action(self.inbox) + with patch.object(self.inbox_action.get_handler(), "supports_group_less", True): + actions, attribution, workflows = filter_recently_fired_workflow_actions( + self.groups | {unsupported_group}, event_data + ) + assert list(actions) == [self.inbox_action] + assert attribution == {self.inbox_action.id: self.inbox.id} + assert workflows == {self.inbox.id} + assert not WorkflowActionGroupStatus.objects.exists() + + def test_shared_action_records_only_eligible_workflows(self) -> None: + shared_group, _ = self.create_workflow_action(self.alert, action=self.inbox_action) + groups = {self.inbox_group, shared_group} + actions, attribution, workflows = filter_recently_fired_workflow_actions( + groups, self.event_data + ) + assert list(actions) == [self.inbox_action] + assert workflows == {self.inbox.id, self.alert.id} + assert attribution == {self.inbox_action.id: min(workflows)} + histories = create_workflow_fire_histories( + actions, self.event_data, eligible_workflow_ids=workflows + ) + assert {history.workflow_id for history in histories} == workflows + + # On the next evaluation only Inbox remains eligible for the shared action. + actions, attribution, workflows = filter_recently_fired_workflow_actions( + groups, self.event_data + ) + assert list(actions) == [self.inbox_action] + assert attribution == {self.inbox_action.id: self.inbox.id} + assert workflows == {self.inbox.id} + histories = create_workflow_fire_histories( + actions, self.event_data, eligible_workflow_ids=workflows + ) + assert {history.workflow_id for history in histories} == {self.inbox.id} + + def test_alert_insert_conflict_does_not_drop_shared_inbox_action(self) -> None: + self.create_data_condition_group_action( + condition_group=self.alert_group, action=self.inbox_action + ) + with patch( + "sentry.workflow_engine.processors.action.update_workflow_action_group_statuses", + return_value=StatusUpdateResult( + updated=0, + created=0, + not_created=[ + (self.alert.id, self.inbox_action.id), + (self.alert.id, self.alert_action.id), + ], + ), + ): + actions, attribution, workflows = filter_recently_fired_workflow_actions( + self.groups, self.event_data + ) + assert list(actions) == [self.inbox_action] + assert attribution == {self.inbox_action.id: self.inbox.id} + assert workflows == {self.inbox.id} + + def test_equivalent_inbox_actions_dispatch_once_per_evaluation(self) -> None: + duplicate = self.create_action(type=Action.Type.EMAIL, config=self.inbox_action.config) + duplicate_group, _ = self.create_workflow_action(self.inbox, action=duplicate) + groups = {self.inbox_group, duplicate_group} + with patch( + "sentry.workflow_engine.processors.action.trigger_action.apply_async" + ) as enqueue: + actions, attribution, _ = filter_recently_fired_workflow_actions( + groups, self.event_data + ) + fire_actions(actions, self.event_data, {}, attribution) + assert enqueue.call_count == 1 + actions, attribution, _ = filter_recently_fired_workflow_actions( + groups, self.event_data + ) + fire_actions(actions, self.event_data, {}, attribution) + assert enqueue.call_count == 2 + assert not WorkflowActionGroupStatus.objects.exists() + + def test_excludes_other_organizations(self) -> None: + foreign = self.create_workflow( + type=Workflow.Type.INBOX, organization=self.create_organization() + ) + foreign_group, _ = self.create_workflow_action(foreign) + actions, attribution, workflows = filter_recently_fired_workflow_actions( + {foreign_group}, self.event_data + ) + assert list(actions) == [] + assert attribution == {} + assert workflows == set() + assert not WorkflowActionGroupStatus.objects.exists() + + def test_mixed_activity_processing_preserves_throttle_and_notification_uuid(self) -> None: + # Prime only the alert's throttle, then evaluate both workflows normally. + filter_recently_fired_workflow_actions({self.alert_group}, self.event_data) + status = WorkflowActionGroupStatus.objects.get() + status.update(date_updated=timezone.now() - timedelta(minutes=1)) + previous_date = status.date_updated + detector = self.create_detector(project=self.project) + with ( + patch( + "sentry.workflow_engine.processors.workflow.get_workflows_by_detectors", + return_value={self.inbox, self.alert}, + ), + patch("sentry.workflow_engine.processors.action.trigger_action.apply_async") as enqueue, + ): + result = process_workflows( + DelayedWorkflowClient(), self.event_data, timezone.now(), detector=detector + ) + assert result.has_triggered_actions() + enqueue.assert_called_once() + params = enqueue.call_args.kwargs["kwargs"] + assert params["action_id"] == self.inbox_action.id + assert params["workflow_id"] == self.inbox.id + assert params["activity_id"] == self.activity.id + history = WorkflowFireHistory.objects.get() + assert history.workflow_id == self.inbox.id + assert params["notification_uuid"] == str(history.notification_uuid) + status.refresh_from_db() + assert status.date_updated == previous_date + assert WorkflowActionGroupStatus.objects.count() == 1 diff --git a/tests/sentry/workflow_engine/processors/test_workflow.py b/tests/sentry/workflow_engine/processors/test_workflow.py index 70e4c9ec69ef..39d5081e12bb 100644 --- a/tests/sentry/workflow_engine/processors/test_workflow.py +++ b/tests/sentry/workflow_engine/processors/test_workflow.py @@ -299,7 +299,7 @@ def test_populate_workflow_env_for_filters( self, mock_filter: MagicMock, mock_fire_actions: MagicMock ) -> None: action = self.create_action() - mock_filter.return_value = (Action.objects.filter(id=action.id), {action.id: 1}) + mock_filter.return_value = (Action.objects.filter(id=action.id), {action.id: 1}, {1}) # this should not pass because the environment is not None self.error_workflow.update(environment=self.group_event.get_environment()) @@ -437,7 +437,7 @@ def test_activity_with_incomplete_trigger_skips_action_processing(self) -> None: patch("sentry.workflow_engine.processors.workflow.enqueue_workflows") as enqueue, patch( "sentry.workflow_engine.processors.action.filter_recently_fired_workflow_actions", - return_value=([], {}), + return_value=([], {}, set()), ) as suppress_actions, ): result = process_workflows( diff --git a/tests/sentry/workflow_engine/test_task.py b/tests/sentry/workflow_engine/test_task.py index 1a0daa874e40..f42ffd57bed9 100644 --- a/tests/sentry/workflow_engine/test_task.py +++ b/tests/sentry/workflow_engine/test_task.py @@ -130,7 +130,7 @@ def test_process_workflow_activity__workflows__no_actions( @mock.patch( "sentry.workflow_engine.processors.action.filter_recently_fired_workflow_actions", - return_value=([], {}), + return_value=([], {}, set()), ) def test_process_workflow_activity(self, mock_filter_actions: mock.MagicMock) -> None: self.workflow = self.create_workflow(organization=self.organization) @@ -175,7 +175,7 @@ def _create_workflow_with_action(self, **workflow_kwargs: Any) -> DataConditionG @mock.patch( "sentry.workflow_engine.processors.action.filter_recently_fired_workflow_actions", - return_value=([], {}), + return_value=([], {}, set()), ) def test_process_workflow_activity__matching_environment( self, mock_filter_actions: mock.MagicMock @@ -200,7 +200,7 @@ def test_process_workflow_activity__matching_environment( @mock.patch( "sentry.workflow_engine.processors.action.filter_recently_fired_workflow_actions", - return_value=([], {}), + return_value=([], {}, set()), ) def test_process_workflow_activity__other_environment( self, mock_filter_actions: mock.MagicMock