chore(ACI): Add logs for triggers to compare both systems (#98758)

Add logs on both the subscription processor when a trigger's threshold
is met or resolved and in workflow engine when the workflow's trigger
conditions are met so that we can compare the two systems. Both logs
have the detector ID so we can match them up and are behind a flag with
a small subset of organizations flagged in so it won't be too much data.

Author: ceorourke

src/sentry/incidents/subscription_processor.py

@@ -325,8 +325,13 @@ def handle_trigger_alerts(
         aggregation_value: float,
         fired_incident_triggers: list[IncidentTrigger],
         metrics_incremented: bool,
+        detector: Detector | None,
     ) -> tuple[list[IncidentTrigger], bool]:
         # OVER/UNDER value trigger
+        has_dual_processing_flag = features.has(
+            "organizations:workflow-engine-metric-alert-dual-processing-logs",
+            self.subscription.project.organization,
+        )
         alert_operator, resolve_operator = self.THRESHOLD_TYPE_OPERATORS[
             AlertRuleThresholdType(self.alert_rule.threshold_type)
         ]
@@ -339,15 +344,20 @@ def handle_trigger_alerts(
                 "incidents.alert_rules.threshold.alert",
                 tags={"detection_type": self.alert_rule.detection_type},
             )
-            if (
-                features.has(
-                    "organizations:workflow-engine-metric-alert-dual-processing-logs",
-                    self.subscription.project.organization,
-                )
-                and not metrics_incremented
-            ):
-                metrics.incr("dual_processing.alert_rules.fire")
-                metrics_incremented = True
+            if has_dual_processing_flag:
+                if detector is not None:
+                    logger.info(
+                        "subscription_processor.alert_triggered",
+                        extra={
+                            "rule_id": self.alert_rule.id,
+                            "detector_id": detector.id,
+                            "organization_id": self.subscription.project.organization.id,
+                            "project_id": self.subscription.project.id,
+                        },
+                    )
+                if not metrics_incremented:
+                    metrics.incr("dual_processing.alert_rules.fire")
+                    metrics_incremented = True
             # triggering a threshold will create an incident and set the status to active
             incident_trigger = self.trigger_alert_threshold(trigger, aggregation_value)
             if incident_trigger is not None:
@@ -364,10 +374,17 @@ def handle_trigger_alerts(
                 "incidents.alert_rules.threshold.resolve",
                 tags={"detection_type": self.alert_rule.detection_type},
             )
-            if features.has(
-                "organizations:workflow-engine-metric-alert-dual-processing-logs",
-                self.subscription.project.organization,
-            ):
+            if has_dual_processing_flag:
+                if detector is not None:
+                    logger.info(
+                        "subscription_processor.alert_triggered",
+                        extra={
+                            "rule_id": self.alert_rule.id,
+                            "detector_id": detector.id,
+                            "organization_id": self.subscription.project.organization.id,
+                            "project_id": self.subscription.project.id,
+                        },
+                    )
                 metrics.incr("dual_processing.alert_rules.resolve")
             incident_trigger = self.trigger_resolve_threshold(trigger, aggregation_value)
 
@@ -615,7 +632,11 @@ def process_update(self, subscription_update: QuerySubscriptionUpdate) -> None:
                             return
 
                         fired_incident_triggers, metrics_incremented = self.handle_trigger_alerts(
-                            trigger, aggregation_value, fired_incident_triggers, metrics_incremented
+                            trigger,
+                            aggregation_value,
+                            fired_incident_triggers,
+                            metrics_incremented,
+                            detector,
                         )
 
                 if fired_incident_triggers:

src/sentry/workflow_engine/processors/workflow.py

@@ -14,7 +14,13 @@
 from sentry.services.eventstore.models import GroupEvent
 from sentry.utils import json
 from sentry.workflow_engine import buffer
-from sentry.workflow_engine.models import Action, DataConditionGroup, Detector, Workflow
+from sentry.workflow_engine.models import (
+    Action,
+    DataConditionGroup,
+    Detector,
+    DetectorWorkflow,
+    Workflow,
+)
 from sentry.workflow_engine.models.workflow_data_condition_group import WorkflowDataConditionGroup
 from sentry.workflow_engine.processors.contexts.workflow_event_context import (
     WorkflowEventContext,
@@ -183,6 +189,23 @@ def evaluate_workflow_triggers(
         else:
             if evaluation:
                 triggered_workflows.add(workflow)
+                if features.has(
+                    "organizations:workflow-engine-metric-alert-dual-processing-logs",
+                    workflow.organization,
+                ):
+                    try:
+                        detector_workflow = DetectorWorkflow.objects.get(workflow_id=workflow.id)
+                        logger.info(
+                            "workflow_engine.process_workflows.workflow_triggered",
+                            extra={
+                                "workflow_id": workflow.id,
+                                "detector_id": detector_workflow.detector_id,
+                                "organization_id": workflow.organization.id,
+                                "project_id": event_data.group.project.id,
+                            },
+                        )
+                    except DetectorWorkflow.DoesNotExist:
+                        continue
 
     metrics_incr(
         "process_workflows.triggered_workflows",

tests/sentry/workflow_engine/processors/test_workflow.py

@@ -11,6 +11,7 @@
 from sentry.services.eventstore.models import GroupEvent
 from sentry.testutils.factories import Factories
 from sentry.testutils.helpers.datetime import before_now, freeze_time
+from sentry.testutils.helpers.features import with_feature
 from sentry.testutils.helpers.options import override_options
 from sentry.testutils.helpers.redis import mock_redis_buffer
 from sentry.testutils.pytest.fixtures import django_db_all
@@ -337,6 +338,20 @@ def test_workflow_trigger(self) -> None:
         )
         assert triggered_workflows == {self.workflow}
 
+    @with_feature("organizations:workflow-engine-metric-alert-dual-processing-logs")
+    @patch("sentry.workflow_engine.processors.workflow.logger")
+    def test_logs_triggered_workflows(self, mock_logger: MagicMock) -> None:
+        evaluate_workflow_triggers({self.workflow}, self.event_data, self.event_start_time)
+        mock_logger.info.assert_called_once_with(
+            "workflow_engine.process_workflows.workflow_triggered",
+            extra={
+                "workflow_id": self.workflow.id,
+                "detector_id": self.detector.id,
+                "organization_id": self.workflow.organization.id,
+                "project_id": self.event_data.group.project.id,
+            },
+        )
+
     def test_workflow_trigger__no_conditions(self) -> None:
         assert self.workflow.when_condition_group
         self.workflow.when_condition_group.conditions.all().delete()