diff --git a/cloud_pipelines_backend/instrumentation/metrics.py b/cloud_pipelines_backend/instrumentation/metrics.py index 6f0ba87..34de2ab 100644 --- a/cloud_pipelines_backend/instrumentation/metrics.py +++ b/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: diff --git a/tests/test_sql_event_listeners.py b/tests/test_sql_event_listeners.py index 832f1eb..b984b63 100644 --- a/tests/test_sql_event_listeners.py +++ b/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"]