feat(observability): record which GPUs each Workload holds

docs/configuration.md

@@ -129,6 +129,13 @@ The API server serves a read-only inventory of runs at
 `/api/v1/inspect/inventory` and a run's logs at
 `/api/v1/inspect/logs?run_id=<id>`. See `/docs` for their fields.
 
+`placements` lists the `Workload`s that hold GPUs now, from their status in
+Kubernetes. Each entry has the node, pod, GPU UUIDs, `placed_at` in Unix
+seconds, and the `run_ids` the `Workload` serves. When the server releases a
+run's workers, it also saves each `Workload`'s last placement to the Redis list
+`open_rl:placements:<runtime>`. The list keeps the newest 200 records for seven
+days.
+
 | Env var | Default | What it does |
 | --- | --- | --- |
 | `OPEN_RL_LOG_STORE` | unset | Where it reads logs. Unset turns log reads off. `cloud-logging` reads Cloud Logging in the project of the server's Google credentials. `cloud-logging:<resource>` reads that Cloud Logging resource instead, such as `projects/<project>/locations/global/buckets/<bucket>/views/<view>`. |

scheduler/controller/api/v1alpha1/workload_types.go

@@ -171,6 +171,20 @@ type WorkloadStatus struct {
 	// the node is genuinely unknown.
 	NodeName string `json:"nodeName,omitempty"`
 
+	// GPUUUIDs are the UUIDs of the claim's allocated devices, which GPU
+	// metrics are keyed by.
+	// +optional
+	GPUUUIDs []string `json:"gpuUUIDs,omitempty"`
+
+	// PlacedAt is when the controller observed the allocation and set NodeName.
+	// +optional
+	PlacedAt *metav1.Time `json:"placedAt,omitempty"`
+
+	// ReleasedAt is when the worker stopped holding its devices. Cleared by
+	// the next placement.
+	// +optional
+	ReleasedAt *metav1.Time `json:"releasedAt,omitempty"`
+
 	// Reason is the most recent human-readable explanation of Phase.
 	Reason string `json:"reason,omitempty"`
 

scheduler/controller/internal/controller/fleet.go

@@ -3,11 +3,14 @@ package controller
 import (
 	"context"
 	"fmt"
+	"path"
 	"sort"
+	"strings"
 
 	corev1 "k8s.io/api/core/v1"
 	resourcev1 "k8s.io/api/resource/v1"
 	"k8s.io/apimachinery/pkg/api/resource"
+	"k8s.io/apimachinery/pkg/types"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/log"
 
@@ -250,6 +253,43 @@ func latestCompletePools(ctx context.Context, slices []resourcev1.ResourceSlice,
 	return keep
 }
 
+// gpuUUIDs resolves the devices allocated to a claim to the UUIDs their
+// ResourceSlices publish, in allocation order. A device without a UUID
+// attribute is left out; an empty result is retried on a later reconcile.
+func (r *WorkloadReconciler) gpuUUIDs(ctx context.Context, claimName string) ([]string, error) {
+	var claim resourcev1.ResourceClaim
+	if err := r.Get(ctx, types.NamespacedName{Namespace: r.Namespace, Name: claimName}, &claim); err != nil {
+		return nil, client.IgnoreNotFound(err)
+	}
+	if claim.Status.Allocation == nil {
+		return nil, nil
+	}
+	var slices resourcev1.ResourceSliceList
+	if err := r.List(ctx, &slices); err != nil {
+		return nil, fmt.Errorf("list resourceslices: %w", err)
+	}
+	type deviceKey struct{ driver, pool, device string }
+	published := map[deviceKey]string{}
+	for _, i := range latestCompletePools(ctx, slices.Items, r.DeviceDriver) {
+		spec := slices.Items[i].Spec
+		for _, device := range spec.Devices {
+			for name, attr := range device.Attributes {
+				// NVIDIA's driver publishes "uuid"; match any domain and case.
+				if attr.StringValue != nil && strings.EqualFold(path.Base(string(name)), "uuid") {
+					published[deviceKey{spec.Driver, spec.Pool.Name, device.Name}] = *attr.StringValue
+				}
+			}
+		}
+	}
+	var uuids []string
+	for _, result := range claim.Status.Allocation.Devices.Results {
+		if uuid, ok := published[deviceKey{result.Driver, result.Pool, result.Device}]; ok {
+			uuids = append(uuids, uuid)
+		}
+	}
+	return uuids, nil
+}
+
 // claimFrom reads a managed claim back into the fleet: identity from
 // metadata, shape from DRA's allocation. An unallocated claim has no shape
 // yet -- the tiers were alternatives, and which one holds is DRA's answer.

scheduler/controller/internal/controller/workload_controller.go

@@ -201,6 +201,16 @@ func (r *WorkloadReconciler) repairPlacementAssignment(ctx context.Context, scop
 				return placementPhaseResult{}, err
 			}
 			scope.claimName = ""
+			// Record the loss now. A re-cut reuses the claim's name, and the
+			// later Pending or Failed patches touch neither claim nor node,
+			// so this is the only transition that releases the placement.
+			if err := r.patchStatus(ctx, worker, func(status *openrlv1alpha1.WorkloadStatus) {
+				status.Phase = openrlv1alpha1.PhasePending
+				status.ClaimName, status.AssignmentID, status.NodeName = "", "", ""
+				status.Reason = "ReplacingLostClaim: the assigned claim no longer exists"
+			}); err != nil {
+				return placementPhaseResult{}, err
+			}
 		}
 	}
 	// One seat per worker. Any other seat under this name is stale, left by
