From 777cbe11702196be180fae01c0767c5f670ae779 Mon Sep 17 00:00:00 2001 From: Yue Chao Qin Date: Tue, 27 Oct 2026 17:03:43 -0600 Subject: [PATCH] feat(observability): label a task whose pods were never created --- .../instrumentation/execution_annotations.py | 37 ++++++-- .../instrumentation/metrics.py | 65 ++++++++++++- tests/test_sql_event_listeners.py | 94 +++++++++++++++++++ 3 files changed, 186 insertions(+), 10 deletions(-) diff --git a/cloud_pipelines_backend/instrumentation/execution_annotations.py b/cloud_pipelines_backend/instrumentation/execution_annotations.py index 154a3d2..b030b4c 100644 --- a/cloud_pipelines_backend/instrumentation/execution_annotations.py +++ b/cloud_pipelines_backend/instrumentation/execution_annotations.py @@ -192,6 +192,17 @@ class PendingReason(str, enum.Enum): rather than adding a label value, so the metric cannot grow a series behind us. """ + # What: the Job exists but has produced no pod at all. + # Fault: admission refused it -- a namespace quota, an RBAC rule or a pod + # security policy. Which one is NOT knowable here: Kubernetes records + # the refusal as a FailedCreate Event, and launcher_data serialises the + # Job and its pods, never Events. Read the Events to find out why. + # K8s: an empty debug_pods. Every task starts this way for the first second + # or two, so this value only means something once the wait is long -- + # see the overwrite note on pending_reason below. + # Example: a team whose namespace quota is already exhausted. + NO_POD_CREATED = "no_pod_created" + # What: nothing has placed this pod on a machine yet. # Fault: the cluster's capacity, not the task's configuration. # K8s: a Pending pod with no containerStatuses at all -- there is nothing to @@ -296,21 +307,31 @@ class PendingReason(str, enum.Enum): def pending_reason(*, launcher_data: dict) -> PendingReason | None: """Why a task is not running yet, in our words rather than Kubernetes'. - a pod with no containerStatuses yet → "waiting_for_node" (a) + a Job with no pods at all → "no_pod_created" (a) + a pod with no containerStatuses yet → "waiting_for_node" (b) waiting.reason "ContainerCreating" → "starting_container" waiting.reason "ImagePullBackOff" → "image_unavailable" waiting.reason "CreateContainerConfigError" → "bad_config" - waiting.reason "SomeNewK8sString" → "other" (b) - a pod whose phase is not Pending → None (c) + waiting.reason "SomeNewK8sString" → "other" (c) + a pod whose phase is not Pending → None (d) + a launcher that is not Kubernetes → None - (a) scheduling has not placed it, so there is no container to report on - (b) a Kubernetes release adding a reason must not add a label value - (c) nothing is waiting, so there is nothing to attribute + (a) admission refused every pod, or the Job is one second old + (b) scheduling has not placed it, so there is no container to report on + (c) a Kubernetes release adding a reason must not add a label value + (d) nothing is waiting, so there is nothing to attribute Reads the serialised pod already persisted on ContainerExecution.launcher_data, so no Kubernetes call and no database read happen here. + + "no_pod_created" is last-stamp-wins, not an observation. A Job is launched with an + empty pod map, so every task is briefly in this state; a healthy one is overwritten + within seconds by the first poll that sees a pod. It survives to be read only when + no pod ever appears, which is the case it is for. """ - root = launcher_data.get("kubernetes_job") or launcher_data.get("kubernetes") or {} + root = launcher_data.get("kubernetes_job") or launcher_data.get("kubernetes") + if root is None: + return None pods = root.get("debug_pods") if pods is None: single = root.get("debug_pod") @@ -327,4 +348,6 @@ def pending_reason(*, launcher_data: dict) -> PendingReason | None: if reason: return _PENDING_REASONS.get(reason, PendingReason.OTHER) return PendingReason.WAITING_FOR_NODE + if not pods: + return PendingReason.NO_POD_CREATED return None diff --git a/cloud_pipelines_backend/instrumentation/metrics.py b/cloud_pipelines_backend/instrumentation/metrics.py index 5d96990..2754bf5 100644 --- a/cloud_pipelines_backend/instrumentation/metrics.py +++ b/cloud_pipelines_backend/instrumentation/metrics.py @@ -351,6 +351,28 @@ def _count_slow_transitions_by_owner(session: orm.Session) -> None: ) +# Two hooks save the pending reason, because the only moment that survives __init__ is +# too late for the metric to read, and the only moment the metric can read is too early +# to survive __init__. +# +# FIRST WRITE -- the row is created EVERY POLL AFTER -- the row is updated +# --------------------------------- ------------------------------------- +# ContainerExecution(launcher_data=..) execution.launcher_data = +# | | +# +- "set" stamps the reason +- "set" stamps the reason +# +- __init__ sets extra_data=None | object is already built, +# | the stamp is GONE x | so nothing overwrites it ok +# v v +# flush session before_commit +# +- before_insert stamps again ok +- the metric READS it ok +# v v +# the INSERT carries the reason flush +# +- before_update would stamp +# HERE, after the read too late +# +# So neither hook can replace the other, and a single flush hook cannot do both. + + @sql_event.listens_for(backend_types_sql.ContainerExecution.launcher_data, "set") def _persist_pending_reason_on_extra_data( container_execution: "backend_types_sql.ContainerExecution", @@ -360,7 +382,7 @@ def _persist_pending_reason_on_extra_data( ) -> None: """Save why a task is waiting onto its row, while it is still waiting. - poll, pod Pending, no containers → extra_data = {"last_pending_reason": "waiting_for_node"} + poll, pod Pending, no containers → extra_data = {"last_pending_reason": "waiting_for_node"} poll, pod Pending, ImagePullBackOff → extra_data = {"last_pending_reason": "image_unavailable"} poll, pod Running → left alone, the last reason stands (a) @@ -371,9 +393,46 @@ def _persist_pending_reason_on_extra_data( the only thing in the system that updates while the status does not. The serialised pod arrives as this event's own argument, so nothing is read from the database. """ - if not value: + _stamp_pending_reason(container_execution=container_execution, launcher_data=value) + + +@sql_event.listens_for(backend_types_sql.ContainerExecution, "before_insert") +def _persist_pending_reason_on_first_write( + _mapper: object, + _connection: object, + container_execution: "backend_types_sql.ContainerExecution", +) -> None: + """Save the reason that arrives with the row itself, before any poll has run. + + ContainerExecution(launcher_data=...) → extra_data written into the same INSERT + + extra_data is declared after launcher_data, so __init__ overwrites the "set" + listener's stamp with its own default of None. This runs at flush instead, where no + declaration order can reach it, and the stamp rides the INSERT at no extra cost. + """ + _stamp_pending_reason( + container_execution=container_execution, + launcher_data=container_execution.launcher_data, + ) + + +def _stamp_pending_reason( + *, + container_execution: "backend_types_sql.ContainerExecution", + launcher_data: dict | None, +) -> None: + """Write the reason onto the row, leaving it alone when there is nothing to say. + + launcher_data None, or no reason found → extra_data untouched + reason already the one stored → extra_data untouched (a) + a different reason → extra_data replaced + + (a) the dict is rewritten rather than mutated, so an unchanged value would + still mark the attribute dirty and force an UPDATE + """ + if not launcher_data: return - reason = execution_annotations.pending_reason(launcher_data=value) + reason = execution_annotations.pending_reason(launcher_data=launcher_data) if reason is None: return extra_data = container_execution.extra_data or {} diff --git a/tests/test_sql_event_listeners.py b/tests/test_sql_event_listeners.py index 18220ef..415050c 100644 --- a/tests/test_sql_event_listeners.py +++ b/tests/test_sql_event_listeners.py @@ -863,3 +863,97 @@ def test_the_slow_counter_carries_why_the_task_was_waiting( attributes = add.call_args.kwargs["attributes"] assert attributes["execution.pending_reason"] == "image_unavailable" assert attributes["execution.status.to"] == "RUNNING" + + +class TestPendingReasonWhenNoPodExists: + """A Job whose pods were refused at admission, and the first write of any task.""" + + @staticmethod + def _no_pods() -> dict: + """What the launcher serialises for a Job that has produced no pod.""" + return {"kubernetes_job": {"debug_pods": {}}} + + def test_a_job_with_no_pods_is_not_silent(self): + assert ( + execution_annotations.pending_reason(launcher_data=self._no_pods()) + == "no_pod_created" + ) + + def test_a_launcher_that_is_not_kubernetes_says_nothing(self): + # Without the explicit None check this would fall through to an empty pod map + # and label every local-process task as a refused Kubernetes pod. + for data in ({}, {"local_process": {"pid": 42}}): + assert execution_annotations.pending_reason(launcher_data=data) is None + + def test_a_pod_that_exists_but_is_not_pending_is_still_none(self): + data = { + "kubernetes_job": {"debug_pods": {"0": {"status": {"phase": "Running"}}}} + } + assert execution_annotations.pending_reason(launcher_data=data) is None + + def test_the_set_listener_alone_would_lose_the_constructor_reason(self): + # Pins why there are two listeners. ContainerExecution declares extra_data + # after launcher_data, so __init__ assigns the reason and then overwrites it + # with extra_data's own default of None. Nothing is flushed here, so only the + # attribute listener has run. Move extra_data above launcher_data and this + # fails -- which is the warning that the insert hook has become dead code. + execution = bts.ContainerExecution( + status=bts.ContainerExecutionStatus.PENDING, + launcher_data=self._no_pods(), + ) + assert execution.launcher_data is not None + assert execution.extra_data is None + + def test_the_reason_supplied_to_the_constructor_is_persisted( + self, session: orm.Session + ): + # The orchestrator passes the first launcher_data as a constructor argument, + # which is not an attribute assignment and never reaches the "set" listener. + execution = bts.ContainerExecution( + status=bts.ContainerExecutionStatus.PENDING, + launcher_data=self._no_pods(), + ) + session.add(execution) + session.commit() + execution_id = execution.id + session.expunge_all() + + reloaded = session.get(bts.ContainerExecution, execution_id) + assert reloaded is not None + assert reloaded.extra_data == { + execution_annotations.PENDING_REASON_KEY: "no_pod_created" + } + + def test_a_later_poll_overwrites_the_reason_from_the_first_write( + self, session: orm.Session + ): + # Every task is launched with an empty pod map, so a healthy one must not stay + # labelled as refused once its pod shows up. + execution = bts.ContainerExecution( + status=bts.ContainerExecutionStatus.PENDING, + launcher_data=self._no_pods(), + ) + session.add(execution) + session.commit() + assert execution.extra_data == { + execution_annotations.PENDING_REASON_KEY: "no_pod_created" + } + + execution.launcher_data = { + "kubernetes_job": { + "debug_pods": { + "0": { + "status": { + "phase": "Pending", + "containerStatuses": [ + {"state": {"waiting": {"reason": "ContainerCreating"}}} + ], + } + } + } + } + } + session.commit() + assert execution.extra_data == { + execution_annotations.PENDING_REASON_KEY: "starting_container" + }