wdgzyjx/myxce
1
0
Fork 0
forked from lei.chen/myxce
Code Pull requests Activity

fix(schedulerclient): parse apiserver OrderMap columnar rows (bug-1491)

Browse source 
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 <noreply@anthropic.com>
This commit is contained in:
lei.chen 2026-08-02 05:01:16 +00:00
parent 4a4ae16725
commit 226259a52b
4 changed files with 294 additions and 56 deletions
Download patch file Download diff file Expand all files Collapse all files

20
.wolf/buglog.json
View file

@ -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"
}
]

12
.wolf/memory.md
View file

@ -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 |

178
src/frontend/schedulerclient/httpclient.go
View file

@ -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"`

140
src/frontend/schedulerclient/ordermap_test.go Normal file
View file

@ -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)
}
}