feat(alerts): run metrics investigations for platform-owned insight alerts

posthog/tasks/alerts/metrics_investigation.py

@@ -14,6 +14,7 @@
 """
 
 import math
+from datetime import datetime
 from typing import TYPE_CHECKING, Any, NamedTuple
 
 import structlog
@@ -93,6 +94,11 @@ def run_metrics_alert_investigation(alert: AlertConfiguration, alert_check: Aler
 
 
 def _run_investigation(alert: AlertConfiguration, alert_check: AlertCheck) -> str:
+    return investigate_metrics_alert(alert, fired_at=alert_check.created_at)
+
+
+def investigate_metrics_alert(alert: AlertConfiguration, *, fired_at: datetime) -> str:
+    """The summary of an investigation of the window around `fired_at`. Raises when it cannot run."""
     # Deferred: the facade pulls in HogQL machinery that must stay off the
     # alerts import path (matching how core defers facade.queries).
     from products.metrics.backend.facade.api import investigate_incident  # noqa: PLC0415
@@ -104,7 +110,7 @@ def _run_investigation(alert: AlertConfiguration, alert_check: AlertCheck) -> st
             team=alert.team,
             context=IncidentContext(
                 metric_name=target.metric_name,
-                fired_at=alert_check.created_at,
+                fired_at=fired_at,
                 service_name=target.service_name,
             ),
         )

products/alerts/backend/platform_investigation.py

@@ -8,14 +8,21 @@
 write, and runs once the evaluation is recorded.
 """
 
+from collections.abc import Callable
+from contextlib import AbstractContextManager, nullcontext
 from dataclasses import replace
 from datetime import UTC, datetime
 from typing import Any, Final
 from uuid import UUID
 
 from django.db import connection, transaction
 
+import structlog
+
+from posthog.schema import AlertState as InsightAlertState
+
 from posthog.dataclasses import frozen
+from posthog.tasks.alerts.metrics_investigation import investigate_metrics_alert, should_investigate_metrics_alert
 from posthog.temporal.alerts.investigation import (
     InvestigationDecision,
     carried_verdict_suppresses,
@@ -36,11 +43,14 @@
 from products.alerts.backend.models.investigation import InsightAlertInvestigation
 from products.alerts.backend.platform_investigation_detail import (
     INVESTIGATION_REQUESTED_KEY,
+    METRICS_INVESTIGATION_REQUESTED_KEY,
     InvestigationClaims,
     InvestigationStart,
+    MetricsInvestigationRun,
     requested_investigation,
 )
 from products.alerts_platform.backend.facade.contracts import (
+    AlertDeliveryRequest,
     AlertEventKind,
     HeldDelivery,
     HeldDeliveryAddress,
@@ -51,8 +61,13 @@
 from products.alerts_platform.backend.facade.enums import HoldReleasedBy
 from products.alerts_platform.backend.facade.lifecycle import AlertState, NotificationAction
 
+logger = structlog.get_logger(__name__)
+
 _CLAIM_LOCK_SQL: Final = "SELECT pg_advisory_xact_lock(hashtext(%s))"
 HOLD_REASON: Final = "investigation"
+METRICS_HOLD_REASON: Final = "metrics_investigation"
+# A metrics investigation is a few bounded queries, so its firing waits minutes at most, not the agent's grace.
+METRICS_HOLD_SECONDS: Final = 5 * 60
 _EVALUATED_KEYS: Final = ("anomaly_scores", "triggered_points", "triggered_dates", "interval", "triggered_metadata")
 # A verdict-change follow-up is recorded under the key of the firing it follows up plus this suffix,
 # so it gets a delivery of its own.
@@ -144,18 +159,45 @@ def with_investigation(check: PlatformAlertCheckInput, evaluation: SourceBatchEv
 
     A firing the agent investigates asks for its claim in `source_detail`, and a gated alert holds the
     firing's delivery until the verdict. A firing past the budget is muted when the firing's last
-    verdict suppresses it. Only an owned check, because a parallel run must never pay for an agent run.
+    verdict suppresses it. A metrics alert's firing asks for its metrics investigation, which holds
+    nothing. Only an owned check, because a parallel run must never pay for an agent run.
     """
     if not check.owned_by_platform or check.legacy_configuration_id is None or len(evaluation.outcomes) != 1:
         return evaluation
     outcome = evaluation.outcomes[0]
     # A skipped check keeps a firing alert's state without evaluating it, so it has nothing to investigate.
     if outcome.skipped or len(outcome.groups) != 1 or outcome.groups[0].new_state != AlertState.FIRING.value:
         return evaluation
-    alert = AlertConfiguration.objects.filter(team_id=check.team_id, id=check.legacy_configuration_id).first()
+    alert = (
+        AlertConfiguration.objects.select_related("insight")
+        .filter(team_id=check.team_id, id=check.legacy_configuration_id)
+        .first()
+    )
     if alert is None:
         return evaluation
     group = outcome.groups[0]
+    was_firing = check.instance().state == AlertState.FIRING.value
+    if should_investigate_metrics_alert(
+        alert,
+        previous_state=InsightAlertState.FIRING if was_firing else None,
+        new_state=InsightAlertState.FIRING,
+    ):
+        group = replace(
+            group, source_detail={**group.source_detail, METRICS_INVESTIGATION_REQUESTED_KEY: str(alert.id)}
+        )
+        # Held, as the insight stack sends a metrics firing with its explanation. The hold also lets
+        # delivery drop the incident trigger of a firing that resolved while it waited.
+        metrics_held = None
+        if evaluation.deliveries:
+            (request,) = evaluation.deliveries
+            metrics_held = HeldDelivery(
+                request=request, reason=METRICS_HOLD_REASON, release_after_seconds=METRICS_HOLD_SECONDS
+            )
+        return replace(
+            evaluation,
+            outcomes=(replace(outcome, groups=(group,), held_delivery=metrics_held),),
+            deliveries=() if metrics_held else evaluation.deliveries,
+        )
     decision = decide_investigation(
         alert,
         firing=True,
@@ -184,13 +226,14 @@ def with_investigation(check: PlatformAlertCheckInput, evaluation: SourceBatchEv
 
 
 def claim_requested_investigations(team_id: int, outcomes: tuple[PlatformAlertOutcome, ...]) -> InvestigationClaims:
-    """Claims the investigation each recorded firing asked for, and returns the agent runs to start.
+    """Claims the investigation each recorded firing asked for, and returns the runs to start.
 
     A refused claim releases the firing's held delivery by the firing's last verdict, as the insight
     stack mutes a check that runs no investigation of its own.
     """
     starts: list[InvestigationStart] = []
     deliveries = []
+    metrics: list[MetricsInvestigationRun] = []
     for outcome in outcomes:
         requested = requested_investigation(outcome)
         if requested is None:
@@ -200,6 +243,27 @@ def claim_requested_investigations(team_id: int, outcomes: tuple[PlatformAlertOu
             # The safety net sends the held firing at its deadline.
             continue
         group = outcome.groups[0]
+        if requested.metrics:
+            claim = claim_investigation(
+                alert,
+                configuration_id=outcome.configuration_id,
+                evaluation_key=outcome.evaluation_key,
+                episode_started_at=requested.episode_started_at,
+                kind=InsightAlertInvestigation.Kind.METRICS,
+                evaluated={"value": group.value},
+            )
+            if claim.claimed:
+                metrics.append(
+                    MetricsInvestigationRun(
+                        team_id=team_id,
+                        investigation_id=str(claim.investigation_id),
+                        configuration_id=str(outcome.configuration_id),
+                        evaluation_key=outcome.evaluation_key,
+                    )
+                )
+            elif outcome.held_delivery is not None and (request := _release_metrics_hold(team_id, outcome)):
+                deliveries.append(request)
+            continue
         claim = claim_investigation(
             alert,
             configuration_id=outcome.configuration_id,
@@ -237,4 +301,67 @@ def claim_requested_investigations(team_id: int, outcomes: tuple[PlatformAlertOu
         )
         if request is not None:
             deliveries.append(request)
-    return InvestigationClaims(starts=tuple(starts), deliveries=tuple(deliveries))
+    return InvestigationClaims(starts=tuple(starts), deliveries=tuple(deliveries), metrics=tuple(metrics))
+
+
+def _release_metrics_hold(team_id: int, outcome: PlatformAlertOutcome) -> AlertDeliveryRequest | None:
+    return release_held(
+        HeldDeliveryAddress(
+            team_id=team_id, configuration_id=str(outcome.configuration_id), evaluation_key=outcome.evaluation_key
+        ),
+        suppress=False,
+        released_by=HoldReleasedBy.CLAIM_REFUSED,
+    )
+
+
+def release_metrics_firing(run: MetricsInvestigationRun) -> AlertDeliveryRequest | None:
+    """Sends the firing a metrics investigation held, once the run ended or could not start. Returns
+    the request the caller starts. A retry gets the same request back."""
+    return release_held(
+        HeldDeliveryAddress(
+            team_id=run.team_id, configuration_id=run.configuration_id, evaluation_key=run.evaluation_key
+        ),
+        suppress=False,
+        released_by=HoldReleasedBy.VERDICT,
+    )
+
+
+def fail_unstarted_metrics_investigation(run: MetricsInvestigationRun) -> None:
+    """Records that a granted metrics investigation never started, because its activity ran out of time."""
+    InsightAlertInvestigation.objects.for_team(run.team_id).filter(
+        id=run.investigation_id, status=InvestigationStatus.PENDING
+    ).update(
+        status=InvestigationStatus.FAILED, error={"message": "The investigation ran out of time before it started."}
+    )
+
+
+def run_metrics_investigation(
+    run: MetricsInvestigationRun, *, query_slot: Callable[[], AbstractContextManager[None]] = nullcontext
+) -> None:
+    """Runs a metrics alert's investigation once, and records its summary or its failure.
+
+    The move from PENDING to RUNNING is the run's lease, so a retry of a timed-out attempt finds the
+    row taken and runs nothing. A run that dies holds RUNNING until the in-flight horizon passes,
+    as an agent run that dies does. Its failure never fails the activity, because the firing it
+    explains has already gone out. `query_slot` holds a ClickHouse slot around the queries, so they
+    share the evaluation's concurrency cap.
+    """
+    rows = InsightAlertInvestigation.objects.for_team(run.team_id).filter(id=run.investigation_id)
+    if not rows.filter(status=InvestigationStatus.PENDING).update(status=InvestigationStatus.RUNNING):
+        return
+    investigation = rows.select_related("alert_configuration__insight", "alert_configuration__team").get()
+    alert = investigation.alert_configuration
+    running = rows.filter(status=InvestigationStatus.RUNNING)
+    try:
+        # The firing's start, because a metrics alert investigates only the check that started firing.
+        with query_slot():
+            summary = investigate_metrics_alert(
+                alert, fired_at=investigation.episode_started_at or investigation.created_at
+            )
+    except Exception as error:
+        logger.exception("platform_metrics_alert_investigation_failed", alert_id=str(alert.id))
+        running.update(
+            status=InvestigationStatus.FAILED, error={"message": (str(error) or type(error).__name__)[:1000]}
+        )
+        return
+    running.update(status=InvestigationStatus.DONE, summary=summary)

products/alerts/backend/platform_investigation_detail.py

@@ -14,6 +14,8 @@
 # The legacy alert whose agent investigates the firing. Recorded with the firing, so history shows
 # which checks asked for an investigation.
 INVESTIGATION_REQUESTED_KEY = "investigation_alert_id"
+# The same, for a metrics alert, whose investigation is a few metric queries rather than an agent run.
+METRICS_INVESTIGATION_REQUESTED_KEY = "metrics_investigation_alert_id"
 
 
 @frozen
@@ -42,24 +44,38 @@ def workflow_inputs(self) -> dict[str, object]:
         }
 
 
+@frozen
+class MetricsInvestigationRun:
+    """A metrics investigation the claim granted, which the workflow runs in an activity of its own."""
+
+    team_id: int
+    investigation_id: str
+    # The firing the run explains, whose held delivery the run releases once it ends.
+    configuration_id: str = ""
+    evaluation_key: str = ""
+
+
 @frozen
 class InvestigationClaims:
     starts: tuple[InvestigationStart, ...] = ()
     # Held firings whose claim was refused, released by the firing's last verdict and due to start.
     deliveries: tuple[AlertDeliveryRequest, ...] = ()
+    metrics: tuple[MetricsInvestigationRun, ...] = ()
 
 
 @frozen
 class RequestedInvestigation:
     alert_id: str
     episode_started_at: datetime | None
+    metrics: bool = False
 
 
 def requested_investigation(outcome: PlatformAlertOutcome) -> RequestedInvestigation | None:
     """The investigation an outcome asks for, or None. Only an ungrouped firing asks for one."""
     for group in outcome.groups:
-        alert_id = group.source_detail.get(INVESTIGATION_REQUESTED_KEY)
-        if alert_id is not None:
-            started_at = group.firing_episode.started_at if group.firing_episode else None
+        started_at = group.firing_episode.started_at if group.firing_episode else None
+        if (alert_id := group.source_detail.get(INVESTIGATION_REQUESTED_KEY)) is not None:
             return RequestedInvestigation(alert_id=str(alert_id), episode_started_at=started_at)
+        if (alert_id := group.source_detail.get(METRICS_INVESTIGATION_REQUESTED_KEY)) is not None:
+            return RequestedInvestigation(alert_id=str(alert_id), episode_started_at=started_at, metrics=True)
     return None

products/alerts/backend/platform_investigation_message.py

@@ -96,7 +96,10 @@ def _read(team_id: int, transition: AnnouncedTransition) -> FiringInvestigation
     value = evaluated.get("value")
     breaches: tuple[str, ...] | None = None
     link: str | None = None
-    if follow_up or released_by == HoldReleasedBy.VERDICT:
+    # Only the agent's verdict rewords a firing. A metrics investigation explains the firing beside
+    # its own breaches, so its release keeps them, also when it failed.
+    agent = kind == InsightAlertInvestigation.Kind.AGENT
+    if follow_up or (agent and released_by == HoldReleasedBy.VERDICT):
         previous_verdict = (
             rows.filter(episode_started_at=episode_started_at, created_at__lt=created_at, verdict__isnull=False)
             .order_by("-created_at")
@@ -116,8 +119,11 @@ def _read(team_id: int, transition: AnnouncedTransition) -> FiringInvestigation
             ).lines
         )
         link = notebook_url(notebook_short_id) if notebook_short_id else None
-    elif released_by == HoldReleasedBy.SAFETY_NET:
+    elif agent and released_by == HoldReleasedBy.SAFETY_NET:
         breaches = tuple(safety_net_breaches(triggered_dates=triggered_dates, value=value))
+    elif not agent and status == InvestigationStatus.DONE and summary:
+        # Every channel reads the breaches, so the explanation goes after them, as the agent's teaser does.
+        breaches = (*(str(line) for line in transition.source_detail.get("breaches") or []), summary)
 
     return FiringInvestigation(
         status=status,

products/alerts/backend/platform_source_cycle.py

@@ -328,22 +328,24 @@
     return token if admitted else None
 
 
-class _ClickHouseSlot:
+class ClickHouseQuerySlot:
     """Holds a ClickHouse slot around one attempt's query, and counts how long it waited for one.
 
     Raises `ClickHouseBudgetFull` when no slot comes free in time. `holder` names one attempt of one
     check, so a retry never takes over the slot of the attempt before it.
     """
 
-    def __init__(self, holder: str) -> None:
+    def __init__(self, holder: str, *, wait_seconds: float | None = None) -> None:
         self.holder = holder
         self.waited_ms = 0
+        # A caller with a budget of its own waits no longer than it, on the same shared cap.
+        self._limit = _QUERY_LIMIT if wait_seconds is None else replace(_QUERY_LIMIT, retry_timeout=wait_seconds)
 
     @contextmanager
     def __call__(self) -> Iterator[None]:
         waiting_since = time.monotonic()
         try:
-            slot = _QUERY_LIMIT.use(self.holder, limit=settings.ALERTS_PLATFORM_INSIGHT_MAX_INFLIGHT_EVALUATIONS)
+            slot = self._limit.use(self.holder, limit=settings.ALERTS_PLATFORM_INSIGHT_MAX_INFLIGHT_EVALUATIONS)
         except ConcurrencyLimitExceeded as error:
             raise ClickHouseBudgetFull() from error
         except Exception as error:
@@ -357,13 +359,13 @@
         finally:
             if slot is not None:
                 try:
-                    _QUERY_LIMIT.release(slot)
+                    self._limit.release(slot)
                 except Exception:
                     # Best effort, so a Redis failure never replaces the query's result. The lease frees the slot.
                     logger.exception("Could not release a ClickHouse slot", holder=self.holder)
 
 
-def _with_capacity_retries[T](run: Callable[[], T], alert: AlertConfiguration, query_slot: _ClickHouseSlot) -> T:
+def _with_capacity_retries[T](run: Callable[[], T], alert: AlertConfiguration, query_slot: ClickHouseQuerySlot) -> T:
     """Runs a check's query, and runs it again after a ClickHouse refusal for load, on the retry
     policy production evaluates the alert's cadence with.
 
@@ -687,7 +689,7 @@
 
 def _decide(
     check: PlatformAlertCheckInput,
     now: datetime,
     *,
     evaluation_id: str,
     attempt: int,
@@ -742,7 +744,7 @@
         alert_calculation_interval=alert.calculation_interval,
         alert_config_type=(alert.config or {}).get("type"),
     )
-    query_slot = _ClickHouseSlot(f"{evaluation_id}:{attempt}")
+    query_slot = ClickHouseQuerySlot(f"{evaluation_id}:{attempt}")
     if check.grouping.mode != GroupingMode.SINGLE:
         return _decide_grouped(check, snapshot, alert, now=now, query_slot=query_slot)
     started_at = time.monotonic()
@@ -901,11 +903,11 @@
 
 def _decide_grouped(
     check: PlatformAlertCheckInput,
     snapshot: AlertSnapshot,
     alert: AlertConfiguration,
     *,
     now: datetime,
-    query_slot: _ClickHouseSlot,
+    query_slot: ClickHouseQuerySlot,
 ) -> PlatformAlertOutcome | None:
     """One check of a breakdown alert, judged per breakdown value through the platform's decision.