feat(observability): label repeat RUNNING attempts on the status transition metric

cloud_pipelines_backend/instrumentation/metrics.py

@@ -84,6 +84,33 @@ class MetricUnit(str, enum.Enum):
 
 _HISTORY_KEY = backend_types_sql.EXECUTION_NODE_EXTRA_DATA_STATUS_HISTORY_KEY
 
+_RUNNING = backend_types_sql.ContainerExecutionStatus.RUNNING.value
+
+
+def _run_attempt(*, history: list[dict[str, str]]) -> str:
+    """Whether the execution is waiting to START or waiting to RESUME.
+
+    Both waits are recorded as PENDING->RUNNING, so without this one preempted task's
+    resumption is indistinguishable from a slow queue -- and p99 over a 5 minute window
+    is the single slowest sample in it.
+
+        QUEUED -> PENDING -> RUNNING                  "first"   never had a machine
+        ... RUNNING -> PENDING -> RUNNING             "repeat"  had one, lost it
+        QUEUED -> PENDING                             "none"    not a start at all
+
+    "first" is how long the orchestrator queued the work. "repeat" is how long a
+    tenant's cluster took to give the machine back, which is not our queue.
+
+    "none" rather than omitting the attribute, because an attribute that disappears
+    forms a second series that no filter on the first one covers.
+
+    "Attempt" counts times the execution held a machine, not times it was retried after
+    an error. Read off the history list the session already holds; no query.
+    """
+    if not history or history[-1]["status"] != _RUNNING:
+        return "none"
+    return "repeat" if any(e["status"] == _RUNNING for e in history[:-1]) else "first"
+
 
 @sql_event.listens_for(orm.Session, "before_commit")
 def _handle_before_commit(session: orm.Session) -> None:
@@ -104,6 +131,7 @@ def _handle_before_commit(session: orm.Session) -> None:
                     attributes={
                         "execution.status.from": prev["status"],
                         "execution.status.to": curr["status"],
+                        "execution.run_attempt": _run_attempt(history=history),
                     },
                 )
             except Exception:

tests/test_sql_event_listeners.py

@@ -66,4 +66,73 @@ def test_second_status_change_records_duration_metric(
         assert mock_record.call_args.kwargs["attributes"] == {
             "execution.status.from": bts.ContainerExecutionStatus.QUEUED,
             "execution.status.to": bts.ContainerExecutionStatus.RUNNING,
+            "execution.run_attempt": "first",
         }
+
+
+class TestRunAttempt:
+    """A preempted execution reaches RUNNING more than once; only the first is a start."""
+
+    @staticmethod
+    def _history(*statuses: str) -> list[dict[str, str]]:
+        return [
+            {"status": s, "first_observed_at": "2024-01-01T00:00:00+00:00"}
+            for s in statuses
+        ]
+
+    @pytest.mark.parametrize(
+        ("statuses", "expected"),
+        [
+            ((), "none"),
+            (("QUEUED",), "none"),
+            (("QUEUED", "PENDING"), "none"),
+            (("QUEUED", "PENDING", "RUNNING"), "first"),
+            (("QUEUED", "PENDING", "RUNNING", "PENDING"), "none"),
+            (("QUEUED", "PENDING", "RUNNING", "PENDING", "RUNNING"), "repeat"),
+            (
+                (
+                    "QUEUED",
+                    "PENDING",
+                    "RUNNING",
+                    "PENDING",
+                    "RUNNING",
+                    "PENDING",
+                    "RUNNING",
+                ),
+                "repeat",
+            ),
+            (("QUEUED", "PENDING", "RUNNING", "SUCCEEDED"), "none"),
+        ],
+    )
+    def test_run_attempt_for_history(
+        self, statuses: tuple[str, ...], expected: str
+    ) -> None:
+        assert metrics._run_attempt(history=self._history(*statuses)) == expected
+
+    def test_resumed_execution_is_labelled_repeat_end_to_end(
+        self, session: orm.Session
+    ) -> None:
+        """The preemption walk: three PENDING->RUNNING samples, first then repeat twice."""
+        node = bts.ExecutionNode(task_spec={})
+        session.add(node)
+        attempts = []
+        for status in [
+            bts.ContainerExecutionStatus.QUEUED,
+            bts.ContainerExecutionStatus.PENDING,
+            bts.ContainerExecutionStatus.RUNNING,
+            bts.ContainerExecutionStatus.PENDING,
+            bts.ContainerExecutionStatus.RUNNING,
+            bts.ContainerExecutionStatus.PENDING,
+            bts.ContainerExecutionStatus.RUNNING,
+        ]:
+            node.container_execution_status = status
+            with unittest.mock.patch.object(
+                metrics.execution_status_transition_duration, "record"
+            ) as mock_record:
+                session.commit()
+            for call in mock_record.call_args_list:
+                attrs = call.kwargs["attributes"]
+                if attrs["execution.status.to"] == bts.ContainerExecutionStatus.RUNNING:
+                    attempts.append(attrs["execution.run_attempt"])
+
+        assert attempts == ["first", "repeat", "repeat"]