feat(gdd): report incremental derived data processing latency

src/sentry/issues/derived/processing.py

@@ -1,6 +1,8 @@
 import enum
 import logging
 import time
+from collections.abc import Sequence
+from dataclasses import dataclass
 from datetime import datetime, timedelta
 from typing import NamedTuple
 
@@ -10,7 +12,7 @@
 from django.utils import timezone
 
 from sentry.issues.derived.aggregators import AGGREGATORS
-from sentry.issues.derived.framework import Pipeline
+from sentry.issues.derived.framework import Pipeline, State
 from sentry.issues.derived.store import GroupDerivedDataStore
 from sentry.issues.derived.tasks import process_group_log_task
 from sentry.issues.models.groupactionlogentry import GroupActionLogEntry
@@ -57,6 +59,37 @@ class ProcessingStrategy(enum.Enum):
     INLINE = "inline"  # try to process all pending actions quickly; fall back to ASYNC
 
 
+@dataclass(frozen=True)
+class DerivedMetrics:
+    """Encapsulates derived-data metric reporting; incremental mode adds per-entry latency."""
+
+    mode: ProcessingStrategy
+    incremental: bool
+
+    def report_batch_processed(
+        self,
+        entries: Sequence[GroupActionLogEntry],
+        result: State,
+    ) -> None:
+        if self.incremental:
+            now = timezone.now()
+            tags = {"mode": self.mode.value}
+            for entry in entries:
+                age_seconds = (now - entry.date_added).total_seconds()
+                metrics.distribution(
+                    "issues.derived.incremental_processing_latency",
+                    age_seconds,
+                    tags=tags,
+                    unit="second",
+                )
+        for f in result.updated:
+            metrics.incr(
+                "issues.derived.feature_updated",
+                sample_rate=1.0,
+                tags={"feature": f.name},
+            )
+
+
 def _ensure_derived(group_id: int, pipeline_hash: str) -> GroupDerivedData:
     """Get or create the GroupDerivedData row for a group.
 
@@ -101,6 +134,7 @@ def _process_batch(
     batch_size: int,
     *,
     persist: bool = True,
+    derived_metrics: DerivedMetrics | None = None,
 ) -> bool:
     """
     Process up to `batch_size` entries for a group. Updates derived in place.
@@ -151,11 +185,8 @@ def _process_batch(
     ).update(cursor_date=last_date, cursor_id=last_id, **state_update)
 
     if updated:
-        # Features updated in this batch (not total; a feature appears at most once per batch)
-        for f in result.updated:
-            metrics.incr(
-                "issues.derived.feature_updated", sample_rate=1.0, tags={"feature": f.name}
-            )
+        if derived_metrics is not None:
+            derived_metrics.report_batch_processed(entries, result)
         derived.cursor_date = last_date
         derived.cursor_id = last_id
         GroupDerivedDataStore.apply_to_instance(derived, state_update)
@@ -204,6 +235,7 @@ def _drain_log(
     *,
     time_limit: timedelta,
     persist: bool = True,
+    derived_metrics: DerivedMetrics | None = None,
 ) -> bool:
     """Process pending log entries into *derived*, batching as needed.
 
@@ -214,7 +246,9 @@ def _drain_log(
     When *persist* is False, batches update only the in-memory object.
     """
     deadline = time.monotonic() + time_limit.total_seconds()
-    while _process_batch(pipeline, derived, batch_size, persist=persist):
+    while _process_batch(
+        pipeline, derived, batch_size, persist=persist, derived_metrics=derived_metrics
+    ):
         if time.monotonic() >= deadline:
             return False
     return True
@@ -230,6 +264,7 @@ def process_group_log(
     batch_size: int = DEFAULT_BATCH_SIZE,
     pipeline: Pipeline[GroupActionLogEntry] | None = None,
     timeout: timedelta | None = None,
+    derived_metrics: DerivedMetrics | None = None,
 ) -> GroupDerivedData:
     """Fully drain all pending entries for a group's row.
 
@@ -243,12 +278,14 @@ def process_group_log(
         derived = _ensure_derived(group_id, p.pipeline_hash)
 
     if timeout is not None:
-        drained = _drain_log(derived, p, batch_size, time_limit=timeout)
+        drained = _drain_log(
+            derived, p, batch_size, time_limit=timeout, derived_metrics=derived_metrics
+        )
         if not drained:
             raise GroupLogTimeout(group_id)
     else:
         # No timeout — drain to completion.
-        while _process_batch(p, derived, batch_size):
+        while _process_batch(p, derived, batch_size, derived_metrics=derived_metrics):
             pass
 
     return derived
@@ -265,12 +302,15 @@ def trigger_group_log_processing(group_id: int, *, strategy: ProcessingStrategy)
       INLINE — try to process all pending actions quickly; fall back to ASYNC
     """
     if strategy is ProcessingStrategy.ASYNC:
