diff --git a/src/frontend/advisor/go.mod b/src/frontend/advisor/go.mod index a4170a68..aab476f8 100644 --- a/src/frontend/advisor/go.mod +++ b/src/frontend/advisor/go.mod @@ -4,12 +4,9 @@ go 1.25.5 replace schedulerclient => ../schedulerclient -require ( - schedulerclient v0.0.0-00010101000000-000000000000 - gopkg.in/yaml.v3 v3.0.1 -) +require schedulerclient v0.0.0-00010101000000-000000000000 require ( - resty.dev/v3 v3.0.0-beta.2 // indirect golang.org/x/net v0.33.0 // indirect + resty.dev/v3 v3.0.0-beta.2 // indirect ) diff --git a/src/frontend/advisor/go.sum b/src/frontend/advisor/go.sum new file mode 100644 index 00000000..a5fed1cd --- /dev/null +++ b/src/frontend/advisor/go.sum @@ -0,0 +1,4 @@ +golang.org/x/net v0.33.0 h1:74SYHlV8BIgHIFC/LrYkOGIwL19eTYXQ5wc6TBuO36I= +golang.org/x/net v0.33.0/go.mod h1:HXLR5J+9DxmrqMwG9qjGCxZ+zKXxBru04zlTvWlWuN4= +resty.dev/v3 v3.0.0-beta.2 h1:xu4mGAdbCLuc3kbk7eddWfWm4JfhwDtdapwss5nCjnQ= +resty.dev/v3 v3.0.0-beta.2/go.mod h1:OgkqiPvTDtOuV4MGZuUDhwOpkY8enjOsjjMzeOHefy4= diff --git a/src/frontend/advisor/internal/detect/detect.go b/src/frontend/advisor/internal/detect/detect.go new file mode 100644 index 00000000..b7b7fe67 --- /dev/null +++ b/src/frontend/advisor/internal/detect/detect.go @@ -0,0 +1,212 @@ +// Package detect holds the advisor's deterministic triggers — cheap, no +// LLM. A detector looks at one snapshot and decides whether a scenario is +// worth a provider round: threshold, then cooldown (don't re-evaluate the +// same scenario too often), then dedup (don't re-evaluate an identical +// situation at all). +package detect + +import ( + "crypto/sha256" + "encoding/hex" + "encoding/json" + "time" + + "advisor/internal/snapshot" +) + +// Scenario names, shared by detector, provider, and log. +const ( + ScenarioPendStall = "pend_stall" + ScenarioWorkerHealth = "worker_health" + ScenarioUtilSkew = "util_skew" + ScenarioExitBurst = "exit_burst" +) + +// Trigger is one fired detection. +type Trigger struct { + Scenario string + // DedupKey hashes the situation (job set + reason set / node set) so an + // unchanged situation is not re-evaluated on every scan. Two rounds with + // the same key produce the second finding only after the cooldown. + DedupKey string + // Fingerprint is a human-readable summary of what matched (logged, and + // what the echo provider re-derives from the state). + Fingerprint string +} + +// Limits configures the detectors. +type Limits struct { + // PendStallAfter: PEND wait counted as stalled (default 5m). + PendStallAfter time.Duration + // PendSpikeMin: absolute PEND count growth that fires without waits. + PendSpikeMin int + // ExitBurstMin: non-zero exits within the window that fire. + ExitBurstMin int + // Cooldowns per scenario; a scenario on cooldown is skipped entirely. + Cooldowns map[string]time.Duration +} + +// DefaultLimits returns the P0 defaults. +func DefaultLimits() Limits { + return Limits{ + PendStallAfter: 5 * time.Minute, + PendSpikeMin: 5, + ExitBurstMin: 3, + Cooldowns: map[string]time.Duration{ + ScenarioPendStall: 10 * time.Minute, + ScenarioWorkerHealth: 30 * time.Minute, + ScenarioUtilSkew: 30 * time.Minute, + ScenarioExitBurst: 30 * time.Minute, + }, + } +} + +// Detector evaluates snapshots against the limits. Stateful: tracks +// per-scenario cooldown and the last dedup key seen. +type Detector struct { + lim Limits + now func() time.Time // injectable clock for tests + + lastFired map[string]time.Time + lastKey map[string]string +} + +// NewDetector returns a Detector with the given limits. +func NewDetector(lim Limits) *Detector { + return &Detector{ + lim: lim, + now: time.Now, + lastFired: map[string]time.Time{}, + lastKey: map[string]string{}, + } +} + +// Detect runs all detectors over the snapshot and returns the scenarios to +// evaluate this round, in stable order. +func (d *Detector) Detect(st *snapshot.State) []Trigger { + var out []Trigger + if t, ok := d.pendStall(st); ok { + out = append(out, t) + } + if t, ok := d.workerHealth(st); ok { + out = append(out, t) + } + if t, ok := d.utilSkew(st); ok { + out = append(out, t) + } + if t, ok := d.exitBurst(st); ok { + out = append(out, t) + } + return out +} + +// fire applies cooldown + dedup; ok=true means the scenario should be +// evaluated this round and the state recorded. +func (d *Detector) fire(scenario, key, fingerprint string) (Trigger, bool) { + if cd, ok := d.lim.Cooldowns[scenario]; ok { + if last, fired := d.lastFired[scenario]; fired && d.now().Sub(last) < cd { + return Trigger{}, false + } + } + // Identical situation as last fired round: skip regardless of cooldown. + if d.lastKey[scenario] == key { + return Trigger{}, false + } + d.lastFired[scenario] = d.now() + d.lastKey[scenario] = key + return Trigger{Scenario: scenario, DedupKey: key, Fingerprint: fingerprint}, true +} + +func (d *Detector) pendStall(st *snapshot.State) (Trigger, bool) { + if len(st.PendTop) == 0 && st.Cluster.PendingJobs < d.lim.PendSpikeMin { + return Trigger{}, false + } + // Key: the stalled job set and their reasons. + key := hashOf(reasonKey(st.PendTop)) + fp := "pending=" + itoa(st.Cluster.PendingJobs) + " stalled=" + itoa(len(st.PendTop)) + return d.fire(ScenarioPendStall, key, fp) +} + +func (d *Detector) workerHealth(st *snapshot.State) (Trigger, bool) { + var bad []string + for _, n := range st.Nodes { + switch n.Status { + case "down", "unknown", "unavail", "closed": + bad = append(bad, n.Name+":"+n.Status) + } + } + if len(bad) == 0 { + return Trigger{}, false + } + key := hashOf(join(bad, ",")) + return d.fire(ScenarioWorkerHealth, key, join(bad, ",")) +} + +func (d *Detector) utilSkew(st *snapshot.State) (Trigger, bool) { + var hot, idle []string + for _, n := range st.Nodes { + if n.CoresTotal < 4 { + continue + } + switch { + case n.Utilization >= 0.8: + hot = append(hot, n.Name) + case n.Utilization <= 0.2: + idle = append(idle, n.Name) + } + } + if len(hot) == 0 || len(idle) == 0 { + return Trigger{}, false + } + key := hashOf("hot=" + join(hot, ",") + ";idle=" + join(idle, ",")) + fp := "hot=[" + join(hot, ",") + "] idle=[" + join(idle, ",") + "]" + return d.fire(ScenarioUtilSkew, key, fp) +} + +func (d *Detector) exitBurst(st *snapshot.State) (Trigger, bool) { + var ids []string + for _, x := range st.RecentExits { + if x.ExitCode != 0 { + ids = append(ids, x.ID+":"+itoa(x.ExitCode)) + } + } + if len(ids) < d.lim.ExitBurstMin { + return Trigger{}, false + } + key := hashOf(join(ids, ",")) + return d.fire(ScenarioExitBurst, key, join(ids, ",")) +} + +// reasonKey builds a stable string from the pending set: job id + reason. +func reasonKey(pend []snapshot.JobState) string { + parts := make([]string, 0, len(pend)) + for _, j := range pend { + r := "-" + if j.Reason != nil { + r = j.Reason.MainReasonName + } + parts = append(parts, j.ID+"="+r) + } + return join(parts, "|") +} + +func hashOf(s string) string { + sum := sha256.Sum256([]byte(s)) + return hex.EncodeToString(sum[:8]) +} + +func join(parts []string, sep string) string { + out := "" + for i, p := range parts { + if i > 0 { + out += sep + } + out += p + } + return out +} + +func itoa(i int) string { + b, _ := json.Marshal(i) + return string(b) +} diff --git a/src/frontend/advisor/internal/detect/detect_test.go b/src/frontend/advisor/internal/detect/detect_test.go new file mode 100644 index 00000000..e1dc4c43 --- /dev/null +++ b/src/frontend/advisor/internal/detect/detect_test.go @@ -0,0 +1,143 @@ +package detect + +import ( + "testing" + "time" + + "advisor/internal/snapshot" +) + +func healthyState() *snapshot.State { + return &snapshot.State{ + Cluster: snapshot.ClusterSum{PendingJobs: 0, RunningJobs: 2, Utilization: 0.5}, + Nodes: []snapshot.NodeState{ + {Name: "t1", Status: "busy", CoresTotal: 8, CoresUsed: 4, Utilization: 0.5}, + {Name: "t2", Status: "busy", CoresTotal: 8, CoresUsed: 4, Utilization: 0.5}, + }, + } +} + +func stalledState() *snapshot.State { + st := healthyState() + st.Cluster.PendingJobs = 2 + st.PendTop = []snapshot.JobState{ + {ID: "1-0-1-0", Reason: &snapshot.JobReason{MainReasonName: "no_resource"}}, + {ID: "1-0-2-0", Reason: &snapshot.JobReason{MainReasonName: "no_resource"}}, + } + return st +} + +// fixedClock lets tests drive cooldown timing. +type fixedClock struct{ t time.Time } + +func (c *fixedClock) Now() time.Time { return c.t } + +func TestDetect_HealthyClusterFiresNothing(t *testing.T) { + d := NewDetector(DefaultLimits()) + if got := d.Detect(healthyState()); len(got) != 0 { + t.Fatalf("healthy state fired: %+v", got) + } +} + +func TestDetect_PendStall(t *testing.T) { + d := NewDetector(DefaultLimits()) + trigs := d.Detect(stalledState()) + if len(trigs) != 1 || trigs[0].Scenario != ScenarioPendStall { + t.Fatalf("want one pend_stall trigger: %+v", trigs) + } + if trigs[0].DedupKey == "" || trigs[0].Fingerprint == "" { + t.Errorf("trigger missing key/fingerprint: %+v", trigs[0]) + } + + // Same situation again: deduped even before cooldown matters. + if again := d.Detect(stalledState()); len(again) != 0 { + t.Fatalf("identical situation re-fired: %+v", again) + } + + // Changed job set → new key → fires again (cooldown 10m must still block + // if the clock hasn't moved; use the clock to jump past it). + d2 := NewDetector(DefaultLimits()) + fc := &fixedClock{t: time.Now()} + d2.now = fc.Now + if got := d2.Detect(stalledState()); len(got) != 1 { + t.Fatalf("first fire failed") + } + st2 := stalledState() + st2.PendTop = append(st2.PendTop, snapshot.JobState{ID: "1-0-3-0", Reason: &snapshot.JobReason{MainReasonName: "no_resource"}}) + if got := d2.Detect(st2); len(got) != 0 { + t.Fatalf("cooldown must block within 10m: %+v", got) + } + fc.t = fc.t.Add(11 * time.Minute) + if got := d2.Detect(st2); len(got) != 1 { + t.Fatalf("after cooldown, changed set must fire: %+v", got) + } +} + +func TestDetect_WorkerHealth(t *testing.T) { + st := healthyState() + st.Nodes[1].Status = "down" + d := NewDetector(DefaultLimits()) + trigs := d.Detect(st) + if len(trigs) != 1 || trigs[0].Scenario != ScenarioWorkerHealth { + t.Fatalf("want worker_health: %+v", trigs) + } + if trigs[0].Fingerprint != "t2:down" { + t.Errorf("fingerprint = %q", trigs[0].Fingerprint) + } + // Same node still down → deduped. Node recovers then breaks again → new key. + if again := d.Detect(st); len(again) != 0 { + t.Fatalf("same outage re-fired: %+v", again) + } + st.Nodes[1].Status = "unknown" + if again := d.Detect(st); len(again) != 0 { + t.Fatalf("cooldown must block second scenario change within window: %+v", again) + } +} + +func TestDetect_UtilSkew(t *testing.T) { + st := healthyState() + st.Nodes[0].Utilization = 0.9 + st.Nodes[0].CoresUsed = 8 + st.Nodes[1].Utilization = 0.1 + st.Nodes[1].CoresUsed = 0 + d := NewDetector(DefaultLimits()) + trigs := d.Detect(st) + if len(trigs) != 1 || trigs[0].Scenario != ScenarioUtilSkew { + t.Fatalf("want util_skew: %+v", trigs) + } +} + +func TestDetect_ExitBurst(t *testing.T) { + st := healthyState() + st.RecentExits = []snapshot.ExitState{ + {ID: "j1", ExitCode: 127}, {ID: "j2", ExitCode: 127}, {ID: "j3", ExitCode: 1}, + } + d := NewDetector(DefaultLimits()) + trigs := d.Detect(st) + if len(trigs) != 1 || trigs[0].Scenario != ScenarioExitBurst { + t.Fatalf("want exit_burst: %+v", trigs) + } + + // Two non-zero exits: below the burst threshold → nothing. + st.RecentExits = st.RecentExits[:2] + d2 := NewDetector(DefaultLimits()) + if got := d2.Detect(st); len(got) != 0 { + t.Fatalf("2 exits must not fire: %+v", got) + } +} + +func TestDetect_MultipleScenariosOneRound(t *testing.T) { + st := stalledState() + st.Nodes[1].Status = "down" + st.Nodes[0].Utilization, st.Nodes[0].CoresUsed = 0.9, 8 + st.Nodes[1].Utilization, st.Nodes[1].CoresUsed = 0.1, 0 + d := NewDetector(DefaultLimits()) + got := d.Detect(st) + scenarios := map[string]bool{} + for _, tr := range got { + scenarios[tr.Scenario] = true + } + if !scenarios[ScenarioPendStall] || !scenarios[ScenarioWorkerHealth] || !scenarios[ScenarioUtilSkew] { + t.Fatalf("want pend_stall+worker_health+util_skew, got %+v", got) + } +} diff --git a/src/frontend/advisor/internal/snapshot/pendreason.go b/src/frontend/advisor/internal/snapshot/pendreason.go new file mode 100644 index 00000000..481ec745 --- /dev/null +++ b/src/frontend/advisor/internal/snapshot/pendreason.go @@ -0,0 +1,95 @@ +package snapshot + +import ( + "encoding/json" + "fmt" +) + +// pendingReason mirrors the apiserver rawPendReason (convert_pure.go), +// surfaced by schedulerclient as the JSON-encoded string +// Extensions["pending_reason"]. +type pendingReason struct { + MainReasonName string `json:"main_reason_name"` +} + +// reasonScenario maps a PREASON_* name to the human-readable scenario. +// This mirrors mcp-server/internal/pendreason.go (module-local copy: the +// advisor and mcp-server are separate modules; lifting the table into +// schedulerclient is a deliberate P1 cleanup, not forgotten). +func reasonScenario(name string) string { + switch name { + case "bad_resreq": + return "the resource request is invalid or cannot be satisfied" + case "no_resource": + return "insufficient resource capacity" + case "match_fail": + return "no host matches the job's host-selection condition" + case "limit_global": + return "a global quota is exhausted" + case "limit_host": + return "a host-level quota is exhausted" + case "queue_closed": + return "the queue is closed (not accepting new jobs)" + case "queue_draining": + return "the queue is draining (no new jobs until it empties)" + case "time_window": + return "outside the queue's dispatch window" + case "host_unavail": + return "the host is unavailable" + default: + return fmt.Sprintf("unknown reason (%s)", name) + } +} + +// parsePendingReason extracts the main reason name + scenario from a job's +// Extensions. Returns nil when absent or unparseable — a missing reason is +// reported as such, never guessed. +func parsePendingReason(ext map[string]any) *JobReason { + raw, ok := ext["pending_reason"] + if !ok { + return nil + } + var pr pendingReason + switch v := raw.(type) { + case string: + if v == "" { + return nil + } + if err := json.Unmarshal([]byte(v), &pr); err != nil { + return nil + } + case map[string]any: + b, _ := json.Marshal(v) + if err := json.Unmarshal(b, &pr); err != nil { + return nil + } + default: + return nil + } + if pr.MainReasonName == "" { + return nil + } + return &JobReason{ + MainReasonName: pr.MainReasonName, + Scenario: reasonScenario(pr.MainReasonName), + } +} + +// round2 keeps snapshot numbers at two decimals so the JSON stays small. +func round2(f float64) float64 { + return float64(int(f*100+0.5*sign(f))) / 100 +} + +func sign(f float64) float64 { + if f < 0 { + return -1 + } + return 1 +} + +func frac(a, b int) float64 { + if b == 0 { + return 0 + } + return float64(a) / float64(b) +} diff --git a/src/frontend/advisor/internal/snapshot/snapshot.go b/src/frontend/advisor/internal/snapshot/snapshot.go new file mode 100644 index 00000000..a4acff88 --- /dev/null +++ b/src/frontend/advisor/internal/snapshot/snapshot.go @@ -0,0 +1,179 @@ +// Package snapshot builds the compact JSON cluster snapshot the providers +// evaluate. It reads through schedulerclient (apiserver REST) and trims +// hard: field whitelists and top-N truncation, because a cheap LLM's +// attention is the scarce resource. +package snapshot + +import ( + "context" + "fmt" + "sort" + "time" + + sclient "schedulerclient" +) + +// Builder pulls cluster state via the schedulerclient. +type Builder struct { + client sclient.Client + // PendTop bounds the pending-job list sent to the model (most-starved + // first). 20 keeps the state under a few KiB. + PendTop int + // StallThreshold is the PEND wait beyond which a job counts as stalled + // (default 5m, matching the pend_stall detector). + StallThreshold time.Duration +} + +// NewBuilder returns a Builder with the defaults applied. +func NewBuilder(client sclient.Client) *Builder { + return &Builder{client: client, PendTop: 20, StallThreshold: 5 * time.Minute} +} + +// State is the snapshot shape (also the echo provider's parse target — the +// two must evolve together; see provider/echo.go echoState). +type State struct { + GeneratedAt string `json:"generated_at"` + Cluster ClusterSum `json:"cluster"` + Nodes []NodeState `json:"nodes"` + PendTop []JobState `json:"pend_top"` + // RecentExits carries jobs that exited since the previous scan + // (exit_burst detector input); empty when none. + RecentExits []ExitState `json:"recent_exits,omitempty"` + Queues []QueueSum `json:"queues,omitempty"` +} + +type ClusterSum struct { + PendingJobs int `json:"pending_jobs"` + RunningJobs int `json:"running_jobs"` + Utilization float64 `json:"utilization"` +} + +type NodeState struct { + Name string `json:"name"` + Status string `json:"status"` + CoresTotal int `json:"cores_total"` + CoresUsed int `json:"cores_used"` + Utilization float64 `json:"utilization"` +} + +type JobState struct { + ID string `json:"id"` + Name string `json:"name"` + Owner string `json:"owner"` + Queue string `json:"queue"` + WaitSeconds float64 `json:"wait_seconds"` + Reason *JobReason `json:"reason,omitempty"` + Cores int `json:"cores,omitempty"` +} + +// JobReason is the parsed pending_reason (main_reason_name + the +// human-readable scenario name from the translation table). +type JobReason struct { + MainReasonName string `json:"main_reason_name,omitempty"` + Scenario string `json:"scenario,omitempty"` +} + +type ExitState struct { + ID string `json:"id"` + Name string `json:"name,omitempty"` + ExitCode int `json:"exit_code"` +} + +type QueueSum struct { + Name string `json:"name"` + PendingJobs int `json:"pending_jobs"` + RunningJobs int `json:"running_jobs"` +} + +// Build fetches jobs/nodes/queues and assembles the trimmed snapshot. The +// previous-exits input lets exit_burst see what changed since the last +// round; pass nil on the first round. +func (b *Builder) Build(ctx context.Context, prevExits []ExitState) (*State, error) { + st := &State{GeneratedAt: time.Now().UTC().Format(time.RFC3339)} + + nodes, err := b.client.ListNodes(ctx) + if err != nil { + return nil, fmt.Errorf("list nodes: %w", err) + } + st.Nodes = make([]NodeState, 0, len(nodes)) + var totalCores, usedCores int + for _, n := range nodes { + util := 0.0 + if n.CoresTotal > 0 { + util = float64(n.CoresUsed) / float64(n.CoresTotal) + } + st.Nodes = append(st.Nodes, NodeState{ + Name: n.Name, Status: n.Status, + CoresTotal: n.CoresTotal, CoresUsed: n.CoresUsed, + Utilization: round2(util), + }) + totalCores += n.CoresTotal + usedCores += n.CoresUsed + } + + // PEND jobs are the pend_stall input. One filtered call per status + // keeps the apiserver query simple. + pend, _, err := b.client.ListJobs(ctx, sclient.JobFilter{Statuses: []sclient.JobStatus{sclient.JobStatusPending}}) + if err != nil { + return nil, fmt.Errorf("list pending jobs: %w", err) + } + run, _, err := b.client.ListJobs(ctx, sclient.JobFilter{Statuses: []sclient.JobStatus{sclient.JobStatusRunning}}) + if err != nil { + return nil, fmt.Errorf("list running jobs: %w", err) + } + st.Cluster = ClusterSum{ + PendingJobs: len(pend), + RunningJobs: len(run), + Utilization: round2(frac(usedCores, totalCores)), + } + + st.PendTop = topPending(pend, b.PendTop, b.StallThreshold) + + if prevExits != nil { + st.RecentExits = prevExits + } + + queues, err := b.client.ListQueues(ctx) + if err == nil { // queues are garnish; never fail the snapshot for them + for _, q := range queues { + st.Queues = append(st.Queues, QueueSum{Name: q.Name, PendingJobs: q.PendingJobs, RunningJobs: q.RunningJobs}) + } + } + + return st, nil +} + +// topPending picks the most-starved pending jobs, keeping only those past +// the stall threshold (the snapshot feeds the pend_stall scenario; fresh +// PEND jobs are noise for the model). +func topPending(pend []*sclient.URMJob, n int, stall time.Duration) []JobState { + type keyed struct { + js JobState + wait time.Duration + } + var out []keyed + for _, j := range pend { + wait := time.Since(j.SubmissionTime) + if wait < stall { + continue + } + js := JobState{ + ID: j.ID, Name: j.Name, Owner: j.Owner, Queue: j.Queue, + WaitSeconds: round2(wait.Seconds()), + Cores: j.Cores, + } + if pr := parsePendingReason(j.Extensions); pr != nil { + js.Reason = pr + } + out = append(out, keyed{js, wait}) + } + sort.Slice(out, func(i, k int) bool { return out[i].wait > out[k].wait }) + if len(out) > n { + out = out[:n] + } + res := make([]JobState, 0, len(out)) + for _, k := range out { + res = append(res, k.js) + } + return res +} diff --git a/src/frontend/advisor/internal/snapshot/snapshot_test.go b/src/frontend/advisor/internal/snapshot/snapshot_test.go new file mode 100644 index 00000000..5b6ad694 --- /dev/null +++ b/src/frontend/advisor/internal/snapshot/snapshot_test.go @@ -0,0 +1,215 @@ +package snapshot + +import ( + "context" + "encoding/json" + "testing" + "time" + + sclient "schedulerclient" +) + +// fakeClient implements the read side of sclient.Client for tests. +type fakeClient struct { + jobs map[sclient.JobStatus][]*sclient.URMJob + nodes []*sclient.URMNode + queues []*sclient.URMQueue +} + +func (f *fakeClient) ListJobs(_ context.Context, filter sclient.JobFilter) ([]*sclient.URMJob, *sclient.Page, error) { + var out []*sclient.URMJob + for _, s := range filter.Statuses { + out = append(out, f.jobs[s]...) + } + return out, &sclient.Page{Total: len(out)}, nil +} + +func (f *fakeClient) ListNodes(_ context.Context) ([]*sclient.URMNode, error) { return f.nodes, nil } +func (f *fakeClient) ListQueues(_ context.Context) ([]*sclient.URMQueue, error) { + return f.queues, nil +} + +func (f *fakeClient) GetJob(context.Context, string) (*sclient.URMJob, error) { + panic("not needed") +} +func (f *fakeClient) GetJobLogs(context.Context, string, sclient.LogOpts) ([]string, error) { + panic("not needed") +} +func (f *fakeClient) GetClusterMetrics(context.Context) (*sclient.Metrics, error) { + panic("not needed") +} +func (f *fakeClient) SubmitJob(context.Context, sclient.JobSpec) (*sclient.SubmitJobResult, error) { + panic("observe-only: writes must be unreachable") +} +func (f *fakeClient) CancelJob(context.Context, string) error { panic("observe-only") } +func (f *fakeClient) HoldJob(context.Context, string) error { panic("observe-only") } +func (f *fakeClient) ReleaseJob(context.Context, string) error { + panic("observe-only") +} +func (f *fakeClient) CheckpointJob(context.Context, string) error { panic("observe-only") } +func (f *fakeClient) RestartJob(context.Context, string) (*sclient.RestartJobResult, error) { + panic("observe-only") +} +func (f *fakeClient) WatchJobs(context.Context) (<-chan sclient.JobEvent, error) { + panic("not needed in snapshot tests") +} + +func testClient() *fakeClient { + old := time.Now().Add(-2 * time.Hour) + recent := time.Now().Add(-30 * time.Second) + return &fakeClient{ + jobs: map[sclient.JobStatus][]*sclient.URMJob{ + sclient.JobStatusPending: { + {ID: "1-0-1-0", Name: "stalled", Owner: "u1", Queue: "normal", SubmissionTime: old, + Cores: 4, Extensions: map[string]any{ + "pending_reason": `{"main_reason":1,"main_reason_name":"no_resource"}`, + }}, + {ID: "1-0-2-0", Name: "fresh", Owner: "u1", Queue: "normal", SubmissionTime: recent}, + {ID: "1-0-3-0", Name: "matchfail", Owner: "u2", Queue: "gpu", SubmissionTime: old, + Extensions: map[string]any{ + "pending_reason": `{"main_reason":2,"main_reason_name":"match_fail"}`, + }}, + }, + sclient.JobStatusRunning: { + {ID: "1-0-9-0", Name: "runner", SubmissionTime: old}, + }, + }, + nodes: []*sclient.URMNode{ + {Name: "t1", Status: "busy", CoresTotal: 8, CoresUsed: 6}, + {Name: "t2", Status: "idle", CoresTotal: 8, CoresUsed: 0}, + }, + queues: []*sclient.URMQueue{ + {Name: "normal", PendingJobs: 2, RunningJobs: 1}, + }, + } +} + +func TestBuild_TrimsAndSorts(t *testing.T) { + b := NewBuilder(testClient()) + st, err := b.Build(context.Background(), nil) + if err != nil { + t.Fatalf("Build: %v", err) + } + + if st.Cluster.PendingJobs != 3 || st.Cluster.RunningJobs != 1 { + t.Errorf("cluster counts = %+v", st.Cluster) + } + // 6/16 cores used. + if st.Cluster.Utilization != 0.38 { + t.Errorf("utilization = %v, want 0.38", st.Cluster.Utilization) + } + + // Fresh PEND job (30s) filtered by stall threshold; two stalled remain, + // most-starved first. Both stalled are 2h old so order is by (equal) + // wait — assert set membership instead. + if len(st.PendTop) != 2 { + t.Fatalf("pend_top = %+v, want 2 stalled jobs", st.PendTop) + } + got := map[string]bool{} + for _, j := range st.PendTop { + got[j.ID] = true + if j.WaitSeconds < 300 { + t.Errorf("%s below stall threshold: %v", j.ID, j.WaitSeconds) + } + } + if !got["1-0-1-0"] || !got["1-0-3-0"] { + t.Errorf("stalled jobs missing: %+v", st.PendTop) + } + + // Reason parsing. + for _, j := range st.PendTop { + if j.ID == "1-0-1-0" && (j.Reason == nil || j.Reason.MainReasonName != "no_resource") { + t.Errorf("reason parse failed: %+v", j.Reason) + } + if j.ID == "1-0-3-0" && (j.Reason == nil || j.Reason.Scenario != "no host matches the job's host-selection condition") { + t.Errorf("scenario translation failed: %+v", j.Reason) + } + if j.ID == "1-0-2-0" { + t.Errorf("fresh job leaked into pend_top") + } + } + + // Nodes. + if len(st.Nodes) != 2 || st.Nodes[0].Utilization != 0.75 { + t.Errorf("nodes = %+v", st.Nodes) + } + + // Queues garnish. + if len(st.Queues) != 1 || st.Queues[0].Name != "normal" { + t.Errorf("queues = %+v", st.Queues) + } +} + +func TestBuild_PendTopCap(t *testing.T) { + fc := testClient() + var pend []*sclient.URMJob + for i := 0; i < 30; i++ { + pend = append(pend, &sclient.URMJob{ + ID: "1-0-" + jsonInt(i) + "-0", SubmissionTime: time.Now().Add(-time.Duration(i+1) * time.Hour), + }) + } + fc.jobs[sclient.JobStatusPending] = pend + b := NewBuilder(fc) + b.PendTop = 5 + st, err := b.Build(context.Background(), nil) + if err != nil { + t.Fatalf("Build: %v", err) + } + if len(st.PendTop) != 5 { + t.Fatalf("cap failed: %d entries", len(st.PendTop)) + } + // Most-starved first: the oldest job (i=29 → 30h) is top. + if st.PendTop[0].ID != "1-0-29-0" { + t.Errorf("top = %s, want oldest", st.PendTop[0].ID) + } +} + +func TestBuild_StateJSONMatchesEchoShape(t *testing.T) { + // The snapshot must decode cleanly into the echo provider's echoState — + // the contract between the two packages. + st, err := NewBuilder(testClient()).Build(context.Background(), []ExitState{{ID: "1-0-7-0", ExitCode: 127}}) + if err != nil { + t.Fatalf("Build: %v", err) + } + raw, err := json.Marshal(st) + if err != nil { + t.Fatalf("marshal: %v", err) + } + var probe struct { + GeneratedAt string `json:"generated_at"` + Cluster struct { + PendingJobs int `json:"pending_jobs"` + } `json:"cluster"` + Nodes []struct { + Name string `json:"name"` + } `json:"nodes"` + PendTop []struct { + ID string `json:"id"` + Reason *struct { + MainReasonName string `json:"main_reason_name"` + Scenario string `json:"scenario"` + } `json:"reason"` + } `json:"pend_top"` + RecentExits []struct { + ID string `json:"id"` + ExitCode int `json:"exit_code"` + } `json:"recent_exits"` + } + if err := json.Unmarshal(raw, &probe); err != nil { + t.Fatalf("state JSON: %v", err) + } + if probe.Cluster.PendingJobs != 3 || len(probe.Nodes) != 2 { + t.Errorf("roundtrip mismatch: %+v", probe) + } + if len(probe.RecentExits) != 1 || probe.RecentExits[0].ID != "1-0-7-0" { + t.Errorf("recent_exits: %+v", probe.RecentExits) + } + if len(probe.PendTop) == 0 || probe.PendTop[0].Reason == nil { + t.Errorf("reason missing in JSON: %+v", probe.PendTop) + } +} + +func jsonInt(i int) string { + b, _ := json.Marshal(i) + return string(b) +}