From 226259a52b57598cfb86bb002df5abf70dc0be10 Mon Sep 17 00:00:00 2001 From: "lei.chen" Date: Sun, 2 Aug 2026 05:01:16 +0000 Subject: [PATCH] fix(schedulerclient): parse apiserver OrderMap columnar rows (bug-1491) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The apiserver returns list rows in OrderMap columnar form — {"Fields":["col",...],"Data":{"col":val,...}} — but the read clients unmarshalled each row as a flat object, so every field came back empty (counts were right, values were all zero/""). Fix: decode rows as apiRow{Fields, Data map[string]json.RawMessage} and pull named fields out of Data via rowStr/rowInt/rowInt64/rowFloat helpers. ListNodes, ListQueues, and getJobs (ListJobs/GetJob) all use apiRow now. TDD: 3 new ordermap_test.go tests built from live-captured apiserver response shapes (red: fields empty → green: parsed). Live-verified against t1: node name=t1 cores=8, 4 queues active, 3 jobs with real fields. Co-Authored-By: Claude --- .wolf/buglog.json | 20 ++ .wolf/memory.md | 12 ++ src/frontend/schedulerclient/httpclient.go | 178 ++++++++++++------ src/frontend/schedulerclient/ordermap_test.go | 140 ++++++++++++++ 4 files changed, 294 insertions(+), 56 deletions(-) create mode 100644 src/frontend/schedulerclient/ordermap_test.go diff --git a/.wolf/buglog.json b/.wolf/buglog.json index 7de982c2..1146fe16 100644 --- a/.wolf/buglog.json +++ b/.wolf/buglog.json @@ -14544,5 +14544,25 @@ ], "occurrences": 2, "last_seen": "2026-08-01" + }, + { + "id": "bug-1491", + "timestamp": "2026-08-02T12:00:00+08:00", + "error_message": "scheduler_list_nodes/queues/jobs MCP read tools return correct entity count but all fields empty (name='', status='unknown', 0 resources); raw_status empty", + "file": "src/frontend/schedulerclient/httpclient.go", + "root_cause": "apiserver /xce/v1/hosts|queues|jobs return rows in utils.OrderMap columnar format {\"Fields\":[...],\"Data\":{...}}, but schedulerclient's apiHost/apiQueue/apiJob structs unmarshal flat JSON objects, so every field is zero-valued. Entity count still correct because len(rows.Rows) works.", + "fix": "schedulerclient now decodes apiserver OrderMap columnar rows: added apiRow{Fields,Data map[string]json.RawMessage} + rowStr/rowInt/rowInt64/rowFloat helpers in httpclient.go; ListNodes/ListQueues/getJobs pull named fields from row.Data. TDD: 3 new ordermap_test.go tests with real captured response shapes (red\u2192green). Live-verified against t1: nodes name=t1 cores=8, 4 queues active, 3 jobs real fields.", + "tags": [ + "schedulerclient", + "ordermap", + "read-parse", + "phase1-verify", + "fixed" + ], + "related_bugs": [ + "bug-1490" + ], + "occurrences": 4, + "last_seen": "2026-08-02" } ] \ No newline at end of file diff --git a/.wolf/memory.md b/.wolf/memory.md index c49ad2ca..884d02de 100644 --- a/.wolf/memory.md +++ b/.wolf/memory.md @@ -7278,3 +7278,15 @@ | 09:56 | Pushed chatbot branch to myxce (2e299f4..9fe14c1, 3 commits) | git push | done | ~0.2k tok | | 09:58 | Wrote src/docs/chatbot-verification.md (cluster deploy + Phase 0-1 verify manual), added to design doc index, committed 56da673 + pushed | docs | done | ~1k tok | | 04:26 | Added §5 "接入 Claude Code" to chatbot-verification.md (verified via claude-code-guide against installed CLI v2.1.220: claude mcp add / .mcp.json schema / scopes / approval gate) | src/docs/chatbot-verification.md | done | ~1k tok | + +## Session: 2026-08-02 04:45 + +| Time | Action | File(s) | Outcome | ~Tokens | +|------|--------|---------|---------|--------| + +## Session: 2026-08-02 04:50 + +| Time | Action | File(s) | Outcome | ~Tokens | +|------|--------|---------|---------|--------| +| 12:05 | 查询集群机器: MCP 只读工具(list_nodes/queues/jobs)返回空字段 → 根因: apiserver 返回 OrderMap 列式格式 {Fields,Data}, schedulerclient 用扁平 struct 解析 | src/frontend/schedulerclient/httpclient.go | 记 bug-1491, 未修 | ~1k tok | +| 05:00 | Fixed bug-1491: schedulerclient read clients now parse apiserver OrderMap columnar rows (apiRow + rowStr/rowInt/rowInt64/rowFloat). TDD: 3 ordermap_test.go red→green; live-verified on t1 (nodes/queues/jobs fields all correct) | src/frontend/schedulerclient/httpclient.go, ordermap_test.go | green | ~1k tok | diff --git a/src/frontend/schedulerclient/httpclient.go b/src/frontend/schedulerclient/httpclient.go index c633a7df..f6fb36f7 100644 --- a/src/frontend/schedulerclient/httpclient.go +++ b/src/frontend/schedulerclient/httpclient.go @@ -124,8 +124,8 @@ func (hc *HTTPClient) ListQueues(ctx context.Context) ([]*URMQueue, error) { } var rows struct { - Count int `json:"count"` - Rows []apiQueue `json:"rows"` + Count int `json:"count"` + Rows []apiRow `json:"rows"` } if err := json.Unmarshal(env.Data, &rows); err != nil { return nil, fmt.Errorf("parse queues: %w", err) @@ -134,19 +134,20 @@ func (hc *HTTPClient) ListQueues(ctx context.Context) ([]*URMQueue, error) { queues := make([]*URMQueue, 0, len(rows.Rows)) for i := range rows.Rows { q := rows.Rows[i] + rawStatus := q.rowStr("queue_status") queues = append(queues, &URMQueue{ - Name: q.QueueName, - Status: QueueStatusFromString(q.QueueStatus), - RunningJobs: q.NumRun, - PendingJobs: q.NumPend, - SuspendedJobs: q.NumSSusp + q.NumUSusp, + Name: q.rowStr("queue_name"), + Status: QueueStatusFromString(rawStatus), + RunningJobs: q.rowInt("num_run"), + PendingJobs: q.rowInt("num_pend"), + SuspendedJobs: q.rowInt("num_ssusp") + q.rowInt("num_ususp"), NodeCount: 0, // apiserver does not expose queue→node mapping directly Extensions: map[string]any{ - "raw_status": q.QueueStatus, - "priority": q.Priority, - "num_ssusp": q.NumSSusp, - "num_ususp": q.NumUSusp, - "num_done": q.NumDone, + "raw_status": rawStatus, + "priority": q.rowInt("priority"), + "num_ssusp": q.rowInt("num_ssusp"), + "num_ususp": q.rowInt("num_ususp"), + "num_done": q.rowInt("num_done"), }, }) } @@ -168,8 +169,8 @@ func (hc *HTTPClient) ListNodes(ctx context.Context) ([]*URMNode, error) { } var rows struct { - Count int `json:"count"` - Rows []apiHost `json:"rows"` + Count int `json:"count"` + Rows []apiRow `json:"rows"` } if err := json.Unmarshal(env.Data, &rows); err != nil { return nil, fmt.Errorf("parse nodes: %w", err) @@ -178,26 +179,28 @@ func (hc *HTTPClient) ListNodes(ctx context.Context) ([]*URMNode, error) { nodes := make([]*URMNode, 0, len(rows.Rows)) for i := range rows.Rows { h := rows.Rows[i] + rawStatus := h.rowStr("host_status") + numRun := h.rowInt("num_run") nodes = append(nodes, &URMNode{ - Name: h.HostName, - Status: NodeStatusFromString(h.HostStatus), - CoresTotal: h.NumCPUs, - CoresUsed: h.NumRun, // running jobs ≈ cores used - MemoryBytesTotal: int64(h.MaxMem) * mbToBytes, + Name: h.rowStr("host_name"), + Status: NodeStatusFromString(rawStatus), + CoresTotal: h.rowInt("num_cpus"), + CoresUsed: numRun, // running jobs ≈ cores used + MemoryBytesTotal: int64(h.rowInt("max_mem")) * mbToBytes, MemoryBytesUsed: 0, // apiserver does not expose live used-memory for hosts GPUCount: 0, // TODO: parse from extensions/topology when available Features: nil, Extensions: map[string]any{ - "raw_status": h.HostStatus, - "host_type": h.HostType, - "host_model": h.HostModel, - "cpu_factor": h.CPUFactor, - "max_slot": h.MaxSlot, - "max_swap_mb": h.MaxSwap, - "max_tmp_mb": h.MaxTmp, - "num_run": h.NumRun, - "num_ssusp": h.NumSSusp, - "num_ususp": h.NumUSusp, + "raw_status": rawStatus, + "host_type": h.rowStr("host_type"), + "host_model": h.rowStr("host_model"), + "cpu_factor": h.rowFloat("cpu_factor"), + "max_slot": h.rowInt("max_slot"), + "max_swap_mb": h.rowInt("max_swap"), + "max_tmp_mb": h.rowInt("max_tmp"), + "num_run": numRun, + "num_ssusp": h.rowInt("num_ssusp"), + "num_ususp": h.rowInt("num_ususp"), }, }) } @@ -285,9 +288,9 @@ func (hc *HTTPClient) getJobs(req *resty.Request, url string, filter JobFilter) } var rows struct { - Count int `json:"count"` - Rows []apiJob `json:"rows"` - PageToken string `json:"page_token"` + Count int `json:"count"` + Rows []apiRow `json:"rows"` + PageToken string `json:"page_token"` } if err := json.Unmarshal(env.Data, &rows); err != nil { return nil, nil, fmt.Errorf("parse jobs: %w", err) @@ -296,7 +299,8 @@ func (hc *HTTPClient) getJobs(req *resty.Request, url string, filter JobFilter) jobs := make([]*URMJob, 0, len(rows.Rows)) for i := range rows.Rows { j := rows.Rows[i] - urmStatus := JobStatusFromString(j.JobStatus) + rawStatus := j.rowStr("job_status") + urmStatus := JobStatusFromString(rawStatus) // Apply URM status filter in post-processing if len(filter.Statuses) > 0 { @@ -314,20 +318,20 @@ func (hc *HTTPClient) getJobs(req *resty.Request, url string, filter JobFilter) // Parse exit code var exitCode *int - if j.ExitStatus != "" { - if code, err := strconv.Atoi(j.ExitStatus); err == nil { + if es := j.rowStr("exit_status"); es != "" { + if code, err := strconv.Atoi(es); err == nil { exitCode = &code } } // Parse timestamps var submitTime time.Time - if j.SubmitTime > 0 { - submitTime = time.Unix(j.SubmitTime, 0) + if st := j.rowInt64("submit_time"); st > 0 { + submitTime = time.Unix(st, 0) } var startTime *time.Time - if j.StartTime > 0 { - t := time.Unix(j.StartTime, 0) + if st := j.rowInt64("start_time"); st > 0 { + t := time.Unix(st, 0) startTime = &t } @@ -337,33 +341,34 @@ func (hc *HTTPClient) getJobs(req *resty.Request, url string, filter JobFilter) walltime := 0 // TODO: parse run_limit string when available // Memory: MaxMem from apiserver is in MB (standard HPC convention). - memBytes := int64(math.Round(j.MaxMem * mbToBytes)) + memBytes := int64(math.Round(j.rowFloat("max_mem") * mbToBytes)) + numProcessors := j.rowInt("num_processors") urmJob := &URMJob{ - ID: strconv.FormatInt(j.JobID, 10), - Name: j.JobName, + ID: strconv.FormatInt(j.rowInt64("job_id"), 10), + Name: j.rowStr("job_name"), Status: urmStatus, - RawStatus: j.JobStatus, - Queue: j.QueueName, - Nodes: j.NumHosts, - Cores: j.NumProcessors, + RawStatus: rawStatus, + Queue: j.rowStr("queue_name"), + Nodes: j.rowInt("num_hosts"), + Cores: numProcessors, GPUs: 0, // TODO: parse from resreq/extensions MemoryBytes: memBytes, WalltimeSeconds: walltime, SubmissionTime: submitTime, StartTime: startTime, ExitCode: exitCode, - Owner: j.SubmitUser, + Owner: j.rowStr("submit_user"), Extensions: map[string]any{ - "command": j.Command, - "project_name": j.ProjectName, - "pending_reason": j.PendingReason, - "resreq": j.Resreq, - "exit_status_raw": j.ExitStatus, - "num_processors": j.NumProcessors, - "submit_time": j.SubmitTime, - "start_time": j.StartTime, - "end_time": j.EndTime, + "command": j.rowStr("command"), + "project_name": j.rowStr("project_name"), + "pending_reason": j.rowStr("pending_reason"), + "resreq": j.rowStr("resreq"), + "exit_status_raw": j.rowStr("exit_status"), + "num_processors": numProcessors, + "submit_time": j.rowInt64("submit_time"), + "start_time": j.rowInt64("start_time"), + "end_time": j.rowInt64("end_time"), }, } @@ -388,6 +393,67 @@ type apiEnvelope struct { Data json.RawMessage `json:"data"` } +// apiRow is one row of an apiserver OrderMap response. The apiserver returns +// list rows in columnar form: {"Fields":["col",...],"Data":{"col":val,...}} +// (bug-1491: a flat-object unmarshal left every field empty). Decode rows as +// apiRow and pull named fields out of Data with the helpers below. +type apiRow struct { + Fields []string `json:"fields"` + Data map[string]json.RawMessage `json:"data"` +} + +// rowStr extracts a string field. Missing, null, or non-string → "". +func (r apiRow) rowStr(key string) string { + raw, ok := r.Data[key] + if !ok || string(raw) == "null" { + return "" + } + var s string + if err := json.Unmarshal(raw, &s); err != nil { + return "" + } + return s +} + +// rowInt extracts an integer field. Missing, null, or non-number → 0. +func (r apiRow) rowInt(key string) int { + raw, ok := r.Data[key] + if !ok || string(raw) == "null" { + return 0 + } + var f float64 + if err := json.Unmarshal(raw, &f); err != nil { + return 0 + } + return int(f) +} + +// rowInt64 extracts a 64-bit integer field. Missing, null, or non-number → 0. +func (r apiRow) rowInt64(key string) int64 { + raw, ok := r.Data[key] + if !ok || string(raw) == "null" { + return 0 + } + var f float64 + if err := json.Unmarshal(raw, &f); err != nil { + return 0 + } + return int64(f) +} + +// rowFloat extracts a float field. Missing, null, or non-number → 0. +func (r apiRow) rowFloat(key string) float64 { + raw, ok := r.Data[key] + if !ok || string(raw) == "null" { + return 0 + } + var f float64 + if err := json.Unmarshal(raw, &f); err != nil { + return 0 + } + return f +} + // apiJob mirrors the fields of apiserver's QueryJob that we need for URM conversion. type apiJob struct { JobID int64 `json:"job_id"` diff --git a/src/frontend/schedulerclient/ordermap_test.go b/src/frontend/schedulerclient/ordermap_test.go new file mode 100644 index 00000000..44004a0e --- /dev/null +++ b/src/frontend/schedulerclient/ordermap_test.go @@ -0,0 +1,140 @@ +package schedulerclient + +import ( + "context" + "io" + "net/http" + "net/http/httptest" + "testing" +) + +// TestOrderMap_RowsUseColumnarShape is the regression guard for bug-1491: +// the apiserver's /xce/v1/hosts|queues|jobs responses are OrderMap columnar +// rows — {"Fields":["col",...],"Data":{"col":val,...}} — NOT flat objects. +// The read clients must decode each row via apiRow and pull named fields out +// of Data, or every field comes back empty (count still correct). +func TestOrderMap_RowsUseColumnarShape(t *testing.T) { + // Capture what schedulerclient actually requests, then answer with a + // real apiserver-shaped hosts response (captured live from t1). + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + io.WriteString(w, `{ + "code": 200, "msg": "Success", + "data": { + "count": 1, + "rows": [{ + "Fields": ["host_name","host_status","host_model","host_type","host_group","flag","cpu_factor","num_cpus","num_cores","num_threads","dispt_window","max_slot","max_mem","max_swap","max_tmp","num_disk","num_run","num_ssusp","num_ususp","num_done","topology","sched_stop"], + "Data": { + "host_name":"t1","host_status":"OK","host_model":"x86_64-linux-gnu", + "host_type":"x86_64-linux-gnu","host_group":"","flag":0,"cpu_factor":1, + "num_cpus":8,"num_cores":0,"num_threads":0,"dispt_window":null, + "max_slot":0,"max_mem":8224821248,"max_swap":1024,"max_tmp":2048, + "num_disk":1,"num_run":0,"num_ssusp":0,"num_ususp":0,"num_done":0, + "topology":null,"sched_stop":null + } + }] + } + }`) + })) + defer srv.Close() + + hc := NewClient(srv.URL, "test-token") + nodes, err := hc.ListNodes(context.Background()) + if err != nil { + t.Fatalf("ListNodes error: %v", err) + } + if len(nodes) != 1 { + t.Fatalf("len(nodes) = %d, want 1", len(nodes)) + } + n := nodes[0] + if n.Name != "t1" { + t.Errorf("node.Name = %q, want t1", n.Name) + } + if n.Status != "idle" { + t.Errorf("node.Status = %q, want idle (OK → idle)", n.Status) + } + if n.CoresTotal != 8 { + t.Errorf("node.CoresTotal = %d, want 8", n.CoresTotal) + } + if n.MemoryBytesTotal != 8224821248*mbToBytes { + t.Errorf("node.MemoryBytesTotal = %d, want %d", n.MemoryBytesTotal, 8224821248*mbToBytes) + } +} + +func TestListQueues_OrderMapShape(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + io.WriteString(w, `{ + "code": 200, "msg": "Success", + "data": { + "count": 4, + "rows": [{ + "Fields": ["queue_name","queue_status","priority","description","num_pend","num_run","num_ssusp","num_ususp","num_done","fs_info"], + "Data": {"queue_name":"night","queue_status":"open:active","priority":30,"description":"","num_pend":0,"num_run":0,"num_ssusp":0,"num_ususp":0,"num_done":0,"fs_info":null} + }] + } + }`) + })) + defer srv.Close() + + hc := NewClient(srv.URL, "test-token") + queues, err := hc.ListQueues(context.Background()) + if err != nil { + t.Fatalf("ListQueues error: %v", err) + } + if len(queues) != 1 { + t.Fatalf("len(queues) = %d, want 1", len(queues)) + } + q := queues[0] + if q.Name != "night" { + t.Errorf("queue.Name = %q, want night", q.Name) + } + if q.Status != "active" { + t.Errorf("queue.Status = %q, want active (open:active → active)", q.Status) + } +} + +func TestListJobs_OrderMapShape(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + io.WriteString(w, `{ + "code": 200, "msg": "Success", + "data": { + "count": 3, + "rows": [{ + "Fields": ["job_id","array_idx","unique_job_id","submit_user","job_status","queue_name","from_host","exec_hosts","job_name","command","submit_time"], + "Data": {"job_id":73929906127372288,"array_idx":0,"unique_job_id":73929906127372288,"submit_user":"scheduler-agent","job_status":"PEND","queue_name":"normal","from_host":"","exec_hosts":null,"job_name":"verify1","command":"sleep 30","submit_time":1785575952} + }] + } + }`) + })) + defer srv.Close() + + hc := NewClient(srv.URL, "test-token") + jobs, page, err := hc.ListJobs(context.Background(), JobFilter{}) + if err != nil { + t.Fatalf("ListJobs error: %v", err) + } + if page.Total != 3 { + t.Errorf("page.Total = %d, want 3", page.Total) + } + if len(jobs) != 1 { + t.Fatalf("len(jobs) = %d, want 1", len(jobs)) + } + j := jobs[0] + if j.ID != "73929906127372288" { + t.Errorf("job.ID = %q, want 73929906127372288", j.ID) + } + if j.Name != "verify1" { + t.Errorf("job.Name = %q, want verify1", j.Name) + } + if j.Status != JobStatusPending { + t.Errorf("job.Status = %q, want pending", j.Status) + } + if j.Queue != "normal" { + t.Errorf("job.Queue = %q, want normal", j.Queue) + } + if j.Owner != "scheduler-agent" { + t.Errorf("job.Owner = %q, want scheduler-agent", j.Owner) + } +}