-        process_group_log_task.delay(group_id)
+        process_group_log_task.delay(group_id, incremental=True)
         return
 
     if strategy is ProcessingStrategy.SYNC:
         try:
-            process_group_log(group_id)
+            process_group_log(
+                group_id,
+                derived_metrics=DerivedMetrics(mode=strategy, incremental=True),
+            )
         except ObjectDoesNotExist:
             pass
         return
@@ -286,12 +326,17 @@ def trigger_group_log_processing(group_id: int, *, strategy: ProcessingStrategy)
         except ObjectDoesNotExist:
             return
 
-        has_more = _process_batch(pipeline, derived, INLINE_BATCH_SIZE)
+        has_more = _process_batch(
+            pipeline,
+            derived,
+            INLINE_BATCH_SIZE,
+            derived_metrics=DerivedMetrics(mode=strategy, incremental=True),
+        )
     if has_more:
         # Derived data will be stale for any code running between now and
         # when the task completes.
         metrics.incr("issues.derived.inline_fallback_to_async")
-        process_group_log_task.delay(group_id)
+        process_group_log_task.delay(group_id, incremental=True)
 
 
 # ---------------------------------------------------------------------------
@@ -430,7 +475,14 @@ def build_and_promote_derived_data(
     result = PromotionResult.CURSOR_BEHIND
     for attempt in range(MAX_PROMOTION_ATTEMPTS):
         remaining = timedelta(seconds=max(0, deadline - time.monotonic()))
-        drained = _drain_log(derived, PIPELINE, batch_size, time_limit=remaining, persist=False)
+        drained = _drain_log(
+            derived,
+            PIPELINE,
+            batch_size,
+            time_limit=remaining,
+            persist=False,
+            derived_metrics=DerivedMetrics(mode=ProcessingStrategy.ASYNC, incremental=False),
+        )
         if not drained:
             _generation_cache.set(current_gen_id, derived)
             raise GroupLogTimeout(group_id, generation_id=current_gen_id)

src/sentry/issues/derived/tasks.py

@@ -52,13 +52,18 @@ def _stale_pipeline_filter(qs: BaseQuerySet[Group], pipeline_hash: str) -> BaseQ
     namespace=issues_tasks,
     silo_mode=SiloMode.CELL,
 )
-def process_group_log_task(group_id: int, **kwargs: object) -> None:
+def process_group_log_task(group_id: int, incremental: bool = False, **kwargs: object) -> None:
     """Drain all pending action log entries for a single group into its derived data."""
-    from sentry.issues.derived.processing import process_group_log
+    from sentry.issues.derived.processing import (
+        DerivedMetrics,
+        ProcessingStrategy,
+        process_group_log,
+    )
     from sentry.models.group import Group
 
+    derived_metrics = DerivedMetrics(mode=ProcessingStrategy.ASYNC, incremental=incremental)
     try:
-        process_group_log(group_id)
+        process_group_log(group_id, derived_metrics=derived_metrics)
     except Group.DoesNotExist:
         logger.info("process_group_log_task.group_not_found", extra={"group_id": group_id})
 
@@ -257,7 +262,13 @@ def process_project_derived_data_batch(
     """
     from taskbroker_client.state import current_task
 
-    from sentry.issues.derived.processing import PIPELINE, GroupLogTimeout, process_group_log
+    from sentry.issues.derived.processing import (
+        PIPELINE,
+        DerivedMetrics,
+        GroupLogTimeout,
+        ProcessingStrategy,
+        process_group_log,
+    )
     from sentry.issues.models.groupderiveddata import GroupDerivedData
     from sentry.models.group import Group
     from sentry.taskworker.selfchain_idempotency import already_spawned, mark_spawned
@@ -296,7 +307,11 @@ def process_project_derived_data_batch(
     for group_id in group_ids:
         remaining = timedelta(seconds=max(0, timeout_seconds - (time.monotonic() - start)))
         try:
-            process_group_log(group_id, timeout=remaining)
+            process_group_log(
+                group_id,
+                timeout=remaining,
+                derived_metrics=DerivedMetrics(mode=ProcessingStrategy.ASYNC, incremental=False),
+            )
             processed += 1
         except Group.DoesNotExist:
             logger.info(

tests/sentry/issues/test_action_log.py

@@ -672,7 +672,7 @@ def test_force_async_derived_dispatches_task(self, mock_task: MagicMock) -> None
         # Derived data was NOT processed inline
         assert not GroupDerivedData.objects.filter(group_id=self.group.id).exists()
         # Task was dispatched instead
-        mock_task.delay.assert_called_once_with(self.group.id)
+        mock_task.delay.assert_called_once_with(self.group.id, incremental=True)
 
     @patch("sentry.issues.derived.processing.process_group_log_task")
     def test_inline_derived_processes_without_task(self, mock_task: MagicMock) -> None: