diff --git a/cloud_pipelines_backend/instrumentation/metrics.py b/cloud_pipelines_backend/instrumentation/metrics.py index ee187b9..6f4446d 100644 --- a/cloud_pipelines_backend/instrumentation/metrics.py +++ b/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}", diff --git a/tests/test_sql_event_listeners.py b/tests/test_sql_event_listeners.py index 67c3142..28b5f36 100644 --- a/tests/test_sql_event_listeners.py +++ b/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") + ]