feat(observability): count slow status transitions by who submitted them

cloud_pipelines_backend/instrumentation/metrics.py

@@ -25,7 +25,9 @@
 import datetime
 import enum
 import logging
+import os
 
+import sqlalchemy as sql
 from opentelemetry import metrics as otel_metrics
 from sqlalchemy import event as sql_event
 from sqlalchemy import orm
@@ -71,6 +73,15 @@ class MetricUnit(str, enum.Enum):
     unit=MetricUnit.SECONDS,
 )
 
+execution_slow_transitions = orchestrator_meter.create_counter(
+    name="execution.status_transition.slow",
+    description=(
+        "Transitions that took longer than the slow threshold, by owner — the metric that"
+        " answers how many distinct submitters a latency alert is affecting"
+    ),
+    unit="{transition}",
+)
+
 execution_status_count = orchestrator_meter.create_observable_gauge(
     name="execution.status.count",
     callbacks=[],
@@ -91,6 +102,14 @@ class MetricUnit(str, enum.Enum):
 # query that filters on a label never silently drops the executions that lack it.
 _UNSET = "none"
 
+# Transitions at or above this are counted by owner. A deployment sets it to whatever its
+# own latency alert fires on, so the counter and the alert describe the same population.
+_SLOW_TRANSITION_SECONDS = float(
+    os.environ.get("TANGLE_SLOW_TRANSITION_SECONDS", "6000")
+)
+
+_SLOW_KEY = "tangle.instrumentation.slow_transitions"
+
 
 def _run_attempt(*, history: list[dict[str, str]]) -> str:
     """Whether the execution is waiting to START or waiting to RESUME.
@@ -228,6 +247,108 @@ def _accelerator(*, execution: backend_types_sql.ExecutionNode) -> tuple[str, st
     return model, str(count)
 
 
+def _owners_of(
+    *, session: orm.Session, node_ids: set[backend_types_sql.IdType]
+) -> dict[backend_types_sql.IdType, str]:
+    """Who submitted the run each of these executions belongs to.
+
+        reads: pipeline_run.created_by
+
+        a node under a run      → "flywheel-production@shopify-applications…"  (a)
+        the run's own root node → the same, by the other half of the union     (b)
+        a node with no run      → absent from the result                       (c)
+
+        (a) through the ancestor link, which is the only route an arbitrary node has
+        (b) a root execution has no ancestor link to itself
+        (c) the caller labels it "none" rather than dropping the point
+
+    One round trip for the whole batch, in the shape the orchestrator already uses to
+    answer this question. An arbitrary node has no foreign key to its run; only the
+    root execution does, so neither half of the union answers it alone.
+    """
+    link = backend_types_sql.ExecutionToAncestorExecutionLink
+    run = backend_types_sql.PipelineRun
+    as_descendant = (
+        sql.select(link.execution_id, run.created_by)
+        .join(link, link.ancestor_execution_id == run.root_execution_id)
+        .where(link.execution_id.in_(node_ids))
+    )
+    as_root = sql.select(run.root_execution_id, run.created_by).where(
+        run.root_execution_id.in_(node_ids)
+    )
+    rows = session.execute(sql.union(as_descendant, as_root)).all()
+    return {node_id: created_by for node_id, created_by in rows if created_by}
+
+
+@sql_event.listens_for(orm.Session, "after_commit")
+def _count_slow_transitions_by_owner(session: orm.Session) -> None:
+    """Count transitions that spent too long waiting, against whoever submitted them.
+
+        reads: pipeline_run.created_by
+
+        THE GATE   from is not RUNNING, and the gap is >= the threshold
+
+          QUEUED  → PENDING    2h   counted   waiting for a machine
+          PENDING → RUNNING    9h   counted   waiting for a machine
+          PENDING → FAILED     7h   counted   waited, then never ran
+          RUNNING → SUCCEEDED  4h   skipped   a job that ran for four hours
+          RUNNING → FAILED     7h   skipped   ran for seven, then failed
+          QUEUED  → PENDING    4s   skipped   under the threshold
+
+        WHAT IS STORED    one series per (from, to, owner), monotonic
+
+          …slow{from="QUEUED", to="PENDING", owner="pipeline-runner@…"}  37
+          …slow{from="QUEUED", to="PENDING", owner="batch-scorer@…"}      6
+          …slow{from="QUEUED", to="PENDING", owner="none"}                3   (a)
+
+        WHAT A HUMAN ASKS    the window lives here, in the query
+
+          ranked list     sum by (owner)(increase(…[1h]))
+                            pipeline-runner@…  18
+                            batch-scorer@…      6
+
+          how many        count(count by (owner)(increase(…[1h]) > 0))
+          people            3
+
+        (a) no run owns the node, or the lookup failed — counted anyway, because
+            this is what an alert fires on and a point missing one label beats
+            no point at all
+
+    The owner read happens after the commit, not during it: ``before_commit`` runs
+    inside the flush in progress, so a SELECT on that session re-enters it.
+    """
+    stashed = session.info.pop(_SLOW_KEY, None)
+    if not stashed:
+        return
+    owners: dict[backend_types_sql.IdType, str] = {}
+    try:
+        with orm.Session(bind=session.get_bind()) as lookup_session:
+            owners = _owners_of(
+                session=lookup_session, node_ids={node_id for node_id, _, _ in stashed}
+            )
+    except Exception:
+        _logger.warning(
+            "Failed to read the owners of %d slow transitions; counting them unlabelled",
+            len(stashed),
+            exc_info=True,
+        )
+    for node_id, status_from, status_to in stashed:
+        try:
+            execution_slow_transitions.add(
+                1,
+                attributes={
+                    "execution.status.from": status_from,
+                    "execution.status.to": status_to,
+                    "owner": owners.get(node_id, _UNSET),
+                },
+            )
+        except Exception:
+            _logger.warning(
+                f"Failed to count slow transition for execution {node_id!r}",
+                exc_info=True,
+            )
+
+
 @sql_event.listens_for(orm.Session, "before_commit")
 def _handle_before_commit(session: orm.Session) -> None:
     for obj in list(session.new) + list(session.dirty):
@@ -258,6 +379,11 @@ def _handle_before_commit(session: orm.Session) -> None:
                         "k8s.namespace.name": namespace,
                     },
                 )
+                waited = (curr_time - prev_time).total_seconds()
+                if prev["status"] != _RUNNING and waited >= _SLOW_TRANSITION_SECONDS:
+                    session.info.setdefault(_SLOW_KEY, []).append(
+                        (obj.id, prev["status"], curr["status"])
+                    )
             except Exception:
                 _logger.warning(
                     f"Failed to record status transition metric for execution {obj.id!r}",

tests/test_sql_event_listeners.py

@@ -1,8 +1,10 @@
 """Tests for SQLAlchemy event listeners in orchestrator_sql and instrumentation.metrics."""
 
+import datetime
 import unittest.mock
 
 import pytest
+from sqlalchemy import event as sql_event
 from sqlalchemy import orm
 
 from cloud_pipelines_backend import backend_types_sql as bts
@@ -460,3 +462,200 @@ def test_resumed_execution_is_labelled_repeat_end_to_end(
                     attempts.append(attrs["execution.run_attempt"])
 
         assert attempts == ["first", "repeat", "repeat"]
+
+
+class TestSlowTransitionCounter:
+    """The slow-path counter, and the rule that the commit path issues no query."""
+
+    SLOW = datetime.timedelta(seconds=metrics._SLOW_TRANSITION_SECONDS + 1)
+    FAST = datetime.timedelta(seconds=metrics._SLOW_TRANSITION_SECONDS - 1)
+
+    @staticmethod
+    def _transition(
+        session: orm.Session,
+        *,
+        gap: datetime.timedelta,
+        status_from: bts.ContainerExecutionStatus = (
+            bts.ContainerExecutionStatus.QUEUED
+        ),
+        status_to: bts.ContainerExecutionStatus = (
+            bts.ContainerExecutionStatus.RUNNING
+        ),
+    ) -> tuple[bts.ExecutionNode, list]:
+        """Move a node between two statuses with a chosen gap, returning the counter calls."""
+        node = bts.ExecutionNode(task_spec={})
+        session.add(node)
+        node.container_execution_status = status_from
+        session.commit()
+
+        history = node.extra_data[metrics._HISTORY_KEY]
+        earlier = (
+            datetime.datetime.fromisoformat(history[-1]["first_observed_at"]) - gap
+        )
+        history[-1]["first_observed_at"] = earlier.isoformat()
+        orm.attributes.flag_modified(node, "extra_data")
+        session.commit()
+
+        node.container_execution_status = status_to
+        with unittest.mock.patch.object(
+            metrics.execution_slow_transitions, "add"
+        ) as mock_add:
+            session.commit()
+        return node, mock_add.call_args_list
+
+    def test_a_slow_transition_is_counted(self, session: orm.Session) -> None:
+        _, calls = self._transition(session, gap=self.SLOW)
+        assert len(calls) == 1
+        assert calls[0].kwargs["attributes"]["execution.status.from"] == "QUEUED"
+        assert calls[0].kwargs["attributes"]["execution.status.to"] == "RUNNING"
+
+    def test_a_fast_transition_is_not(self, session: orm.Session) -> None:
+        assert self._transition(session, gap=self.FAST)[1] == []
+
+    @pytest.mark.parametrize(
+        "status_to",
+        [
+            bts.ContainerExecutionStatus.SUCCEEDED,
+            bts.ContainerExecutionStatus.FAILED,
+        ],
+        ids=["succeeded", "failed"],
+    )
+    def test_a_long_run_is_not_a_long_wait(
+        self, session: orm.Session, status_to: bts.ContainerExecutionStatus
+    ) -> None:
+        """Once a task is RUNNING the clock measures work, not queueing."""
+        _, calls = self._transition(
+            session,
+            gap=self.SLOW,
+            status_from=bts.ContainerExecutionStatus.RUNNING,
+            status_to=status_to,
+        )
+        assert calls == []
+
+    def test_a_long_wait_that_never_ran_is_counted(self, session: orm.Session) -> None:
+        """Leaving PENDING for FAILED still means the submitter waited."""
+        _, calls = self._transition(
+            session,
+            gap=self.SLOW,
+            status_from=bts.ContainerExecutionStatus.PENDING,
+            status_to=bts.ContainerExecutionStatus.FAILED,
+        )
+        assert len(calls) == 1
+        assert calls[0].kwargs["attributes"]["execution.status.from"] == "PENDING"
+
+    def test_the_owner_is_the_runs_creator(self, session: orm.Session) -> None:
+        node = bts.ExecutionNode(task_spec={})
+        session.add(
+            bts.PipelineRun(root_execution=node, created_by="alice@example.com")
+        )
+        session.commit()
+
+        node.container_execution_status = bts.ContainerExecutionStatus.QUEUED
+        session.commit()
+        history = node.extra_data[metrics._HISTORY_KEY]
+        history[-1]["first_observed_at"] = (
+            datetime.datetime.fromisoformat(history[-1]["first_observed_at"])
+            - self.SLOW
+        ).isoformat()
+        orm.attributes.flag_modified(node, "extra_data")
+        session.commit()
+
+        node.container_execution_status = bts.ContainerExecutionStatus.RUNNING
+        with unittest.mock.patch.object(
+            metrics.execution_slow_transitions, "add"
+        ) as mock_add:
+            session.commit()
+        assert mock_add.call_args.kwargs["attributes"]["owner"] == "alice@example.com"
+
+    def test_a_node_under_a_run_resolves_through_the_ancestor_link(
+        self, session: orm.Session
+    ) -> None:
+        """An arbitrary node has no foreign key to its run; only the root does."""
+        root = bts.ExecutionNode(task_spec={})
+        child = bts.ExecutionNode(task_spec={})
+        session.add(bts.PipelineRun(root_execution=root, created_by="bob@example.com"))
+        session.add(
+            bts.ExecutionToAncestorExecutionLink(
+                ancestor_execution=root, execution=child
+            )
+        )
+        session.commit()
+
+        child.container_execution_status = bts.ContainerExecutionStatus.QUEUED
+        session.commit()
+        history = child.extra_data[metrics._HISTORY_KEY]
+        history[-1]["first_observed_at"] = (
+            datetime.datetime.fromisoformat(history[-1]["first_observed_at"])
+            - self.SLOW
+        ).isoformat()
+        orm.attributes.flag_modified(child, "extra_data")
+        session.commit()
+
+        child.container_execution_status = bts.ContainerExecutionStatus.RUNNING
+        with unittest.mock.patch.object(
+            metrics.execution_slow_transitions, "add"
+        ) as mock_add:
+            session.commit()
+        assert mock_add.call_args.kwargs["attributes"]["owner"] == "bob@example.com"
+
+    def test_a_node_with_no_run_is_labelled_not_dropped(
+        self, session: orm.Session
+    ) -> None:
+        _, calls = self._transition(session, gap=self.SLOW)
+        assert calls[0].kwargs["attributes"]["owner"] == "none"
+
+    def test_a_failed_owner_read_still_counts_the_point(
+        self, session: orm.Session
+    ) -> None:
+        """An alert fires on this counter, so a point missing a label beats no point."""
+        with unittest.mock.patch.object(
+            metrics, "_owners_of", side_effect=RuntimeError("database is gone")
+        ):
+            _, calls = self._transition(session, gap=self.SLOW)
+        assert len(calls) == 1
+        assert calls[0].kwargs["attributes"]["owner"] == "none"
+
+    def test_the_commit_path_issues_no_query(self, session: orm.Session) -> None:
+        """before_commit runs inside the flush; a SELECT there re-enters it."""
+        statements: list[str] = []
+
+        @sql_event.listens_for(session.get_bind(), "before_cursor_execute")
+        def _record(conn, cursor, statement, *args):  # noqa: ANN001, ANN202
+            statements.append(statement)
+
+        node = bts.ExecutionNode(task_spec={})
+        session.add(node)
+        node.container_execution_status = bts.ContainerExecutionStatus.QUEUED
+        session.commit()
+        history = node.extra_data[metrics._HISTORY_KEY]
+        history[-1]["first_observed_at"] = (
+            datetime.datetime.fromisoformat(history[-1]["first_observed_at"])
+            - self.SLOW
+        ).isoformat()
+        orm.attributes.flag_modified(node, "extra_data")
+        session.commit()
+
+        # Everything the commit path ran before the owner lookup, which is the first
+        # thing that happens once the transaction has closed.
+        issued_before_the_owner_read: list[str] = []
+
+        def _snapshot(**_kwargs: object) -> dict:
+            issued_before_the_owner_read.extend(statements)
+            return {}
+
+        node.container_execution_status = bts.ContainerExecutionStatus.RUNNING
+        statements.clear()
+        with unittest.mock.patch.object(
+            metrics, "_owners_of", side_effect=_snapshot
+        ) as mock_owners:
+            session.commit()
+
+        mock_owners.assert_called_once()
+        assert (
+            issued_before_the_owner_read
+        ), "the status change itself should be a write"
+        assert not [
+            s
+            for s in issued_before_the_owner_read
+            if s.lstrip().upper().startswith("SELECT")
+        ]