feat(observability): label the platform, accelerator, cluster and namespace a task ran on

cloud_pipelines_backend/instrumentation/execution_annotations.py

@@ -7,9 +7,11 @@
 """
 
 import json
+import re
 from typing import NamedTuple
 
 from .. import backend_types_sql as bts
+from ..launchers import common_annotations
 from ..launchers import kubernetes_launchers
 
 # Aliased, not respelled: a launcher renaming a key cannot silently empty these readers.
@@ -20,21 +22,29 @@
 # Set at routing time rather than by a launcher, so no launcher module owns it.
 CLOUD_PROVIDER_KEY = "cloud-pipelines.net/orchestration/cloud_provider"
 
+# Read by the deployment's pod builder, which puts such a task in a namespace of its
+# own rather than the cluster's.
+GOOGLE_SERVICE_ACCOUNT_KEY = "cloud-pipelines.net/launchers/google/service_account"
+
 
 class Target(NamedTuple):
     """Where a task runs, once its annotation has been resolved.
 
-        Target("google", "lt3")                   a cluster the config names
-        Target("nebius", "h200")                  ditto
-        Target("davies", "davies",                a provider with no clusters of
+        Target("google", "us-ce1-lt3",            a cluster the config names,
+               "kueue-jobs-staging")              with the namespace it runs in
+        Target("nebius", "b300-cluster",
+               "tangle-jobs-production")
+        Target("davies", "davies", "none",        a provider with no clusters of
                cluster_from_annotation=<key>)     its own; each task names one
 
     ``cluster_from_annotation`` lets the deployment register which annotation key its
     tasks use for that, so a deployment-specific key never appears in this repository.
+    ``namespace`` is the cluster's own; a task carrying a service account overrides it.
     """
 
     provider: str
     cluster: str
+    namespace: str = "none"
     cluster_from_annotation: str | None = None
 
 
@@ -114,6 +124,39 @@ def target(*, execution: "bts.ExecutionNode") -> Target | None:
     return resolved._replace(cluster=named or "none")
 
 
+def namespace(*, execution: "bts.ExecutionNode") -> str | None:
+    """The namespace a task's own service account puts it in, or None for its cluster's.
+
+        reads: cloud-pipelines.net/launchers/google/service_account
+               + cloud-pipelines.net/pipeline_runs/created_by
+
+        service_account="sidekick-oasis-runner@s5y-sidekick-52ee
+                         .iam.gserviceaccount.com", created_by set
+                                   → "sidekick-oasis-runner-at-s5y-sidekick-52ee"
+        service_account set, created_by absent  → None   (a)
+        neither                                 → None   (a)
+
+        (a) the pod lands in the cluster's namespace instead, which this module
+            cannot see; the caller falls back to Target.namespace
+
+    Both annotations are required because the deployment's pod builder only creates a
+    per-account namespace when it has both. Deriving it here duplicates that rule
+    rather than querying Kubernetes, which no reader in this module may do.
+    """
+    annotations = _annotations(execution=execution)
+    account = annotations.get(GOOGLE_SERVICE_ACCOUNT_KEY)
+    created_by = annotations.get(
+        common_annotations.PIPELINE_RUN_CREATED_BY_ANNOTATION_KEY
+    )
+    if not account or not created_by:
+        return None
+    return re.sub(
+        r"[^-a-z0-9]",
+        "-",
+        account.lower().removesuffix(".iam.gserviceaccount.com").replace("@", "-at-"),
+    ).strip("-")
+
+
 def accelerators(*, execution: "bts.ExecutionNode") -> tuple[str, int] | None:
     """The accelerator model and count a task asked for.
 

cloud_pipelines_backend/instrumentation/metrics.py

@@ -31,6 +31,7 @@
 from sqlalchemy import orm
 
 from .. import backend_types_sql
+from . import execution_annotations
 from . import execution_tracing
 
 _logger = logging.getLogger(__name__)
@@ -86,6 +87,10 @@ class MetricUnit(str, enum.Enum):
 
 _RUNNING = backend_types_sql.ContainerExecutionStatus.RUNNING.value
 
+# Every label an execution contributes has one of these when the task said nothing, so a
+# query that filters on a label never silently drops the executions that lack it.
+_UNSET = "none"
+
 
 def _run_attempt(*, history: list[dict[str, str]]) -> str:
     """Whether the execution is waiting to START or waiting to RESUME.
@@ -112,6 +117,117 @@ def _run_attempt(*, history: list[dict[str, str]]) -> str:
     return "repeat" if any(e["status"] == _RUNNING for e in history[:-1]) else "first"
 
 
+def _cloud_provider(*, execution: backend_types_sql.ExecutionNode) -> str:
+    """Whose machines ran the work.
+
+        reads: cloud-pipelines.net/orchestration/cloud_provider
+
+        annotation          target map        label
+        "lt3"            →  google / lt3   →  "google"
+        "gke-lt3"        →  google / lt3   →  "google"   (a)
+        "h200"           →  nebius / h200  →  "nebius"
+        "davies"         →  davies / b300  →  "davies"   (b)
+        "retired-tpu"    →  not in the map →  "none"     (c)
+        no annotation    →  --             →  "none"     (c)
+
+        (a) an alias for the row above -- one series, not two
+        (b) davies declares no clusters, so the task names one in a second
+            annotation the deployment registered
+        (c) labelled rather than omitted: a vanishing attribute forms a second
+            series that no filter on the first one covers
+
+    The map decides, not the annotation. The config section is `gke` and the label
+    is `google`, because it answers whose machines, not which config stanza.
+    """
+    resolved = execution_annotations.target(execution=execution)
+    return _UNSET if resolved is None else resolved.provider
+
+
+def _namespace(*, execution: backend_types_sql.ExecutionNode) -> str:
+    """Which Kubernetes namespace the work ran in.
+
+        reads: cloud-pipelines.net/launchers/google/service_account
+               + cloud-pipelines.net/pipeline_runs/created_by
+               falling back to the namespace registered with the target
+
+        a task with its own service account → "sidekick-oasis-runner-at-s5y-…"  (a)
+        no service account, target "lt3"    → "kueue-jobs-staging"
+        no service account, target "eo9"    → "tangle-jobs-production"
+        an unrouted or missing target       → "none"
+
+        (a) ~39 such namespaces exist and the set grows with onboarding: each is a
+            function of a submitter's service account, so none is declared anywhere
+            in this repository or the deployment's config
+
+    Two origins, and only the first is visible in the annotations. A task carrying a
+    service account gets a namespace of its own; everything else lands in its
+    cluster's, which the deployment registers with the target.
+    """
+    own = execution_annotations.namespace(execution=execution)
+    if own is not None:
+        return own
+    resolved = execution_annotations.target(execution=execution)
+    return _UNSET if resolved is None else resolved.namespace
+
+
+def _cluster(*, execution: backend_types_sql.ExecutionNode) -> str:
+    """Which cluster inside that platform.
+
+        reads: cloud-pipelines.net/orchestration/cloud_provider
+
+        annotation       target map             label
+        "lt3"         →  google / us-ce1-lt3  → "us-ce1-lt3"
+        "eo9"         →  google / us-ce1-eo9  → "us-ce1-eo9"
+        "h200"        →  nebius / b300-cluster→ "b300-cluster"
+        "davies"      →  davies / named by    → whatever the task named   (a)
+                                   the task
+        "retired-tpu" →  not in the map       → "none"
+        no annotation →  --                   → "none"
+
+        (a) davies declares no clusters of its own, so the value is whatever the
+            submitter typed, and "none" when the task named nothing
+
+    The deployment registers these already slugified: config labels are display
+    strings like "B300 cluster", which is legal in a label value and unpleasant in
+    a query. Resolved through the map, never read off the annotation -- the
+    annotation names a target, which is not the same thing as a cluster.
+    """
+    resolved = execution_annotations.target(execution=execution)
+    return _UNSET if resolved is None else resolved.cluster
+
+
+def _accelerator(*, execution: backend_types_sql.ExecutionNode) -> tuple[str, str]:
+    """Which accelerator the task asked for, and how many.
+
+        reads: cloud-pipelines.net/launchers/generic/resources.accelerators
+
+        annotation                          type                   count
+        '{"NVIDIA-B300-SXM6-PC":"8"}'    →  "NVIDIA-B300-SXM6-PC"  "8"
+        '{"NVIDIA-B300-SXM6-PC": "8"}'   →  "NVIDIA-B300-SXM6-PC"  "8"    (a)
+        '{"B300-SXM6-PC":"1"}'           →  "B300-SXM6-PC"         "1"    (b)
+        '{"NVIDIA-B300-SXM6-PC":"0"}'    →  "NVIDIA-B300-SXM6-PC"  "0"
+        absent  or  '{}'                 →  "none"                 "0"
+        '{"A":"1","B":"1"}'              →  "unparseable"          "0"    (c)
+        'not json'                       →  "unparseable"          "0"    (c)
+
+        (a) one space from the row above -- one series, not two
+        (b) the capacity API's spelling of the same GPU; both kept as written
+        (c) labelled, not dropped, so a bad annotation shows as a spike rather
+            than as missing traffic
+
+    Parsing is what collapses the spacings: labelling the raw string would split one
+    model across several series.
+    """
+    try:
+        requested = execution_annotations.accelerators(execution=execution)
+    except ValueError:
+        return "unparseable", "0"
+    if requested is None:
+        return _UNSET, "0"
+    model, count = requested
+    return model, str(count)
+
+
 @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):
@@ -125,13 +241,21 @@ def _handle_before_commit(session: orm.Session) -> None:
             curr = history[-1]
             prev_time = datetime.datetime.fromisoformat(prev["first_observed_at"])
             curr_time = datetime.datetime.fromisoformat(curr["first_observed_at"])
+            accelerator_type, accelerator_count = _accelerator(execution=obj)
+            cluster = _cluster(execution=obj)
+            namespace = _namespace(execution=obj)
             try:
                 execution_status_transition_duration.record(
                     (curr_time - prev_time).total_seconds(),
                     attributes={
                         "execution.status.from": prev["status"],
                         "execution.status.to": curr["status"],
                         "execution.run_attempt": _run_attempt(history=history),
+                        "execution.cloud_provider": _cloud_provider(execution=obj),
+                        "execution.accelerator_type": accelerator_type,
+                        "execution.accelerator_count": accelerator_count,
+                        "execution.cluster": cluster,
+                        "k8s.namespace.name": namespace,
                     },
                 )
             except Exception: