feat(observability): label a task whose pods were never created

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

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 = <new pod>
+#     |                                         |
+#     +- "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 {}

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"
+        }