From fd61f11bca69dbb06f2ae812be69c16e93075693 Mon Sep 17 00:00:00 2001 From: amshithnair Date: Tue, 28 Jul 2026 03:00:51 +0530 Subject: [PATCH 1/7] feat(telemetry): add NodeTelemetry and GPUTelemetry data models - Introduce GPUTelemetry struct with utilization, memory, ECC fields - Introduce NodeTelemetry struct with GPU slice, topology, staleness - Add Validate(), DeepCopy(), IsStale(), IsStaleAt() methods - Define sentinel constants (UnknownUtilization, UnknownMemory) - SchemaVersion field for future-proof struct evolution --- pkg/telemetry/model.go | 242 +++++++++++++++++++++++++++++++++++++++++ 1 file changed, 242 insertions(+) create mode 100644 pkg/telemetry/model.go diff --git a/pkg/telemetry/model.go b/pkg/telemetry/model.go new file mode 100644 index 0000000..d91af10 --- /dev/null +++ b/pkg/telemetry/model.go @@ -0,0 +1,242 @@ +// Package telemetry defines the core data types shared across Phase 3 components: +// the telemetry agent, the in-memory store, and (in Phase 4) the CRD controller. +// +// This file adds the structured snapshot model on top of the GPUStatsProvider +// interface defined in provider.go. It deliberately imports nothing outside the +// standard library so the model is usable on every platform and in every test. +package telemetry + +import ( + "fmt" + "time" +) + +// SchemaVersion is incremented whenever the NodeTelemetry or GPUTelemetry +// structures gain or remove fields in a backward-incompatible way. Consumers +// that cache records should discard any record whose SchemaVersion differs from +// the current value. +const SchemaVersion = 1 + +// UnknownUtilization is the sentinel value written to GPUTelemetry.UtilizationPct +// when the provider returned an error for that GPU and no real value is available. +// It is strictly outside the valid [0.0, 1.0] range so consumers can distinguish +// "unknown" from "idle". +const UnknownUtilization float64 = -1.0 + +// UnknownMemory is the sentinel value written to GPUTelemetry.FreeMemoryMB and +// GPUTelemetry.TotalMemoryMB when the provider returned an error. +const UnknownMemory int64 = -1 + +// --------------------------------------------------------------------------- +// GPUTelemetry — per-GPU metrics at a single point in time +// --------------------------------------------------------------------------- + +// GPUTelemetry holds all telemetry values for a single GPU at the moment the +// enclosing NodeTelemetry was collected. +// +// Fields whose corresponding provider call failed are populated with the +// sentinel constants (UnknownUtilization, UnknownMemory) and their "Known" +// companion field is set to false. Consumers must check the Known fields +// before using the numeric values in safety-critical paths. +type GPUTelemetry struct { + // GPUID is the identifier used to query the GPUStatsProvider (e.g. "gpu-0"). + GPUID string + + // UtilizationPct is the GPU compute utilisation in [0.0, 1.0]. + // Set to UnknownUtilization (-1) when UtilizationKnown is false. + UtilizationPct float64 + + // UtilizationKnown is false when the provider returned an error for this GPU's + // utilisation query. Consumers must treat UnknownUtilization as worst-case. + UtilizationKnown bool + + // FreeMemoryMB is the number of megabytes currently free on this GPU. + // Set to UnknownMemory (-1) when MemoryKnown is false. + FreeMemoryMB int64 + + // TotalMemoryMB is the total installed memory in megabytes. + // Set to UnknownMemory (-1) when MemoryKnown is false. + TotalMemoryMB int64 + + // MemoryKnown is false when the provider returned an error for this GPU's + // memory query. + MemoryKnown bool + + // ECCErrors is the total number of ECC errors observed on this GPU. + // Zero is valid (no errors); the value is only meaningful when ECCKnown is true. + ECCErrors uint64 + + // ECCKnown is false when the provider returned an error for this GPU's ECC query. + ECCKnown bool +} + +// IsHealthy returns true when all three telemetry signals are known for this GPU. +// A GPU with any unknown signal should be treated conservatively by the scorer. +func (g GPUTelemetry) IsHealthy() bool { + return g.UtilizationKnown && g.MemoryKnown && g.ECCKnown +} + +// Validate returns an error if the GPUTelemetry contains logically inconsistent +// values. It does not validate sentinel values — those are valid when the +// corresponding Known flag is false. +func (g GPUTelemetry) Validate() error { + if g.GPUID == "" { + return fmt.Errorf("GPUTelemetry: GPUID must not be empty") + } + if g.UtilizationKnown { + if g.UtilizationPct < 0.0 || g.UtilizationPct > 1.0 { + return fmt.Errorf("GPUTelemetry %s: UtilizationPct %.4f is outside [0.0, 1.0]", + g.GPUID, g.UtilizationPct) + } + } + if g.MemoryKnown { + if g.FreeMemoryMB < 0 { + return fmt.Errorf("GPUTelemetry %s: FreeMemoryMB %d must not be negative", + g.GPUID, g.FreeMemoryMB) + } + if g.TotalMemoryMB <= 0 { + return fmt.Errorf("GPUTelemetry %s: TotalMemoryMB %d must be positive", + g.GPUID, g.TotalMemoryMB) + } + if g.FreeMemoryMB > g.TotalMemoryMB { + return fmt.Errorf("GPUTelemetry %s: FreeMemoryMB %d exceeds TotalMemoryMB %d", + g.GPUID, g.FreeMemoryMB, g.TotalMemoryMB) + } + } + return nil +} + +// DeepCopy returns a fully independent copy of this GPUTelemetry. +// All fields are value types so a simple struct copy is sufficient. +func (g GPUTelemetry) DeepCopy() GPUTelemetry { + // All fields are scalars — struct assignment is already a deep copy. + return g +} + +// --------------------------------------------------------------------------- +// NodeTelemetry — complete per-node snapshot +// --------------------------------------------------------------------------- + +// NodeTelemetry is the complete telemetry snapshot for a single Kubernetes node +// at a single point in time. It is the primary unit of data exchanged between +// the polling agent and the telemetry store. +// +// Immutability contract: once written to the store, a NodeTelemetry value must +// not be mutated. The store returns deep copies on every read to enforce this. +type NodeTelemetry struct { + // NodeID is the Kubernetes node name. It is the primary key in the store. + NodeID string + + // SchemaVersion records the version of this struct at collection time. + // Consumers should discard records with a version they do not recognise. + SchemaVersion int + + // GPUs is the ordered slice of per-GPU telemetry snapshots. + // The order matches the GPUIDs slice supplied to the agent configuration. + GPUs []GPUTelemetry + + // Topology is the inter-GPU interconnect topology matrix for this node, + // as returned by GPUStatsProvider.GetTopology. A zero-value TopologyMatrix + // (nil Matrix) indicates the topology was unavailable at collection time. + Topology TopologyMatrix + + // CollectedAt is the wall-clock time at which this snapshot was assembled + // by the agent. It is used to compute staleness. + CollectedAt time.Time + + // Stale is set to true by TelemetryStore.MarkStale when + // time.Since(CollectedAt) exceeds the configured staleness threshold. + // It is never set to true by the agent itself — only the store sweeper + // changes this field post-collection. + Stale bool +} + +// IsStale returns true when the age of this snapshot exceeds threshold. +// This is a pure, side-effect-free helper; it does not mutate the record. +// The store's MarkStale method calls this to decide which records to flag. +func (n NodeTelemetry) IsStale(threshold time.Duration) bool { + if n.CollectedAt.IsZero() { + // A zero timestamp is treated as infinitely stale. + return true + } + return time.Since(n.CollectedAt) > threshold +} + +// IsStaleAt returns true when the age of this snapshot relative to now exceeds +// threshold. The now parameter is injected so callers (and tests) can provide +// a deterministic clock. +func (n NodeTelemetry) IsStaleAt(now time.Time, threshold time.Duration) bool { + if n.CollectedAt.IsZero() { + return true + } + return now.Sub(n.CollectedAt) > threshold +} + +// Validate returns an error if the NodeTelemetry contains logically inconsistent +// or incomplete required fields. +func (n NodeTelemetry) Validate() error { + if n.NodeID == "" { + return fmt.Errorf("NodeTelemetry: NodeID must not be empty") + } + if n.CollectedAt.IsZero() { + return fmt.Errorf("NodeTelemetry %s: CollectedAt must not be zero", n.NodeID) + } + if n.SchemaVersion <= 0 { + return fmt.Errorf("NodeTelemetry %s: SchemaVersion must be positive, got %d", + n.NodeID, n.SchemaVersion) + } + seenIDs := make(map[string]struct{}, len(n.GPUs)) + for i, gpu := range n.GPUs { + if err := gpu.Validate(); err != nil { + return fmt.Errorf("NodeTelemetry %s GPU[%d]: %w", n.NodeID, i, err) + } + if _, dup := seenIDs[gpu.GPUID]; dup { + return fmt.Errorf("NodeTelemetry %s: duplicate GPUID %q at index %d", + n.NodeID, gpu.GPUID, i) + } + seenIDs[gpu.GPUID] = struct{}{} + } + return nil +} + +// DeepCopy returns a fully independent copy of this NodeTelemetry. +// All nested slices and maps are recursively copied so the caller cannot +// accidentally alias the store's internal data. +func (n NodeTelemetry) DeepCopy() NodeTelemetry { + out := NodeTelemetry{ + NodeID: n.NodeID, + SchemaVersion: n.SchemaVersion, + CollectedAt: n.CollectedAt, + Stale: n.Stale, + } + + // Deep-copy the GPU slice. + if n.GPUs != nil { + out.GPUs = make([]GPUTelemetry, len(n.GPUs)) + for i, g := range n.GPUs { + out.GPUs[i] = g.DeepCopy() + } + } + + // Deep-copy the topology matrix. + out.Topology = deepCopyTopologyMatrix(n.Topology) + + return out +} + +// deepCopyTopologyMatrix returns an independent copy of a TopologyMatrix. +// Both the outer and inner maps are newly allocated. +func deepCopyTopologyMatrix(src TopologyMatrix) TopologyMatrix { + if src.Matrix == nil { + return TopologyMatrix{} + } + dst := make(map[string]map[string]int, len(src.Matrix)) + for outerKey, innerMap := range src.Matrix { + dstInner := make(map[string]int, len(innerMap)) + for innerKey, val := range innerMap { + dstInner[innerKey] = val + } + dst[outerKey] = dstInner + } + return TopologyMatrix{Matrix: dst} +} From 1cd55429f0e10fc55bcbe603fbe6a85d86856737 Mon Sep 17 00:00:00 2001 From: amshithnair Date: Tue, 28 Jul 2026 03:01:12 +0530 Subject: [PATCH 2/7] test(telemetry): add comprehensive tests for data model validation and deep copy - 23 test functions covering GPUTelemetry and NodeTelemetry - Validate happy/sad paths: empty GPUID, out-of-range utilization, negative memory, free-exceeds-total, duplicate GPU IDs - DeepCopy independence: GPU slice, topology maps, nil cases - IsStale/IsStaleAt boundary and deterministic tests - TopologyMatrix deep copy edge case --- pkg/telemetry/model_test.go | 319 ++++++++++++++++++++++++++++++++++++ 1 file changed, 319 insertions(+) create mode 100644 pkg/telemetry/model_test.go diff --git a/pkg/telemetry/model_test.go b/pkg/telemetry/model_test.go new file mode 100644 index 0000000..dc933af --- /dev/null +++ b/pkg/telemetry/model_test.go @@ -0,0 +1,319 @@ +package telemetry + +import ( + "testing" + "time" +) + +// --------------------------------------------------------------------------- +// GPUTelemetry.Validate +// --------------------------------------------------------------------------- + +func TestGPUTelemetry_Validate_Happy(t *testing.T) { + g := GPUTelemetry{ + GPUID: "gpu-0", + UtilizationPct: 0.5, + UtilizationKnown: true, + FreeMemoryMB: 4096, + TotalMemoryMB: 8192, + MemoryKnown: true, + ECCErrors: 0, + ECCKnown: true, + } + if err := g.Validate(); err != nil { + t.Errorf("expected no error, got: %v", err) + } +} + +func TestGPUTelemetry_Validate_EmptyGPUID(t *testing.T) { + g := GPUTelemetry{GPUID: ""} + if err := g.Validate(); err == nil { + t.Error("expected error for empty GPUID") + } +} + +func TestGPUTelemetry_Validate_UtilizationOutOfRange(t *testing.T) { + tests := []struct { + name string + util float64 + }{ + {"negative", -0.1}, + {"above 1", 1.001}, + {"way above", 99.0}, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + g := GPUTelemetry{GPUID: "gpu-0", UtilizationPct: tc.util, UtilizationKnown: true} + if err := g.Validate(); err == nil { + t.Errorf("expected error for utilization %.3f", tc.util) + } + }) + } +} + +func TestGPUTelemetry_Validate_UnknownUtilization_NoError(t *testing.T) { + // When UtilizationKnown is false the sentinel value should not be validated. + g := GPUTelemetry{GPUID: "gpu-0", UtilizationPct: UnknownUtilization, UtilizationKnown: false} + if err := g.Validate(); err != nil { + t.Errorf("unexpected error for unknown utilization: %v", err) + } +} + +func TestGPUTelemetry_Validate_NegativeFreeMemory(t *testing.T) { + g := GPUTelemetry{GPUID: "gpu-0", FreeMemoryMB: -1, TotalMemoryMB: 8192, MemoryKnown: true} + if err := g.Validate(); err == nil { + t.Error("expected error for negative FreeMemoryMB") + } +} + +func TestGPUTelemetry_Validate_ZeroTotalMemory(t *testing.T) { + g := GPUTelemetry{GPUID: "gpu-0", FreeMemoryMB: 0, TotalMemoryMB: 0, MemoryKnown: true} + if err := g.Validate(); err == nil { + t.Error("expected error for zero TotalMemoryMB") + } +} + +func TestGPUTelemetry_Validate_FreeExceedsTotal(t *testing.T) { + g := GPUTelemetry{GPUID: "gpu-0", FreeMemoryMB: 9000, TotalMemoryMB: 8192, MemoryKnown: true} + if err := g.Validate(); err == nil { + t.Error("expected error for FreeMemoryMB > TotalMemoryMB") + } +} + +// --------------------------------------------------------------------------- +// GPUTelemetry.IsHealthy +// --------------------------------------------------------------------------- + +func TestGPUTelemetry_IsHealthy(t *testing.T) { + healthy := GPUTelemetry{ + GPUID: "gpu-0", UtilizationKnown: true, MemoryKnown: true, ECCKnown: true, + } + if !healthy.IsHealthy() { + t.Error("expected IsHealthy() = true") + } + partial := GPUTelemetry{GPUID: "gpu-0", UtilizationKnown: true, MemoryKnown: false, ECCKnown: true} + if partial.IsHealthy() { + t.Error("expected IsHealthy() = false when MemoryKnown is false") + } +} + +// --------------------------------------------------------------------------- +// GPUTelemetry.DeepCopy +// --------------------------------------------------------------------------- + +func TestGPUTelemetry_DeepCopy_Independence(t *testing.T) { + orig := GPUTelemetry{ + GPUID: "gpu-0", + UtilizationPct: 0.3, + UtilizationKnown: true, + FreeMemoryMB: 4096, + TotalMemoryMB: 8192, + MemoryKnown: true, + ECCErrors: 5, + ECCKnown: true, + } + cp := orig.DeepCopy() + cp.UtilizationPct = 0.9 + cp.ECCErrors = 999 + if orig.UtilizationPct != 0.3 { + t.Errorf("original UtilizationPct mutated: got %v", orig.UtilizationPct) + } + if orig.ECCErrors != 5 { + t.Errorf("original ECCErrors mutated: got %v", orig.ECCErrors) + } +} + +// --------------------------------------------------------------------------- +// NodeTelemetry.Validate +// --------------------------------------------------------------------------- + +func validNodeTelemetry() NodeTelemetry { + return NodeTelemetry{ + NodeID: "node-1", + SchemaVersion: SchemaVersion, + CollectedAt: time.Now(), + GPUs: []GPUTelemetry{ + {GPUID: "gpu-0", UtilizationKnown: false, MemoryKnown: false, ECCKnown: false}, + }, + } +} + +func TestNodeTelemetry_Validate_Happy(t *testing.T) { + n := validNodeTelemetry() + if err := n.Validate(); err != nil { + t.Errorf("unexpected error: %v", err) + } +} + +func TestNodeTelemetry_Validate_EmptyNodeID(t *testing.T) { + n := validNodeTelemetry() + n.NodeID = "" + if err := n.Validate(); err == nil { + t.Error("expected error for empty NodeID") + } +} + +func TestNodeTelemetry_Validate_ZeroCollectedAt(t *testing.T) { + n := validNodeTelemetry() + n.CollectedAt = time.Time{} + if err := n.Validate(); err == nil { + t.Error("expected error for zero CollectedAt") + } +} + +func TestNodeTelemetry_Validate_InvalidSchemaVersion(t *testing.T) { + n := validNodeTelemetry() + n.SchemaVersion = 0 + if err := n.Validate(); err == nil { + t.Error("expected error for SchemaVersion=0") + } + n.SchemaVersion = -1 + if err := n.Validate(); err == nil { + t.Error("expected error for negative SchemaVersion") + } +} + +func TestNodeTelemetry_Validate_DuplicateGPUID(t *testing.T) { + n := validNodeTelemetry() + n.GPUs = []GPUTelemetry{ + {GPUID: "gpu-0"}, + {GPUID: "gpu-0"}, // duplicate + } + if err := n.Validate(); err == nil { + t.Error("expected error for duplicate GPUID") + } +} + +func TestNodeTelemetry_Validate_InvalidGPU(t *testing.T) { + n := validNodeTelemetry() + n.GPUs = []GPUTelemetry{ + {GPUID: ""}, // invalid — empty GPUID + } + if err := n.Validate(); err == nil { + t.Error("expected error for GPU with empty GPUID") + } +} + +// --------------------------------------------------------------------------- +// NodeTelemetry.IsStale / IsStaleAt +// --------------------------------------------------------------------------- + +func TestNodeTelemetry_IsStale_ZeroTimestamp(t *testing.T) { + n := NodeTelemetry{CollectedAt: time.Time{}} + if !n.IsStale(30 * time.Second) { + t.Error("zero CollectedAt should always be stale") + } +} + +func TestNodeTelemetry_IsStale_Fresh(t *testing.T) { + n := NodeTelemetry{CollectedAt: time.Now()} + if n.IsStale(30 * time.Second) { + t.Error("just-created record should not be stale with 30s threshold") + } +} + +func TestNodeTelemetry_IsStaleAt_Deterministic(t *testing.T) { + base := time.Date(2024, 1, 1, 12, 0, 0, 0, time.UTC) + n := NodeTelemetry{CollectedAt: base} + threshold := 30 * time.Second + + // Exactly at threshold: not stale (strict >) + notYet := base.Add(threshold) + if n.IsStaleAt(notYet, threshold) { + t.Error("should not be stale at exactly threshold boundary") + } + + // One nanosecond over: stale + justOver := base.Add(threshold + 1) + if !n.IsStaleAt(justOver, threshold) { + t.Error("should be stale one ns over threshold") + } +} + +// --------------------------------------------------------------------------- +// NodeTelemetry.DeepCopy — independence tests +// --------------------------------------------------------------------------- + +func TestNodeTelemetry_DeepCopy_GPUSliceIndependence(t *testing.T) { + orig := NodeTelemetry{ + NodeID: "node-1", + SchemaVersion: 1, + CollectedAt: time.Now(), + GPUs: []GPUTelemetry{ + {GPUID: "gpu-0", UtilizationPct: 0.2, UtilizationKnown: true}, + {GPUID: "gpu-1", UtilizationPct: 0.8, UtilizationKnown: true}, + }, + } + cp := orig.DeepCopy() + cp.GPUs[0].UtilizationPct = 0.99 + cp.GPUs[1].GPUID = "mutated" + + if orig.GPUs[0].UtilizationPct != 0.2 { + t.Errorf("original GPU[0] utilisation mutated: got %.2f", orig.GPUs[0].UtilizationPct) + } + if orig.GPUs[1].GPUID != "gpu-1" { + t.Errorf("original GPU[1] GPUID mutated: got %q", orig.GPUs[1].GPUID) + } +} + +func TestNodeTelemetry_DeepCopy_TopologyIndependence(t *testing.T) { + orig := NodeTelemetry{ + NodeID: "node-1", + SchemaVersion: 1, + CollectedAt: time.Now(), + Topology: TopologyMatrix{ + Matrix: map[string]map[string]int{ + "gpu-0": {"gpu-1": 1}, + }, + }, + } + cp := orig.DeepCopy() + cp.Topology.Matrix["gpu-0"]["gpu-1"] = 999 + cp.Topology.Matrix["gpu-2"] = map[string]int{"gpu-3": 5} + + if orig.Topology.Matrix["gpu-0"]["gpu-1"] != 1 { + t.Errorf("original topology mutated: gpu-0→gpu-1 = %d", orig.Topology.Matrix["gpu-0"]["gpu-1"]) + } + if _, exists := orig.Topology.Matrix["gpu-2"]; exists { + t.Error("original topology should not have gpu-2 key") + } +} + +func TestNodeTelemetry_DeepCopy_NilTopology(t *testing.T) { + orig := NodeTelemetry{ + NodeID: "node-1", + SchemaVersion: 1, + CollectedAt: time.Now(), + Topology: TopologyMatrix{Matrix: nil}, + } + cp := orig.DeepCopy() + if cp.Topology.Matrix != nil { + t.Error("deep copy of nil topology should remain nil") + } +} + +func TestNodeTelemetry_DeepCopy_NilGPUs(t *testing.T) { + orig := NodeTelemetry{ + NodeID: "node-1", + SchemaVersion: 1, + CollectedAt: time.Now(), + GPUs: nil, + } + cp := orig.DeepCopy() + if cp.GPUs != nil { + t.Error("deep copy of nil GPUs slice should remain nil") + } +} + +// --------------------------------------------------------------------------- +// deepCopyTopologyMatrix (internal, tested via DeepCopy) +// --------------------------------------------------------------------------- + +func TestDeepCopyTopologyMatrix_Empty(t *testing.T) { + src := TopologyMatrix{Matrix: map[string]map[string]int{}} + dst := deepCopyTopologyMatrix(src) + dst.Matrix["new-key"] = map[string]int{"x": 1} + if _, exists := src.Matrix["new-key"]; exists { + t.Error("source matrix was mutated via deep copy result") + } +} From 88b5feb3b2f3606bced0a9beb56e9c88425687d0 Mon Sep 17 00:00:00 2001 From: amshithnair Date: Tue, 28 Jul 2026 03:01:24 +0530 Subject: [PATCH 3/7] feat(api): add GPUNodeStatus CRD Go types with hand-written DeepCopy - Define GPUNodeStatus, GPUNodeStatusSpec, GPUNodeStatusStatus, GPUStatusEntry, and GPUNodeStatusList structs - JSON tags match CRD YAML field names for serialization - Hand-written DeepCopyInto/DeepCopyObject methods (no code-gen) - GPU slice deep copy creates independent backing arrays - Implements runtime.Object interface for client-go compatibility --- pkg/api/v1alpha1/deepcopy.go | 176 +++++++++++++++++++++++++++++++++++ pkg/api/v1alpha1/types.go | 150 +++++++++++++++++++++++++++++ 2 files changed, 326 insertions(+) create mode 100644 pkg/api/v1alpha1/deepcopy.go create mode 100644 pkg/api/v1alpha1/types.go diff --git a/pkg/api/v1alpha1/deepcopy.go b/pkg/api/v1alpha1/deepcopy.go new file mode 100644 index 0000000..84efd45 --- /dev/null +++ b/pkg/api/v1alpha1/deepcopy.go @@ -0,0 +1,176 @@ +// Code in this file provides DeepCopy methods for all API types in v1alpha1. +// +// These are hand-written rather than code-generated (no controller-gen or +// kubebuilder toolchain is required for Phase 3). Each method produces a +// fully independent copy — no map or slice aliases the original. +// +// When Phase 4 migrates to real k8s.io/apimachinery types, controller-gen +// will regenerate these automatically and this file can be deleted. +package v1alpha1 + +// --------------------------------------------------------------------------- +// TypeMeta +// --------------------------------------------------------------------------- + +// DeepCopyInto copies all fields of src into out. +func (in *TypeMeta) DeepCopyInto(out *TypeMeta) { + *out = *in +} + +// DeepCopy returns a deep copy of this TypeMeta. +func (in *TypeMeta) DeepCopy() *TypeMeta { + if in == nil { + return nil + } + out := new(TypeMeta) + in.DeepCopyInto(out) + return out +} + +// --------------------------------------------------------------------------- +// ObjectMeta +// --------------------------------------------------------------------------- + +// DeepCopyInto copies all fields of src into out, including map allocations. +func (in *ObjectMeta) DeepCopyInto(out *ObjectMeta) { + *out = *in + if in.Labels != nil { + out.Labels = make(map[string]string, len(in.Labels)) + for k, v := range in.Labels { + out.Labels[k] = v + } + } + if in.Annotations != nil { + out.Annotations = make(map[string]string, len(in.Annotations)) + for k, v := range in.Annotations { + out.Annotations[k] = v + } + } +} + +// DeepCopy returns a deep copy of this ObjectMeta. +func (in *ObjectMeta) DeepCopy() *ObjectMeta { + if in == nil { + return nil + } + out := new(ObjectMeta) + in.DeepCopyInto(out) + return out +} + +// --------------------------------------------------------------------------- +// GPUStatusEntry +// --------------------------------------------------------------------------- + +// DeepCopyInto copies all fields of src into out. +// All fields are scalars so a struct assignment is sufficient. +func (in *GPUStatusEntry) DeepCopyInto(out *GPUStatusEntry) { + *out = *in +} + +// DeepCopy returns a deep copy of this GPUStatusEntry. +func (in *GPUStatusEntry) DeepCopy() *GPUStatusEntry { + if in == nil { + return nil + } + out := new(GPUStatusEntry) + in.DeepCopyInto(out) + return out +} + +// --------------------------------------------------------------------------- +// GPUNodeStatusSpec +// --------------------------------------------------------------------------- + +// DeepCopyInto copies all fields of src into out, including slice allocation. +func (in *GPUNodeStatusSpec) DeepCopyInto(out *GPUNodeStatusSpec) { + *out = *in + if in.GPUIDs != nil { + out.GPUIDs = make([]string, len(in.GPUIDs)) + copy(out.GPUIDs, in.GPUIDs) + } +} + +// DeepCopy returns a deep copy of this GPUNodeStatusSpec. +func (in *GPUNodeStatusSpec) DeepCopy() *GPUNodeStatusSpec { + if in == nil { + return nil + } + out := new(GPUNodeStatusSpec) + in.DeepCopyInto(out) + return out +} + +// --------------------------------------------------------------------------- +// GPUNodeStatusStatus +// --------------------------------------------------------------------------- + +// DeepCopyInto copies all fields of src into out, including slice allocation. +func (in *GPUNodeStatusStatus) DeepCopyInto(out *GPUNodeStatusStatus) { + *out = *in + if in.GPUs != nil { + out.GPUs = make([]GPUStatusEntry, len(in.GPUs)) + for i := range in.GPUs { + in.GPUs[i].DeepCopyInto(&out.GPUs[i]) + } + } +} + +// DeepCopy returns a deep copy of this GPUNodeStatusStatus. +func (in *GPUNodeStatusStatus) DeepCopy() *GPUNodeStatusStatus { + if in == nil { + return nil + } + out := new(GPUNodeStatusStatus) + in.DeepCopyInto(out) + return out +} + +// --------------------------------------------------------------------------- +// GPUNodeStatus +// --------------------------------------------------------------------------- + +// DeepCopyInto copies all fields of src into out. +func (in *GPUNodeStatus) DeepCopyInto(out *GPUNodeStatus) { + *out = *in + in.TypeMeta.DeepCopyInto(&out.TypeMeta) + in.ObjectMeta.DeepCopyInto(&out.ObjectMeta) + in.Spec.DeepCopyInto(&out.Spec) + in.Status.DeepCopyInto(&out.Status) +} + +// DeepCopy returns a deep copy of this GPUNodeStatus. +func (in *GPUNodeStatus) DeepCopy() *GPUNodeStatus { + if in == nil { + return nil + } + out := new(GPUNodeStatus) + in.DeepCopyInto(out) + return out +} + +// --------------------------------------------------------------------------- +// GPUNodeStatusList +// --------------------------------------------------------------------------- + +// DeepCopyInto copies all fields of src into out, including slice allocation. +func (in *GPUNodeStatusList) DeepCopyInto(out *GPUNodeStatusList) { + *out = *in + in.TypeMeta.DeepCopyInto(&out.TypeMeta) + if in.Items != nil { + out.Items = make([]GPUNodeStatus, len(in.Items)) + for i := range in.Items { + in.Items[i].DeepCopyInto(&out.Items[i]) + } + } +} + +// DeepCopy returns a deep copy of this GPUNodeStatusList. +func (in *GPUNodeStatusList) DeepCopy() *GPUNodeStatusList { + if in == nil { + return nil + } + out := new(GPUNodeStatusList) + in.DeepCopyInto(out) + return out +} diff --git a/pkg/api/v1alpha1/types.go b/pkg/api/v1alpha1/types.go new file mode 100644 index 0000000..4e86214 --- /dev/null +++ b/pkg/api/v1alpha1/types.go @@ -0,0 +1,150 @@ +// Package v1alpha1 defines the Go API types for the GPUNodeStatus custom resource. +// +// # Design note — no k8s.io imports +// +// Phase 3 intentionally avoids importing k8s.io/apimachinery so that the +// entire project continues to compile and test without a Kubernetes toolchain. +// The TypeMeta and ObjectMeta structs below are lightweight local equivalents +// that carry the same JSON/YAML tags as their upstream counterparts. +// +// Phase 4 will replace these stubs with the real k8s.io/apimachinery types +// when it adds client-go and the CRD controller. The field names and JSON tags +// are kept identical so the migration is a pure import-path substitution. +// +// CRD identity: +// +// Group: gpu.amshithnair.dev +// Version: v1alpha1 +// Resource: gpunodestatuses +// Kind: GPUNodeStatus +package v1alpha1 + +import "time" + +// --------------------------------------------------------------------------- +// Meta stubs (replaced by k8s.io/apimachinery in Phase 4) +// --------------------------------------------------------------------------- + +// TypeMeta describes the type and API version of an object. +// Mirrors k8s.io/apimachinery/pkg/apis/meta/v1.TypeMeta. +type TypeMeta struct { + // Kind is the object type (e.g. "GPUNodeStatus"). + Kind string `json:"kind,omitempty"` + // APIVersion is the group/version string (e.g. "gpu.amshithnair.dev/v1alpha1"). + APIVersion string `json:"apiVersion,omitempty"` +} + +// ObjectMeta holds standard object metadata. +// Mirrors the subset of k8s.io/apimachinery/pkg/apis/meta/v1.ObjectMeta +// needed for basic CRD operations. +type ObjectMeta struct { + // Name is the unique name of the object within its namespace. + Name string `json:"name,omitempty"` + // Namespace is the namespace of the object. GPUNodeStatus objects use the + // "gpu-scheduler" namespace by convention. + Namespace string `json:"namespace,omitempty"` + // Labels is a set of key/value pairs attached to the object. + Labels map[string]string `json:"labels,omitempty"` + // Annotations is a set of arbitrary key/value pairs. + Annotations map[string]string `json:"annotations,omitempty"` + // ResourceVersion is used for optimistic concurrency. + ResourceVersion string `json:"resourceVersion,omitempty"` + // Generation is a monotonically increasing sequence number. + Generation int64 `json:"generation,omitempty"` +} + +// --------------------------------------------------------------------------- +// GPUNodeStatus — the CRD kind +// --------------------------------------------------------------------------- + +// GPUNodeStatus is a custom resource that represents the current GPU health +// and telemetry state of a single Kubernetes node. +// +// One GPUNodeStatus object exists per GPU-enabled node. The telemetry agent +// (DaemonSet) running on that node updates the Status subresource on every +// poll cycle. The scheduler plugin reads the Status to build NodeScoreInput. +// +// CRD group: gpu.amshithnair.dev +// CRD version: v1alpha1 +// CRD resource: gpunodestatuses +type GPUNodeStatus struct { + TypeMeta `json:",inline"` + ObjectMeta `json:"metadata,omitempty"` + + // Spec describes the static configuration of the GPU node. + Spec GPUNodeStatusSpec `json:"spec"` + + // Status is updated by the telemetry agent and carries live GPU metrics. + Status GPUNodeStatusStatus `json:"status,omitempty"` +} + +// GPUNodeStatusSpec contains the static, operator-configured fields for a +// GPU node. These are set once during node registration and do not change +// between poll cycles. +type GPUNodeStatusSpec struct { + // NodeName is the Kubernetes node name this object represents. + // It must match the node's metadata.name exactly. + NodeName string `json:"nodeName"` + + // GPUIDs is the ordered list of GPU identifiers on this node. + // These IDs are used as keys when querying the GPUStatsProvider. + GPUIDs []string `json:"gpuIDs"` +} + +// GPUNodeStatusStatus is the status subresource updated by the telemetry agent. +// All fields in this struct are written by the agent and are read-only for +// other consumers. +type GPUNodeStatusStatus struct { + // GPUs is the list of per-GPU telemetry snapshots from the most recent poll. + GPUs []GPUStatusEntry `json:"gpus,omitempty"` + + // LastUpdated is the RFC3339 timestamp of the most recent successful poll. + LastUpdated time.Time `json:"lastUpdated,omitempty"` + + // Stale is true when the LastUpdated timestamp is older than the configured + // staleness threshold. The store's MarkStale sweep sets this field. + Stale bool `json:"stale"` + + // SchemaVersion records the version of the GPUStatusEntry schema in use. + // Consumers should discard records with an unrecognised version. + SchemaVersion int `json:"schemaVersion"` +} + +// GPUStatusEntry holds the telemetry snapshot for a single GPU within a +// GPUNodeStatus.Status. +type GPUStatusEntry struct { + // GPUID is the GPU identifier (e.g. "gpu-0"). + GPUID string `json:"gpuID"` + + // UtilizationPct is the GPU compute utilisation in [0.0, 1.0]. + // -1 when UtilizationKnown is false. + UtilizationPct float64 `json:"utilizationPct"` + + // UtilizationKnown is false when the provider returned an error for this GPU. + UtilizationKnown bool `json:"utilizationKnown"` + + // FreeMemoryMB is the number of megabytes currently free. + // -1 when MemoryKnown is false. + FreeMemoryMB int64 `json:"freeMemoryMB"` + + // TotalMemoryMB is the total installed memory in megabytes. + // -1 when MemoryKnown is false. + TotalMemoryMB int64 `json:"totalMemoryMB"` + + // MemoryKnown is false when the provider returned an error for memory queries. + MemoryKnown bool `json:"memoryKnown"` + + // ECCErrors is the total ECC error count. + ECCErrors uint64 `json:"eccErrors"` + + // ECCKnown is false when the provider returned an error for ECC queries. + ECCKnown bool `json:"eccKnown"` +} + +// GPUNodeStatusList is a list of GPUNodeStatus objects, used for LIST operations. +type GPUNodeStatusList struct { + TypeMeta `json:",inline"` + + // Items is the list of GPUNodeStatus objects. + Items []GPUNodeStatus `json:"items"` +} From 38fba990f0d3a7198e7f449efd8dea24b0206812 Mon Sep 17 00:00:00 2001 From: amshithnair Date: Tue, 28 Jul 2026 03:01:37 +0530 Subject: [PATCH 4/7] feat(store): add thread-safe TelemetryStore with RWMutex and tests - Implement Set/Get/Delete/List/Snapshot/Len/MarkStale/MarkStaleAt - sync.RWMutex: exclusive write lock, shared read lock - Deep-copy on all read and write boundaries (no aliasing) - 20 tests: CRUD, staleness sweep, snapshot independence, 4 concurrent stress tests (reads+writes, mark-stale, snapshot, set+delete) --- pkg/store/store.go | 148 ++++++++++++++++ pkg/store/store_test.go | 381 ++++++++++++++++++++++++++++++++++++++++ 2 files changed, 529 insertions(+) create mode 100644 pkg/store/store.go create mode 100644 pkg/store/store_test.go diff --git a/pkg/store/store.go b/pkg/store/store.go new file mode 100644 index 0000000..5aa355a --- /dev/null +++ b/pkg/store/store.go @@ -0,0 +1,148 @@ +// Package store provides a thread-safe, in-memory telemetry store keyed by +// Kubernetes node name. +// +// # Concurrency model +// +// All mutable state is guarded by a single sync.RWMutex. Read operations +// (Get, List, Snapshot, Len) acquire a shared read-lock and release it before +// returning — they never hold the lock while performing heap allocations. +// Write operations (Set, Delete, MarkStale) acquire an exclusive write-lock. +// +// Every value returned from a read operation is a deep copy of the stored +// record. This prevents callers from inadvertently mutating internal state +// and eliminates the need for callers to take their own locks. +// +// # Staleness sweeps +// +// MarkStale performs a full scan of all records under a write-lock. For the +// expected cluster size (≤ 1 000 GPU nodes) this is O(n) and completes in +// microseconds. If the cluster grows beyond that, Phase 4 can switch to a +// time-sorted priority queue and an O(log n) sweep. +package store + +import ( + "sync" + "time" + + "github.com/amshithnair/gpu-aware-scheduler/pkg/telemetry" +) + +// TelemetryStore is a thread-safe in-memory store of NodeTelemetry records, +// keyed by node name. +// +// The zero value is not usable; construct with NewTelemetryStore. +type TelemetryStore struct { + mu sync.RWMutex + records map[string]telemetry.NodeTelemetry +} + +// NewTelemetryStore allocates and returns a ready-to-use TelemetryStore. +func NewTelemetryStore() *TelemetryStore { + return &TelemetryStore{ + records: make(map[string]telemetry.NodeTelemetry), + } +} + +// Set stores a deep copy of t keyed by t.NodeID, replacing any previous record +// for that node. Set is safe to call concurrently with any other method. +// +// The deep copy ensures the store owns its data and the caller retains full +// ownership of the original NodeTelemetry without risk of aliasing. +func (s *TelemetryStore) Set(t telemetry.NodeTelemetry) { + cp := t.DeepCopy() + s.mu.Lock() + s.records[cp.NodeID] = cp + s.mu.Unlock() +} + +// Get returns a deep copy of the NodeTelemetry for nodeID and true, or the +// zero NodeTelemetry and false if no record exists for nodeID. +// +// The returned copy is fully independent of the store's internal state; +// callers may freely mutate it without affecting other goroutines. +func (s *TelemetryStore) Get(nodeID string) (telemetry.NodeTelemetry, bool) { + s.mu.RLock() + rec, ok := s.records[nodeID] + s.mu.RUnlock() + if !ok { + return telemetry.NodeTelemetry{}, false + } + return rec.DeepCopy(), true +} + +// Delete removes the record for nodeID from the store. +// It is a no-op if no record exists for nodeID. +func (s *TelemetryStore) Delete(nodeID string) { + s.mu.Lock() + delete(s.records, nodeID) + s.mu.Unlock() +} + +// List returns a slice of deep copies of all records currently in the store. +// The ordering of entries in the returned slice is non-deterministic (map +// iteration order). The returned slice and all its elements are independent +// of the store's internal state. +func (s *TelemetryStore) List() []telemetry.NodeTelemetry { + s.mu.RLock() + out := make([]telemetry.NodeTelemetry, 0, len(s.records)) + for _, rec := range s.records { + out = append(out, rec.DeepCopy()) + } + s.mu.RUnlock() + return out +} + +// Snapshot returns a deep copy of the entire store as a map[nodeID]NodeTelemetry. +// The returned map is fully independent of the store's internal state. +// An empty store returns a non-nil, empty map. +func (s *TelemetryStore) Snapshot() map[string]telemetry.NodeTelemetry { + s.mu.RLock() + out := make(map[string]telemetry.NodeTelemetry, len(s.records)) + for k, rec := range s.records { + out[k] = rec.DeepCopy() + } + s.mu.RUnlock() + return out +} + +// Len returns the number of records currently in the store. +func (s *TelemetryStore) Len() int { + s.mu.RLock() + n := len(s.records) + s.mu.RUnlock() + return n +} + +// MarkStale sweeps all records and sets Stale=true on any record whose +// CollectedAt timestamp is older than threshold. Records that are already +// marked stale are left unchanged (no unnecessary write). +// +// MarkStale acquires a write-lock for the duration of the sweep. For typical +// cluster sizes (≤ 1 000 nodes) this completes in microseconds. +// +// The agent calls MarkStale after every successful poll cycle so that nodes +// that have stopped reporting are flagged promptly. +func (s *TelemetryStore) MarkStale(threshold time.Duration) { + now := time.Now() + s.mu.Lock() + for id, rec := range s.records { + if !rec.Stale && rec.IsStaleAt(now, threshold) { + rec.Stale = true + s.records[id] = rec + } + } + s.mu.Unlock() +} + +// MarkStaleAt is identical to MarkStale but accepts an explicit now timestamp. +// It exists so tests can inject a deterministic clock without real time passing. +func (s *TelemetryStore) MarkStaleAt(now time.Time, threshold time.Duration) { + s.mu.Lock() + for id, rec := range s.records { + if !rec.Stale && rec.IsStaleAt(now, threshold) { + rec.Stale = true + s.records[id] = rec + } + } + s.mu.Unlock() +} diff --git a/pkg/store/store_test.go b/pkg/store/store_test.go new file mode 100644 index 0000000..0f52780 --- /dev/null +++ b/pkg/store/store_test.go @@ -0,0 +1,381 @@ +package store_test + +import ( + "fmt" + "sync" + "testing" + "time" + + "github.com/amshithnair/gpu-aware-scheduler/pkg/store" + "github.com/amshithnair/gpu-aware-scheduler/pkg/telemetry" +) + +// --------------------------------------------------------------------------- +// Helpers +// --------------------------------------------------------------------------- + +func makeRecord(nodeID string, age time.Duration) telemetry.NodeTelemetry { + return telemetry.NodeTelemetry{ + NodeID: nodeID, + SchemaVersion: telemetry.SchemaVersion, + CollectedAt: time.Now().Add(-age), + GPUs: []telemetry.GPUTelemetry{ + {GPUID: "gpu-0", UtilizationPct: 0.5, UtilizationKnown: true}, + }, + } +} + +func freshRecord(nodeID string) telemetry.NodeTelemetry { + return makeRecord(nodeID, 0) +} + +// --------------------------------------------------------------------------- +// Basic CRUD +// --------------------------------------------------------------------------- + +func TestStore_SetAndGet(t *testing.T) { + s := store.NewTelemetryStore() + rec := freshRecord("node-1") + s.Set(rec) + + got, ok := s.Get("node-1") + if !ok { + t.Fatal("expected record to be present after Set") + } + if got.NodeID != "node-1" { + t.Errorf("want NodeID=node-1, got %q", got.NodeID) + } +} + +func TestStore_Get_UnknownNode(t *testing.T) { + s := store.NewTelemetryStore() + rec, ok := s.Get("does-not-exist") + if ok { + t.Error("expected ok=false for unknown node") + } + if rec.NodeID != "" { + t.Errorf("expected zero-value NodeTelemetry, got NodeID=%q", rec.NodeID) + } +} + +func TestStore_Set_Replaces(t *testing.T) { + s := store.NewTelemetryStore() + r1 := freshRecord("node-1") + r1.SchemaVersion = 1 + s.Set(r1) + + r2 := freshRecord("node-1") + r2.SchemaVersion = 2 + s.Set(r2) + + got, ok := s.Get("node-1") + if !ok { + t.Fatal("expected record after second Set") + } + if got.SchemaVersion != 2 { + t.Errorf("want SchemaVersion=2, got %d", got.SchemaVersion) + } +} + +func TestStore_Delete(t *testing.T) { + s := store.NewTelemetryStore() + s.Set(freshRecord("node-1")) + s.Delete("node-1") + + _, ok := s.Get("node-1") + if ok { + t.Error("expected record to be absent after Delete") + } +} + +func TestStore_Delete_NoOp_UnknownNode(t *testing.T) { + s := store.NewTelemetryStore() + // Must not panic. + s.Delete("does-not-exist") +} + +// --------------------------------------------------------------------------- +// Len +// --------------------------------------------------------------------------- + +func TestStore_Len(t *testing.T) { + s := store.NewTelemetryStore() + if s.Len() != 0 { + t.Errorf("expected 0, got %d", s.Len()) + } + s.Set(freshRecord("node-1")) + s.Set(freshRecord("node-2")) + if s.Len() != 2 { + t.Errorf("expected 2, got %d", s.Len()) + } + s.Delete("node-1") + if s.Len() != 1 { + t.Errorf("expected 1, got %d", s.Len()) + } +} + +// --------------------------------------------------------------------------- +// List +// --------------------------------------------------------------------------- + +func TestStore_List_Empty(t *testing.T) { + s := store.NewTelemetryStore() + list := s.List() + if list == nil { + t.Error("List() must never return nil") + } + if len(list) != 0 { + t.Errorf("expected 0 items, got %d", len(list)) + } +} + +func TestStore_List_AllNodes(t *testing.T) { + s := store.NewTelemetryStore() + for i := 0; i < 5; i++ { + s.Set(freshRecord(fmt.Sprintf("node-%d", i))) + } + list := s.List() + if len(list) != 5 { + t.Errorf("expected 5 items, got %d", len(list)) + } +} + +func TestStore_List_DeepCopy(t *testing.T) { + s := store.NewTelemetryStore() + rec := freshRecord("node-1") + s.Set(rec) + + list := s.List() + list[0].NodeID = "mutated" + + got, _ := s.Get("node-1") + if got.NodeID != "node-1" { + t.Errorf("store was mutated via List() result: got %q", got.NodeID) + } +} + +// --------------------------------------------------------------------------- +// Snapshot +// --------------------------------------------------------------------------- + +func TestStore_Snapshot_Empty(t *testing.T) { + s := store.NewTelemetryStore() + snap := s.Snapshot() + if snap == nil { + t.Error("Snapshot() must return a non-nil map") + } +} + +func TestStore_Snapshot_Independence(t *testing.T) { + s := store.NewTelemetryStore() + s.Set(freshRecord("node-1")) + s.Set(freshRecord("node-2")) + + snap := s.Snapshot() + snap["node-1"] = telemetry.NodeTelemetry{NodeID: "mutated"} + snap["node-3"] = freshRecord("node-3") + + if s.Len() != 2 { + t.Errorf("store size changed after mutating Snapshot: got %d", s.Len()) + } + got, _ := s.Get("node-1") + if got.NodeID != "node-1" { + t.Errorf("store entry mutated via Snapshot: got %q", got.NodeID) + } +} + +// --------------------------------------------------------------------------- +// MarkStale / MarkStaleAt +// --------------------------------------------------------------------------- + +func TestStore_MarkStaleAt_OldRecord(t *testing.T) { + s := store.NewTelemetryStore() + old := makeRecord("node-1", 60*time.Second) // 60 seconds old + s.Set(old) + + now := time.Now() + threshold := 30 * time.Second + s.MarkStaleAt(now, threshold) + + got, _ := s.Get("node-1") + if !got.Stale { + t.Error("expected record to be marked stale") + } +} + +func TestStore_MarkStaleAt_FreshRecord(t *testing.T) { + s := store.NewTelemetryStore() + fresh := freshRecord("node-1") + s.Set(fresh) + + now := time.Now() + threshold := 30 * time.Second + s.MarkStaleAt(now, threshold) + + got, _ := s.Get("node-1") + if got.Stale { + t.Error("expected fresh record to remain non-stale") + } +} + +func TestStore_MarkStaleAt_AlreadyStale_NotModified(t *testing.T) { + s := store.NewTelemetryStore() + old := makeRecord("node-1", 120*time.Second) + old.Stale = true + s.Set(old) + + // After first sweep it should still be stale. + s.MarkStaleAt(time.Now(), 30*time.Second) + got, _ := s.Get("node-1") + if !got.Stale { + t.Error("already-stale record should remain stale") + } +} + +func TestStore_MarkStaleAt_MixedRecords(t *testing.T) { + s := store.NewTelemetryStore() + s.Set(makeRecord("node-fresh", 0)) + s.Set(makeRecord("node-old", 120*time.Second)) + + s.MarkStaleAt(time.Now(), 30*time.Second) + + fresh, _ := s.Get("node-fresh") + old, _ := s.Get("node-old") + + if fresh.Stale { + t.Error("fresh node should not be stale") + } + if !old.Stale { + t.Error("old node should be stale") + } +} + +// --------------------------------------------------------------------------- +// Deep-copy enforcement — Get returns independent copies +// --------------------------------------------------------------------------- + +func TestStore_Get_DeepCopy(t *testing.T) { + s := store.NewTelemetryStore() + rec := telemetry.NodeTelemetry{ + NodeID: "node-1", + SchemaVersion: 1, + CollectedAt: time.Now(), + Topology: telemetry.TopologyMatrix{ + Matrix: map[string]map[string]int{ + "gpu-0": {"gpu-1": 1}, + }, + }, + } + s.Set(rec) + + got, _ := s.Get("node-1") + // Mutate the copy's topology. + got.Topology.Matrix["gpu-0"]["gpu-1"] = 999 + + // Re-read from store — must be unaffected. + got2, _ := s.Get("node-1") + if got2.Topology.Matrix["gpu-0"]["gpu-1"] != 1 { + t.Errorf("store topology mutated via Get result: got %d", got2.Topology.Matrix["gpu-0"]["gpu-1"]) + } +} + +// --------------------------------------------------------------------------- +// Concurrency / race tests +// --------------------------------------------------------------------------- + +func TestStore_ConcurrentReadsAndWrites(t *testing.T) { + s := store.NewTelemetryStore() + const numNodes = 10 + const goroutines = 20 + + // Pre-populate + for i := 0; i < numNodes; i++ { + s.Set(freshRecord(fmt.Sprintf("node-%d", i))) + } + + var wg sync.WaitGroup + for g := 0; g < goroutines; g++ { + wg.Add(1) + go func(id int) { + defer wg.Done() + nodeID := fmt.Sprintf("node-%d", id%numNodes) + for i := 0; i < 100; i++ { + if i%3 == 0 { + s.Set(freshRecord(nodeID)) + } else if i%3 == 1 { + s.Get(nodeID) + } else { + s.List() + } + } + }(g) + } + wg.Wait() +} + +func TestStore_ConcurrentMarkStale(t *testing.T) { + s := store.NewTelemetryStore() + for i := 0; i < 5; i++ { + s.Set(makeRecord(fmt.Sprintf("node-%d", i), 60*time.Second)) + } + + var wg sync.WaitGroup + for i := 0; i < 10; i++ { + wg.Add(1) + go func() { + defer wg.Done() + s.MarkStale(30 * time.Second) + }() + } + wg.Wait() + + // All records should be stale after concurrent sweeps. + for _, rec := range s.List() { + if !rec.Stale { + t.Errorf("node %q should be stale after concurrent MarkStale calls", rec.NodeID) + } + } +} + +func TestStore_ConcurrentSnapshot(t *testing.T) { + s := store.NewTelemetryStore() + for i := 0; i < 5; i++ { + s.Set(freshRecord(fmt.Sprintf("node-%d", i))) + } + + var wg sync.WaitGroup + for i := 0; i < 20; i++ { + wg.Add(1) + go func(id int) { + defer wg.Done() + snap := s.Snapshot() + // Mutate the snapshot — must not affect the store. + for k := range snap { + n := snap[k] + n.NodeID = "mutated" + snap[k] = n + } + }(i) + } + wg.Wait() +} + +func TestStore_ConcurrentSetAndDelete(t *testing.T) { + s := store.NewTelemetryStore() + var wg sync.WaitGroup + + for i := 0; i < 50; i++ { + wg.Add(2) + nodeID := fmt.Sprintf("node-%d", i%5) + go func() { + defer wg.Done() + s.Set(freshRecord(nodeID)) + }() + go func() { + defer wg.Done() + s.Delete(nodeID) + }() + } + wg.Wait() + // Must not panic or race; exact final state is non-deterministic. +} From 7e310cd9efc558e76f76914cfa12e68e9fd88418 Mon Sep 17 00:00:00 2001 From: amshithnair Date: Tue, 28 Jul 2026 03:01:52 +0530 Subject: [PATCH 5/7] feat(agent): add polling telemetry agent with retry logic and tests - Agent.Run(ctx) polls GPUStatsProvider on configurable interval - Per-GPU retry with MaxRetries and RetryDelay - Sentinel values (Known=false) on exhausted retries - MetricsHook callbacks: OnPollComplete, OnStoreWrite - Context cancellation marks all store records stale (1ns threshold) - 17 tests: constructor validation, PollOnce happy/error paths, topology error non-fatal, multi-GPU independence, RunOneTick, context cancellation, shutdown staleness, MetricsHook, race test --- pkg/agent/agent.go | 406 +++++++++++++++++++++++++++++++ pkg/agent/agent_test.go | 511 ++++++++++++++++++++++++++++++++++++++++ 2 files changed, 917 insertions(+) create mode 100644 pkg/agent/agent.go create mode 100644 pkg/agent/agent_test.go diff --git a/pkg/agent/agent.go b/pkg/agent/agent.go new file mode 100644 index 0000000..3cca785 --- /dev/null +++ b/pkg/agent/agent.go @@ -0,0 +1,406 @@ +// Package agent implements the GPU telemetry polling agent. +// +// The agent runs a ticker loop, polling a GPUStatsProvider on every tick and +// writing the resulting NodeTelemetry snapshot to a TelemetryStore. It then +// calls MarkStale on the store so that nodes that have stopped reporting are +// flagged promptly. +// +// # Lifecycle +// +// 1. Construct with NewAgent — validates configuration, returns an error for +// invalid inputs. +// 2. Call Run(ctx) — blocks until ctx is cancelled, then returns ctx.Err(). +// 3. Call PollOnce(ctx) directly in tests to exercise a single poll cycle +// without starting the ticker loop. +// +// # Error handling +// +// Per-GPU errors from the provider are retried up to AgentConfig.MaxRetries +// times with AgentConfig.RetryDelay between attempts. If all retries fail, +// the GPU's telemetry fields are populated with conservative sentinel values +// (UnknownUtilization, UnknownMemory) and the Known flags are set to false. +// The snapshot is still written to the store — a partial snapshot is always +// preferable to no snapshot for staleness tracking. +// +// # Concurrency +// +// Agent.Run runs a single goroutine. All mutable state is local to the +// goroutine or owned by the injected TelemetryStore (which is itself +// concurrency-safe). The Agent struct itself has no exported mutable fields +// and does not need external synchronisation. +package agent + +import ( + "context" + "fmt" + "time" + + "github.com/amshithnair/gpu-aware-scheduler/pkg/store" + "github.com/amshithnair/gpu-aware-scheduler/pkg/telemetry" +) + +// --------------------------------------------------------------------------- +// Configuration +// --------------------------------------------------------------------------- + +// DefaultPollInterval is used when AgentConfig.PollInterval is zero. +const DefaultPollInterval = 10 * time.Second + +// DefaultStalenessMultiplier is multiplied by PollInterval to derive the +// default StalenessThreshold when none is provided. +const DefaultStalenessMultiplier = 3 + +// DefaultMaxRetries is the default number of times a per-GPU provider call +// is retried before the GPU is marked as unknown. +const DefaultMaxRetries = 3 + +// DefaultRetryDelay is the pause between consecutive retry attempts. +const DefaultRetryDelay = 500 * time.Millisecond + +// MetricsHook carries optional callback functions invoked by the agent after +// significant events. All fields are optional; nil callbacks are silently +// skipped. Implementations must be non-blocking — the agent does not start a +// goroutine for hook invocations. +type MetricsHook struct { + // OnPollComplete is called after each complete poll cycle. + // duration is the wall-clock time the poll took. + // errCount is the number of GPUs for which at least one provider call + // failed after all retries. + OnPollComplete func(duration time.Duration, errCount int) + + // OnStoreWrite is called immediately after a NodeTelemetry snapshot is + // written to the store. + OnStoreWrite func(nodeID string) +} + +// AgentConfig holds all configuration for an Agent instance. +type AgentConfig struct { + // NodeID is the Kubernetes node name this agent reports for. + // Required — must be non-empty. + NodeID string + + // GPUIDs is the ordered list of GPU identifiers to poll. + // Required — must be non-empty. + GPUIDs []string + + // PollInterval is the time between consecutive poll cycles. + // Defaults to DefaultPollInterval (10s) when zero. + PollInterval time.Duration + + // StalenessThreshold is the age beyond which a NodeTelemetry record is + // considered stale. Defaults to DefaultStalenessMultiplier × PollInterval + // when zero. + StalenessThreshold time.Duration + + // MaxRetries is the number of times a failing provider call is retried + // per GPU per poll cycle. Defaults to DefaultMaxRetries. + MaxRetries int + + // RetryDelay is the pause between retry attempts. + // Defaults to DefaultRetryDelay. + RetryDelay time.Duration + + // Metrics holds optional hook callbacks. All fields are optional. + Metrics MetricsHook + + // clock is the time source used for CollectedAt and retries. + // Defaults to time.Now. Exposed for testing via the functional option + // pattern; not part of the public API surface (lowercase). + clock func() time.Time +} + +// --------------------------------------------------------------------------- +// Agent +// --------------------------------------------------------------------------- + +// Agent polls a GPUStatsProvider on a fixed interval and writes NodeTelemetry +// snapshots to a TelemetryStore. +type Agent struct { + cfg AgentConfig + provider telemetry.GPUStatsProvider + store *store.TelemetryStore +} + +// NewAgent validates cfg and returns a ready-to-use Agent. +// Returns an error if any required configuration field is missing or invalid. +func NewAgent(cfg AgentConfig, provider telemetry.GPUStatsProvider, st *store.TelemetryStore) (*Agent, error) { + if cfg.NodeID == "" { + return nil, fmt.Errorf("agent: NodeID must not be empty") + } + if len(cfg.GPUIDs) == 0 { + return nil, fmt.Errorf("agent: GPUIDs must not be empty") + } + if provider == nil { + return nil, fmt.Errorf("agent: provider must not be nil") + } + if st == nil { + return nil, fmt.Errorf("agent: store must not be nil") + } + + // Apply defaults. + if cfg.PollInterval <= 0 { + cfg.PollInterval = DefaultPollInterval + } + if cfg.StalenessThreshold <= 0 { + cfg.StalenessThreshold = time.Duration(DefaultStalenessMultiplier) * cfg.PollInterval + } + if cfg.MaxRetries <= 0 { + cfg.MaxRetries = DefaultMaxRetries + } + if cfg.RetryDelay <= 0 { + cfg.RetryDelay = DefaultRetryDelay + } + if cfg.clock == nil { + cfg.clock = time.Now + } + + return &Agent{cfg: cfg, provider: provider, store: st}, nil +} + +// Run starts the agent's poll loop. It blocks until ctx is cancelled, then +// performs a final MarkStale sweep and returns ctx.Err(). +// +// Run is designed to be called in a dedicated goroutine: +// +// go func() { _ = agent.Run(ctx) }() +func (a *Agent) Run(ctx context.Context) error { + ticker := time.NewTicker(a.cfg.PollInterval) + defer ticker.Stop() + + for { + select { + case <-ctx.Done(): + // Graceful shutdown: mark all store entries stale so consumers + // immediately see that this agent is no longer reporting. + // Threshold of 1ns means age > 1ns — true for any real timestamp. + a.store.MarkStale(1) + return ctx.Err() + case <-ticker.C: + a.runOneTick(ctx) + } + } +} + +// runOneTick executes a single poll-and-store cycle. Errors are handled +// internally (per-GPU conservative defaults); this method never returns an +// error to the caller. +func (a *Agent) runOneTick(ctx context.Context) { + start := a.cfg.clock() + snapshot, errCount := a.poll(ctx) + duration := time.Since(start) + + a.store.Set(snapshot) + a.store.MarkStale(a.cfg.StalenessThreshold) + + if a.cfg.Metrics.OnStoreWrite != nil { + a.cfg.Metrics.OnStoreWrite(snapshot.NodeID) + } + if a.cfg.Metrics.OnPollComplete != nil { + a.cfg.Metrics.OnPollComplete(duration, errCount) + } +} + +// PollOnce executes a single poll cycle: queries the provider for every GPU +// and assembles a NodeTelemetry snapshot. It does NOT write to the store — +// that is done by runOneTick. PollOnce is exported so integration tests can +// exercise the collection logic in isolation. +// +// The returned errCount is the number of GPUs for which at least one telemetry +// signal could not be obtained after all retries. +func (a *Agent) PollOnce(ctx context.Context) (telemetry.NodeTelemetry, int) { + return a.poll(ctx) +} + +// poll is the internal implementation shared by runOneTick and PollOnce. +func (a *Agent) poll(ctx context.Context) (telemetry.NodeTelemetry, int) { + gpus := make([]telemetry.GPUTelemetry, 0, len(a.cfg.GPUIDs)) + errCount := 0 + + for _, gpuID := range a.cfg.GPUIDs { + // Check for context cancellation before each GPU to allow fast shutdown + // when the GPU list is long. + select { + case <-ctx.Done(): + // Return what we have so far with conservative defaults for the + // remaining GPUs. + remaining := len(a.cfg.GPUIDs) - len(gpus) + errCount += remaining + for i := len(gpus); i < len(a.cfg.GPUIDs); i++ { + gpus = append(gpus, conservativeGPU(a.cfg.GPUIDs[i])) + } + return a.buildSnapshot(gpus), errCount + default: + } + + gpu, gpuErr := a.collectGPU(ctx, gpuID) + if gpuErr { + errCount++ + } + gpus = append(gpus, gpu) + } + + // Collect topology (best-effort; errors produce a zero TopologyMatrix). + topo, _ := a.provider.GetTopology(a.cfg.NodeID) + + snap := telemetry.NodeTelemetry{ + NodeID: a.cfg.NodeID, + SchemaVersion: telemetry.SchemaVersion, + GPUs: gpus, + Topology: topo, + CollectedAt: a.cfg.clock(), + Stale: false, + } + return snap, errCount +} + +// buildSnapshot assembles a NodeTelemetry with the already-collected GPU slice +// and a fresh topology query. Used when the context is cancelled mid-poll. +func (a *Agent) buildSnapshot(gpus []telemetry.GPUTelemetry) telemetry.NodeTelemetry { + topo, _ := a.provider.GetTopology(a.cfg.NodeID) + return telemetry.NodeTelemetry{ + NodeID: a.cfg.NodeID, + SchemaVersion: telemetry.SchemaVersion, + GPUs: gpus, + Topology: topo, + CollectedAt: a.cfg.clock(), + Stale: false, + } +} + +// collectGPU queries all three telemetry signals for a single GPU with retry +// logic. Returns the populated GPUTelemetry and a boolean indicating whether +// any signal failed after all retries. +func (a *Agent) collectGPU(ctx context.Context, gpuID string) (telemetry.GPUTelemetry, bool) { + gpu := telemetry.GPUTelemetry{GPUID: gpuID} + anyError := false + + // --- Utilisation --- + util, err := a.retryFloat(ctx, func() (float64, error) { + return a.provider.GetUtilization(gpuID) + }) + if err == nil { + gpu.UtilizationPct = util + gpu.UtilizationKnown = true + } else { + gpu.UtilizationPct = telemetry.UnknownUtilization + gpu.UtilizationKnown = false + anyError = true + } + + // --- Memory --- + freeMem, err := a.retryInt64(ctx, func() (int64, error) { + return a.provider.GetFreeMemoryMB(gpuID) + }) + // We need TotalMemoryMB too. The interface does not expose it directly; + // derive it from FreeMemoryMB only (FakeProvider stores exact values). + // The real NVML provider will be updated in Phase 4 to return total memory. + // For now, store free memory and mark known/unknown appropriately. + if err == nil { + gpu.FreeMemoryMB = freeMem + gpu.TotalMemoryMB = freeMem // Phase 4 TODO: query total separately + gpu.MemoryKnown = true + } else { + gpu.FreeMemoryMB = telemetry.UnknownMemory + gpu.TotalMemoryMB = telemetry.UnknownMemory + gpu.MemoryKnown = false + anyError = true + } + + // --- ECC --- + eccCount, err := a.retryUint64(ctx, func() (uint64, error) { + return a.provider.GetECCErrorCount(gpuID) + }) + if err == nil { + gpu.ECCErrors = eccCount + gpu.ECCKnown = true + } else { + gpu.ECCErrors = 0 + gpu.ECCKnown = false + anyError = true + } + + return gpu, anyError +} + +// conservativeGPU returns a GPUTelemetry populated with conservative sentinel +// values for a GPU that could not be queried (e.g. context cancelled). +func conservativeGPU(gpuID string) telemetry.GPUTelemetry { + return telemetry.GPUTelemetry{ + GPUID: gpuID, + UtilizationPct: telemetry.UnknownUtilization, + UtilizationKnown: false, + FreeMemoryMB: telemetry.UnknownMemory, + TotalMemoryMB: telemetry.UnknownMemory, + MemoryKnown: false, + ECCErrors: 0, + ECCKnown: false, + } +} + +// --------------------------------------------------------------------------- +// Retry helpers — one per return type to avoid interface{} / reflection +// --------------------------------------------------------------------------- + +func (a *Agent) retryFloat(ctx context.Context, fn func() (float64, error)) (float64, error) { + var ( + val float64 + err error + ) + for attempt := 0; attempt <= a.cfg.MaxRetries; attempt++ { + val, err = fn() + if err == nil { + return val, nil + } + if attempt < a.cfg.MaxRetries { + a.sleep(ctx, a.cfg.RetryDelay) + } + } + return 0, err +} + +func (a *Agent) retryInt64(ctx context.Context, fn func() (int64, error)) (int64, error) { + var ( + val int64 + err error + ) + for attempt := 0; attempt <= a.cfg.MaxRetries; attempt++ { + val, err = fn() + if err == nil { + return val, nil + } + if attempt < a.cfg.MaxRetries { + a.sleep(ctx, a.cfg.RetryDelay) + } + } + return 0, err +} + +func (a *Agent) retryUint64(ctx context.Context, fn func() (uint64, error)) (uint64, error) { + var ( + val uint64 + err error + ) + for attempt := 0; attempt <= a.cfg.MaxRetries; attempt++ { + val, err = fn() + if err == nil { + return val, nil + } + if attempt < a.cfg.MaxRetries { + a.sleep(ctx, a.cfg.RetryDelay) + } + } + return 0, err +} + +// sleep pauses for d, but returns immediately if ctx is cancelled. +func (a *Agent) sleep(ctx context.Context, d time.Duration) { + if d <= 0 { + return + } + timer := time.NewTimer(d) + defer timer.Stop() + select { + case <-ctx.Done(): + case <-timer.C: + } +} diff --git a/pkg/agent/agent_test.go b/pkg/agent/agent_test.go new file mode 100644 index 0000000..5a0b90f --- /dev/null +++ b/pkg/agent/agent_test.go @@ -0,0 +1,511 @@ +package agent_test + +import ( + "context" + "fmt" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/amshithnair/gpu-aware-scheduler/pkg/agent" + "github.com/amshithnair/gpu-aware-scheduler/pkg/store" + "github.com/amshithnair/gpu-aware-scheduler/pkg/telemetry" +) + +// --------------------------------------------------------------------------- +// Helpers +// --------------------------------------------------------------------------- + +// buildFake returns a FakeGPUStatsProvider pre-populated for n GPUs with +// healthy telemetry values. +func buildFake(nodeID string, n int) *telemetry.FakeGPUStatsProvider { + f := telemetry.NewFakeGPUStatsProvider() + for i := 0; i < n; i++ { + id := fmt.Sprintf("gpu-%d", i) + f.Utilization[id] = float64(i) * 0.1 + f.FreeMemoryMB[id] = int64(8192 - i*1024) + f.ECCErrorCount[id] = uint64(i * 2) + } + f.Topologies[nodeID] = telemetry.TopologyMatrix{ + Matrix: map[string]map[string]int{ + "gpu-0": {"gpu-1": 1}, + "gpu-1": {"gpu-0": 1}, + }, + } + return f +} + +// gpuIDs returns ["gpu-0", ..., "gpu-(n-1)"]. +func gpuIDs(n int) []string { + ids := make([]string, n) + for i := range ids { + ids[i] = fmt.Sprintf("gpu-%d", i) + } + return ids +} + +// fastConfig returns a config with tiny intervals so tests don't have to wait. +func fastConfig(nodeID string, n int) agent.AgentConfig { + return agent.AgentConfig{ + NodeID: nodeID, + GPUIDs: gpuIDs(n), + PollInterval: 5 * time.Millisecond, + StalenessThreshold: 50 * time.Millisecond, + MaxRetries: 1, + RetryDelay: 0, + } +} + +// --------------------------------------------------------------------------- +// NewAgent validation +// --------------------------------------------------------------------------- + +func TestNewAgent_RequiresNodeID(t *testing.T) { + cfg := fastConfig("", 2) + _, err := agent.NewAgent(cfg, buildFake("", 2), store.NewTelemetryStore()) + if err == nil { + t.Error("expected error for empty NodeID") + } +} + +func TestNewAgent_RequiresGPUIDs(t *testing.T) { + cfg := fastConfig("node-1", 0) + cfg.GPUIDs = nil + _, err := agent.NewAgent(cfg, buildFake("node-1", 0), store.NewTelemetryStore()) + if err == nil { + t.Error("expected error for nil GPUIDs") + } +} + +func TestNewAgent_RequiresProvider(t *testing.T) { + cfg := fastConfig("node-1", 2) + _, err := agent.NewAgent(cfg, nil, store.NewTelemetryStore()) + if err == nil { + t.Error("expected error for nil provider") + } +} + +func TestNewAgent_RequiresStore(t *testing.T) { + cfg := fastConfig("node-1", 2) + _, err := agent.NewAgent(cfg, buildFake("node-1", 2), nil) + if err == nil { + t.Error("expected error for nil store") + } +} + +func TestNewAgent_AppliesDefaults(t *testing.T) { + cfg := agent.AgentConfig{ + NodeID: "node-1", + GPUIDs: gpuIDs(1), + // PollInterval, MaxRetries, RetryDelay all zero → should use defaults + } + a, err := agent.NewAgent(cfg, buildFake("node-1", 1), store.NewTelemetryStore()) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if a == nil { + t.Fatal("expected non-nil agent") + } +} + +// --------------------------------------------------------------------------- +// PollOnce — happy path +// --------------------------------------------------------------------------- + +func TestAgent_PollOnce_HappyPath(t *testing.T) { + nodeID := "node-1" + fake := buildFake(nodeID, 2) + cfg := fastConfig(nodeID, 2) + st := store.NewTelemetryStore() + + a, err := agent.NewAgent(cfg, fake, st) + if err != nil { + t.Fatalf("NewAgent: %v", err) + } + + snap, errCount := a.PollOnce(context.Background()) + + if errCount != 0 { + t.Errorf("expected 0 errors, got %d", errCount) + } + if snap.NodeID != nodeID { + t.Errorf("want NodeID=%q, got %q", nodeID, snap.NodeID) + } + if len(snap.GPUs) != 2 { + t.Errorf("want 2 GPUs, got %d", len(snap.GPUs)) + } + for _, gpu := range snap.GPUs { + if !gpu.UtilizationKnown { + t.Errorf("GPU %s: UtilizationKnown should be true", gpu.GPUID) + } + if !gpu.MemoryKnown { + t.Errorf("GPU %s: MemoryKnown should be true", gpu.GPUID) + } + if !gpu.ECCKnown { + t.Errorf("GPU %s: ECCKnown should be true", gpu.GPUID) + } + } + if snap.CollectedAt.IsZero() { + t.Error("CollectedAt must not be zero") + } + if snap.SchemaVersion != telemetry.SchemaVersion { + t.Errorf("want SchemaVersion=%d, got %d", telemetry.SchemaVersion, snap.SchemaVersion) + } +} + +// --------------------------------------------------------------------------- +// PollOnce — per-GPU provider errors +// --------------------------------------------------------------------------- + +func TestAgent_PollOnce_PerGPU_ProviderError(t *testing.T) { + nodeID := "node-1" + // Only gpu-0 is configured in the fake; gpu-1 will return errors. + fake := telemetry.NewFakeGPUStatsProvider() + fake.Utilization["gpu-0"] = 0.3 + fake.FreeMemoryMB["gpu-0"] = 4096 + fake.ECCErrorCount["gpu-0"] = 0 + // gpu-1 has no entries → all calls return errors + + cfg := fastConfig(nodeID, 2) + cfg.MaxRetries = 1 + cfg.RetryDelay = 0 + + a, _ := agent.NewAgent(cfg, fake, store.NewTelemetryStore()) + snap, errCount := a.PollOnce(context.Background()) + + if errCount != 1 { + t.Errorf("expected 1 errored GPU, got %d", errCount) + } + + // gpu-0 should be known + gpu0 := snap.GPUs[0] + if !gpu0.UtilizationKnown { + t.Errorf("gpu-0 should have known utilization") + } + + // gpu-1 should be unknown (conservative) + gpu1 := snap.GPUs[1] + if gpu1.UtilizationKnown { + t.Errorf("gpu-1 utilization should be unknown after provider error") + } + if gpu1.UtilizationPct != telemetry.UnknownUtilization { + t.Errorf("gpu-1 utilization should be sentinel, got %v", gpu1.UtilizationPct) + } +} + +func TestAgent_PollOnce_AllGPUs_ProviderError(t *testing.T) { + // Empty fake — all calls fail. + fake := telemetry.NewFakeGPUStatsProvider() + cfg := fastConfig("node-1", 2) + cfg.MaxRetries = 1 + cfg.RetryDelay = 0 + + a, _ := agent.NewAgent(cfg, fake, store.NewTelemetryStore()) + snap, errCount := a.PollOnce(context.Background()) + + if errCount != 2 { + t.Errorf("expected 2 errored GPUs, got %d", errCount) + } + for _, gpu := range snap.GPUs { + if gpu.UtilizationKnown || gpu.MemoryKnown || gpu.ECCKnown { + t.Errorf("GPU %s: all Known flags should be false after total provider failure", gpu.GPUID) + } + } +} + +// --------------------------------------------------------------------------- +// PollOnce — topology error is non-fatal +// --------------------------------------------------------------------------- + +func TestAgent_PollOnce_TopologyError_NonFatal(t *testing.T) { + nodeID := "node-1" + fake := telemetry.NewFakeGPUStatsProvider() + fake.Utilization["gpu-0"] = 0.1 + fake.FreeMemoryMB["gpu-0"] = 4096 + fake.ECCErrorCount["gpu-0"] = 0 + // No topology for nodeID → GetTopology returns error + + cfg := fastConfig(nodeID, 1) + a, _ := agent.NewAgent(cfg, fake, store.NewTelemetryStore()) + snap, errCount := a.PollOnce(context.Background()) + + // Topology error must not count as a GPU error. + if errCount != 0 { + t.Errorf("topology error should not increment GPU errCount, got %d", errCount) + } + if snap.Topology.Matrix != nil { + t.Error("topology should be zero-value when provider returns error") + } +} + +// --------------------------------------------------------------------------- +// PollOnce writes to store +// --------------------------------------------------------------------------- + +func TestAgent_RunOneTick_WritesToStore(t *testing.T) { + nodeID := "node-1" + fake := buildFake(nodeID, 2) + cfg := fastConfig(nodeID, 2) + st := store.NewTelemetryStore() + + a, _ := agent.NewAgent(cfg, fake, st) + a.PollOnce(context.Background()) // does NOT write to store — that's runOneTick + + // Verify by running the agent briefly. + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Millisecond) + defer cancel() + go func() { _ = a.Run(ctx) }() + time.Sleep(20 * time.Millisecond) + + rec, ok := st.Get(nodeID) + if !ok { + t.Fatal("expected record in store after agent ran") + } + if rec.NodeID != nodeID { + t.Errorf("want NodeID=%q, got %q", nodeID, rec.NodeID) + } +} + +// --------------------------------------------------------------------------- +// Context cancellation / graceful shutdown +// --------------------------------------------------------------------------- + +func TestAgent_Run_ContextCancellation(t *testing.T) { + nodeID := "node-1" + fake := buildFake(nodeID, 2) + cfg := fastConfig(nodeID, 2) + st := store.NewTelemetryStore() + + a, _ := agent.NewAgent(cfg, fake, st) + + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan error, 1) + go func() { + done <- a.Run(ctx) + }() + + cancel() + + select { + case err := <-done: + if err != context.Canceled { + t.Errorf("expected context.Canceled, got %v", err) + } + case <-time.After(500 * time.Millisecond): + t.Error("agent did not stop within 500ms after context cancellation") + } +} + +func TestAgent_Run_ShutdownMarksStale(t *testing.T) { + nodeID := "node-1" + fake := buildFake(nodeID, 1) + cfg := fastConfig(nodeID, 1) + st := store.NewTelemetryStore() + + a, _ := agent.NewAgent(cfg, fake, st) + + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan struct{}) + go func() { + _ = a.Run(ctx) + close(done) + }() + + // Wait for at least one poll cycle to write a record. + deadline := time.Now().Add(200 * time.Millisecond) + for time.Now().Before(deadline) { + if _, ok := st.Get(nodeID); ok { + break + } + time.Sleep(2 * time.Millisecond) + } + if _, ok := st.Get(nodeID); !ok { + t.Fatal("expected record in store before shutdown") + } + + // Cancel and wait for agent to finish. + cancel() + select { + case <-done: + case <-time.After(500 * time.Millisecond): + t.Fatal("agent did not stop within 500ms") + } + + // After shutdown the agent calls MarkStale(0), which marks all records stale + // regardless of their CollectedAt timestamp. + rec, ok := st.Get(nodeID) + if !ok { + // Record might have been deleted — that's also acceptable; skip assertion. + return + } + if !rec.Stale { + t.Error("record should be stale after agent shutdown (MarkStale(0) called)") + } +} + +// --------------------------------------------------------------------------- +// MetricsHook callbacks +// --------------------------------------------------------------------------- + +func TestAgent_MetricsHook_OnPollComplete(t *testing.T) { + nodeID := "node-1" + fake := buildFake(nodeID, 2) + cfg := fastConfig(nodeID, 2) + + var mu sync.Mutex + pollCount := 0 + lastErrCount := -1 + + cfg.Metrics = agent.MetricsHook{ + OnPollComplete: func(d time.Duration, errCount int) { + mu.Lock() + pollCount++ + lastErrCount = errCount + mu.Unlock() + }, + } + + st := store.NewTelemetryStore() + a, _ := agent.NewAgent(cfg, fake, st) + + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Millisecond) + defer cancel() + go func() { _ = a.Run(ctx) }() + time.Sleep(25 * time.Millisecond) + + mu.Lock() + count := pollCount + errC := lastErrCount + mu.Unlock() + + if count == 0 { + t.Error("OnPollComplete should have been called at least once") + } + if errC != 0 { + t.Errorf("expected 0 errors in hook, got %d", errC) + } +} + +func TestAgent_MetricsHook_OnStoreWrite(t *testing.T) { + nodeID := "node-1" + fake := buildFake(nodeID, 1) + cfg := fastConfig(nodeID, 1) + + var writeCount int64 + cfg.Metrics = agent.MetricsHook{ + OnStoreWrite: func(nid string) { + atomic.AddInt64(&writeCount, 1) + }, + } + + st := store.NewTelemetryStore() + a, _ := agent.NewAgent(cfg, fake, st) + + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Millisecond) + defer cancel() + go func() { _ = a.Run(ctx) }() + time.Sleep(25 * time.Millisecond) + + if atomic.LoadInt64(&writeCount) == 0 { + t.Error("OnStoreWrite should have been called at least once") + } +} + +// --------------------------------------------------------------------------- +// Staleness written after poll +// --------------------------------------------------------------------------- + +func TestAgent_StalenessThreshold_Applied(t *testing.T) { + nodeID := "node-1" + fake := buildFake(nodeID, 1) + cfg := fastConfig(nodeID, 1) + cfg.StalenessThreshold = 5 * time.Millisecond + cfg.PollInterval = 5 * time.Millisecond + st := store.NewTelemetryStore() + + // Manually insert a very old record. + old := telemetry.NodeTelemetry{ + NodeID: nodeID, + SchemaVersion: telemetry.SchemaVersion, + CollectedAt: time.Now().Add(-1 * time.Hour), + } + st.Set(old) + + a, _ := agent.NewAgent(cfg, fake, st) + + // Run one full tick (poll + MarkStale). + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Millisecond) + defer cancel() + go func() { _ = a.Run(ctx) }() + time.Sleep(20 * time.Millisecond) + + rec, ok := st.Get(nodeID) + if !ok { + t.Fatal("expected record in store") + } + // After at least one poll cycle the agent writes a fresh record and then + // MarkStale runs. The fresh record should NOT be stale. + if rec.Stale { + t.Error("fresh record written by agent should not be stale immediately after poll") + } +} + +// --------------------------------------------------------------------------- +// Concurrent agent and store readers (race test) +// --------------------------------------------------------------------------- + +func TestAgent_Race_ConcurrentReaders(t *testing.T) { + nodeID := "node-1" + fake := buildFake(nodeID, 4) + cfg := fastConfig(nodeID, 4) + cfg.PollInterval = 2 * time.Millisecond + st := store.NewTelemetryStore() + + a, _ := agent.NewAgent(cfg, fake, st) + + ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond) + defer cancel() + go func() { _ = a.Run(ctx) }() + + var wg sync.WaitGroup + for i := 0; i < 20; i++ { + wg.Add(1) + go func() { + defer wg.Done() + for j := 0; j < 50; j++ { + st.Get(nodeID) + st.List() + st.Snapshot() + } + }() + } + wg.Wait() +} + +// --------------------------------------------------------------------------- +// Multiple GPUs — all collected independently +// --------------------------------------------------------------------------- + +func TestAgent_PollOnce_MultipleGPUs_Independent(t *testing.T) { + nodeID := "node-1" + fake := buildFake(nodeID, 4) + cfg := fastConfig(nodeID, 4) + + a, _ := agent.NewAgent(cfg, fake, store.NewTelemetryStore()) + snap, errCount := a.PollOnce(context.Background()) + + if errCount != 0 { + t.Errorf("expected 0 errors, got %d", errCount) + } + if len(snap.GPUs) != 4 { + t.Fatalf("expected 4 GPUs, got %d", len(snap.GPUs)) + } + // Each GPU should have its own utilisation value (0, 0.1, 0.2, 0.3). + for i, gpu := range snap.GPUs { + expectedUtil := float64(i) * 0.1 + if gpu.UtilizationPct != expectedUtil { + t.Errorf("GPU[%d] util: want %.2f, got %.2f", i, expectedUtil, gpu.UtilizationPct) + } + } +} From ca5d71faf8ebbb8f6e3985b4c306e3fa213c5435 Mon Sep 17 00:00:00 2001 From: amshithnair Date: Tue, 28 Jul 2026 03:02:05 +0530 Subject: [PATCH 6/7] feat(manifests): add GPUNodeStatus CRD with status subresource - apiextensions.k8s.io/v1 CustomResourceDefinition - Group: gpu.amshithnair.dev, version: v1alpha1 - Status subresource enabled (agent-only writes to .status) - OpenAPI v3 schema with full validation constraints - Printer columns: Node, Stale, LastUpdated, Age - Short name: gpuns, scope: Namespaced --- manifests/gpunodestatus-crd.yaml | 153 +++++++++++++++++++++++++++++++ 1 file changed, 153 insertions(+) create mode 100644 manifests/gpunodestatus-crd.yaml diff --git a/manifests/gpunodestatus-crd.yaml b/manifests/gpunodestatus-crd.yaml new file mode 100644 index 0000000..caa64c0 --- /dev/null +++ b/manifests/gpunodestatus-crd.yaml @@ -0,0 +1,153 @@ +--- +# GPUNodeStatus Custom Resource Definition +# +# Group: gpu.amshithnair.dev +# Version: v1alpha1 +# Resource: gpunodestatuses +# Kind: GPUNodeStatus +# +# One GPUNodeStatus object exists per GPU-enabled Kubernetes node. +# The telemetry agent (DaemonSet) updates .status on every poll cycle. +# The scheduler plugin reads .status to build a NodeScoreInput. +# +# Apply: +# kubectl apply -f manifests/gpunodestatus-crd.yaml +# +# Dry-run (no cluster required): +# kubectl apply --dry-run=client -f manifests/gpunodestatus-crd.yaml + +apiVersion: apiextensions.k8s.io/v1 +kind: CustomResourceDefinition +metadata: + name: gpunodestatuses.gpu.amshithnair.dev + labels: + app.kubernetes.io/name: gpu-aware-scheduler + app.kubernetes.io/component: telemetry-crd + app.kubernetes.io/version: v1alpha1 +spec: + group: gpu.amshithnair.dev + names: + kind: GPUNodeStatus + listKind: GPUNodeStatusList + plural: gpunodestatuses + singular: gpunodestatus + shortNames: + - gpuns + scope: Namespaced + versions: + - name: v1alpha1 + served: true + storage: true + # Status subresource: only the telemetry agent writes .status. + # kubectl edit cannot overwrite telemetry data accidentally. + subresources: + status: {} + additionalPrinterColumns: + - name: Node + type: string + description: The Kubernetes node this object represents + jsonPath: .spec.nodeName + - name: Stale + type: boolean + description: Whether telemetry data is stale + jsonPath: .status.stale + - name: LastUpdated + type: date + description: Time of last successful telemetry poll + jsonPath: .status.lastUpdated + - name: Age + type: date + jsonPath: .metadata.creationTimestamp + schema: + openAPIV3Schema: + type: object + description: > + GPUNodeStatus represents the current GPU health and telemetry state + of a single Kubernetes node. + required: + - spec + properties: + apiVersion: + type: string + kind: + type: string + metadata: + type: object + spec: + type: object + description: > + Static operator-configured fields. Set once at node registration. + required: + - nodeName + - gpuIDs + properties: + nodeName: + type: string + description: > + Kubernetes node name. Must match the node's metadata.name. + minLength: 1 + maxLength: 253 + gpuIDs: + type: array + description: > + Ordered list of GPU identifiers on this node. + minItems: 1 + maxItems: 64 + items: + type: string + minLength: 1 + maxLength: 64 + status: + type: object + description: > + Updated by the telemetry agent on every poll cycle. + Do not edit manually. + properties: + gpus: + type: array + description: Per-GPU telemetry from the most recent poll. + items: + type: object + required: + - gpuID + properties: + gpuID: + type: string + minLength: 1 + utilizationPct: + type: number + description: > + GPU compute utilisation in [0.0, 1.0]. + -1 when utilizationKnown is false. + minimum: -1.0 + maximum: 1.0 + utilizationKnown: + type: boolean + freeMemoryMB: + type: integer + description: Free memory in MB. -1 when memoryKnown is false. + minimum: -1 + totalMemoryMB: + type: integer + description: Total memory in MB. -1 when memoryKnown is false. + minimum: -1 + memoryKnown: + type: boolean + eccErrors: + type: integer + description: Total ECC error count. + minimum: 0 + eccKnown: + type: boolean + lastUpdated: + type: string + format: date-time + description: RFC3339 timestamp of the most recent successful poll. + stale: + type: boolean + description: > + True when lastUpdated is older than the staleness threshold. + schemaVersion: + type: integer + description: Version of the GPUStatusEntry schema. + minimum: 1 From 02f99ef05442153eb40aba3ce31d6ecc9706bff5 Mon Sep 17 00:00:00 2001 From: amshithnair Date: Tue, 28 Jul 2026 03:02:20 +0530 Subject: [PATCH 7/7] docs: add Phase 3 documentation, verification scripts, and manual checklist - docs/phase3.md: architecture, data model, store/agent design, concurrency model, staleness model, failure handling, testing guide, future extensions, known limitations, rollback notes - scripts/phase3-test.ps1 + .sh: automated quality gate (fmt, vet, lint, build, test, race) - scripts/phase3-manual.ps1 + .sh: cluster CRD lifecycle verification - checklists/phase3-manual-checklist.md: 60-test inventory across 3 packages with step-by-step sign-off procedure --- checklists/phase3-manual-checklist.md | 390 +++++++++++++++++++++ docs/phase3.md | 481 ++++++++++++++++++++++++++ scripts/phase3-manual.ps1 | 261 ++++++++++++++ scripts/phase3-manual.sh | 180 ++++++++++ scripts/phase3-test.ps1 | 243 +++++++++++++ scripts/phase3-test.sh | 170 +++++++++ 6 files changed, 1725 insertions(+) create mode 100644 checklists/phase3-manual-checklist.md create mode 100644 docs/phase3.md create mode 100644 scripts/phase3-manual.ps1 create mode 100644 scripts/phase3-manual.sh create mode 100644 scripts/phase3-test.ps1 create mode 100644 scripts/phase3-test.sh diff --git a/checklists/phase3-manual-checklist.md b/checklists/phase3-manual-checklist.md new file mode 100644 index 0000000..af6b1a1 --- /dev/null +++ b/checklists/phase3-manual-checklist.md @@ -0,0 +1,390 @@ +# Phase 3 Manual Verification Guide + +> Complete every item in order. Check each box only after you have personally observed the +> described outcome. Do not proceed to Phase 4 until all boxes are checked. + +--- + +## Prerequisites + +- [ ] Go 1.22 or later is installed. + - Verify: `go version` prints `go1.22` or higher. +- [ ] You are in the repository root directory. + - Verify: `Test-Path go.mod` returns `True` (PowerShell) or `ls go.mod` succeeds (bash). +- [ ] `go.sum` exists (should have been created in Phase 1). +- [ ] Phase 1 and Phase 2 tests still pass. + - Verify: `go test ./pkg/telemetry/... ./pkg/scoring/... -count=1` shows `ok` for both packages. + +--- + +## Section 1 — File Existence + +Confirm every Phase 3 file is present on disk. + +**Data Model & Telemetry** +- [ ] `pkg/telemetry/model.go` exists. +- [ ] `pkg/telemetry/model_test.go` exists. + +**CRD Go Types** +- [ ] `pkg/api/v1alpha1/types.go` exists. +- [ ] `pkg/api/v1alpha1/deepcopy.go` exists. + +**Telemetry Store** +- [ ] `pkg/store/store.go` exists. +- [ ] `pkg/store/store_test.go` exists. + +**Polling Agent** +- [ ] `pkg/agent/agent.go` exists. +- [ ] `pkg/agent/agent_test.go` exists. + +**CRD Manifest** +- [ ] `manifests/gpunodestatus-crd.yaml` exists. + +**Documentation** +- [ ] `docs/phase3.md` exists. + +**Verification Scripts** +- [ ] `scripts/phase3-test.ps1` exists. +- [ ] `scripts/phase3-test.sh` exists. +- [ ] `scripts/phase3-manual.ps1` exists. +- [ ] `scripts/phase3-manual.sh` exists. + +--- + +## Section 2 — Module Correctness + +- [ ] Open `go.mod`. Confirm the module line reads: + ``` + module github.com/amshithnair/gpu-aware-scheduler + ``` +- [ ] Confirm the `go` directive is `1.22` or higher. +- [ ] Run `go build ./...` — no errors. All Phase 3 packages compile cleanly. +- [ ] Run `go vet ./...` — no warnings or errors. + +--- + +## Section 3 — Automated Script Execution (Windows) + +Run the PowerShell verification script and observe its output. + +```powershell +cd "C:\Users\Amshith Nair\gpu-aware-scheduler" +.\scripts\phase3-test.ps1 +``` + +- [ ] **Step 1 — Validate tools**: Script detects `go` and prints its version. Missing optional + tools (kubectl, kind, docker, golangci-lint) are reported as `[SKIP]`, not `[FAIL]`. +- [ ] **Step 2 — go fmt**: Script prints `[PASS] STEP 2: go fmt ./...` with no reformatted files. +- [ ] **Step 3 — go vet**: Script prints `[PASS] STEP 3: go vet ./...` with no issues. +- [ ] **Step 4 — golangci-lint**: Either prints `[PASS]` or `[SKIP]` (if not installed). + Both are acceptable for verification. +- [ ] **Step 5 — go build**: Script prints `[PASS] STEP 5: go build ./...` with no errors. +- [ ] **Step 6 — go test**: Script prints `[PASS] STEP 6: go test ./... -v -count=1 -cover`. + - The `-v` output shows every test function ending with `--- PASS`. + - No test ends with `--- FAIL` or `--- SKIP`. + - Coverage numbers are printed for each package. + - Final lines include: + ``` + ok github.com/amshithnair/gpu-aware-scheduler/pkg/agent ... + ok github.com/amshithnair/gpu-aware-scheduler/pkg/scoring ... + ok github.com/amshithnair/gpu-aware-scheduler/pkg/store ... + ok github.com/amshithnair/gpu-aware-scheduler/pkg/telemetry ... + ``` +- [ ] **Step 7 — go test -race**: Either `[PASS]` (Linux/macOS with CGo) or + `[SKIP] CGO_ENABLED=0` (Windows without C compiler). Both are acceptable. +- [ ] **Summary banner**: Script prints `ALL PHASE 3 CHECKS PASSED`. +- [ ] **Exit code**: Script exits with code 0. + - Verify: `echo $LASTEXITCODE` prints `0`. + +--- + +## Section 4 — Individual Test Inventory + +Open the `-v` test output from Step 6 and confirm each of the following test functions appeared +and passed. Cross off each one as you find it. + +### 4A — `pkg/telemetry` — `model_test.go` (23 tests) + +**GPUTelemetry.Validate** +- [ ] `TestGPUTelemetry_Validate_Happy` +- [ ] `TestGPUTelemetry_Validate_EmptyGPUID` +- [ ] `TestGPUTelemetry_Validate_UtilizationOutOfRange` (3 sub-tests: negative, above 1, way above) +- [ ] `TestGPUTelemetry_Validate_UnknownUtilization_NoError` +- [ ] `TestGPUTelemetry_Validate_NegativeFreeMemory` +- [ ] `TestGPUTelemetry_Validate_ZeroTotalMemory` +- [ ] `TestGPUTelemetry_Validate_FreeExceedsTotal` + +**GPUTelemetry helpers** +- [ ] `TestGPUTelemetry_IsHealthy` +- [ ] `TestGPUTelemetry_DeepCopy_Independence` + +**NodeTelemetry.Validate** +- [ ] `TestNodeTelemetry_Validate_Happy` +- [ ] `TestNodeTelemetry_Validate_EmptyNodeID` +- [ ] `TestNodeTelemetry_Validate_ZeroCollectedAt` +- [ ] `TestNodeTelemetry_Validate_InvalidSchemaVersion` +- [ ] `TestNodeTelemetry_Validate_DuplicateGPUID` +- [ ] `TestNodeTelemetry_Validate_InvalidGPU` + +**NodeTelemetry.IsStale** +- [ ] `TestNodeTelemetry_IsStale_ZeroTimestamp` +- [ ] `TestNodeTelemetry_IsStale_Fresh` +- [ ] `TestNodeTelemetry_IsStaleAt_Deterministic` + +**NodeTelemetry.DeepCopy** +- [ ] `TestNodeTelemetry_DeepCopy_GPUSliceIndependence` +- [ ] `TestNodeTelemetry_DeepCopy_TopologyIndependence` +- [ ] `TestNodeTelemetry_DeepCopy_NilTopology` +- [ ] `TestNodeTelemetry_DeepCopy_NilGPUs` + +**TopologyMatrix deep copy** +- [ ] `TestDeepCopyTopologyMatrix_Empty` + +--- + +### 4B — `pkg/store` — `store_test.go` (20 tests) + +**CRUD operations** +- [ ] `TestStore_SetAndGet` +- [ ] `TestStore_Get_UnknownNode` +- [ ] `TestStore_Set_Replaces` +- [ ] `TestStore_Delete` +- [ ] `TestStore_Delete_NoOp_UnknownNode` +- [ ] `TestStore_Len` + +**List / Snapshot** +- [ ] `TestStore_List_Empty` +- [ ] `TestStore_List_AllNodes` +- [ ] `TestStore_List_DeepCopy` +- [ ] `TestStore_Snapshot_Empty` +- [ ] `TestStore_Snapshot_Independence` + +**Staleness sweep** +- [ ] `TestStore_MarkStaleAt_OldRecord` +- [ ] `TestStore_MarkStaleAt_FreshRecord` +- [ ] `TestStore_MarkStaleAt_AlreadyStale_NotModified` +- [ ] `TestStore_MarkStaleAt_MixedRecords` + +**Deep-copy contract** +- [ ] `TestStore_Get_DeepCopy` + +**Concurrency** +- [ ] `TestStore_ConcurrentReadsAndWrites` +- [ ] `TestStore_ConcurrentMarkStale` +- [ ] `TestStore_ConcurrentSnapshot` +- [ ] `TestStore_ConcurrentSetAndDelete` + +--- + +### 4C — `pkg/agent` — `agent_test.go` (17 tests) + +**Constructor validation** +- [ ] `TestNewAgent_RequiresNodeID` +- [ ] `TestNewAgent_RequiresGPUIDs` +- [ ] `TestNewAgent_RequiresProvider` +- [ ] `TestNewAgent_RequiresStore` +- [ ] `TestNewAgent_AppliesDefaults` + +**PollOnce** +- [ ] `TestAgent_PollOnce_HappyPath` +- [ ] `TestAgent_PollOnce_PerGPU_ProviderError` +- [ ] `TestAgent_PollOnce_AllGPUs_ProviderError` +- [ ] `TestAgent_PollOnce_TopologyError_NonFatal` +- [ ] `TestAgent_PollOnce_MultipleGPUs_Independent` + +**Run loop & lifecycle** +- [ ] `TestAgent_RunOneTick_WritesToStore` +- [ ] `TestAgent_Run_ContextCancellation` +- [ ] `TestAgent_Run_ShutdownMarksStale` + +**MetricsHook** +- [ ] `TestAgent_MetricsHook_OnPollComplete` +- [ ] `TestAgent_MetricsHook_OnStoreWrite` + +**Staleness** +- [ ] `TestAgent_StalenessThreshold_Applied` + +**Concurrency** +- [ ] `TestAgent_Race_ConcurrentReaders` + +--- + +**Total expected: 60 test functions** across 3 packages (23 + 20 + 17). + +> [!NOTE] +> The `pkg/telemetry/provider_test.go` tests (Phase 1) also run during `go test ./...` but are +> not counted here — they were verified in the Phase 1 checklist. Similarly for `pkg/scoring` +> (Phase 2). + +--- + +## Section 5 — CRD Manifest Validation + +### 5A — Structure review (manual code read) + +Open `manifests/gpunodestatus-crd.yaml` and verify: + +- [ ] `apiVersion` is `apiextensions.k8s.io/v1`. +- [ ] `metadata.name` is `gpunodestatuses.gpu.amshithnair.dev`. +- [ ] `spec.group` is `gpu.amshithnair.dev`. +- [ ] `spec.names.kind` is `GPUNodeStatus`. +- [ ] `spec.names.plural` is `gpunodestatuses`. +- [ ] `spec.names.shortNames` includes `gpuns`. +- [ ] `spec.scope` is `Namespaced`. +- [ ] `spec.versions[0].name` is `v1alpha1`. +- [ ] `spec.versions[0].served` is `true`. +- [ ] `spec.versions[0].storage` is `true`. +- [ ] `subresources.status: {}` is present (status subresource enabled). +- [ ] `additionalPrinterColumns` includes columns for: `Node`, `Stale`, `LastUpdated`, `Age`. + +### 5B — Schema fields (manual code read) + +Verify the OpenAPI v3 schema contains: + +- [ ] `.spec.nodeName` — type: string, required, minLength: 1. +- [ ] `.spec.gpuIDs` — type: array of strings, required, minItems: 1, maxItems: 64. +- [ ] `.status.gpus` — type: array of objects with fields: + `gpuID`, `utilizationPct`, `utilizationKnown`, `freeMemoryMB`, `totalMemoryMB`, + `memoryKnown`, `eccErrors`, `eccKnown`. +- [ ] `.status.lastUpdated` — type: string, format: date-time. +- [ ] `.status.stale` — type: boolean. +- [ ] `.status.schemaVersion` — type: integer, minimum: 1. + +### 5C — Dry-run validation + +```powershell +kubectl apply --dry-run=client -f manifests/gpunodestatus-crd.yaml +``` + +- [ ] Command succeeds with output: + ``` + customresourcedefinition.apiextensions.k8s.io/gpunodestatuses.gpu.amshithnair.dev configured (dry run) + ``` + If kubectl is not available, mark as `[SKIP]` — this will be verified during cluster testing. + +--- + +## Section 6 — CRD Go Types Validation (manual code read) + +Open `pkg/api/v1alpha1/types.go` and verify: + +- [ ] `GPUNodeStatus` struct has `TypeMeta`, `ObjectMeta`, `Spec`, and `Status` fields. +- [ ] `GPUNodeStatusSpec` has `NodeName string` and `GPUIDs []string`. +- [ ] `GPUNodeStatusStatus` has `GPUs []GPUStatusEntry`, `LastUpdated string`, `Stale bool`, + `SchemaVersion int`. +- [ ] `GPUStatusEntry` mirrors the CRD schema fields (gpuID, utilizationPct, etc.). +- [ ] JSON tags on all fields match the CRD YAML field names exactly. + +Open `pkg/api/v1alpha1/deepcopy.go` and verify: + +- [ ] `DeepCopyInto` methods exist for all CRD types. +- [ ] `DeepCopyObject` method exists on `GPUNodeStatus`, returning `runtime.Object`. +- [ ] GPU slice deep copy creates a new slice (not just a header copy). + +--- + +## Section 7 — Manual Cluster Verification (optional) + +> [!IMPORTANT] +> This section requires a live Kubernetes cluster (kind, minikube, or real). +> If no cluster is available, mark all items as `[SKIP — no cluster]`. +> The automated scripts in Sections 3–4 already cover code correctness. + +### 7A — Run the manual script + +**Windows:** +```powershell +.\scripts\phase3-manual.ps1 +``` + +**Linux / macOS:** +```bash +./scripts/phase3-manual.sh +``` + +### 7B — Verify script sections + +- [ ] **Section A — CRD Lifecycle**: + - [ ] A1: Dry-run passes (`[PASS]`). + - [ ] A2: CRD applied to cluster. + - [ ] A3: `kubectl get crd gpunodestatuses.gpu.amshithnair.dev` returns the CRD. + - [ ] A4: `kubectl describe crd` shows the schema and printer columns. + +- [ ] **Section B — Sample Object**: + - [ ] B1: `gpu-scheduler` namespace created. + - [ ] B2: Sample `GPUNodeStatus` object `node-gpu-01` created. + - [ ] B3: `kubectl get gpuns -n gpu-scheduler` shows the object with printer columns + (Node, Stale, LastUpdated, Age). + - [ ] B4: `kubectl describe gpuns node-gpu-01` shows Spec with nodeName and gpuIDs. + - [ ] B5: Status subresource patched — stale=false, two GPU entries visible. + - [ ] B6: `kubectl get gpuns node-gpu-01 -o yaml` shows populated `.status` with GPU data. + +- [ ] **Section C — Cleanup** (bash script only; PS1 skips directly to cleanup): + - [ ] Stale flag set to `true` via status patch. + - [ ] `kubectl get gpuns` shows `Stale = true` in printer column. + +- [ ] **Section D — Cleanup**: + - [ ] Sample object deleted. + - [ ] CRD deleted from cluster. + - [ ] CRD confirmed removed (`[PASS]`). + +- [ ] **Summary**: Script prints `PHASE 3 MANUAL VERIFICATION COMPLETE`. + +--- + +## Section 8 — Cross-Phase Regression Check + +Confirm that Phase 3 did not break Phase 1 or Phase 2. + +```powershell +go test ./pkg/telemetry/... ./pkg/scoring/... -v -count=1 +``` + +- [ ] `pkg/telemetry` — all Phase 1 tests pass (18 tests). +- [ ] `pkg/scoring` — all Phase 2 tests pass. +- [ ] No import cycle errors or compilation failures. + +--- + +## Section 9 — Documentation Review + +- [ ] Open `docs/phase3.md`. Confirm it contains all of the following sections: + - Overview + - Architecture diagram + - Telemetry Flow + - Data Model (GPUTelemetry + NodeTelemetry tables) + - Store Design (API, deep-copy contract, lock granularity) + - Agent Design (AgentConfig, MetricsHook, error handling policy) + - CRD Explanation (identity, status subresource, namespace convention, example YAML) + - Concurrency Model + - Staleness Model + - Failure Handling + - Testing Guide + - Future Extension Points + - Known Limitations + - Rollback Notes + +--- + +## Section 10 — Output Artifact + +- [ ] Copy the full terminal output from Section 3 (phase3-test.ps1) and save it for the + Phase 3 verification record. +- [ ] If Section 7 was executed, copy the full terminal output from the manual cluster + verification script and append it to the verification record. +- [ ] Confirm the pasted output matches what you actually observed (no editing of results). + +--- + +## Phase 3 Sign-off + +- [ ] All items in Sections 1–6 and Section 8–9 are checked. +- [ ] Section 7 is either fully checked or all items marked `[SKIP — no cluster]`. +- [ ] Section 10 output artifact is saved. +- [ ] Zero test failures across all packages. + +**Phase 3 is complete. You may proceed to Phase 4.** + +--- + +*Checklist version: Phase 3 — created 2026-07-28* diff --git a/docs/phase3.md b/docs/phase3.md new file mode 100644 index 0000000..0f7925e --- /dev/null +++ b/docs/phase3.md @@ -0,0 +1,481 @@ +# Phase 3 -- Telemetry Store & GPU Telemetry Agent + +## Table of Contents + +1. [Overview](#overview) +2. [Architecture](#architecture) +3. [Telemetry Flow](#telemetry-flow) +4. [Data Model](#data-model) +5. [Store Design](#store-design) +6. [Agent Design](#agent-design) +7. [CRD Explanation](#crd-explanation) +8. [Concurrency Model](#concurrency-model) +9. [Staleness Model](#staleness-model) +10. [Failure Handling](#failure-handling) +11. [Testing Guide](#testing-guide) +12. [Future Extension Points](#future-extension-points) +13. [Known Limitations](#known-limitations) +14. [Rollback Notes](#rollback-notes) + +--- + +## Overview + +Phase 3 adds the infrastructure layer between the GPU hardware abstraction +(Phase 1 `GPUStatsProvider`) and the scoring engine (Phase 2 `Score()`). + +It delivers three things: + +1. **Structured data model** (`NodeTelemetry`, `GPUTelemetry`) — a typed, + versioned, deep-copyable snapshot of one node's GPU state. +2. **Thread-safe in-memory store** (`TelemetryStore`) — fast-read, safe for + concurrent access by multiple scheduler goroutines. +3. **Polling agent** (`Agent`) — ticks on a configurable interval, queries + `GPUStatsProvider`, writes snapshots to the store, and sweeps for stale + records. + +Phase 4 will bridge the store to the scoring engine and add the real +Kubernetes CRD write path. + +--- + +## Architecture + +``` + GPUStatsProvider (FakeProvider / NVMLProvider) + | + | interface boundary (Phase 1) + v + pkg/agent.Agent + - ticker: PollInterval (default 10s) + - per tick: PollOnce() -> NodeTelemetry snapshot + - store.Set(snapshot) + - store.MarkStale(StalenessThreshold) + - MetricsHook callbacks + | + v + pkg/store.TelemetryStore + - map[nodeID]NodeTelemetry (sync.RWMutex) + - Set / Get / Delete / List / Snapshot / MarkStale + | + | (Phase 4 adds CRD write path here) + v + pkg/api/v1alpha1.GPUNodeStatus (CRD Go types) + - Spec: nodeName, gpuIDs + - Status: per-GPU entries, lastUpdated, stale, schemaVersion + | + v + manifests/gpunodestatus-crd.yaml + - apiextensions.k8s.io/v1 CRD + - group: gpu.amshithnair.dev + - status subresource enabled +``` + +--- + +## Telemetry Flow + +``` +[1] Agent.Run(ctx) starts ticker + +[2] ticker fires every PollInterval + +[3] Agent.PollOnce(ctx): + for each gpuID in cfg.GPUIDs: + util <- provider.GetUtilization(gpuID) [retried up to MaxRetries] + mem <- provider.GetFreeMemoryMB(gpuID) [retried] + ecc <- provider.GetECCErrorCount(gpuID) [retried] + on error: sentinel value + Known=false + topo <- provider.GetTopology(nodeID) [best-effort, no retry] + assemble NodeTelemetry{CollectedAt: now} + +[4] store.Set(snapshot) -- write-lock, deep copy stored + +[5] store.MarkStale(threshold) -- write-lock, sweeps all records + +[6] MetricsHook.OnStoreWrite() + MetricsHook.OnPollComplete(duration, errCount) + +[7] repeat from [2] until ctx.Done() + +[8] ctx cancelled: + store.MarkStale(1ns) -- all records immediately stale + return ctx.Err() +``` + +--- + +## Data Model + +### `GPUTelemetry` (`pkg/telemetry/model.go`) + +| Field | Type | Description | +|-------|------|-------------| +| `GPUID` | `string` | GPU identifier (matches provider key) | +| `UtilizationPct` | `float64` | Compute utilisation [0.0-1.0]; -1 if unknown | +| `UtilizationKnown` | `bool` | False if provider returned an error | +| `FreeMemoryMB` | `int64` | Free memory in MB; -1 if unknown | +| `TotalMemoryMB` | `int64` | Total memory in MB; -1 if unknown | +| `MemoryKnown` | `bool` | False if provider returned an error | +| `ECCErrors` | `uint64` | Cumulative ECC error count | +| `ECCKnown` | `bool` | False if provider returned an error | + +**Sentinel constants:** +- `UnknownUtilization = -1.0` -- outside [0,1], distinguishable from idle +- `UnknownMemory = -1` -- negative, never a valid MB value + +### `NodeTelemetry` + +| Field | Type | Description | +|-------|------|-------------| +| `NodeID` | `string` | Kubernetes node name (store key) | +| `SchemaVersion` | `int` | Bumped on breaking struct changes | +| `GPUs` | `[]GPUTelemetry` | One entry per configured GPU | +| `Topology` | `TopologyMatrix` | Inter-GPU topology (may be zero) | +| `CollectedAt` | `time.Time` | Wall clock at snapshot assembly | +| `Stale` | `bool` | Set by store sweep, never by agent | + +### Key methods + +```go +n.Validate() error // guards invariants +n.DeepCopy() NodeTelemetry // full independence guarantee +n.IsStale(threshold time.Duration) bool // uses time.Now() +n.IsStaleAt(now time.Time, threshold) bool // deterministic, test-friendly +``` + +--- + +## Store Design + +`TelemetryStore` wraps `map[string]NodeTelemetry` behind `sync.RWMutex`. + +### API + +```go +func NewTelemetryStore() *TelemetryStore +func (s *TelemetryStore) Set(t NodeTelemetry) +func (s *TelemetryStore) Get(nodeID string) (NodeTelemetry, bool) +func (s *TelemetryStore) Delete(nodeID string) +func (s *TelemetryStore) List() []NodeTelemetry +func (s *TelemetryStore) Snapshot() map[string]NodeTelemetry +func (s *TelemetryStore) Len() int +func (s *TelemetryStore) MarkStale(threshold time.Duration) +func (s *TelemetryStore) MarkStaleAt(now time.Time, threshold time.Duration) +``` + +### Deep-copy contract + +Every value leaving the store (`Get`, `List`, `Snapshot`) is a deep copy. +Every value entering the store (`Set`) is deep-copied before storage. +No caller can alias internal store data. + +### Lock granularity + +| Operation | Lock | +|-----------|------| +| `Set`, `Delete`, `MarkStale`, `MarkStaleAt` | `Lock()` (exclusive write) | +| `Get`, `List`, `Snapshot`, `Len` | `RLock()` (shared read) | + +The lock is released before returning from every method; callers never hold +the store's lock. + +--- + +## Agent Design + +### Configuration (`AgentConfig`) + +| Field | Default | Description | +|-------|---------|-------------| +| `NodeID` | required | Kubernetes node name | +| `GPUIDs` | required | GPU IDs to poll | +| `PollInterval` | 10s | Time between poll cycles | +| `StalenessThreshold` | 3x PollInterval | Age threshold for stale marking | +| `MaxRetries` | 3 | Per-GPU retry attempts | +| `RetryDelay` | 500ms | Pause between retries | +| `Metrics` | nil (no-op) | Hook callbacks | + +### MetricsHook + +```go +type MetricsHook struct { + OnPollComplete func(duration time.Duration, errCount int) + OnStoreWrite func(nodeID string) +} +``` + +All callbacks are optional (nil = skip). They are invoked synchronously and +must be non-blocking. Phase 4 can connect them to Prometheus counters. + +### Error handling policy + +| Error source | Behaviour | +|---|---| +| Single GPU, single signal | Retry up to `MaxRetries` with `RetryDelay` | +| Single GPU, all retries exhausted | Sentinel value + `Known=false` for that signal | +| All GPUs fail | Snapshot written with all `Known=false`; still written to store | +| Topology fails | Zero `TopologyMatrix`; not counted in `errCount` | +| Context cancelled mid-poll | Conservative defaults for remaining GPUs; snapshot written | + +--- + +## CRD Explanation + +### Identity + +``` +apiVersion: apiextensions.k8s.io/v1 +group: gpu.amshithnair.dev +version: v1alpha1 +kind: GPUNodeStatus +plural: gpunodestatuses +short: gpuns +scope: Namespaced +``` + +### Status subresource + +The `.status` subresource is enabled. This means: + +- `kubectl apply` / `kubectl edit` can only update `.spec`. +- Only the telemetry agent (using the status subresource endpoint + `/apis/gpu.amshithnair.dev/v1alpha1/namespaces/*/gpunodestatuses/*/status`) + can write `.status`. +- The scheduler plugin reads `.status` via the informer cache. + +### Namespace convention + +GPUNodeStatus objects are created in the `gpu-scheduler` namespace by +convention. All `kubectl` commands in the manual scripts use +`-n gpu-scheduler`. + +### Example object + +```yaml +apiVersion: gpu.amshithnair.dev/v1alpha1 +kind: GPUNodeStatus +metadata: + name: node-gpu-01 + namespace: gpu-scheduler +spec: + nodeName: node-gpu-01 + gpuIDs: + - gpu-0 + - gpu-1 +status: + schemaVersion: 1 + stale: false + lastUpdated: "2024-01-15T10:30:00Z" + gpus: + - gpuID: gpu-0 + utilizationPct: 0.23 + utilizationKnown: true + freeMemoryMB: 71680 + totalMemoryMB: 81920 + memoryKnown: true + eccErrors: 0 + eccKnown: true + - gpuID: gpu-1 + utilizationPct: 0.81 + utilizationKnown: true + freeMemoryMB: 20480 + totalMemoryMB: 81920 + memoryKnown: true + eccErrors: 3 + eccKnown: true +``` + +--- + +## Concurrency Model + +``` +Writer goroutine (Agent.Run): + store.Set() -- Lock / Unlock + store.MarkStale() -- Lock / Unlock + +Reader goroutines (Phase 4 scheduler, N concurrent): + store.Get() -- RLock / RUnlock + store.List() -- RLock / RUnlock + store.Snapshot() -- RLock / RUnlock +``` + +Key properties: + +- Multiple scheduler goroutines can read concurrently without blocking each + other (`sync.RWMutex` allows concurrent `RLock`). +- A single writer (agent) never starves readers on modern Go runtimes (Go's + `sync.RWMutex` is write-preferring to prevent writer starvation). +- Returned values are deep copies; the caller holds no lock reference. +- No `sync/atomic`, no channels, no `unsafe` -- just a well-understood mutex. + +--- + +## Staleness Model + +``` +Stale = time.Since(record.CollectedAt) > StalenessThreshold +``` + +| Component | Role | +|-----------|------| +| `NodeTelemetry.CollectedAt` | Set by agent at snapshot time (`time.Now()`) | +| `NodeTelemetry.Stale` | Set by store sweep; never set by agent | +| `NodeTelemetry.IsStaleAt(now, threshold)` | Pure helper used by `MarkStale` | +| `TelemetryStore.MarkStale(threshold)` | Called by agent after every tick | +| `AgentConfig.StalenessThreshold` | Default: 3 x PollInterval | + +**Lifecycle of a stale record:** + +``` +[agent polls] --> record.Stale = false, CollectedAt = now +[time passes -- agent is healthy] --> MarkStale leaves record.Stale = false +[agent stops reporting] --> time.Since(CollectedAt) grows +[threshold exceeded] --> MarkStale sets record.Stale = true +[Phase 4 scorer reads] --> NodeScoreInput.Stale = true --> Score() = 0.0 +``` + +**Shutdown behaviour:** + +When `ctx` is cancelled, `Run()` calls `store.MarkStale(1)` (1ns threshold), +which marks all records stale regardless of age. This ensures the scheduler +immediately stops favouring a node whose agent has shut down. + +--- + +## Failure Handling + +| Scenario | Result | +|----------|--------| +| Provider error, one GPU, one signal | Retry; if all retries fail: sentinel + Known=false | +| Provider error, all GPUs | Full snapshot with all Known=false; CollectedAt is fresh | +| Agent crashes (process dies) | Records remain; become stale after StalenessThreshold; scorer returns 0 | +| Agent restart | New fresh records overwrite stale ones; Stale resets to false | +| Store.Get unknown node | Returns zero NodeTelemetry, ok=false; no panic | +| Concurrent Set + Delete on same node | Last writer wins; mutex prevents corruption | + +--- + +## Testing Guide + +### Run all Phase 3 tests + +```powershell +# Windows +go test ./pkg/telemetry/... ./pkg/store/... ./pkg/agent/... -v -count=1 -cover +``` + +```bash +# Linux / macOS (race detector available) +go test -race ./pkg/telemetry/... ./pkg/store/... ./pkg/agent/... -v -count=1 -cover +``` + +### Coverage by package + +| Package | Test file | Key scenarios | +|---------|-----------|---------------| +| `pkg/telemetry` | `model_test.go` | Validate happy/sad, DeepCopy independence (GPU slice, topology maps, nil), IsStale boundary | +| `pkg/store` | `store_test.go` | CRUD, Len, List/Snapshot deep-copy, MarkStaleAt mixed records, all concurrent tests | +| `pkg/agent` | `agent_test.go` | NewAgent validation, PollOnce happy/error paths, context cancellation, MetricsHook, staleness, multi-GPU independence, concurrent readers race test | + +### Verify race safety on Linux + +```bash +go test -race ./pkg/store/... ./pkg/agent/... -count=1 +``` + +Expected: `ok ... [no race conditions]` + +### CRD dry-run validation + +```bash +# Requires kubectl pointed at any cluster or --dry-run=client +kubectl apply --dry-run=client -f manifests/gpunodestatus-crd.yaml +``` + +Expected: `customresourcedefinition.apiextensions.k8s.io/gpunodestatuses.gpu.amshithnair.dev configured (dry run)` + +--- + +## Future Extension Points + +### 1. CRD write path (Phase 4) + +`Agent.runOneTick` currently writes only to the in-memory store. +Phase 4 will add a `CRDWriter` interface so the agent can also patch +the `GPUNodeStatus.Status` subresource via `client-go`: + +```go +type CRDWriter interface { + UpdateStatus(ctx context.Context, t NodeTelemetry) error +} +``` + +The agent already has all the data; only the write path needs adding. + +### 2. Prometheus metrics + +`MetricsHook.OnPollComplete` and `OnStoreWrite` accept function values. +Phase 4 can inject Prometheus counter/histogram implementations: + +```go +cfg.Metrics = agent.MetricsHook{ + OnPollComplete: func(d time.Duration, errCount int) { + pollDuration.Observe(d.Seconds()) + if errCount > 0 { pollErrors.Add(float64(errCount)) } + }, +} +``` + +### 3. Adaptive staleness threshold + +`StalenessThreshold` is a scalar constant. It could become a function of +observed poll jitter — widen the threshold when the agent is running slowly, +tighten it under normal conditions. + +### 4. Priority-queue staleness sweep + +`MarkStale` currently O(n). For clusters > 1 000 GPU nodes, replace the +full scan with a min-heap sorted by `CollectedAt`. O(log n) per tick. + +### 5. Multi-node agent + +Current design: one `Agent` per node (DaemonSet pod). If a cluster needs +a centralised agent, `AgentConfig.GPUIDs` can carry cross-node GPU IDs and +the store key (`NodeID`) disambiguates them. No API change required. + +--- + +## Known Limitations + +| Item | Detail | +|------|--------| +| `TotalMemoryMB` = `FreeMemoryMB` | Phase 3 lacks a `GetTotalMemoryMB` provider method. Phase 4 will add it. Until then, the memory ratio is 1.0 (all free), which is conservative-optimistic. | +| Race detector on Windows | `go test -race` requires CGo, disabled by default on Windows without MinGW. Race safety is enforced by design (`sync.RWMutex`) and verified on Linux CI. | +| No CRD write path yet | Phase 3 stores telemetry in memory only. Phase 4 adds `client-go` + status subresource update. | +| No persistent telemetry | A process restart clears the store. Phase 4's CRD path provides durability via `etcd`. | + +--- + +## Rollback Notes + +Phase 3 adds new packages and files only. No Phase 1 or Phase 2 files were modified. + +To roll back Phase 3 entirely: + +```bash +rm -rf pkg/telemetry/model.go +rm -rf pkg/telemetry/model_test.go +rm -rf pkg/api/ +rm -rf pkg/store/ +rm -rf pkg/agent/ +rm -rf manifests/ +rm -f docs/phase3.md +rm -f scripts/phase3-test.ps1 +rm -f scripts/phase3-test.sh +rm -f scripts/phase3-manual.ps1 +rm -f scripts/phase3-manual.sh +``` + +`go build ./...` will pass cleanly after removal. Phase 1 and Phase 2 are unaffected. diff --git a/scripts/phase3-manual.ps1 b/scripts/phase3-manual.ps1 new file mode 100644 index 0000000..ce2cfeb --- /dev/null +++ b/scripts/phase3-manual.ps1 @@ -0,0 +1,261 @@ +#Requires -Version 5.1 +<# +.SYNOPSIS + Phase 3 manual cluster verification script (Windows / PowerShell). + +.DESCRIPTION + Verifies the GPUNodeStatus CRD on a live kind or real Kubernetes cluster. + Applies the CRD, creates a sample object, patches its status, then cleans up. + + IMPORTANT: Only run against a TEST cluster, not production. + +.NOTES + Run from the repository root: + .\scripts\phase3-manual.ps1 +#> + +$ErrorActionPreference = "Stop" + +$RepoRoot = Split-Path -Parent $PSScriptRoot +Set-Location $RepoRoot + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +function Write-Section { + param([string]$Text) + Write-Host "" + Write-Host ("=" * 64) -ForegroundColor Cyan + Write-Host " $Text" -ForegroundColor Cyan + Write-Host ("=" * 64) -ForegroundColor Cyan +} + +function Write-Step { + param([string]$Number, [string]$Description) + Write-Host "" + Write-Host "[$Number] $Description" -ForegroundColor Yellow + Write-Host ("-" * 48) -ForegroundColor DarkGray +} + +function Assert-ExitCode { + param([string]$Label) + if ($LASTEXITCODE -ne 0) { + Write-Host "[FAIL] $Label (exit code $LASTEXITCODE)" -ForegroundColor Red + exit 1 + } + Write-Host "[PASS] $Label" -ForegroundColor Green +} + +# --------------------------------------------------------------------------- +# Pre-flight +# --------------------------------------------------------------------------- + +Write-Section "Phase 3 -- Manual Cluster Verification" +Write-Host " Working dir : $RepoRoot" +Write-Host "" + +if (-not (Get-Command kubectl -ErrorAction SilentlyContinue)) { + Write-Host "[SKIP] kubectl not found on PATH." -ForegroundColor Yellow + Write-Host " Install kubectl and configure it, then re-run." -ForegroundColor Yellow + exit 0 +} + +Write-Host " kubectl : $(Get-Command kubectl | Select-Object -ExpandProperty Source)" + +$ctx = kubectl config current-context 2>&1 +if ($LASTEXITCODE -ne 0) { + Write-Host "" + Write-Host "[SKIP] No kubectl context configured. Create a kind cluster first:" -ForegroundColor Yellow + Write-Host " kind create cluster --name gpu-scheduler-dev" -ForegroundColor Cyan + exit 0 +} + +Write-Host " context : $ctx" +Write-Host "" +Write-Host " WARNING: This script applies and removes a CRD from your cluster." -ForegroundColor Yellow +Write-Host " Only run against a TEST cluster, not production." -ForegroundColor Yellow +Write-Host "" +Write-Host " Press ENTER to continue or Ctrl+C to abort..." -ForegroundColor Yellow +$null = Read-Host + +# --------------------------------------------------------------------------- +# SECTION A -- CRD lifecycle +# --------------------------------------------------------------------------- + +Write-Section "SECTION A: CRD Lifecycle" + +Write-Step "A1" "Dry-run: validate CRD manifest without applying to cluster" +Write-Host " kubectl apply --dry-run=client -f manifests/gpunodestatus-crd.yaml" +kubectl apply --dry-run=client -f manifests/gpunodestatus-crd.yaml +Assert-ExitCode "A1: CRD dry-run validation" + +Write-Step "A2" "Apply CRD to cluster" +Write-Host " kubectl apply -f manifests/gpunodestatus-crd.yaml" +kubectl apply -f manifests/gpunodestatus-crd.yaml +Assert-ExitCode "A2: CRD applied" + +# Wait briefly for the CRD to be established before creating objects. +Write-Host " Waiting 3s for CRD to be established..." -ForegroundColor DarkGray +Start-Sleep -Seconds 3 + +Write-Step "A3" "Verify CRD is registered" +Write-Host " kubectl get crd gpunodestatuses.gpu.amshithnair.dev" +kubectl get crd gpunodestatuses.gpu.amshithnair.dev +Assert-ExitCode "A3: CRD registered" + +Write-Step "A4" "Describe CRD (schema and printer columns)" +Write-Host " kubectl describe crd gpunodestatuses.gpu.amshithnair.dev" +kubectl describe crd gpunodestatuses.gpu.amshithnair.dev +Assert-ExitCode "A4: CRD described" + +# --------------------------------------------------------------------------- +# SECTION B -- Namespace and sample object +# --------------------------------------------------------------------------- + +Write-Section "SECTION B: Sample GPUNodeStatus Object" + +Write-Step "B1" "Create gpu-scheduler namespace (idempotent)" +Write-Host " kubectl create namespace gpu-scheduler --dry-run=client -o yaml | kubectl apply -f -" +kubectl create namespace gpu-scheduler --dry-run=client -o yaml | kubectl apply -f - +Assert-ExitCode "B1: namespace ready" + +Write-Step "B2" "Create sample GPUNodeStatus object" +$sampleYaml = @" +apiVersion: gpu.amshithnair.dev/v1alpha1 +kind: GPUNodeStatus +metadata: + name: node-gpu-01 + namespace: gpu-scheduler + labels: + gpu-aware-scheduler/phase: "3" +spec: + nodeName: node-gpu-01 + gpuIDs: + - gpu-0 + - gpu-1 +"@ +Write-Host " kubectl apply -f - (inline YAML)" +$sampleYaml | kubectl apply -f - +Assert-ExitCode "B2: GPUNodeStatus created" + +Write-Step "B3" "List GPUNodeStatus objects (printer columns)" +Write-Host " kubectl get gpuns -n gpu-scheduler" +kubectl get gpuns -n gpu-scheduler +Assert-ExitCode "B3: list GPUNodeStatus" + +Write-Step "B4" "Describe the sample GPUNodeStatus" +Write-Host " kubectl describe gpuns node-gpu-01 -n gpu-scheduler" +kubectl describe gpuns node-gpu-01 -n gpu-scheduler +Assert-ExitCode "B4: describe GPUNodeStatus" + +Write-Step "B5" "Patch status subresource (simulates telemetry agent write)" +Write-Host " kubectl patch gpuns node-gpu-01 -n gpu-scheduler --subresource=status --type=merge -f " +# Write patch to a temp file to avoid PowerShell quote-mangling of JSON args. +$patchFile = Join-Path $env:TEMP "gpuns-status-patch.json" +@' +{ + "status": { + "schemaVersion": 1, + "stale": false, + "lastUpdated": "2024-01-15T10:30:00Z", + "gpus": [ + { + "gpuID": "gpu-0", + "utilizationPct": 0.23, + "utilizationKnown": true, + "freeMemoryMB": 71680, + "totalMemoryMB": 81920, + "memoryKnown": true, + "eccErrors": 0, + "eccKnown": true + }, + { + "gpuID": "gpu-1", + "utilizationPct": 0.81, + "utilizationKnown": true, + "freeMemoryMB": 20480, + "totalMemoryMB": 81920, + "memoryKnown": true, + "eccErrors": 3, + "eccKnown": true + } + ] + } +} +'@ | Set-Content -Path $patchFile -Encoding UTF8 +kubectl patch gpuns node-gpu-01 -n gpu-scheduler --subresource=status --type=merge --patch-file $patchFile +Assert-ExitCode "B5: status patched" +Remove-Item $patchFile -ErrorAction SilentlyContinue + +Write-Step "B6" "Read back the full GPUNodeStatus (verify status populated)" +Write-Host " kubectl get gpuns node-gpu-01 -n gpu-scheduler -o yaml" +kubectl get gpuns node-gpu-01 -n gpu-scheduler -o yaml +Assert-ExitCode "B6: status read back" + +# --------------------------------------------------------------------------- +# SECTION C -- Stale simulation +# --------------------------------------------------------------------------- + +Write-Section "SECTION C: Stale Simulation" + +Write-Step "C1" "Patch stale=true (simulates store staleness sweep)" +Write-Host " kubectl patch gpuns node-gpu-01 -n gpu-scheduler --subresource=status --type=merge ..." +$stalePatchFile = Join-Path $env:TEMP "gpuns-stale-patch.json" +'{"status":{"stale":true}}' | Set-Content -Path $stalePatchFile -Encoding UTF8 +kubectl patch gpuns node-gpu-01 -n gpu-scheduler --subresource=status --type=merge --patch-file $stalePatchFile +Assert-ExitCode "C1: stale flag set" +Remove-Item $stalePatchFile -ErrorAction SilentlyContinue + +Write-Step "C2" "Verify stale=true visible in printer columns" +Write-Host " kubectl get gpuns -n gpu-scheduler" +kubectl get gpuns -n gpu-scheduler +Assert-ExitCode "C2: stale visible" + +# --------------------------------------------------------------------------- +# SECTION D -- Cleanup +# --------------------------------------------------------------------------- + +Write-Section "SECTION D: Cleanup" + +Write-Step "D1" "Delete sample GPUNodeStatus" +Write-Host " kubectl delete gpuns node-gpu-01 -n gpu-scheduler --ignore-not-found" +kubectl delete gpuns node-gpu-01 -n gpu-scheduler --ignore-not-found +Assert-ExitCode "D1: sample object deleted" + +Write-Step "D2" "Delete CRD from cluster" +Write-Host " kubectl delete crd gpunodestatuses.gpu.amshithnair.dev --ignore-not-found" +kubectl delete crd gpunodestatuses.gpu.amshithnair.dev --ignore-not-found +Assert-ExitCode "D2: CRD deleted" + +Write-Step "D3" "Verify CRD is gone" +Write-Host " kubectl get crd gpunodestatuses.gpu.amshithnair.dev (expect NotFound)" +# Temporarily allow non-zero exit codes so kubectl's NotFound (exit 1) doesn't +# terminate the script under $ErrorActionPreference = "Stop". +$check = "" +try { + $check = kubectl get crd gpunodestatuses.gpu.amshithnair.dev 2>&1 +} catch { + $check = $_.ToString() +} +if ($check -match "NotFound" -or $check -match "not found") { + Write-Host "[PASS] D3: CRD successfully removed" -ForegroundColor Green +} elseif ($LASTEXITCODE -eq 0) { + Write-Host "[WARN] D3: CRD may still be terminating" -ForegroundColor Yellow + Write-Host $check +} else { + Write-Host "[WARN] D3: unexpected kubectl response" -ForegroundColor Yellow + Write-Host $check +} + +# --------------------------------------------------------------------------- +# Summary +# --------------------------------------------------------------------------- + +Write-Host "" +Write-Host ("=" * 64) -ForegroundColor Green +Write-Host " PHASE 3 MANUAL VERIFICATION COMPLETE" -ForegroundColor Green +Write-Host ("=" * 64) -ForegroundColor Green +Write-Host "" +Write-Host " Paste this output into the Phase 3 verification record." +Write-Host "" diff --git a/scripts/phase3-manual.sh b/scripts/phase3-manual.sh new file mode 100644 index 0000000..4464cdf --- /dev/null +++ b/scripts/phase3-manual.sh @@ -0,0 +1,180 @@ +#!/usr/bin/env bash +# ============================================================================= +# Phase 3 manual cluster verification script -- Linux / macOS +# +# Usage: +# ./scripts/phase3-manual.sh +# +# Prerequisites: +# - A running Kubernetes cluster (kind or real) +# - kubectl configured and pointing at that cluster +# - The CRD manifest at manifests/gpunodestatus-crd.yaml +# +# WARNING: This script applies and then removes a CRD from your cluster. +# Only run against a test cluster, not production. +# ============================================================================= + +set -euo pipefail + +CYAN='\033[0;36m' +GREEN='\033[0;32m' +YELLOW='\033[1;33m' +RED='\033[0;31m' +RESET='\033[0m' + +section() { echo -e "\n${CYAN}================================================================${RESET}"; + echo -e "${CYAN} $*${RESET}"; + echo -e "${CYAN}================================================================${RESET}"; } +step() { echo -e "\n${YELLOW}[$1] $2${RESET}"; + echo -e "${CYAN}------------------------------------------------${RESET}"; } +pass() { echo -e "${GREEN}[PASS]${RESET} $*"; } +fail() { echo -e "${RED}[FAIL]${RESET} $*"; exit 1; } +warn() { echo -e "${YELLOW}[WARN]${RESET} $*"; } + +# --------------------------------------------------------------------------- +# Resolve repo root +# --------------------------------------------------------------------------- + +SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +REPO_ROOT="$(cd "${SCRIPT_DIR}/.." && pwd)" +cd "${REPO_ROOT}" + +# --------------------------------------------------------------------------- +# Pre-flight +# --------------------------------------------------------------------------- + +section "Phase 3 -- Manual Cluster Verification" +echo " Working dir : ${REPO_ROOT}" +echo "" + +if ! command -v kubectl &>/dev/null; then + warn "kubectl not found. Install kubectl and configure it to point at your cluster." + exit 0 +fi + +echo " kubectl : $(kubectl version --client --short 2>/dev/null | head -1 || true)" +echo " context : $(kubectl config current-context)" +echo "" +echo -e "${YELLOW} WARNING: This script applies and then removes a CRD from your${RESET}" +echo -e "${YELLOW} cluster. Only run against a test cluster, not production.${RESET}" +echo "" +read -r -p " Press ENTER to continue or Ctrl+C to abort..." + +# --------------------------------------------------------------------------- +# SECTION A -- CRD lifecycle +# --------------------------------------------------------------------------- + +section "SECTION A: CRD Lifecycle" + +step "A1" "Dry-run: validate CRD manifest without applying to cluster" +echo " Running: kubectl apply --dry-run=client -f manifests/gpunodestatus-crd.yaml" +kubectl apply --dry-run=client -f manifests/gpunodestatus-crd.yaml +pass "CRD manifest is valid" + +step "A2" "Apply CRD to cluster" +kubectl apply -f manifests/gpunodestatus-crd.yaml +pass "CRD applied" + +step "A3" "Verify CRD is registered" +kubectl get crd gpunodestatuses.gpu.amshithnair.dev +pass "CRD registered" + +step "A4" "Describe CRD (inspect schema and printer columns)" +kubectl describe crd gpunodestatuses.gpu.amshithnair.dev +pass "CRD describe complete" + +# --------------------------------------------------------------------------- +# SECTION B -- Namespace and sample object +# --------------------------------------------------------------------------- + +section "SECTION B: Sample GPUNodeStatus Object" + +step "B1" "Create gpu-scheduler namespace (idempotent)" +kubectl create namespace gpu-scheduler --dry-run=client -o yaml | kubectl apply -f - +pass "namespace ready" + +step "B2" "Create sample GPUNodeStatus object" +cat <<'EOF' | kubectl apply -f - +apiVersion: gpu.amshithnair.dev/v1alpha1 +kind: GPUNodeStatus +metadata: + name: node-gpu-01 + namespace: gpu-scheduler + labels: + gpu-aware-scheduler/phase: "3" +spec: + nodeName: node-gpu-01 + gpuIDs: + - gpu-0 + - gpu-1 +EOF +pass "GPUNodeStatus created" + +step "B3" "List GPUNodeStatus objects (printer columns: Node, Stale, LastUpdated)" +kubectl get gpuns -n gpu-scheduler +pass "list complete" + +step "B4" "Describe the sample GPUNodeStatus object" +kubectl describe gpuns node-gpu-01 -n gpu-scheduler +pass "describe complete" + +step "B5" "Patch status subresource (simulates telemetry agent write)" +# The --subresource=status flag is required because status subresource is enabled. +kubectl patch gpuns node-gpu-01 -n gpu-scheduler \ + --subresource=status \ + --type=merge \ + --patch='{"status":{"schemaVersion":1,"stale":false,"lastUpdated":"2024-01-15T10:30:00Z","gpus":[{"gpuID":"gpu-0","utilizationPct":0.23,"utilizationKnown":true,"freeMemoryMB":71680,"totalMemoryMB":81920,"memoryKnown":true,"eccErrors":0,"eccKnown":true},{"gpuID":"gpu-1","utilizationPct":0.81,"utilizationKnown":true,"freeMemoryMB":20480,"totalMemoryMB":81920,"memoryKnown":true,"eccErrors":3,"eccKnown":true}]}}' +pass "status patched" + +step "B6" "Read updated GPUNodeStatus (should show stale=false and GPU data)" +kubectl get gpuns node-gpu-01 -n gpu-scheduler -o yaml +pass "status read back" + +# --------------------------------------------------------------------------- +# SECTION C -- Stale simulation +# --------------------------------------------------------------------------- + +section "SECTION C: Stale Simulation" + +step "C1" "Patch status with stale=true (simulates staleness sweep result)" +kubectl patch gpuns node-gpu-01 -n gpu-scheduler \ + --subresource=status \ + --type=merge \ + --patch='{"status":{"stale":true}}' +pass "stale flag set" + +step "C2" "Verify stale=true is visible" +kubectl get gpuns -n gpu-scheduler +pass "stale visible in printer columns" + +# --------------------------------------------------------------------------- +# SECTION D -- Cleanup +# --------------------------------------------------------------------------- + +section "SECTION D: Cleanup" + +step "D1" "Delete sample GPUNodeStatus" +kubectl delete gpuns node-gpu-01 -n gpu-scheduler --ignore-not-found +pass "sample object deleted" + +step "D2" "Delete CRD from cluster" +kubectl delete crd gpunodestatuses.gpu.amshithnair.dev --ignore-not-found +pass "CRD deleted" + +step "D3" "Verify CRD is gone" +if kubectl get crd gpunodestatuses.gpu.amshithnair.dev &>/dev/null; then + warn "CRD may still be terminating (etcd propagation delay -- check again in a few seconds)" +else + pass "CRD successfully removed" +fi + +# --------------------------------------------------------------------------- +# Summary +# --------------------------------------------------------------------------- + +echo -e "\n${GREEN}================================================================${RESET}" +echo -e "${GREEN} PHASE 3 MANUAL VERIFICATION COMPLETE${RESET}" +echo -e "${GREEN}================================================================${RESET}" +echo "" +echo " Paste this output into the Phase 3 verification record." +echo "" diff --git a/scripts/phase3-test.ps1 b/scripts/phase3-test.ps1 new file mode 100644 index 0000000..b4b840c --- /dev/null +++ b/scripts/phase3-test.ps1 @@ -0,0 +1,243 @@ +#Requires -Version 5.1 +<# +.SYNOPSIS + Phase 3 verification script for the GPU-Aware Scheduler telemetry layer. + +.DESCRIPTION + Runs the full Phase 3 quality gate: + 1. Validate required tools (go, git, kubectl, kind, docker, golangci-lint) + 2. go fmt ./... + 3. go vet ./... + 4. golangci-lint run (SKIP if not installed or SKIP_LINT=1) + 5. go build ./... + 6. go test ./... (all packages) + 7. go test -race ./... (SKIP on Windows where CGO_ENABLED=0) + + Does NOT deploy anything, apply any manifests, or modify any cluster. + +.NOTES + Run from the repository root: + .\scripts\phase3-test.ps1 + + Environment overrides: + $env:SKIP_LINT = "1" bypass golangci-lint + $env:SKIP_RACE = "1" bypass race detector (auto-set on Windows) +#> + +$ErrorActionPreference = "Stop" + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +function Write-Header { + param([string]$Text) + Write-Host "" + Write-Host ("-" * 56) -ForegroundColor Cyan + Write-Host " $Text" -ForegroundColor Cyan + Write-Host ("-" * 56) -ForegroundColor Cyan +} + +function Write-Pass { + param([string]$Step) + Write-Host "[PASS] $Step" -ForegroundColor Green +} + +function Write-Fail { + param([string]$Step, [string]$Detail) + Write-Host "[FAIL] $Step" -ForegroundColor Red + if ($Detail) { Write-Host " $Detail" -ForegroundColor Red } +} + +function Write-Skip { + param([string]$Step, [string]$Reason) + Write-Host "[SKIP] $Step -- $Reason" -ForegroundColor Yellow +} + +# --------------------------------------------------------------------------- +# Resolve repository root +# --------------------------------------------------------------------------- + +$RepoRoot = Split-Path -Parent $PSScriptRoot +Set-Location $RepoRoot + +# --------------------------------------------------------------------------- +# Banner +# --------------------------------------------------------------------------- + +Write-Host "" +Write-Host ("=" * 64) -ForegroundColor Cyan +Write-Host " GPU-Aware Scheduler - Phase 3 Verification" -ForegroundColor Cyan +Write-Host ("=" * 64) -ForegroundColor Cyan + +$goExe = Get-Command go -ErrorAction SilentlyContinue +if (-not $goExe) { + Write-Fail "go binary" "go not found on PATH. Install from https://go.dev/dl/" + exit 1 +} +Write-Host " Go binary : $($goExe.Source)" +Write-Host " Working dir : $RepoRoot" +Write-Host ("=" * 64) -ForegroundColor Cyan + +# --------------------------------------------------------------------------- +# STEP 1 -- Validate tools +# --------------------------------------------------------------------------- + +Write-Header "STEP 1: validate tools" + +$goVersion = go version 2>&1 +Write-Host " go : $goVersion" + +$gitExe = Get-Command git -ErrorAction SilentlyContinue +if ($gitExe) { + Write-Host " git : $(git --version 2>&1)" +} else { + Write-Skip "git" "not found (not required for unit tests)" +} + +$kubectlExe = Get-Command kubectl -ErrorAction SilentlyContinue +if ($kubectlExe) { + $kubectlVersion = kubectl version --client -o json 2>&1 | ConvertFrom-Json -ErrorAction SilentlyContinue + if ($kubectlVersion) { + Write-Host " kubectl : $($kubectlVersion.clientVersion.gitVersion)" + } else { + Write-Host " kubectl : $(kubectl version --client 2>&1 | Select-Object -First 1)" + } +} else { + Write-Skip "kubectl" "not found -- CRD dry-run not available (run phase3-manual.ps1 separately)" +} + +$kindExe = Get-Command kind -ErrorAction SilentlyContinue +if ($kindExe) { + Write-Host " kind : $(kind version 2>&1)" +} else { + Write-Skip "kind" "not found -- integration cluster not available" +} + +$dockerExe = Get-Command docker -ErrorAction SilentlyContinue +if ($dockerExe) { + Write-Host " docker : $(docker --version 2>&1)" +} else { + Write-Skip "docker" "not found -- kind cluster creation not available" +} + +$lintExe = Get-Command golangci-lint -ErrorAction SilentlyContinue +$skipLint = ($env:SKIP_LINT -eq "1") -or (-not $lintExe) +if ($lintExe) { + Write-Host " golangci-lint: $(golangci-lint version 2>&1 | Select-Object -First 1)" +} else { + Write-Skip "golangci-lint" "not found -- install from https://golangci-lint.run/usage/install/" +} + +# Race detector requires CGO -- auto-detect. +$cgoEnabled = go env CGO_ENABLED 2>&1 +$skipRace = ($env:SKIP_RACE -eq "1") -or ($cgoEnabled.Trim() -eq "0") +if ($skipRace) { + Write-Skip "race detector" "CGO_ENABLED=0 (Windows without C compiler). Race safety enforced by sync.RWMutex design; verify with 'go test -race' on Linux." +} + +Write-Pass "STEP 1: validate tools" + +# --------------------------------------------------------------------------- +# STEP 2 -- go fmt +# --------------------------------------------------------------------------- + +Write-Header "STEP 2: go fmt ./..." +$fmtOut = go fmt ./... 2>&1 +if ($LASTEXITCODE -ne 0) { + Write-Host $fmtOut + Write-Fail "STEP 2: go fmt ./..." "exit code $LASTEXITCODE" + exit 1 +} +if ($fmtOut) { + Write-Host $fmtOut + Write-Fail "STEP 2: go fmt ./..." "go fmt reformatted files -- run 'go fmt ./...' locally" + exit 1 +} +Write-Pass "STEP 2: go fmt ./..." + +# --------------------------------------------------------------------------- +# STEP 3 -- go vet +# --------------------------------------------------------------------------- + +Write-Header "STEP 3: go vet ./..." +$vetOut = go vet ./... 2>&1 +if ($LASTEXITCODE -ne 0) { + Write-Host $vetOut + Write-Fail "STEP 3: go vet ./..." "vet reported issues" + exit 1 +} +if ($vetOut) { Write-Host $vetOut } +Write-Pass "STEP 3: go vet ./..." + +# --------------------------------------------------------------------------- +# STEP 4 -- golangci-lint +# --------------------------------------------------------------------------- + +Write-Header "STEP 4: golangci-lint run" +if ($skipLint) { + Write-Skip "STEP 4: golangci-lint run" "binary not found or SKIP_LINT=1" +} else { + $lintOut = golangci-lint run ./... 2>&1 + if ($LASTEXITCODE -ne 0) { + Write-Host $lintOut + Write-Fail "STEP 4: golangci-lint run" "linter reported issues" + exit 1 + } + if ($lintOut) { Write-Host $lintOut } + Write-Pass "STEP 4: golangci-lint run" +} + +# --------------------------------------------------------------------------- +# STEP 5 -- go build +# --------------------------------------------------------------------------- + +Write-Header "STEP 5: go build ./..." +$buildOut = go build ./... 2>&1 +if ($LASTEXITCODE -ne 0) { + Write-Host $buildOut + Write-Fail "STEP 5: go build ./..." "build failed" + exit 1 +} +if ($buildOut) { Write-Host $buildOut } +Write-Pass "STEP 5: go build ./..." + +# --------------------------------------------------------------------------- +# STEP 6 -- go test ./... +# --------------------------------------------------------------------------- + +Write-Header "STEP 6: go test ./... -v -count=1 -cover" +go test ./... -v -count=1 -cover +if ($LASTEXITCODE -ne 0) { + Write-Fail "STEP 6: go test ./... -v -count=1 -cover" "one or more tests failed" + exit 1 +} +Write-Pass "STEP 6: go test ./... -v -count=1 -cover" + +# --------------------------------------------------------------------------- +# STEP 7 -- go test -race +# --------------------------------------------------------------------------- + +Write-Header "STEP 7: go test -race ./..." +if ($skipRace) { + Write-Skip "STEP 7: go test -race ./..." "CGO_ENABLED=0 -- race detector unavailable on this platform" +} else { + go test -race ./... -count=1 + if ($LASTEXITCODE -ne 0) { + Write-Fail "STEP 7: go test -race ./..." "race conditions detected" + exit 1 + } + Write-Pass "STEP 7: go test -race ./..." +} + +# --------------------------------------------------------------------------- +# Summary +# --------------------------------------------------------------------------- + +Write-Host "" +Write-Host ("=" * 64) -ForegroundColor Green +Write-Host " ALL PHASE 3 CHECKS PASSED" -ForegroundColor Green +Write-Host ("=" * 64) -ForegroundColor Green +Write-Host "" +Write-Host " Next step: run .\scripts\phase3-manual.ps1 for cluster verification." +Write-Host "" diff --git a/scripts/phase3-test.sh b/scripts/phase3-test.sh new file mode 100644 index 0000000..95906e3 --- /dev/null +++ b/scripts/phase3-test.sh @@ -0,0 +1,170 @@ +#!/usr/bin/env bash +# ============================================================================= +# Phase 3 verification script -- GPU-Aware Scheduler telemetry layer +# +# Usage: +# ./scripts/phase3-test.sh +# +# Environment overrides: +# SKIP_LINT=1 bypass golangci-lint +# SKIP_RACE=1 bypass race detector +# ============================================================================= + +set -euo pipefail + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +CYAN='\033[0;36m' +GREEN='\033[0;32m' +YELLOW='\033[1;33m' +RED='\033[0;31m' +RESET='\033[0m' + +header() { echo -e "\n${CYAN}--------------------------------------------------------${RESET}"; + echo -e "${CYAN} $*${RESET}"; + echo -e "${CYAN}--------------------------------------------------------${RESET}"; } +pass() { echo -e "${GREEN}[PASS]${RESET} $*"; } +fail() { echo -e "${RED}[FAIL]${RESET} $*"; exit 1; } +skip() { echo -e "${YELLOW}[SKIP]${RESET} $*"; } + +# --------------------------------------------------------------------------- +# Resolve repo root +# --------------------------------------------------------------------------- + +SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +REPO_ROOT="$(cd "${SCRIPT_DIR}/.." && pwd)" +cd "${REPO_ROOT}" + +# --------------------------------------------------------------------------- +# Banner +# --------------------------------------------------------------------------- + +echo -e "\n${CYAN}================================================================${RESET}" +echo -e "${CYAN} GPU-Aware Scheduler -- Phase 3 Verification${RESET}" +echo -e "${CYAN}================================================================${RESET}" +echo " Working dir : ${REPO_ROOT}" + +# --------------------------------------------------------------------------- +# STEP 1 -- Validate tools +# --------------------------------------------------------------------------- + +header "STEP 1: validate tools" + +if ! command -v go &>/dev/null; then + fail "go not found. Install from https://go.dev/dl/" +fi +echo " go : $(go version)" + +if command -v git &>/dev/null; then + echo " git : $(git --version)" +else + skip "git -- not found (not required for unit tests)" +fi + +if command -v kubectl &>/dev/null; then + echo " kubectl : $(kubectl version --client --short 2>/dev/null | head -1 || true)" +else + skip "kubectl -- not found. CRD dry-run unavailable (run phase3-manual.sh separately)" +fi + +if command -v kind &>/dev/null; then + echo " kind : $(kind version)" +else + skip "kind -- not found. Integration cluster unavailable." +fi + +if command -v docker &>/dev/null; then + echo " docker : $(docker --version)" +else + skip "docker -- not found. kind cluster creation unavailable." +fi + +SKIP_LINT="${SKIP_LINT:-0}" +if command -v golangci-lint &>/dev/null; then + echo " golangci-lint: $(golangci-lint version 2>&1 | head -1)" +else + SKIP_LINT=1 + skip "golangci-lint -- not found. Install: https://golangci-lint.run/usage/install/" +fi + +SKIP_RACE="${SKIP_RACE:-0}" +CGO_ENABLED_VAL="$(go env CGO_ENABLED)" +if [ "${CGO_ENABLED_VAL}" = "0" ]; then + SKIP_RACE=1 + skip "race detector -- CGO_ENABLED=0. Race safety enforced by sync.RWMutex; verify with 'go test -race' on Linux." +fi + +pass "STEP 1: validate tools" + +# --------------------------------------------------------------------------- +# STEP 2 -- go fmt +# --------------------------------------------------------------------------- + +header "STEP 2: go fmt ./..." +FMT_OUT="$(go fmt ./... 2>&1)" +if [ -n "${FMT_OUT}" ]; then + echo "${FMT_OUT}" + fail "go fmt reformatted files. Run 'go fmt ./...' locally before re-running." +fi +pass "STEP 2: go fmt ./..." + +# --------------------------------------------------------------------------- +# STEP 3 -- go vet +# --------------------------------------------------------------------------- + +header "STEP 3: go vet ./..." +go vet ./... +pass "STEP 3: go vet ./..." + +# --------------------------------------------------------------------------- +# STEP 4 -- golangci-lint +# --------------------------------------------------------------------------- + +header "STEP 4: golangci-lint run" +if [ "${SKIP_LINT}" = "1" ]; then + skip "STEP 4: golangci-lint run -- binary not found or SKIP_LINT=1" +else + golangci-lint run ./... + pass "STEP 4: golangci-lint run" +fi + +# --------------------------------------------------------------------------- +# STEP 5 -- go build +# --------------------------------------------------------------------------- + +header "STEP 5: go build ./..." +go build ./... +pass "STEP 5: go build ./..." + +# --------------------------------------------------------------------------- +# STEP 6 -- go test ./... +# --------------------------------------------------------------------------- + +header "STEP 6: go test ./... -v -count=1 -cover" +go test ./... -v -count=1 -cover +pass "STEP 6: go test ./... -v -count=1 -cover" + +# --------------------------------------------------------------------------- +# STEP 7 -- go test -race +# --------------------------------------------------------------------------- + +header "STEP 7: go test -race ./..." +if [ "${SKIP_RACE}" = "1" ]; then + skip "STEP 7: go test -race ./... -- CGO_ENABLED=0 or SKIP_RACE=1" +else + go test -race ./... -count=1 + pass "STEP 7: go test -race ./..." +fi + +# --------------------------------------------------------------------------- +# Summary +# --------------------------------------------------------------------------- + +echo -e "\n${GREEN}================================================================${RESET}" +echo -e "${GREEN} ALL PHASE 3 CHECKS PASSED${RESET}" +echo -e "${GREEN}================================================================${RESET}" +echo "" +echo " Next step: run ./scripts/phase3-manual.sh for cluster verification." +echo ""