diff --git a/src/sentry/issues/derived/processing.py b/src/sentry/issues/derived/processing.py index 3666851a247b..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,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) diff --git a/src/sentry/issues/derived/tasks.py b/src/sentry/issues/derived/tasks.py index d879691c2c72..627b3391073d 100644 --- a/src/sentry/issues/derived/tasks.py +++ b/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( 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: