aci(feat): Use propagated timestamp to track triggering latency (#97612)

Create a timestamp before process_workflows_event and use it after alert
triggering to report latency.

Author: kcons

pyproject.toml

@@ -644,9 +644,11 @@ module = [
     "sentry.web.frontend.cli",
     "sentry.web.frontend.csv",
     "sentry.web.frontend.mixins.*",
+    "sentry.workflow_engine.buffer.*",
     "sentry.workflow_engine.handlers.condition.*",
     "sentry.workflow_engine.migrations.*",
     "sentry.workflow_engine.processors.*",
+    "sentry.workflow_engine.tasks.*",
     "sentry.workflow_engine.typings.*",
     "sentry.workflow_engine.utils.*",
     "sentry_plugins.base",

src/sentry/rules/processing/delayed_processing.py

@@ -4,7 +4,7 @@
 from collections import defaultdict
 from dataclasses import dataclass
 from datetime import datetime, timedelta, timezone
-from typing import Any, DefaultDict, NamedTuple
+from typing import Any, DefaultDict, NamedTuple, NotRequired, TypedDict
 
 import sentry_sdk
 from celery import Task
@@ -50,7 +50,6 @@
 from sentry.taskworker.namespaces import issues_tasks
 from sentry.taskworker.retry import Retry
 from sentry.utils import json, metrics
-from sentry.utils.dates import ensure_aware
 from sentry.utils.iterators import chunked
 from sentry.utils.lazy_service_wrapper import LazyServiceWrapper
 from sentry.utils.retries import ConditionalRetryPolicy, exponential_delay
@@ -243,25 +242,48 @@ def bulk_fetch_events(event_ids: list[str], project_id: int) -> dict[str, Event]
     }
 
 
+class EventData(TypedDict):
+    event_id: str
+    occurrence_id: NotRequired[str | None]
+    start_timestamp: NotRequired[datetime | None]
+
+
 def parse_rulegroup_to_event_data(
     rulegroup_to_event_data: dict[str, str],
-) -> dict[tuple[int, int], dict[str, str]]:
+) -> dict[tuple[int, int], EventData]:
     parsed_rulegroup_to_event_data = {}
     for rule_group, instance_data in rulegroup_to_event_data.items():
         event_data = json.loads(instance_data)
+        if ts_string := event_data.get("start_timestamp"):
+            try:
+                # Handle ISO format with timezone info
+                event_data["start_timestamp"] = datetime.fromisoformat(ts_string)
+            except (ValueError, TypeError):
+                try:
+                    # Fallback to manual parsing if needed
+                    event_data["start_timestamp"] = datetime.strptime(
+                        ts_string, "%Y-%m-%dT%H:%M:%S.%fZ"
+                    ).replace(tzinfo=timezone.utc)
+                except ValueError:
+                    logger.exception(
+                        "delayed_processing.invalid_start_timestamp",
+                        extra={"rule_group": rule_group, "start_timestamp": ts_string},
+                    )
+                    del event_data["start_timestamp"]
+
         rule_id, group_id = rule_group.split(":")
         parsed_rulegroup_to_event_data[(int(rule_id), int(group_id))] = event_data
     return parsed_rulegroup_to_event_data
 
 
 def build_group_to_groupevent(
     log_config: LogConfig,
-    parsed_rulegroup_to_event_data: dict[tuple[int, int], dict[str, str]],
+    parsed_rulegroup_to_event_data: dict[tuple[int, int], EventData],
     bulk_event_id_to_events: dict[str, Event],
     bulk_occurrence_id_to_occurrence: dict[str, IssueOccurrence],
     group_id_to_group: dict[int, Group],
     project_id: int,
-) -> dict[Group, GroupEvent]:
+) -> dict[Group, tuple[GroupEvent, datetime | None]]:
 
     project = fetch_project(project_id)
     if project:
@@ -278,11 +300,12 @@ def build_group_to_groupevent(
                     "project_id": project_id,
                 },
             )
-    group_to_groupevent = {}
+    group_to_groupevent: dict[Group, tuple[GroupEvent, datetime | None]] = {}
 
     for rule_group, instance_data in parsed_rulegroup_to_event_data.items():
         event_id = instance_data.get("event_id")
         occurrence_id = instance_data.get("occurrence_id")
+        start_timestamp = instance_data.get("start_timestamp")
 
         if event_id is None:
             logger.info(
@@ -312,16 +335,16 @@ def build_group_to_groupevent(
         group_event = event.for_group(group)
         if occurrence_id:
             group_event.occurrence = bulk_occurrence_id_to_occurrence.get(occurrence_id)
-        group_to_groupevent[group] = group_event
+        group_to_groupevent[group] = (group_event, start_timestamp)
     return group_to_groupevent
 
 
 def get_group_to_groupevent(
     log_config: LogConfig,
-    parsed_rulegroup_to_event_data: dict[tuple[int, int], dict[str, str]],
+    parsed_rulegroup_to_event_data: dict[tuple[int, int], EventData],
     project_id: int,
     group_ids: set[int],
-) -> dict[Group, GroupEvent]:
+) -> dict[Group, tuple[GroupEvent, datetime | None]]:
     groups = Group.objects.filter(id__in=group_ids)
     group_id_to_group = {group.id: group for group in groups}
 
@@ -491,7 +514,7 @@ def get_rules_to_fire(
 def fire_rules(
     log_config: LogConfig,
     rules_to_fire: DefaultDict[Rule, set[int]],
-    parsed_rulegroup_to_event_data: dict[tuple[int, int], dict[str, str]],
+    parsed_rulegroup_to_event_data: dict[tuple[int, int], EventData],
     alert_rules: list[Rule],
     project: Project,
 ) -> None:
@@ -511,7 +534,8 @@ def fire_rules(
         )
         if log_config.num_events_issue_debugging or log_config.workflow_engine_process_workflows:
             serialized_groups = {
-                group.id: group_event.event_id for group, group_event in group_to_groupevent.items()
+                group.id: group_event.event_id
+                for group, (group_event, _) in group_to_groupevent.items()
             }
             logger.info(
                 "delayed_processing.group_to_groupevent",
@@ -574,14 +598,13 @@ def fire_rules(
                         continue
 
                     notification_uuid = str(uuid.uuid4())
-                    groupevent = group_to_groupevent[group]
-                    metrics.timing(
-                        "rule_fire_history.latency",
-                        (
-                            datetime.now(tz=timezone.utc) - ensure_aware(groupevent.datetime)
-                        ).total_seconds(),
-                        tags={"delayed": True, "group_type": group.issue_type.slug},
-                    )
+                    groupevent, start_timestamp = group_to_groupevent[group]
+                    if start_timestamp:
+                        metrics.timing(
+                            "rule_fire_history.latency",
+                            (datetime.now(tz=timezone.utc) - start_timestamp).total_seconds(),
+                            tags={"delayed": True, "group_type": group.issue_type.slug},
+                        )
                     rule_fire_history = history.record(
                         rule, group, groupevent.event_id, notification_uuid
                     )

src/sentry/rules/processing/processor.py

@@ -4,7 +4,7 @@
 import random
 import uuid
 from collections.abc import Callable, Collection, Mapping, MutableMapping, Sequence
-from datetime import timedelta
+from datetime import datetime, timedelta
 from random import randrange
 from typing import Any
 
@@ -28,7 +28,6 @@
 from sentry.services.eventstore.models import GroupEvent
 from sentry.types.rules import RuleFuture
 from sentry.utils import json, metrics
-from sentry.utils.dates import ensure_aware
 from sentry.utils.hashlib import hash_values
 from sentry.utils.safe import safe_execute
 
@@ -213,10 +212,12 @@ def __init__(
         is_new_group_environment: bool,
         has_reappeared: bool,
         has_escalated: bool = False,
+        start_timestamp: datetime | None = None,
     ) -> None:
         self.event = event
         self.group = event.group
         self.project = event.project
+        self.start_timestamp = start_timestamp or timezone.now()
 
         self.is_new = is_new
         self.is_regression = is_regression
@@ -282,7 +283,11 @@ def enqueue_rule(self, rule: Rule) -> None:
         buffer.backend.push_to_sorted_set(PROJECT_ID_BUFFER_LIST_KEY, rule.project.id)
 
         value = json.dumps(
-            {"event_id": self.event.event_id, "occurrence_id": self.event.occurrence_id}
+            {
+                "event_id": self.event.event_id,
+                "occurrence_id": self.event.occurrence_id,
+                "start_timestamp": self.start_timestamp,
+            }
         )
         buffer.backend.push_to_hash(
             model=Project,
@@ -417,7 +422,7 @@ def apply_rule(self, rule: Rule, status: GroupRuleStatus) -> None:
         notification_uuid = str(uuid.uuid4())
         metrics.timing(
             "rule_fire_history.latency",
-            (timezone.now() - ensure_aware(self.event.datetime)).total_seconds(),
+            (timezone.now() - self.start_timestamp).total_seconds(),
             tags={"delayed": False, "group_type": self.group.issue_type.slug},
         )
         rule_fire_history = history.record(rule, self.group, self.event.event_id, notification_uuid)

src/sentry/tasks/post_process.py

@@ -1012,6 +1012,7 @@ def process_workflow_engine(job: PostProcessJob) -> None:
                 group_state=job["group_state"],
                 has_reappeared=job["has_reappeared"],
                 has_escalated=job["has_escalated"],
+                start_timestamp_seconds=time(),
             ),
             headers={"sentry-propagate-traces": False},
         )

src/sentry/workflow_engine/processors/delayed_workflow.py

@@ -619,7 +619,7 @@ def get_group_to_groupevent(
     event_data: EventRedisData,
     groups_to_dcgs: dict[GroupId, set[DataConditionGroup]],
     project: Project,
-) -> dict[Group, GroupEvent]:
+) -> dict[Group, tuple[GroupEvent, datetime | None]]:
     groups = Group.objects.filter(id__in=event_data.group_ids)
     group_id_to_group = {group.id: group for group in groups}
 
@@ -638,7 +638,7 @@ def get_group_to_groupevent(
         group_id: {dcg.id for dcg in dcgs} for group_id, dcgs in groups_to_dcgs.items()
     }
 
-    group_to_groupevent: dict[Group, GroupEvent] = {}
+    group_to_groupevent: dict[Group, tuple[GroupEvent, datetime | None]] = {}
     for key, instance in event_data.events.items():
         if key.dcg_ids.intersection(groups_to_dcg_ids.get(key.group_id, set())):
             event = bulk_event_id_to_events.get(instance.event_id)
@@ -652,7 +652,7 @@ def get_group_to_groupevent(
                 group_event.occurrence = bulk_occurrence_id_to_occurrence.get(
                     instance.occurrence_id
                 )
-            group_to_groupevent[group] = group_event
+            group_to_groupevent[group] = (group_event, instance.timestamp)
 
     return group_to_groupevent
 
@@ -661,10 +661,10 @@ def get_group_to_groupevent(
 def fire_actions_for_groups(
     organization: Organization,
     groups_to_fire: dict[GroupId, set[DataConditionGroup]],
-    group_to_groupevent: dict[Group, GroupEvent],
+    group_to_groupevent: dict[Group, tuple[GroupEvent, datetime | None]],
 ) -> None:
     serialized_groups = {
-        group.id: group_event.event_id for group, group_event in group_to_groupevent.items()
+        group.id: group_event.event_id for group, (group_event, _) in group_to_groupevent.items()
     }
     logger.info(
         "workflow_engine.delayed_workflow.fire_actions_for_groups",
@@ -675,7 +675,9 @@ def fire_actions_for_groups(
     )
 
     # Bulk fetch detectors
-    event_id_to_detector = get_detectors_by_groupevents_bulk(list(group_to_groupevent.values()))
+    event_id_to_detector = get_detectors_by_groupevents_bulk(
+        [group_event for group_event, _ in group_to_groupevent.values()]
+    )
 
     # Feature check caching to keep us within the trace budget.
     trigger_actions_ff = features.has("organizations:workflow-engine-trigger-actions", organization)
@@ -697,7 +699,7 @@ def fire_actions_for_groups(
         logger,
         threshold=timedelta(seconds=40),
     ) as tracker:
-        for group, group_event in group_to_groupevent.items():
+        for group, (group_event, start_timestamp) in group_to_groupevent.items():
             with tracker.track(str(group.id)), log_context.new_context(group_id=group.id):
                 workflow_event_data = WorkflowEventData(event=group_event, group=group)
                 detector = event_id_to_detector.get(group_event.event_id)
@@ -729,6 +731,7 @@ def fire_actions_for_groups(
                     workflow_event_data,
                     should_trigger_actions(group_event.group.type),
                     is_delayed=True,
+                    start_timestamp=start_timestamp,
                 )
 
                 event_id = (

src/sentry/workflow_engine/processors/workflow.py

@@ -7,7 +7,6 @@
 import sentry_sdk
 from django.db import router, transaction
 from django.db.models import Q
-from django.utils import timezone
 
 from sentry import features
 from sentry.models.activity import Activity
@@ -144,7 +143,9 @@ def enqueue_workflows(
 
 @sentry_sdk.trace
 def evaluate_workflow_triggers(
-    workflows: set[Workflow], event_data: WorkflowEventData
+    workflows: set[Workflow],
+    event_data: WorkflowEventData,
+    event_start_time: datetime,
 ) -> tuple[set[Workflow], dict[Workflow, DelayedWorkflowItem]]:
     """
     Returns a tuple of (triggered_workflows, queue_items_by_workflow)
@@ -155,7 +156,6 @@ def evaluate_workflow_triggers(
     """
     triggered_workflows: set[Workflow] = set()
     queue_items_by_workflow: dict[Workflow, DelayedWorkflowItem] = {}
-    current_time = timezone.now()
 
     for workflow in workflows:
         evaluation, remaining_conditions = workflow.evaluate_trigger_conditions(event_data)
@@ -168,7 +168,7 @@ def evaluate_workflow_triggers(
                     delayed_when_group_id=workflow.when_condition_group_id,
                     delayed_if_group_ids=[],
                     passing_if_group_ids=[],
-                    timestamp=current_time,
+                    timestamp=event_start_time,
                 )
             else:
                 """
@@ -226,6 +226,7 @@ def evaluate_workflows_action_filters(
     workflows: set[Workflow],
     event_data: WorkflowEventData,
     queue_items_by_workflow: dict[Workflow, DelayedWorkflowItem],
+    event_start_time: datetime,
 ) -> tuple[set[DataConditionGroup], dict[Workflow, DelayedWorkflowItem]]:
     """
     Evaluate the action filters for the given workflows.
@@ -244,7 +245,6 @@ def evaluate_workflows_action_filters(
     }
 
     filtered_action_groups: set[DataConditionGroup] = set()
-    current_time = timezone.now()
 
     for action_condition, workflow in action_conditions_to_workflow.items():
         env = (
@@ -271,7 +271,7 @@ def evaluate_workflows_action_filters(
                         delayed_if_group_ids=[action_condition.id],
                         event=event_data.event,
                         passing_if_group_ids=[],
-                        timestamp=current_time,
+                        timestamp=event_start_time,
                     )
             else:
                 # We should not include activity updates in delayed conditions,
@@ -388,7 +388,9 @@ def _get_associated_workflows(
 
 @log_context.root()
 def process_workflows(
-    event_data: WorkflowEventData, detector: Detector | None = None
+    event_data: WorkflowEventData,
+    event_start_time: datetime,
+    detector: Detector | None = None,
 ) -> set[Workflow]:
     """
     This method will get the detector based on the event, and then gather the associated workflows.
@@ -444,14 +446,14 @@ def process_workflows(
         return set()
 
     triggered_workflows, queue_items_by_workflow_id = evaluate_workflow_triggers(
-        workflows, event_data
+        workflows, event_data, event_start_time
     )
     if not triggered_workflows and not queue_items_by_workflow_id:
         # if there aren't any triggered workflows, there's no action filters to evaluate
         return set()
 
     actions_to_trigger, queue_items_by_workflow_id = evaluate_workflows_action_filters(
-        triggered_workflows, event_data, queue_items_by_workflow_id
+        triggered_workflows, event_data, queue_items_by_workflow_id, event_start_time
     )
     enqueue_workflows(queue_items_by_workflow_id)
     actions = filter_recently_fired_workflow_actions(actions_to_trigger, event_data)
@@ -463,7 +465,12 @@ def process_workflows(
 
     should_trigger_actions = should_fire_workflow_actions(organization, event_data.group.type)
     create_workflow_fire_histories(
-        detector, actions, event_data, should_trigger_actions, is_delayed=False
+        detector,
+        actions,
+        event_data,
+        should_trigger_actions,
+        is_delayed=False,
+        start_timestamp=event_start_time,
     )
     fire_actions(actions, detector, event_data)