From fce1b4fc39842621d02396859ef48c8c994f12ce Mon Sep 17 00:00:00 2001 From: Kyle Consalus Date: Fri, 31 Jul 2026 13:14:16 -0700 Subject: [PATCH 1/3] feat(gdd): Track incremental processing latency --- src/sentry/issues/derived/processing.py | 32 +++++++++++++++++++------ src/sentry/issues/derived/tasks.py | 7 +++--- 2 files changed, 29 insertions(+), 10 deletions(-) diff --git a/src/sentry/issues/derived/processing.py b/src/sentry/issues/derived/processing.py index 3666851a247b..f209b49fa41e 100644 --- a/src/sentry/issues/derived/processing.py +++ b/src/sentry/issues/derived/processing.py @@ -101,6 +101,7 @@ def _process_batch( batch_size: int, *, persist: bool = True, + processing_mode: ProcessingStrategy | None = None, ) -> bool: """ Process up to `batch_size` entries for a group. Updates derived in place. @@ -151,6 +152,17 @@ def _process_batch( ).update(cursor_date=last_date, cursor_id=last_id, **state_update) if updated: + if processing_mode is not None: + now = timezone.now() + tags = {"mode": processing_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", + ) # Features updated in this batch (not total; a feature appears at most once per batch) for f in result.updated: metrics.incr( @@ -204,6 +216,7 @@ def _drain_log( *, time_limit: timedelta, persist: bool = True, + processing_mode: ProcessingStrategy | None = None, ) -> bool: """Process pending log entries into *derived*, batching as needed. @@ -214,7 +227,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, processing_mode=processing_mode + ): if time.monotonic() >= deadline: return False return True @@ -230,6 +245,7 @@ def process_group_log( batch_size: int = DEFAULT_BATCH_SIZE, pipeline: Pipeline[GroupActionLogEntry] | None = None, timeout: timedelta | None = None, + processing_mode: ProcessingStrategy | None = None, ) -> GroupDerivedData: """Fully drain all pending entries for a group's row. @@ -243,12 +259,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, processing_mode=processing_mode + ) 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, processing_mode=processing_mode): pass return derived @@ -265,12 +283,12 @@ 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, processing_mode=strategy) except ObjectDoesNotExist: pass return @@ -286,12 +304,12 @@ 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, processing_mode=strategy) 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) # --------------------------------------------------------------------------- diff --git a/src/sentry/issues/derived/tasks.py b/src/sentry/issues/derived/tasks.py index d879691c2c72..9967dddf2625 100644 --- a/src/sentry/issues/derived/tasks.py +++ b/src/sentry/issues/derived/tasks.py @@ -52,13 +52,14 @@ 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 ProcessingStrategy, process_group_log from sentry.models.group import Group + mode = ProcessingStrategy.ASYNC if incremental else None try: - process_group_log(group_id) + process_group_log(group_id, processing_mode=mode) except Group.DoesNotExist: logger.info("process_group_log_task.group_not_found", extra={"group_id": group_id}) From e1a98ed4c3a94080d05f6f67dc824193e911dc72 Mon Sep 17 00:00:00 2001 From: Kyle Consalus Date: Fri, 31 Jul 2026 17:23:07 -0700 Subject: [PATCH 2/3] use a helper class --- src/sentry/issues/derived/processing.py | 86 +++++++++++++++++-------- src/sentry/issues/derived/tasks.py | 24 +++++-- 2 files changed, 79 insertions(+), 31 deletions(-) diff --git a/src/sentry/issues/derived/processing.py b/src/sentry/issues/derived/processing.py index f209b49fa41e..e2279670f719 100644 --- a/src/sentry/issues/derived/processing.py +++ b/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,7 +134,7 @@ def _process_batch( batch_size: int, *, persist: bool = True, - processing_mode: ProcessingStrategy | None = None, + derived_metrics: DerivedMetrics | None = None, ) -> bool: """ Process up to `batch_size` entries for a group. Updates derived in place. @@ -152,22 +185,8 @@ def _process_batch( ).update(cursor_date=last_date, cursor_id=last_id, **state_update) if updated: - if processing_mode is not None: - now = timezone.now() - tags = {"mode": processing_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", - ) - # 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) @@ -216,7 +235,7 @@ def _drain_log( *, time_limit: timedelta, persist: bool = True, - processing_mode: ProcessingStrategy | None = None, + derived_metrics: DerivedMetrics | None = None, ) -> bool: """Process pending log entries into *derived*, batching as needed. @@ -228,7 +247,7 @@ def _drain_log( """ deadline = time.monotonic() + time_limit.total_seconds() while _process_batch( - pipeline, derived, batch_size, persist=persist, processing_mode=processing_mode + pipeline, derived, batch_size, persist=persist, derived_metrics=derived_metrics ): if time.monotonic() >= deadline: return False @@ -245,7 +264,7 @@ def process_group_log( batch_size: int = DEFAULT_BATCH_SIZE, pipeline: Pipeline[GroupActionLogEntry] | None = None, timeout: timedelta | None = None, - processing_mode: ProcessingStrategy | None = None, + derived_metrics: DerivedMetrics | None = None, ) -> GroupDerivedData: """Fully drain all pending entries for a group's row. @@ -260,13 +279,13 @@ def process_group_log( if timeout is not None: drained = _drain_log( - derived, p, batch_size, time_limit=timeout, processing_mode=processing_mode + 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, processing_mode=processing_mode): + while _process_batch(p, derived, batch_size, derived_metrics=derived_metrics): pass return derived @@ -288,7 +307,10 @@ def trigger_group_log_processing(group_id: int, *, strategy: ProcessingStrategy) if strategy is ProcessingStrategy.SYNC: try: - process_group_log(group_id, processing_mode=strategy) + process_group_log( + group_id, + derived_metrics=DerivedMetrics(mode=strategy, incremental=True), + ) except ObjectDoesNotExist: pass return @@ -304,7 +326,12 @@ def trigger_group_log_processing(group_id: int, *, strategy: ProcessingStrategy) except ObjectDoesNotExist: return - has_more = _process_batch(pipeline, derived, INLINE_BATCH_SIZE, processing_mode=strategy) + 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. @@ -448,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) diff --git a/src/sentry/issues/derived/tasks.py b/src/sentry/issues/derived/tasks.py index 9967dddf2625..627b3391073d 100644 --- a/src/sentry/issues/derived/tasks.py +++ b/src/sentry/issues/derived/tasks.py @@ -54,12 +54,16 @@ def _stale_pipeline_filter(qs: BaseQuerySet[Group], pipeline_hash: str) -> BaseQ ) 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 ProcessingStrategy, process_group_log + from sentry.issues.derived.processing import ( + DerivedMetrics, + ProcessingStrategy, + process_group_log, + ) from sentry.models.group import Group - mode = ProcessingStrategy.ASYNC if incremental else None + derived_metrics = DerivedMetrics(mode=ProcessingStrategy.ASYNC, incremental=incremental) try: - process_group_log(group_id, processing_mode=mode) + 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}) @@ -258,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 @@ -297,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( From 8c0370c0608d86af7b15797828826de63ac97bdd Mon Sep 17 00:00:00 2001 From: Kyle Consalus Date: Fri, 31 Jul 2026 17:44:09 -0700 Subject: [PATCH 3/3] fix tes --- tests/sentry/issues/test_action_log.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/sentry/issues/test_action_log.py b/tests/sentry/issues/test_action_log.py index af6c28149d90..c6e51e7c33e2 100644 --- a/tests/sentry/issues/test_action_log.py +++ b/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: