feat(observability): label a wait with why the task was not running

cloud_pipelines_backend/instrumentation/execution_annotations.py

@@ -6,6 +6,7 @@
 load would put a SELECT in the commit path.
 """
 
+import enum
 import json
 import re
 from typing import NamedTuple
@@ -182,3 +183,148 @@ def accelerators(*, execution: "bts.ExecutionNode") -> tuple[str, int] | None:
         raise ValueError(f"Multiple accelerator types were specified: {requested=}")
     model, quantity = next(iter(requested.items()))
     return model, int(quantity)
+
+
+class PendingReason(str, enum.Enum):
+    """Why a task is not running yet, in our words rather than Kubernetes'.
+
+    A closed set. A Kubernetes release that invents a new waiting reason widens OTHER
+    rather than adding a label value, so the metric cannot grow a series behind us.
+    """
+
+    # 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
+    #          report on until something schedules it, so this one has no entry in
+    #          the mapping below.
+    # Example: a job asking for an accelerator (e.g. a B300 GPU) while every one
+    #          of them is already taken.
+    WAITING_FOR_NODE = "waiting_for_node"
+
+    # What:    the pod has a machine and the kubelet is bringing it up.
+    # Fault:   nobody's, until it lasts.
+    # Example: a very large image, still pulling ten minutes in.
+    STARTING_CONTAINER = "starting_container"
+
+    # What:    the image cannot be fetched.
+    # Fault:   the task's -- a tag that does not exist, or credentials the registry
+    #          rejected. Kubernetes does not separate a typo from a permissions
+    #          failure here, so neither can we.
+    # Example: a task pinned to an image tag that was never pushed.
+    IMAGE_UNAVAILABLE = "image_unavailable"
+
+    # What:    the pod spec names something missing or unreadable.
+    # Fault:   the task's own configuration or permissions, not the cluster's.
+    # Example: a renamed Secret that one pipeline still references.
+    BAD_CONFIG = "bad_config"
+
+    # What:    a waiting reason we have not mapped.
+    # Fault:   unknown, which is the point -- it stays out of the label set until
+    #          someone decides which bucket it belongs in.
+    # Example: a string a future Kubernetes release invents.
+    OTHER = "other"
+
+
+# Kubernetes' open vocabulary on the left, ours on the right. What each of ours means
+# is on the enum above; these are the strings the kubelet actually reports.
+#
+# The pod lifecycle they come from, top to bottom:
+#
+#   pod created
+#     |
+#     +-- nothing schedules it ................... (no containerStatuses)
+#     v
+#   scheduled to a node
+#     |
+#     +-- image reference malformed .............. InvalidImageName
+#     v
+#   init containers running .................... PodInitializing
+#     |
+#     v
+#   main container being created ............... ContainerCreating
+#     |
+#     +-- pull failed ............................ ErrImagePull
+#     |     +-- and kept failing ................. ImagePullBackOff
+#     +-- manifest unreadable .................... ImageInspectError
+#     +-- registry silent ........................ RegistryUnavailable
+#     +-- Secret or ConfigMap missing ............ CreateContainerConfigError
+#     +-- runtime refused the spec ............... CreateContainerError
+#     |
+#     v
+#   RUNNING
+#
+# The common path, not a state machine: ContainerCreating covers the image pull too,
+# so the kubelet may report either it or ErrImagePull for the same moment.
+_PENDING_REASONS = {
+    # The kubelet is bringing the pod up, in this order:
+    #   PodInitializing ──then──► ContainerCreating
+    #   init containers           the main container
+    #
+    # Init containers are still running, so the main container has not started.
+    "PodInitializing": PendingReason.STARTING_CONTAINER,
+    # The main container is being created: pulling layers, mounting volumes.
+    "ContainerCreating": PendingReason.STARTING_CONTAINER,
+    # The image did not arrive.
+    #   ErrImagePull ──retries──► ImagePullBackOff     one incident, two names
+    #
+    # The first pull attempt failed.
+    "ErrImagePull": PendingReason.IMAGE_UNAVAILABLE,
+    # The back-off loop the above decays into after repeated failures.
+    "ImagePullBackOff": PendingReason.IMAGE_UNAVAILABLE,
+    # The image was fetched but its manifest could not be read.
+    "ImageInspectError": PendingReason.IMAGE_UNAVAILABLE,
+    # The registry did not answer at all.
+    "RegistryUnavailable": PendingReason.IMAGE_UNAVAILABLE,
+    # The spec is wrong or not permitted, caught at two different moments:
+    #   InvalidImageName                                before any pull
+    #   CreateContainerConfigError, CreateContainerError at container creation
+    #
+    # The image reference is malformed, so nothing is ever pulled.
+    "InvalidImageName": PendingReason.BAD_CONFIG,
+    # A Secret, ConfigMap or token the spec names is missing or unreadable.
+    "CreateContainerConfigError": PendingReason.BAD_CONFIG,
+    # The runtime refused to create the container from this spec.
+    "CreateContainerError": PendingReason.BAD_CONFIG,
+}
+
+
+# Written onto ContainerExecution.extra_data while a task waits, read back when it stops
+# waiting -- by then the pod is running and the reason is gone from it.
+PENDING_REASON_KEY = "last_pending_reason"
+
+
+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)
+        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)
+
+        (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
+
+    Reads the serialised pod already persisted on ContainerExecution.launcher_data, so
+    no Kubernetes call and no database read happen here.
+    """
+    root = launcher_data.get("kubernetes_job") or launcher_data.get("kubernetes") or {}
+    pods = root.get("debug_pods")
+    if pods is None:
+        single = root.get("debug_pod")
+        pods = {"0": single} if single else {}
+    for pod in pods.values():
+        status = (pod or {}).get("status") or {}
+        if status.get("phase") != "Pending":
+            continue
+        container_statuses = status.get("containerStatuses")
+        if not container_statuses:
+            return PendingReason.WAITING_FOR_NODE
+        waiting = (container_statuses[0].get("state") or {}).get("waiting") or {}
+        reason = waiting.get("reason")
+        if reason:
+            return _PENDING_REASONS.get(reason, PendingReason.OTHER)
+        return PendingReason.WAITING_FOR_NODE
+    return None

cloud_pipelines_backend/instrumentation/metrics.py

@@ -324,22 +324,24 @@ def _count_slow_transitions_by_owner(session: orm.Session) -> None:
     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}
+                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:
+    for node_id, status_from, status_to, pending_reason 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),
+                    "execution.pending_reason": pending_reason,
                 },
             )
         except Exception:
@@ -349,6 +351,63 @@ def _count_slow_transitions_by_owner(session: orm.Session) -> None:
             )
 
 
+@sql_event.listens_for(backend_types_sql.ContainerExecution.launcher_data, "set")
+def _persist_pending_reason_on_extra_data(
+    container_execution: "backend_types_sql.ContainerExecution",
+    value: dict | None,
+    _old_value: object,
+    _initiator: object,
+) -> 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, ImagePullBackOff → extra_data = {"last_pending_reason": "image_unavailable"}
+        poll, pod Running                   → left alone, the last reason stands   (a)
+
+        (a) by the time a task stops waiting its pod is running and the reason is gone,
+            so the transition can only read what an earlier poll wrote
+
+    The orchestrator assigns launcher_data on every sweep that sees a change, which is
+    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:
+        return
+    reason = execution_annotations.pending_reason(launcher_data=value)
+    if reason is None:
+        return
+    extra_data = container_execution.extra_data or {}
+    if extra_data.get(execution_annotations.PENDING_REASON_KEY) == reason.value:
+        return
+    container_execution.extra_data = {
+        **extra_data,
+        # .value, not the member: extra_data is a JSON column and an attribute is a
+        # string, so the enum stops at this boundary.
+        execution_annotations.PENDING_REASON_KEY: reason.value,
+    }
+
+
+def _pending_reason(*, execution: "backend_types_sql.ExecutionNode") -> str:
+    """Why this task was waiting, as stamped by the most recent poll.
+
+        the ContainerExecution is already in the session  → "image_unavailable"
+        it would have to be fetched                       → "none"   (a)
+        nothing ever stamped one                          → "none"
+
+        (a) this runs in before_commit, where a lazy load would put a SELECT inside the
+            flush in progress; an unlabelled point beats a re-entrant query
+
+    reads: ContainerExecution.extra_data["last_pending_reason"]
+    """
+    if "container_execution" in sql.inspect(execution).unloaded:
+        return _UNSET
+    container_execution = execution.container_execution
+    if container_execution is None:
+        return _UNSET
+    extra_data = container_execution.extra_data or {}
+    return extra_data.get(execution_annotations.PENDING_REASON_KEY) or _UNSET
+
+
 @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):
@@ -365,6 +424,7 @@ def _handle_before_commit(session: orm.Session) -> None:
             accelerator_type, accelerator_count = _accelerator(execution=obj)
             cluster = _cluster(execution=obj)
             namespace = _namespace(execution=obj)
+            pending_reason = _pending_reason(execution=obj)
             try:
                 execution_status_transition_duration.record(
                     (curr_time - prev_time).total_seconds(),
@@ -377,12 +437,13 @@ def _handle_before_commit(session: orm.Session) -> None:
                         "execution.accelerator_count": accelerator_count,
                         "execution.cluster": cluster,
                         "k8s.namespace.name": namespace,
+                        "execution.pending_reason": pending_reason,
                     },
                 )
                 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"])
+                        (obj.id, prev["status"], curr["status"], pending_reason)
                     )
             except Exception:
                 _logger.warning(