@@ -512,6 +522,14 @@ func (r *WorkloadReconciler) teardown(ctx context.Context, worker *openrlv1alpha
 			return ctrl.Result{}, err
 		}
 	}
+	if err := r.patchStatus(ctx, worker, func(status *openrlv1alpha1.WorkloadStatus) {
+		if status.PlacedAt != nil && status.ReleasedAt == nil {
+			now := metav1.Now()
+			status.ReleasedAt = &now
+		}
+	}); err != nil {
+		return ctrl.Result{}, err
+	}
 	controllerutil.RemoveFinalizer(worker, workerFinalizer)
 	return ctrl.Result{}, r.Update(ctx, worker)
 }
@@ -694,9 +712,19 @@ func (r *WorkloadReconciler) reportPod(ctx context.Context, worker *openrlv1alph
 			node = claim.Node
 		}
 	}
+	var uuids []string
+	if node != "" && len(worker.Status.GPUUUIDs) == 0 {
+		var err error
+		if uuids, err = r.gpuUUIDs(ctx, claimName); err != nil {
+			return err
+		}
+	}
 
 	return r.patchStatus(ctx, worker, func(status *openrlv1alpha1.WorkloadStatus) {
 		status.Phase, status.ClaimName, status.PodName, status.NodeName, status.Reason = phase, claimName, podName, node, reason
+		if len(uuids) > 0 {
+			status.GPUUUIDs = uuids
+		}
 		// The footprint is DRA's answer, recordable only once the
 		// allocation is observed: which tier held, and what each device
 		// carries under it.
@@ -715,6 +743,7 @@ func (r *WorkloadReconciler) reportPod(ctx context.Context, worker *openrlv1alph
 func (r *WorkloadReconciler) patchStatus(ctx context.Context, worker *openrlv1alpha1.Workload, mutate func(*openrlv1alpha1.WorkloadStatus)) error {
 	before := worker.Status.DeepCopy()
 	mutate(&worker.Status)
+	trackPlacement(before, &worker.Status, metav1.Now())
 	worker.Status.ObservedGeneration = worker.Generation
 	// Skip no-op writes: they would re-trigger the watch forever.
 	if apiequality.Semantic.DeepEqual(before, &worker.Status) {
@@ -730,6 +759,20 @@ func (r *WorkloadReconciler) patchStatus(ctx context.Context, worker *openrlv1al
 	return nil
 }
 
+// trackPlacement stamps the placement interval on every status write. A worker
+// is placed when NodeName is set and released when NodeName clears or its claim
+// changes. Teardown releases it explicitly, since nothing else changes there.
+func trackPlacement(before, after *openrlv1alpha1.WorkloadStatus, now metav1.Time) {
+	held := before.NodeName != "" && before.PlacedAt != nil && before.ReleasedAt == nil
+	if held && (after.NodeName == "" || after.ClaimName != before.ClaimName) {
+		after.NodeName, after.GPUUUIDs, after.ReleasedAt = "", nil, &now
+		return
+	}
+	if after.NodeName != "" && (after.NodeName != before.NodeName || after.PlacedAt == nil) {
+		after.PlacedAt, after.ReleasedAt = &now, nil
+	}
+}
+
 func (r *WorkloadReconciler) markPending(ctx context.Context, worker *openrlv1alpha1.Workload, reason string) error {
 	return r.patchStatus(ctx, worker, func(status *openrlv1alpha1.WorkloadStatus) {
 		status.Phase, status.Reason = openrlv1alpha1.PhasePending, reason

scheduler/controller/internal/controller/workload_controller_test.go

@@ -79,10 +79,12 @@ func enabledNode() []client.Object {
 	nodeName := testNode
 	product := "NVIDIA RTX PRO 6000 Blackwell"
 	device := func(name string) resourcev1.Device {
+		uuid := "GPU-" + name
 		return resourcev1.Device{
 			Name: name,
 			Attributes: map[resourcev1.QualifiedName]resourcev1.DeviceAttribute{
 				"productName": {StringValue: &product},
+				"uuid":        {StringValue: &uuid},
 			},
 			Capacity: map[resourcev1.QualifiedName]resourcev1.DeviceCapacity{
 				"memory": {Value: resource.MustParse("96Gi")},
@@ -1248,3 +1250,35 @@ func TestMultiGPUWorkloadCutsAnExactCountExclusiveClaim(t *testing.T) {
 		t.Fatalf("status = %+v, want no claim and a reason naming four devices", waiting.Status)
 	}
 }
+
+func TestWorkerStatusRecordsItsPlacementInterval(t *testing.T) {
+	r := newReconciler(t, append(enabledNode(), trainerWorker("w-a", "model-a"))...)
+	settle(t, r, "w-a")
+	allocateClaim(t, r, claimOf(t, r, "w-a"))
+	runReconcile(t, r, "w-a")
+
+	placed := getWorker(t, r, "w-a").Status
+	if placed.NodeName != testNode || placed.PlacedAt == nil || placed.ReleasedAt != nil {
+		t.Fatalf("status = %+v, want placed on %s and not released", placed, testNode)
+	}
+	if got := placed.GPUUUIDs; len(got) != 1 || got[0] != "GPU-gpu-0" {
+		t.Fatalf("gpuUUIDs = %v, want the allocated device's UUID", got)
+	}
+
+	// A second finalizer keeps the worker readable after teardown lets go.
+	held := getWorker(t, r, "w-a")
+	held.Finalizers = append(held.Finalizers, "test.openrl.io/hold")
+	if err := r.Update(context.Background(), held); err != nil {
+		t.Fatal(err)
+	}
+	if err := r.Delete(context.Background(), held); err != nil {
+		t.Fatal(err)
+	}
+	runReconcile(t, r, "w-a")
+	runReconcile(t, r, "w-a")
+
+	released := getWorker(t, r, "w-a").Status
+	if released.ReleasedAt == nil || !released.PlacedAt.Equal(placed.PlacedAt) {
+		t.Fatalf("status = %+v, want releasedAt stamped and placedAt kept", released)
+	}
+}

scheduler/deploy/base/00-workload-crd.yaml

@@ -8773,6 +8773,13 @@ spec:
                   the pools registered.
                 format: int32
                 type: integer
+              gpuUUIDs:
+                description: |-
+                  GPUUUIDs are the UUIDs of the claim's allocated devices, which GPU
+                  metrics are keyed by.
+                items:
+                  type: string
+                type: array
               memoryPerDevice:
                 description: |-
                   MemoryPerDevice is what each of those devices must provide, i.e. Memory
@@ -8797,13 +8804,24 @@ spec:
                 - Succeeded
                 - Failed
                 type: string
+              placedAt:
+                description: PlacedAt is when the controller observed the allocation
+                  and set NodeName.
+                format: date-time
+                type: string
               podName:
                 description: PodName is the worker pod, once created.
                 type: string
               reason:
                 description: Reason is the most recent human-readable explanation
                   of Phase.
                 type: string
+              releasedAt:
+                description: |-
+                  ReleasedAt is when the worker stopped holding its devices. Cleared by
+                  the next placement.
+                format: date-time
+                type: string
             type: object
         required:
         - spec

src/server/observability/fleet.py

@@ -98,4 +98,25 @@ async def read() -> dict:
   runs.sort(key=lambda r: -r["created_at"])
   pods = [{"name": p["name"], "problem": p["problem"], "restarts": p["restarts"]} for p in cluster["pods"] if p["problem"]]
   view = {"available": cluster["available"], "nodes": cluster["nodes"], "pods": pods}
-  return {"observed_at": observed_at, "errors": errors, "runs": runs, "cluster": view}
+  return {"observed_at": observed_at, "errors": errors, "runs": runs, "cluster": view, "placements": placements(cluster["workloads"], metas)}
+
+
+def placements(workloads: list[dict] | None, metas: dict[str, str]) -> list[dict]:
+  """The Workloads holding GPUs now, each with the runs it serves: those whose runtime is its model ID and whose kind
+  matches. A shared runtime serves only its active runs."""
+  served = []
+  for name, raw in metas.items():
+    run_id = name.removeprefix("open_rl:model_meta:")
+    try:
+      meta = decode_model_metadata(raw)
+    except Exception:
+      continue  # read reports the unreadable record
+    if meta.status == "active" or not meta.shares_runtime():
+      served.append((meta.runtime(run_id), "lora" if meta.fine_tuning_type == "lora" else "fft", run_id))
+  fields = ("name", "role", "node", "pod", "gpu_uuids", "placed_at")
+  return [
+    {field: w[field] for field in fields}
+    | {"run_ids": sorted(r for runtime, kind, r in served if (runtime, kind) == (w["model_id"], w["training_kind"]))}
+    for w in workloads or []
+    if w["holds_gpus"]
+  ]

src/server/observability/kubernetes.py

@@ -59,7 +59,7 @@ def kubernetes_api() -> tuple:
 
 async def kubernetes() -> Cluster:
   try:
-    from server.scheduler_worker_manager import worker_state
+    from server.scheduler_worker_manager import holds_gpus, placement_of, worker_state
 
     core, custom = kubernetes_api()
   except Exception as exc:
@@ -97,5 +97,7 @@ async def items(name: str, call, *args) -> list[dict] | None:
     spec, status = item["spec"], item.get("status") or {}
     deleting = bool(item["metadata"].get("deletionTimestamp"))
     state, _ = worker_state(phase=status.get("phase") or "Pending", reason=status.get("reason") or "", node=status.get("nodeName"), deleting=deleting)
-    result["workloads"].append({"model_id": spec["modelID"], "training_kind": spec.get("trainingKind"), "state": state})
+    placement = placement_of(item)
+    row = {"model_id": spec["modelID"], "training_kind": spec.get("trainingKind"), "state": state, "holds_gpus": holds_gpus(placement)}
+    result["workloads"].append(row | placement)
   return result