apiserver: derive array stats from live job store; fix CLI stats decode
type=array_stats (xce jobs list -A) read store.ArrayJobs, which has no
production writer, so it always answered zero rows. Stats are now derived
at query time by scanning store.Jobs and aggregating per array base ID
(controllers/arraystats.go), mirroring hosts.go's computeHostSlotUsage
precedent: counters can never drift from the element rows.
- GetJobsInfo/GetJobByIDs array_stats branches run the derived rows
through the same ExecuteQuery filter/sort/paginate pipeline as store
queries; unique_job_id rides in the field list so processJobData can
annotate job_id_human (SelectFields drops unlisted fields).
- processJobData: new rowInt64 helper (uint64 bit-preserved + float64/
json.Number tolerant) replaces brittle bare type assertions for
unique_job_id/array_idx reads; dead code removed (StackLimit
wrong-source copy, unreachable index error).
- QueryData(TableArrayJob) now returns a guard error instead of always
answering zero rows from the never-written store.
- xcecli: ArrayJobStatsData got UnmarshalJSON via UnmarshalOrderMapRow —
the server's OrderMap wrapper ({Fields,Data}) decoded to all-zero
fields, which is why table mode showed 0/- rows against live arrays.
- analysis.md CL5 marked fixed with the real root cause.
Cluster-verified on t1/t2/t3: submit→PEND→RUN→DONE per element, kill
path, job_id filter, and per-status counters all live; all daemons
healthy. Unit suites: apiserver + xcecli green (CGO_ENABLED=0).
Co-Authored-By: Claude Code <noreply@anthropic.com>
This commit is contained in:
parent
9dbb75adb3
commit
a9c8044165
9 changed files with 529 additions and 39 deletions
101
src/frontend/apiserver/controllers/arraystats.go
Normal file
101
src/frontend/apiserver/controllers/arraystats.go
Normal file
|
|
@ -0,0 +1,101 @@
|
||||||
|
package controllers
|
||||||
|
|
||||||
|
import (
|
||||||
|
"apiserver/backend"
|
||||||
|
"apiserver/models"
|
||||||
|
"apiserver/store"
|
||||||
|
"apiserver/utils"
|
||||||
|
"sort"
|
||||||
|
)
|
||||||
|
|
||||||
|
// Array statistics rows (type=array_stats, `xce jobs list -A`) are derived
|
||||||
|
// from the live job store at query time, mirroring the bug-1743 pattern for
|
||||||
|
// HostInfo.NumRun: store.ArrayJobs has no production writer (the controller's
|
||||||
|
// array push and every single-job update land in store.Jobs only), so a
|
||||||
|
// store-backed ArrayJobInfo would stay empty forever. Deriving per query keeps
|
||||||
|
// a single source of truth -- the counters can never drift from the element
|
||||||
|
// rows the other views show.
|
||||||
|
//
|
||||||
|
// The array identity (base job ID, name, user, queue, ...) comes from each
|
||||||
|
// element's own fields; every element of an array carries the same submit spec
|
||||||
|
// (the controller expands one shared descriptor), so the first element seen
|
||||||
|
// for a base ID supplies them.
|
||||||
|
|
||||||
|
// arrayAgg accumulates one array's counters while scanning store.Jobs.
|
||||||
|
type arrayAgg struct {
|
||||||
|
info models.ArrayJobInfo
|
||||||
|
}
|
||||||
|
|
||||||
|
// deriveArrayJobStats scans store.Jobs and aggregates every array element row
|
||||||
|
// into one models.ArrayJobInfo per array base job ID, sorted by job_id
|
||||||
|
// (the stats view's default sort). Ordinary (non-array) jobs are skipped.
|
||||||
|
func deriveArrayJobStats() []models.ArrayJobInfo {
|
||||||
|
aggs := make(map[int64]*arrayAgg)
|
||||||
|
for _, job := range store.Jobs.List() {
|
||||||
|
raw := uint64(job.Unique_JobID)
|
||||||
|
if !backend.IsArrayJob(raw) {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
base := int64(backend.ExtractArrayBaseID(raw))
|
||||||
|
agg, ok := aggs[base]
|
||||||
|
if !ok {
|
||||||
|
agg = &arrayAgg{info: models.ArrayJobInfo{
|
||||||
|
JobID: base,
|
||||||
|
ArrayJobName: job.JobName,
|
||||||
|
SubmitUser: job.SubmitUser,
|
||||||
|
QueueName: job.QueueName,
|
||||||
|
UserGroup: job.UserGroup,
|
||||||
|
JobGroup: job.JobGroup,
|
||||||
|
JobDescription: job.JobDescription,
|
||||||
|
JobName: job.JobName,
|
||||||
|
ProjectName: job.ProjectName,
|
||||||
|
LicenseProject: job.LicenseProject,
|
||||||
|
SLA: job.SLA,
|
||||||
|
}}
|
||||||
|
aggs[base] = agg
|
||||||
|
}
|
||||||
|
agg.info.NumJobs++
|
||||||
|
// Classify through JobStateStr so the counters always agree with the
|
||||||
|
// status string every other view shows (suspend states are stored as
|
||||||
|
// RUN|USUSP / RUN|SSUSP / PEND|PSUSP combinations there).
|
||||||
|
switch models.JobStateStr(job.JobStatus) {
|
||||||
|
case "USUSP":
|
||||||
|
agg.info.NumUSusp++
|
||||||
|
case "SSUSP":
|
||||||
|
agg.info.NumSSusp++
|
||||||
|
case "PSUSP":
|
||||||
|
agg.info.NumPSusp++
|
||||||
|
case "RUN":
|
||||||
|
agg.info.NumRun++
|
||||||
|
case "PEND":
|
||||||
|
agg.info.NumPend++
|
||||||
|
case "EXIT":
|
||||||
|
agg.info.NumExit++
|
||||||
|
case "DONE":
|
||||||
|
agg.info.NumDone++
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
out := make([]models.ArrayJobInfo, 0, len(aggs))
|
||||||
|
for _, agg := range aggs {
|
||||||
|
out = append(out, agg.info)
|
||||||
|
}
|
||||||
|
sort.Slice(out, func(i, j int) bool { return out[i].JobID < out[j].JobID })
|
||||||
|
return out
|
||||||
|
}
|
||||||
|
|
||||||
|
// arrayStatsToOrderMaps converts derived stats rows to the OrderMap shape
|
||||||
|
// QueryData hands to ExecuteQuery (mirrors store.StoreToOrderMaps).
|
||||||
|
func arrayStatsToOrderMaps(rows []models.ArrayJobInfo) []utils.OrderMap {
|
||||||
|
result := make([]utils.OrderMap, 0, len(rows))
|
||||||
|
for _, row := range rows {
|
||||||
|
m := store.StructToOrderMap(row)
|
||||||
|
// unique_job_id = the base ID, so processJobData's job_id_human
|
||||||
|
// annotation (which reads unique_job_id) covers stats rows the same
|
||||||
|
// way it covers ordinary job rows: `jobs list -A` rows carry the
|
||||||
|
// dash-quad human ID every other view shows.
|
||||||
|
m.Data["unique_job_id"] = row.JobID
|
||||||
|
result = append(result, m)
|
||||||
|
}
|
||||||
|
return result
|
||||||
|
}
|
||||||
160
src/frontend/apiserver/controllers/arraystats_test.go
Normal file
160
src/frontend/apiserver/controllers/arraystats_test.go
Normal file
|
|
@ -0,0 +1,160 @@
|
||||||
|
package controllers
|
||||||
|
|
||||||
|
import (
|
||||||
|
"testing"
|
||||||
|
|
||||||
|
"apiserver/backend"
|
||||||
|
"apiserver/models"
|
||||||
|
"apiserver/store"
|
||||||
|
"apiserver/utils"
|
||||||
|
|
||||||
|
"go.uber.org/zap"
|
||||||
|
)
|
||||||
|
|
||||||
|
func init() {
|
||||||
|
utils.Log = zap.NewNop()
|
||||||
|
}
|
||||||
|
|
||||||
|
// Array statistics (type=array_stats, `xce jobs list -A`) are derived from
|
||||||
|
// the live job store -- store.ArrayJobs has no production writer, so a
|
||||||
|
// store-backed read always answered zero rows. These tests pin the derived
|
||||||
|
// counters, identity fields and the human-ID annotation.
|
||||||
|
|
||||||
|
// arrayStatsBase / arrayStatsBase2 are array-capable base IDs (tag 0, low 15
|
||||||
|
// bits clear), as generateArrayJobID hands out.
|
||||||
|
var arrayStatsBase = []uint64{
|
||||||
|
uint64(3)<<40 | uint64(1)<<20,
|
||||||
|
uint64(4)<<40 | uint64(1)<<20,
|
||||||
|
}
|
||||||
|
|
||||||
|
// mkArrayElement builds an array element row as the controller's push path
|
||||||
|
// stores it: unique_job_id = MakeElementID(base, idx), job_id = base.
|
||||||
|
func mkArrayElement(base uint64, idx int, status string) models.JobInfo {
|
||||||
|
uid := backend.MakeElementID(base, idx)
|
||||||
|
return models.JobInfo{
|
||||||
|
Unique_JobID: int64(uid),
|
||||||
|
JobID: int64(base),
|
||||||
|
ArrayIdx: idx,
|
||||||
|
JobStatus: models.JOBSTAT[status],
|
||||||
|
JobName: "arrv",
|
||||||
|
SubmitUser: "alice",
|
||||||
|
QueueName: "normal",
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestDeriveArrayJobStats_CountsPerStatus(t *testing.T) {
|
||||||
|
store.InitStores()
|
||||||
|
|
||||||
|
// Array base[0]: 2 DONE, 1 RUN, 1 PEND.
|
||||||
|
for _, e := range []struct {
|
||||||
|
idx int
|
||||||
|
status string
|
||||||
|
}{
|
||||||
|
{1, "DONE"}, {2, "DONE"}, {3, "RUN"}, {4, "PEND"},
|
||||||
|
} {
|
||||||
|
store.Jobs.Put(uint64(mkArrayElement(arrayStatsBase[0], e.idx, e.status).Unique_JobID),
|
||||||
|
mkArrayElement(arrayStatsBase[0], e.idx, e.status))
|
||||||
|
}
|
||||||
|
// Array base[1]: 1 EXIT.
|
||||||
|
store.Jobs.Put(uint64(mkArrayElement(arrayStatsBase[1], 1, "EXIT").Unique_JobID),
|
||||||
|
mkArrayElement(arrayStatsBase[1], 1, "EXIT"))
|
||||||
|
// An ordinary job must not appear in the stats.
|
||||||
|
store.Jobs.Put(999, models.JobInfo{Unique_JobID: 999, JobID: 999,
|
||||||
|
JobStatus: models.JOBSTAT["RUN"], JobName: "plain"})
|
||||||
|
|
||||||
|
rows := deriveArrayJobStats()
|
||||||
|
if len(rows) != 2 {
|
||||||
|
t.Fatalf("expected 2 array rows (ordinary job excluded), got %d: %+v", len(rows), rows)
|
||||||
|
}
|
||||||
|
|
||||||
|
first := rows[0]
|
||||||
|
if first.JobID != int64(arrayStatsBase[0]) {
|
||||||
|
t.Fatalf("rows sorted by job_id: first = %d, want %d", first.JobID, arrayStatsBase[0])
|
||||||
|
}
|
||||||
|
if first.NumJobs != 4 || first.NumDone != 2 || first.NumRun != 1 || first.NumPend != 1 {
|
||||||
|
t.Errorf("array base[0] counters: jobs=%d done=%d run=%d pend=%d, want 4/2/1/1",
|
||||||
|
first.NumJobs, first.NumDone, first.NumRun, first.NumPend)
|
||||||
|
}
|
||||||
|
second := rows[1]
|
||||||
|
if second.JobID != int64(arrayStatsBase[1]) || second.NumJobs != 1 || second.NumExit != 1 {
|
||||||
|
t.Errorf("array base[1] counters: jobs=%d exit=%d, want 1/1", second.NumJobs, second.NumExit)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestDeriveArrayJobStats_IdentityFromFirstElement(t *testing.T) {
|
||||||
|
store.InitStores()
|
||||||
|
|
||||||
|
e := mkArrayElement(arrayStatsBase[0], 2, "RUN")
|
||||||
|
e.UserGroup = "ug1"
|
||||||
|
e.ProjectName = "proj"
|
||||||
|
store.Jobs.Put(uint64(e.Unique_JobID), e)
|
||||||
|
// A later element with different identity fields must not win (elements
|
||||||
|
// of one array share the submit spec; first-seen supplies the identity).
|
||||||
|
e2 := mkArrayElement(arrayStatsBase[0], 1, "DONE")
|
||||||
|
e2.SubmitUser = "bob"
|
||||||
|
store.Jobs.Put(uint64(e2.Unique_JobID), e2)
|
||||||
|
|
||||||
|
rows := deriveArrayJobStats()
|
||||||
|
if len(rows) != 1 {
|
||||||
|
t.Fatalf("expected 1 array row, got %d", len(rows))
|
||||||
|
}
|
||||||
|
got := rows[0]
|
||||||
|
if got.SubmitUser != "alice" || got.UserGroup != "ug1" || got.ProjectName != "proj" {
|
||||||
|
t.Errorf("identity fields: user=%q ug=%q proj=%q, want alice/ug1/proj",
|
||||||
|
got.SubmitUser, got.UserGroup, got.ProjectName)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestDeriveArrayJobStats_SuspendSubStates(t *testing.T) {
|
||||||
|
store.InitStores()
|
||||||
|
|
||||||
|
for _, e := range []struct {
|
||||||
|
idx int
|
||||||
|
status int
|
||||||
|
}{
|
||||||
|
// Suspend states are stored as RUN|USUSP / PEND|PSUSP
|
||||||
|
// combinations (JobStateStr's classification; the controller
|
||||||
|
// ORs the suspend bit onto the base state). SSUSP currently has
|
||||||
|
// no producer, so it is covered by the USUSP case symmetrically.
|
||||||
|
{1, models.JOBSTAT["RUN"] | models.JOBSTAT["USUSP"]},
|
||||||
|
{2, models.JOBSTAT["RUN"] | models.JOBSTAT["SSUSP"]},
|
||||||
|
} {
|
||||||
|
ele := mkArrayElement(arrayStatsBase[0], e.idx, "RUN")
|
||||||
|
ele.JobStatus = e.status
|
||||||
|
store.Jobs.Put(uint64(ele.Unique_JobID), ele)
|
||||||
|
}
|
||||||
|
psusp := mkArrayElement(arrayStatsBase[0], 3, "PEND")
|
||||||
|
psusp.JobStatus = models.JOBSTAT["PEND"] | models.JOBSTAT["PSUSP"]
|
||||||
|
store.Jobs.Put(uint64(psusp.Unique_JobID), psusp)
|
||||||
|
|
||||||
|
rows := deriveArrayJobStats()
|
||||||
|
if len(rows) != 1 {
|
||||||
|
t.Fatalf("expected 1 array row, got %d", len(rows))
|
||||||
|
}
|
||||||
|
got := rows[0]
|
||||||
|
if got.NumJobs != 3 || got.NumSSusp != 1 || got.NumUSusp != 1 || got.NumPSusp != 1 ||
|
||||||
|
got.NumRun != 0 || got.NumPend != 0 {
|
||||||
|
t.Errorf("suspend counters: jobs=%d ssusp=%d ususp=%d psusp=%d run=%d pend=%d",
|
||||||
|
got.NumJobs, got.NumSSusp, got.NumUSusp, got.NumPSusp, got.NumRun, got.NumPend)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestArrayStatsToOrderMaps_AnnotatesHumanID(t *testing.T) {
|
||||||
|
stats := []models.ArrayJobInfo{{JobID: int64(arrayStatsBase[0]), NumJobs: 2, NumDone: 2}}
|
||||||
|
rows := arrayStatsToOrderMaps(stats)
|
||||||
|
if len(rows) != 1 {
|
||||||
|
t.Fatalf("expected 1 row, got %d", len(rows))
|
||||||
|
}
|
||||||
|
// The conversion carries unique_job_id (the array base ID); processJobData
|
||||||
|
// turns it into job_id_human at the HTTP layer, mirroring element rows.
|
||||||
|
got, ok := rows[0].Data["unique_job_id"]
|
||||||
|
if !ok {
|
||||||
|
t.Fatal("unique_job_id missing from stats row")
|
||||||
|
}
|
||||||
|
if got != int64(arrayStatsBase[0]) {
|
||||||
|
t.Errorf("unique_job_id = %v, want %d", got, arrayStatsBase[0])
|
||||||
|
}
|
||||||
|
if got, _ := rows[0].Data["num_jobs"].(int); got != 2 {
|
||||||
|
t.Errorf("num_jobs = %v, want 2", rows[0].Data["num_jobs"])
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
@ -325,7 +325,10 @@ func (j JobController) GetJobsInfo(ctx *gin.Context) {
|
||||||
// ---- check fields ----
|
// ---- check fields ----
|
||||||
// get fields and do check, if not set, use default display columns
|
// get fields and do check, if not set, use default display columns
|
||||||
if typeValue == "array_stats" {
|
if typeValue == "array_stats" {
|
||||||
fields = models.ARRAY_JOB_DEFAULT_COLUMN
|
// unique_job_id must ride along in the field list: ExecuteQuery's
|
||||||
|
// SelectFields drops anything not requested, and processJobData
|
||||||
|
// annotates job_id_human from it.
|
||||||
|
fields = append(append([]string{}, models.ARRAY_JOB_DEFAULT_COLUMN...), "unique_job_id")
|
||||||
} else {
|
} else {
|
||||||
if fieldsStr != "" {
|
if fieldsStr != "" {
|
||||||
fields = utils.ParseFields(fieldsStr)
|
fields = utils.ParseFields(fieldsStr)
|
||||||
|
|
@ -478,7 +481,13 @@ func (j JobController) GetJobsInfo(ctx *gin.Context) {
|
||||||
var jqueryReply []utils.OrderMap // Declare jqueryReply
|
var jqueryReply []utils.OrderMap // Declare jqueryReply
|
||||||
pagetoken := "" // Initialize pagetoken with an empty string or appropriate value
|
pagetoken := "" // Initialize pagetoken with an empty string or appropriate value
|
||||||
if typeValue == "array_stats" {
|
if typeValue == "array_stats" {
|
||||||
jqueryReply, pagetoken, err = services.QueryData(services.TableArrayJob, queryOpts, fields)
|
// Stats rows are derived from the live job store (arraystats.go):
|
||||||
|
// store.ArrayJobs has no production writer, so the query used to
|
||||||
|
// always answer zero rows (`xce jobs list -A` → "no job found").
|
||||||
|
jqueryReply = arrayStatsToOrderMaps(deriveArrayJobStats())
|
||||||
|
// Pagination over derived rows: run the same filter → sort →
|
||||||
|
// paginate → select pipeline the store path would.
|
||||||
|
jqueryReply, pagetoken = store.ExecuteQuery(jqueryReply, queryOpts, fields)
|
||||||
} else {
|
} else {
|
||||||
jqueryReply, pagetoken, err = services.QueryData(services.TableJobInfo, queryOpts, fields)
|
jqueryReply, pagetoken, err = services.QueryData(services.TableJobInfo, queryOpts, fields)
|
||||||
}
|
}
|
||||||
|
|
@ -684,7 +693,8 @@ func (j JobController) GetJobByIDs(ctx *gin.Context) {
|
||||||
|
|
||||||
var fields []string
|
var fields []string
|
||||||
if typeValue == "array_stats" {
|
if typeValue == "array_stats" {
|
||||||
fields = models.ARRAY_JOB_DEFAULT_COLUMN
|
// same unique_job_id requirement as GetJobsInfo's array_stats path
|
||||||
|
fields = append(append([]string{}, models.ARRAY_JOB_DEFAULT_COLUMN...), "unique_job_id")
|
||||||
} else {
|
} else {
|
||||||
if fieldsStr != "" {
|
if fieldsStr != "" {
|
||||||
fields = utils.ParseFields(fieldsStr)
|
fields = utils.ParseFields(fieldsStr)
|
||||||
|
|
@ -765,7 +775,11 @@ func (j JobController) GetJobByIDs(ctx *gin.Context) {
|
||||||
var jqueryReply []utils.OrderMap // Declare jqueryReply
|
var jqueryReply []utils.OrderMap // Declare jqueryReply
|
||||||
|
|
||||||
if typeValue == "array_stats" {
|
if typeValue == "array_stats" {
|
||||||
jqueryReply, page_token, err = services.QueryData(services.TableArrayJob, queryOpts, fields)
|
// Stats rows are derived from the live job store
|
||||||
|
// (arraystats.go): store.ArrayJobs has no production writer,
|
||||||
|
// so the query used to always answer zero rows.
|
||||||
|
jqueryReply = arrayStatsToOrderMaps(deriveArrayJobStats())
|
||||||
|
jqueryReply, page_token = store.ExecuteQuery(jqueryReply, queryOpts, fields)
|
||||||
} else {
|
} else {
|
||||||
jqueryReply, page_token, err = services.QueryData(services.TableJobInfo, queryOpts, fields)
|
jqueryReply, page_token, err = services.QueryData(services.TableJobInfo, queryOpts, fields)
|
||||||
}
|
}
|
||||||
|
|
@ -1531,9 +1545,6 @@ func preCheckJobRequest(req *models.JobSubmitRequest) (n int, isArray bool, err
|
||||||
if req.ShellLimit.DataLimit.MB != 0 {
|
if req.ShellLimit.DataLimit.MB != 0 {
|
||||||
req.DataLimit = req.ShellLimit.DataLimit
|
req.DataLimit = req.ShellLimit.DataLimit
|
||||||
}
|
}
|
||||||
if req.ShellLimit.StackLimit.MB != 0 {
|
|
||||||
req.StackLimit = req.ShellLimit.DataLimit
|
|
||||||
}
|
|
||||||
if req.ShellLimit.StackLimit.MB != 0 {
|
if req.ShellLimit.StackLimit.MB != 0 {
|
||||||
req.StackLimit = req.ShellLimit.StackLimit
|
req.StackLimit = req.ShellLimit.StackLimit
|
||||||
}
|
}
|
||||||
|
|
@ -1844,9 +1855,6 @@ func createArrayName(jobName string) (arrayName string, indices []int, slimit in
|
||||||
if err != nil || index < 1 {
|
if err != nil || index < 1 {
|
||||||
return "", nil, -1, fmt.Errorf("index must be a positive integer")
|
return "", nil, -1, fmt.Errorf("index must be a positive integer")
|
||||||
}
|
}
|
||||||
if err != nil {
|
|
||||||
return "", nil, -1, fmt.Errorf("invalid index value: %s", part)
|
|
||||||
}
|
|
||||||
if _, exists := seenIndices[index]; !exists {
|
if _, exists := seenIndices[index]; !exists {
|
||||||
seenIndices[index] = struct{}{}
|
seenIndices[index] = struct{}{}
|
||||||
indices = append(indices, index)
|
indices = append(indices, index)
|
||||||
|
|
@ -3015,12 +3023,23 @@ func normalizeJobStatus(v any) (int, bool) {
|
||||||
// produces as int, panicking every control request into an HTTP 500 before it
|
// produces as int, panicking every control request into an HTTP 500 before it
|
||||||
// reached the controller.
|
// reached the controller.
|
||||||
func jobArrayIdx(row utils.OrderMap) int {
|
func jobArrayIdx(row utils.OrderMap) int {
|
||||||
if v, ok := row.Data["array_idx"]; ok {
|
n, _ := rowInt64(row, "array_idx")
|
||||||
if n, ok := normalizeJobInt64(v); ok {
|
return int(n)
|
||||||
return int(n)
|
}
|
||||||
}
|
|
||||||
|
// rowInt64 reads an integer column of a store row as int64, tolerating every
|
||||||
|
// numeric representation the store or a JSON decode path produces (int,
|
||||||
|
// int32, int64, float64, json.Number; uint64 bit-pattern preserved, as
|
||||||
|
// rowUint64 does). Returns ok=false when the column is absent or not numeric.
|
||||||
|
func rowInt64(row utils.OrderMap, col string) (int64, bool) {
|
||||||
|
v, ok := row.Data[col]
|
||||||
|
if !ok {
|
||||||
|
return 0, false
|
||||||
}
|
}
|
||||||
return 0
|
if u, isUint64 := v.(uint64); isUint64 {
|
||||||
|
return int64(u), true
|
||||||
|
}
|
||||||
|
return normalizeJobInt64(v)
|
||||||
}
|
}
|
||||||
|
|
||||||
func processJobData(rows []utils.OrderMap, pagetoken string) utils.QueryResult {
|
func processJobData(rows []utils.OrderMap, pagetoken string) utils.QueryResult {
|
||||||
|
|
@ -3031,21 +3050,13 @@ func processJobData(rows []utils.OrderMap, pagetoken string) utils.QueryResult {
|
||||||
// format, which parses only that form. Data correctness lives here,
|
// format, which parses only that form. Data correctness lives here,
|
||||||
// not in the clients. FormatJobID is the apiserver's own C fmt_jobid
|
// not in the clients. FormatJobID is the apiserver's own C fmt_jobid
|
||||||
// equivalent (backend.FormatJobID).
|
// equivalent (backend.FormatJobID).
|
||||||
if uid, exist := row.Data["unique_job_id"]; exist {
|
if jobID, ok := rowInt64(row, "unique_job_id"); ok && jobID != 0 {
|
||||||
var jobID int64
|
// Human-readable job ID (cluster-apiserver-timestamp-seq) so
|
||||||
switch v := uid.(type) {
|
// clients (MCP server, CLI) can cross-reference with the
|
||||||
case int64:
|
// xce_test/xcecli dash format, which parses only that form. Data
|
||||||
jobID = v
|
// correctness lives here, not in the clients. FormatJobID is the
|
||||||
case int:
|
// apiserver's own C fmt_jobid equivalent (backend.FormatJobID).
|
||||||
jobID = int64(v)
|
row.Data["job_id_human"] = backend.FormatJobID(jobID)
|
||||||
case int32:
|
|
||||||
jobID = int64(v)
|
|
||||||
case float64:
|
|
||||||
jobID = int64(v)
|
|
||||||
}
|
|
||||||
if jobID != 0 {
|
|
||||||
row.Data["job_id_human"] = backend.FormatJobID(jobID)
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
if v, exist := row.Data["job_status"]; exist {
|
if v, exist := row.Data["job_status"]; exist {
|
||||||
|
|
@ -3059,16 +3070,16 @@ func processJobData(rows []utils.OrderMap, pagetoken string) utils.QueryResult {
|
||||||
if jobName, exist := row.Data["job_name"]; exist {
|
if jobName, exist := row.Data["job_name"]; exist {
|
||||||
// Use IsArrayJob to detect array children, which covers
|
// Use IsArrayJob to detect array children, which covers
|
||||||
// the first child (idx=0) that the old array_idx > 0 check missed.
|
// the first child (idx=0) that the old array_idx > 0 check missed.
|
||||||
var isArray bool
|
// The ID read goes through the tolerant rowInt64 (bug-1490's
|
||||||
if uid, ok := row.Data["unique_job_id"].(int64); ok {
|
// lesson: a bare .(int64) assertion silently skips the row when
|
||||||
|
// the store or a JSON decode path produced another numeric
|
||||||
|
// representation).
|
||||||
|
isArray := false
|
||||||
|
if uid, ok := rowInt64(row, "unique_job_id"); ok {
|
||||||
isArray = backend.IsArrayJob(uint64(uid))
|
isArray = backend.IsArrayJob(uint64(uid))
|
||||||
}
|
}
|
||||||
if isArray {
|
if isArray {
|
||||||
if arrayIdx, ok := row.Data["array_idx"].(int); ok {
|
row.Data["job_name"] = fmt.Sprintf("%s[%d]", jobName, jobArrayIdx(row))
|
||||||
row.Data["job_name"] = fmt.Sprintf("%s[%d]", jobName, arrayIdx)
|
|
||||||
} else if arrayIdx64, ok := row.Data["array_idx"].(int64); ok {
|
|
||||||
row.Data["job_name"] = fmt.Sprintf("%s[%d]", jobName, arrayIdx64)
|
|
||||||
}
|
|
||||||
} else {
|
} else {
|
||||||
row.Data["job_name"] = jobName
|
row.Data["job_name"] = jobName
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -973,6 +973,111 @@ func assert400(t *testing.T, name string, w *httptest.ResponseRecorder, want str
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// arrayStatsRows decodes the rows of a type=array_stats list reply.
|
||||||
|
func arrayStatsRows(t *testing.T, w *httptest.ResponseRecorder) []map[string]interface{} {
|
||||||
|
t.Helper()
|
||||||
|
if w.Code != http.StatusOK {
|
||||||
|
t.Fatalf("status = %d, want 200; body: %s", w.Code, w.Body.String())
|
||||||
|
}
|
||||||
|
dec := json.NewDecoder(bytes.NewReader(w.Body.Bytes()))
|
||||||
|
dec.UseNumber()
|
||||||
|
var raw struct {
|
||||||
|
Data struct {
|
||||||
|
Count int `json:"count"`
|
||||||
|
Rows []struct {
|
||||||
|
Data map[string]interface{} `json:"Data"`
|
||||||
|
} `json:"rows"`
|
||||||
|
} `json:"data"`
|
||||||
|
}
|
||||||
|
if err := dec.Decode(&raw); err != nil {
|
||||||
|
t.Fatalf("decode array_stats reply: %v; body: %s", err, w.Body.String())
|
||||||
|
}
|
||||||
|
rows := make([]map[string]interface{}, 0, len(raw.Data.Rows))
|
||||||
|
for _, r := range raw.Data.Rows {
|
||||||
|
rows = append(rows, r.Data)
|
||||||
|
}
|
||||||
|
if raw.Data.Count != len(rows) {
|
||||||
|
t.Errorf("count = %d, rows = %d", raw.Data.Count, len(rows))
|
||||||
|
}
|
||||||
|
return rows
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestJobHandler_ArrayStatsDerivedFromJobStore covers `xce jobs list -A`
|
||||||
|
// (type=array_stats): rows are derived from the live job store, one per
|
||||||
|
// array, with counters matching the elements' states and a job_id_human
|
||||||
|
// annotation (store.ArrayJobs has no production writer).
|
||||||
|
func TestJobHandler_ArrayStatsDerivedFromJobStore(t *testing.T) {
|
||||||
|
f := newHandlerFixture(t)
|
||||||
|
s := seedQueryJobs(f)
|
||||||
|
x, y := s.X, s.Y
|
||||||
|
X := backend.FormatJobID(int64(x))
|
||||||
|
|
||||||
|
// Move element 2 of X to DONE so its counters must shift accordingly
|
||||||
|
// (seedArrayElements stored all four as RUN).
|
||||||
|
elem2 := backend.MakeElementID(x, 2)
|
||||||
|
got, _ := store.Jobs.Get(elem2)
|
||||||
|
got.JobStatus = models.JOBSTAT["DONE"]
|
||||||
|
store.Jobs.Put(elem2, got)
|
||||||
|
|
||||||
|
// GetJobsInfo?type=array_stats lists every seeded array with counters.
|
||||||
|
w := f.do(http.MethodGet, "/xce/v1/jobs?type=array_stats", nil)
|
||||||
|
rows := arrayStatsRows(t, w)
|
||||||
|
if len(rows) != 4 {
|
||||||
|
t.Fatalf("array_stats rows = %d, want 4 (arrays X, Y, Z, W)", len(rows))
|
||||||
|
}
|
||||||
|
|
||||||
|
byJobID := map[uint64]map[string]interface{}{}
|
||||||
|
for _, row := range rows {
|
||||||
|
n, ok := row["job_id"].(json.Number)
|
||||||
|
if !ok {
|
||||||
|
t.Fatalf("stats row without job_id: %v", row)
|
||||||
|
}
|
||||||
|
id, _ := strconv.ParseUint(n.String(), 10, 64)
|
||||||
|
byJobID[id] = row
|
||||||
|
}
|
||||||
|
|
||||||
|
xRow := byJobID[x]
|
||||||
|
if xRow == nil {
|
||||||
|
t.Fatalf("array X missing from stats: %v", rows)
|
||||||
|
}
|
||||||
|
// X has elements 1,2,3,4: 3 RUN + 1 DONE after the overwrite.
|
||||||
|
for field, want := range map[string]float64{
|
||||||
|
"num_jobs": 4, "num_run": 3, "num_done": 1,
|
||||||
|
"num_pend": 0, "num_exit": 0,
|
||||||
|
} {
|
||||||
|
n, _ := xRow[field].(json.Number)
|
||||||
|
if got, _ := n.Float64(); got != want {
|
||||||
|
t.Errorf("X.%s = %v, want %v", field, n.String(), want)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if human, _ := xRow["job_id_human"].(string); human != X {
|
||||||
|
t.Errorf("X.job_id_human = %q, want %q", human, X)
|
||||||
|
}
|
||||||
|
|
||||||
|
yRow := byJobID[y]
|
||||||
|
if yRow == nil {
|
||||||
|
t.Fatalf("array Y missing from stats: %v", rows)
|
||||||
|
}
|
||||||
|
if n, _ := yRow["num_jobs"].(json.Number); n.String() != "2" {
|
||||||
|
t.Errorf("Y.num_jobs = %v, want 2", yRow["num_jobs"])
|
||||||
|
}
|
||||||
|
|
||||||
|
// job_id filter (the ARRAY_JOB_SUPPORT_FILTERS path) narrows to one row.
|
||||||
|
w = f.do(http.MethodGet, "/xce/v1/jobs?type=array_stats&filter="+
|
||||||
|
url.QueryEscape("{job_id:eq:"+X+"}"), nil)
|
||||||
|
rows = arrayStatsRows(t, w)
|
||||||
|
if len(rows) != 1 {
|
||||||
|
t.Fatalf("filtered array_stats rows = %d, want 1", len(rows))
|
||||||
|
}
|
||||||
|
|
||||||
|
// GetJobByIDs?type=array_stats answers the same derived row.
|
||||||
|
w = f.do(http.MethodGet, "/xce/v1/jobs/"+url.PathEscape(X)+"?type=array_stats", nil)
|
||||||
|
rows = arrayStatsRows(t, w)
|
||||||
|
if len(rows) != 1 {
|
||||||
|
t.Fatalf("GetJobByIDs array_stats rows = %d, want 1; body: %s", len(rows), w.Body.String())
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
// TestJobHandler_QuerySelections covers the GetJobsInfo rows of the design's
|
// TestJobHandler_QuerySelections covers the GetJobsInfo rows of the design's
|
||||||
// Testing Strategy (Query).
|
// Testing Strategy (Query).
|
||||||
func TestJobHandler_QuerySelections(t *testing.T) {
|
func TestJobHandler_QuerySelections(t *testing.T) {
|
||||||
|
|
|
||||||
|
|
@ -3,6 +3,7 @@ package controllers
|
||||||
import (
|
import (
|
||||||
"testing"
|
"testing"
|
||||||
|
|
||||||
|
"apiserver/backend"
|
||||||
"apiserver/utils"
|
"apiserver/utils"
|
||||||
|
|
||||||
"go.uber.org/zap"
|
"go.uber.org/zap"
|
||||||
|
|
@ -55,3 +56,66 @@ func TestProcessJobData_NoJobIDHumanWhenZero(t *testing.T) {
|
||||||
t.Error("job_id_human should not be set for zero job id")
|
t.Error("job_id_human should not be set for zero job id")
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// The array-child [idx] name annotation must survive every numeric
|
||||||
|
// representation a store row or JSON decode path can produce. The old code
|
||||||
|
// asserted .(int64)/.(int) and silently left the suffix off for anything
|
||||||
|
// else (bug-1490's lesson, this time as a silent wrong value).
|
||||||
|
func TestProcessJobData_ArrayNameSuffix_NumericRepresentations(t *testing.T) {
|
||||||
|
// 1-0-1786585485-0[2]: an array-capable base with the array tag and
|
||||||
|
// index 2 packed in, as backend.MakeElementID composes it.
|
||||||
|
uid := int64(backend.MakeElementID(uint64(1)<<40|uint64(1)<<20, 2))
|
||||||
|
|
||||||
|
for name, tc := range map[string]struct {
|
||||||
|
uid any
|
||||||
|
idx any
|
||||||
|
}{
|
||||||
|
"int64/int": {uid, int(2)},
|
||||||
|
"int64/int64": {uid, int64(2)},
|
||||||
|
"float64/int": {float64(uid), int(2)},
|
||||||
|
"float64/int64": {float64(uid), int64(2)},
|
||||||
|
"float64/float": {float64(uid), float64(2)},
|
||||||
|
} {
|
||||||
|
rows := []utils.OrderMap{
|
||||||
|
{
|
||||||
|
Fields: []string{"unique_job_id", "job_name"},
|
||||||
|
Data: map[string]interface{}{
|
||||||
|
"unique_job_id": tc.uid,
|
||||||
|
"job_name": "arrv",
|
||||||
|
},
|
||||||
|
},
|
||||||
|
}
|
||||||
|
// array_idx rides the row when present (store path) and defaults to
|
||||||
|
// jobArrayIdx's 0 when absent (some JSON decode paths).
|
||||||
|
rows[0].Data["array_idx"] = tc.idx
|
||||||
|
res := processJobData(rows, "")
|
||||||
|
if got, _ := res.Rows[0].Data["job_name"].(string); got != "arrv[2]" {
|
||||||
|
t.Errorf("%s: job_name = %q, want arrv[2]", name, got)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// uint64 unique_job_id (a representation rowUint64 handles elsewhere) must
|
||||||
|
// still be recognized as an array child and formatted as a human ID.
|
||||||
|
func TestProcessJobData_Uint64UniqueJobID(t *testing.T) {
|
||||||
|
uid := backend.MakeElementID(uint64(1)<<40|uint64(1)<<20, 3)
|
||||||
|
rows := []utils.OrderMap{
|
||||||
|
{
|
||||||
|
Fields: []string{"unique_job_id", "job_name", "array_idx"},
|
||||||
|
Data: map[string]interface{}{
|
||||||
|
"unique_job_id": uid,
|
||||||
|
"job_name": "arrv",
|
||||||
|
"array_idx": uint64(3),
|
||||||
|
},
|
||||||
|
},
|
||||||
|
}
|
||||||
|
res := processJobData(rows, "")
|
||||||
|
row := res.Rows[0]
|
||||||
|
if got, _ := row.Data["job_name"].(string); got != "arrv[3]" {
|
||||||
|
t.Errorf("job_name = %v, want arrv[3]", row.Data["job_name"])
|
||||||
|
}
|
||||||
|
want := backend.FormatJobID(int64(uid))
|
||||||
|
if got, _ := row.Data["job_id_human"].(string); got != want {
|
||||||
|
t.Errorf("job_id_human = %v, want %v", row.Data["job_id_human"], want)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
|
||||||
|
|
@ -40,7 +40,11 @@ func QueryData(tableName string, queryOpts *utils.QueryOptions, fields []string)
|
||||||
case TableClusterInfo:
|
case TableClusterInfo:
|
||||||
rows = store.StoreToOrderMaps(store.Clusters)
|
rows = store.StoreToOrderMaps(store.Clusters)
|
||||||
case TableArrayJob:
|
case TableArrayJob:
|
||||||
rows = store.StoreToOrderMaps(store.ArrayJobs)
|
// Array stats (type=array_stats) are derived from the live job store
|
||||||
|
// by the controllers package (arraystats.go) -- store.ArrayJobs has
|
||||||
|
// no production writer, so reading it here always answered zero rows.
|
||||||
|
// This case stays reachable only if a future caller forgets that.
|
||||||
|
return nil, "", fmt.Errorf("array stats are derived by the caller; QueryData(TableArrayJob) is not supported")
|
||||||
case TableJobExecHosts:
|
case TableJobExecHosts:
|
||||||
rows = store.StoreToOrderMaps(store.JobExecHosts)
|
rows = store.StoreToOrderMaps(store.JobExecHosts)
|
||||||
case TableRolePermission:
|
case TableRolePermission:
|
||||||
|
|
|
||||||
|
|
@ -152,7 +152,7 @@ None of these exist in the current codebase. The Swagger UI will show completely
|
||||||
| **CL2** | TLS verification disabled | `request.go` `SetRootCAs` | `InsecureSkipVerify: true` always set; `loadRootCAs` is defined but never called |
|
| **CL2** | TLS verification disabled | `request.go` `SetRootCAs` | `InsecureSkipVerify: true` always set; `loadRootCAs` is defined but never called |
|
||||||
| **CL3** | Token header space inconsistency | `request.go` | `SubmitJob` uses `" "+token`; all other methods use `""+token` |
|
| **CL3** | Token header space inconsistency | `request.go` | `SubmitJob` uses `" "+token`; all other methods use `""+token` |
|
||||||
| **CL4** | No retry on transient errors | `request.go` | Single attempt; network blips cause immediate failure |
|
| **CL4** | No retry on transient errors | `request.go` | Single attempt; network blips cause immediate failure |
|
||||||
| **CL5** | `xce jobs` array stats has no display | `cmd/jobs/list.go` | `?type=array_stats` is sent but output formatter doesn't handle `ArrayJobStatsData` |
|
| **CL5** | ~~`xce jobs` array stats has no display~~ **FIXED** | `cmd/jobs/list.go` | Display path exists (`ArrayJobsPagePrinter`); the real gap was server-side — `?type=array_stats` read `store.ArrayJobs` which has no writer and always answered zero rows. Stats are now derived from the live job store (controllers/arraystats.go). |
|
||||||
|
|
||||||
---
|
---
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -507,6 +507,19 @@ type ArrayJobStatsData struct {
|
||||||
NumPSusp int `json:"num_psusp"`
|
NumPSusp int `json:"num_psusp"`
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// UnmarshalJSON decodes both the OrderMap wire format the server sends for
|
||||||
|
// store-backed rows ({"Fields":[...],"Data":{...}}) and a bare object, so
|
||||||
|
// array_stats replies decode past the wrapper instead of landing all-zero.
|
||||||
|
func (s *ArrayJobStatsData) UnmarshalJSON(data []byte) error {
|
||||||
|
type alias ArrayJobStatsData
|
||||||
|
var a alias
|
||||||
|
if err := UnmarshalOrderMapRow(data, &a); err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
*s = ArrayJobStatsData(a)
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
// job array statistics response data structure
|
// job array statistics response data structure
|
||||||
type ArrayJobsStatsResponse struct {
|
type ArrayJobsStatsResponse struct {
|
||||||
Code int `json:"code"`
|
Code int `json:"code"`
|
||||||
|
|
|
||||||
|
|
@ -93,3 +93,35 @@ func TestPendingReasonUnmarshalUnknownObject(t *testing.T) {
|
||||||
t.Fatalf("unknown object: want raw fallback, got %q", string(p))
|
t.Fatalf("unknown object: want raw fallback, got %q", string(p))
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func TestArrayJobStatsData_UnmarshalOrderMapWrapper(t *testing.T) {
|
||||||
|
// The server sends store rows as OrderMap wrappers; without the custom
|
||||||
|
// UnmarshalJSON every stats field decoded to zero (CL5's "no display"
|
||||||
|
// symptom: xce jobs list -A showed 0/- rows against live arrays).
|
||||||
|
row := `{"Fields":["job_id","array_job_name","submit_user","num_jobs"],
|
||||||
|
"Data":{"job_id":73937137023582208,
|
||||||
|
"array_job_name":"arrv",
|
||||||
|
"submit_user":"alice",
|
||||||
|
"num_jobs":4,
|
||||||
|
"num_done":1,
|
||||||
|
"num_run":3}}`
|
||||||
|
var s ArrayJobStatsData
|
||||||
|
if err := json.Unmarshal([]byte(row), &s); err != nil {
|
||||||
|
t.Fatalf("ArrayJobStatsData unmarshal: %v", err)
|
||||||
|
}
|
||||||
|
if s.JobID != 73937137023582208 || s.ArrayJobName != "arrv" || s.SubmitUser != "alice" ||
|
||||||
|
s.NumJobs != 4 || s.NumDone != 1 || s.NumRun != 3 {
|
||||||
|
t.Fatalf("decoded stats: %+v", s)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestArrayJobStatsData_UnmarshalBareObject(t *testing.T) {
|
||||||
|
// Bare objects (no wrapper) must keep decoding as before.
|
||||||
|
var s ArrayJobStatsData
|
||||||
|
if err := json.Unmarshal([]byte(`{"job_id":7,"num_pend":2}`), &s); err != nil {
|
||||||
|
t.Fatalf("bare object unmarshal: %v", err)
|
||||||
|
}
|
||||||
|
if s.JobID != 7 || s.NumPend != 2 {
|
||||||
|
t.Fatalf("decoded stats: %+v", s)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue