From 893812e7aa895c52fb11dad82266ff27a715f8d6 Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Thu, 16 Apr 2026 11:25:03 +0000 Subject: [PATCH] feat: P0-1e expand handleConfigPush with host config and cluster dispatch Wires the remaining cluster_config_t data into the notification pipeline: - dispatchConfigPush() extracted as a pure-Go helper (CGO-free, testable) that drives OnQueueUpdated, OnResourceUpdated, OnHostConfigUpdated, and OnClusterUpdated from a single config push event - handleConfigPush() delegates to dispatchConfigPush via the new helper, eliminating the previous inline loops - OnHostConfigUpdated and OnClusterUpdated added to NotificationHandler interface and implemented with partial-merge semantics in DefaultHandler - rawHostConfigData/rawHostConfigToModel and rawClusterData/rawClusterToModel added to convert_pure.go with their CGO extraction counterparts in convert.go Safety fixes from code review: - OnHostUpdated, OnHostConfigUpdated, OnClusterUpdated now copy *incoming to a local merged value before mutation, closing the TOCTOU window between store.Get and store.Put under concurrent pushes - cClusterConfQueues and cClusterConfResources gain nil-pointer guards matching the existing guard in cClusterConfHostConfigs - handleConfigPush gains an explicit gHandler nil check - cClusterConfToClusterInfo returns nil on empty clustername (prevents phantom empty-key write to store.Clusters); handleConfigPush logs a warning when nil - 36 unit tests passing (CGO_ENABLED=0) --- src/docs/apiserver-backend-tasks.md | 117 ++++++ src/frontend/apiserver/backend/clib.go | 19 +- src/frontend/apiserver/backend/convert.go | 46 ++- .../apiserver/backend/convert_pure.go | 50 +++ .../apiserver/backend/convert_test.go | 198 ++++++++++ src/frontend/apiserver/backend/handler.go | 84 ++++- .../apiserver/backend/handler_test.go | 345 ++++++++++++++++++ .../apiserver/backend/notification.go | 2 + 8 files changed, 850 insertions(+), 11 deletions(-) create mode 100644 src/docs/apiserver-backend-tasks.md create mode 100644 src/frontend/apiserver/backend/handler_test.go diff --git a/src/docs/apiserver-backend-tasks.md b/src/docs/apiserver-backend-tasks.md new file mode 100644 index 0000000..6605c46 --- /dev/null +++ b/src/docs/apiserver-backend-tasks.md @@ -0,0 +1,117 @@ +# API Server Backend Enhancement — Task Tracker + +## Status Legend +- [x] Completed +- [ ] Pending +- [>] In Progress (blocked) + +--- + +## P0-1: Handle Cluster Config Push + +- [x] **#1 P0-1a: Add HostConfig + Cluster fields to models** + - `models/clusters.go` — Added `HeartbeatInterval`, `Vcpus`, `InactiveHeartbeatCount` to `ClusterInfo` + - `models/hosts.go` — Changed `HostStateStr(0)` from `"OK"` to `"UNKNOWN"` + - `models/clusters_test.go` — New: verifies new fields serialize + existing fields unchanged + - `models/hosts_test.go` — New: verifies `HostStateStr` for zero, flags, non-flag, combined + - All 15 tests passing + +- [x] **#2 P0-1b: Add pure Go conversion functions + TDD tests** + - `backend/convert_pure.go` — Added `rawHostConfigData` + `rawHostConfigToModel()`, `rawClusterData` + `rawClusterToModel()` + - `backend/convert_test.go` — 8 new table-driven tests: zero-value ints, empty strings, normal values, worker/config-sourced field isolation + - `TODO(dispatchwindow)` comment added per plan spec + - `DisptWindow` guard test added to catch accidental future CGO layer changes + - All 26 backend tests passing + +- [x] **#3 P0-1c: Add CGO conversion layer in convert.go** + - `backend/convert.go` — Added `cHostConfigToModel()`, `cClusterConfHostConfigs()`, `cClusterConfToClusterInfo()` + - `cResdefToModel` intentionally skipped — `xce_resdef_t` doesn't exist in C codebase and `ResourceInfo` has no store/consumer (YAGNI); existing `cResourceToModel` → `ResourceDefinitionInfo` already handles config push + - `cClusterConfResources` already existed — no change needed + - CGO compilation verified passing + - All unit tests still passing + +- [x] **#4 P0-1d: Implement handlers with partial merge + TDD** + - `backend/notification.go` — Added `OnClusterUpdated(cluster *models.ClusterInfo)` and `OnHostConfigUpdated(host *models.HostInfo)` to `NotificationHandler` interface + - `backend/handler.go` — Implemented partial-merge `OnClusterUpdated` (preserves Version, Status, ControllerName, NumServers, Resources, HostTypes, HostModels, ClusterAdmins), `OnHostConfigUpdated` (preserves HostStatus, CPUFactor, NumCPUs, NumCores, NumThreads, MaxMem, MaxSwap, MaxTmp, NumDisk, NumRun, NumSSusp, NumUSusp, NumDone, Topology, SchedStop), fixed `OnHostUpdated` (preserves HostModel, HostType, MaxSlot, DisptWindow via two-value `Get()`) + - `backend/handler_test.go` — 6 new tests: new host, partial merge for each of OnHostConfigUpdated, OnClusterUpdated, OnHostUpdated + - All backend tests passing (32 total) + +- [x] **#5 P0-1e: Expand handleConfigPush** (unblocked) + - `backend/clib.go` — Expand `handleConfigPush()` to call `cClusterConfHostConfigs()` and `cClusterConfToClusterInfo()` + - Call `gHandler.OnHostConfigUpdated()` per host config, `gHandler.OnClusterUpdated()` for cluster settings + +--- + +## P0-2: Host Control via Backend + +- [ ] **#6 P0-2a: C-side plumbing for host control** + - `schema/controller.capnp` — Add `HostControl` struct + `hostctrlmsg @21` + - `include/typemap.h` — Add `xce_hostctrl_t* hostctrlmsg;` to body union + - `include/client.h` — Add `xcecli_hostctrl()` declaration + - `include/xceapi.h` — Add `xce_hostctrl()` declaration + - `include/request.h` — Add `xce_hostctrl_req()` declaration + - `lib/client.c` — Implement `xcecli_hostctrl()` + - `lib/xceapi.c` — Implement `xce_hostctrl()` with `cbuf_get()` wait + - `lib/request.c` — Implement `xce_hostctrl_req()` (validate HOST_OPEN/HOST_CLOSE only) + - `lib/helper.c` — Add `Request_body_hostctrlmsg` case to `xce_request_body()` + `xce_request_msg_name()` + - `controller/cluster.c` — Add `OPCode_hostcontrol` case → `do_hostcontrol_op()` + - `controller/worker.c` — Implement `do_hostcontrol_op()` calling `worker_ctrl()` + - `controller/include/cluster.h` — Declare `do_hostcontrol_op()` + - Regenerate capnp: `ninja -C build` + +- [>] **#7 P0-2b: Implement Go-side ControlHost + fix allFlag race** (blocked by #6) + - `backend/clib.go` — Replace `ControlHost()` stub with real implementation via `xcecli_hostctrl()` + - `controllers/hosts.go` — Fix data race: remove `var allFlag bool`, return from `preCheckHostControlRequest` as local + - `controllers/queues.go` — Same allFlag data race fix (lines 534-536 / 465) + +- [ ] **#8 P0-2c: Register routes + RBAC for host control/select** + - `router/routers.go` — `POST /xce/v1/hosts/control` → RBAC `host:control` (cluster_admin) + - `router/routers.go` — `POST /xce/v1/hosts/select` → RBAC `host:view` (all authenticated) + - `config/rbac.yaml` — Add `host:control` and `host:view` action entries + +- [ ] **#9 P0-2d: Update stub.go with new function stubs** + - `backend/stub.go` — Keep `ControlHost` stub (return `errNoCGO`), add `StartHeartbeat(ctx context.Context){}`, `SetHeartbeatInterval(secs int){}`, `EvalSelectForHost(...)` returning `errNoCGO`, `Cleanup(){}` + - Add `"context"` to import block + +--- + +## P1-1: Handle Resource Value Push + +- [ ] **#10 P1-1: Handle resource value push (OPCode_notify)** + - `backend/clib.go` — Add CGO val helpers to preamble (`xce_val_which`, `xce_val_i32`, etc.) + - `backend/clib.go` — Add `OPCode_notify` case at top-level switch in `go_push_handler()` (NOT inside `handlePush()`) + - `backend/clib.go` — Add `handleNotifyData()` for `NotifyData_data_schedUpdateResvals` + - `backend/convert.go` — Add `cResvalToDistribution()` — switch on `Value_u_*` constants, all 7 cases separate + +--- + +## P1-2: Host Resource Requirement Select + +- [ ] **#11 P1-2a: C select expression parser (PEG + eval)** + - `include/resreq.h` — Replace empty `xce_resreq_select_` with expression tree types, add enums, add extern declarations + - `lib/xce_select.peg` — New PEG grammar for select boolean expressions (right-recursive) + - `lib/resreq.c` — Implement `parse_select()`, `eval_select()`, `free_select()`; update `kill_primary_resreq()` + - `lib/meson.build` — Add `xce_select.peg` to explicit source list + +- [>] **#12 P1-2b: Go EvalSelectForHost + FilterHostByResReqSelect** (blocked by #11) + - `backend/clib.go` — Add `EvalSelectForHost()` using `cgo.Handle` for callback context, add `#include "resreq.h"`, extern declarations, `xce_select_destroy` helper + - `controllers/hosts.go` — Implement `FilterHostByResReqSelect`, update `preCheckHostRequest()` limit from 4094→1024 bytes + +--- + +## P1-3: Heartbeat Keep-Alive + +- [ ] **#13 P1-3: Heartbeat keep-alive implementation** + - `backend/clib.go` — Add `heartbeatIntervalSecs int32` (default 30), `SetHeartbeatInterval()`, `heartbeatFn` var, `StartHeartbeat(ctx context.Context)` + - `backend/handler.go` — `OnClusterUpdated` calls `SetHeartbeatInterval(cluster.HeartbeatInterval)` after store update + - `cmd/main.go` — Call `backend.StartHeartbeat(ctx)` after `backend.RegisterToBackend(...)` + +--- + +## Final + +- [>] **#14 Build and test verification** (blocked by all others) + - `make test` in `src/frontend/` passes + - `ninja -C build` for C components passes + - RBAC verification + - Input safety verification (1024-byte limit, 256-host limit) \ No newline at end of file diff --git a/src/frontend/apiserver/backend/clib.go b/src/frontend/apiserver/backend/clib.go index bc5d599..f3ec255 100644 --- a/src/frontend/apiserver/backend/clib.go +++ b/src/frontend/apiserver/backend/clib.go @@ -213,6 +213,9 @@ func handleDataset(d *C.dataset_t) C.int { // handleConfigPush processes a cluster configuration push. func handleConfigPush(cfgdata *C.xce_configdata_t) C.int { + if gHandler == nil { + return 0 + } if cfgdata.data_which != C.ConfigData_data_clusterconfig { return 0 } @@ -222,14 +225,16 @@ func handleConfigPush(cfgdata *C.xce_configdata_t) C.int { return 0 } - for _, q := range cClusterConfQueues(conf) { - qi := q - gHandler.OnQueueUpdated(&qi) - } - for _, r := range cClusterConfResources(conf) { - ri := r - gHandler.OnResourceUpdated(&ri) + cluster := cClusterConfToClusterInfo(conf) + if cluster == nil { + logger.Warn("handleConfigPush: cluster_config_t has empty clustername; skipping OnClusterUpdated") } + dispatchConfigPush(gHandler, + cClusterConfQueues(conf), + cClusterConfResources(conf), + cClusterConfHostConfigs(conf), + cluster, + ) return 0 } diff --git a/src/frontend/apiserver/backend/convert.go b/src/frontend/apiserver/backend/convert.go index f544b94..97ca3e2 100644 --- a/src/frontend/apiserver/backend/convert.go +++ b/src/frontend/apiserver/backend/convert.go @@ -107,7 +107,7 @@ func cResourceToModel(res *C.xce_resource_t) *models.ResourceDefinitionInfo { // cClusterConfQueues extracts queue models from a cluster_config_t. func cClusterConfQueues(conf *C.cluster_config_t) []models.QueueInfo { n := int(conf.nqueues) - if n == 0 { + if n == 0 || conf.queues == nil { return nil } qps := unsafe.Slice(conf.queues, n) @@ -123,7 +123,7 @@ func cClusterConfQueues(conf *C.cluster_config_t) []models.QueueInfo { // cClusterConfResources extracts resource definition models from a cluster_config_t. func cClusterConfResources(conf *C.cluster_config_t) []models.ResourceDefinitionInfo { n := int(conf.nres) - if n == 0 { + if n == 0 || conf.resources == nil { return nil } rps := unsafe.Slice(conf.resources, n) @@ -136,6 +136,48 @@ func cClusterConfResources(conf *C.cluster_config_t) []models.ResourceDefinition return out } +// cHostConfigToModel extracts C xce_hostconfig_t fields and maps them to HostInfo. +func cHostConfigToModel(hc *C.xce_hostconfig_t) *models.HostInfo { + return rawHostConfigToModel(rawHostConfigData{ + name: goString(hc.name), + hostmodel: goString(hc.hostmodel), + hosttype: goString(hc.hosttype), + mxj: int(hc.mxj), + }) +} + +// cClusterConfHostConfigs extracts host config models from a cluster_config_t. +func cClusterConfHostConfigs(conf *C.cluster_config_t) []models.HostInfo { + n := int(conf.nhconfigs) + if n == 0 || conf.hconfigs == nil { + return nil + } + hps := unsafe.Slice(conf.hconfigs, n) + out := make([]models.HostInfo, 0, n) + for _, hp := range hps { + if hp != nil { + out = append(out, *cHostConfigToModel(hp)) + } + } + return out +} + +// cClusterConfToClusterInfo extracts C cluster_config_t cluster-level fields and maps them to ClusterInfo. +// Returns nil when clustername is empty so dispatchConfigPush skips the OnClusterUpdated call. +// C struct field names are snake_case (clustername, heartbeat_interval, etc.) — not Cap'n Proto names. +func cClusterConfToClusterInfo(conf *C.cluster_config_t) *models.ClusterInfo { + name := goString(conf.clustername) + if name == "" { + return nil + } + return rawClusterToModel(rawClusterData{ + name: name, + heartbeatInterval: int(conf.heartbeat_interval), + vcpus: int(conf.vcpus), + inactiveHeartbeatCount: int(conf.inactive_heartbeat_count), + }) +} + // modelJobToSubmit converts a Go JobInfo to C xce_job_submit_t heap structs. // The caller must free all returned C memory by calling freeJobSubmit. func modelJobToSubmit(job *models.JobInfo) (*C.xce_job_submit_t, *C.job_submit_data_t, *C.job_startinfo_t) { diff --git a/src/frontend/apiserver/backend/convert_pure.go b/src/frontend/apiserver/backend/convert_pure.go index e228a85..f171ef0 100644 --- a/src/frontend/apiserver/backend/convert_pure.go +++ b/src/frontend/apiserver/backend/convert_pure.go @@ -231,3 +231,53 @@ func rawResourceToModel(r rawResourceData) *models.ResourceDefinitionInfo { Interval: r.interval, } } + +// rawHostConfigData mirrors the C xce_hostconfig_t fields used by the Go model. +// Only fields that map to models.HostInfo are included: +// - name → HostName, hostmodel → HostModel, hosttype → HostType, mxj → MaxSlot +// +// resources (xce_config_t**) and thresholds (xce_res_threshold_t**) have no HostInfo field and are excluded. +// dispatchwindow (xce_timewindow_config_t*) is deferred — DisptWindow is left as empty string. +type rawHostConfigData struct { + name string + hostmodel string + hosttype string + mxj int +} + +// rawHostConfigToModel maps a rawHostConfigData to a models.HostInfo. +// Config-sourced fields only; worker-sourced fields remain zero. +func rawHostConfigToModel(r rawHostConfigData) *models.HostInfo { + return &models.HostInfo{ + HostName: r.name, + HostModel: r.hostmodel, + HostType: r.hosttype, + MaxSlot: r.mxj, + // TODO(dispatchwindow): xce_hostconfig_t.dispatchwindow is xce_timewindow_config_t* + // (nsegs int + segs[] with open/close time fields). Serializing it requires a Go + // mirror struct and json.Marshal. Deferred: no API endpoint currently consumes + // DisptWindow. Implement when /xce/v1/hosts returns dispatch window data. + } +} + +// rawClusterData mirrors the C cluster_config_t fields used by the Go model. +// Only config-sourced fields are included (heartbeat_interval, vcpus, inactive_heartbeat_count). +// Built-in/sourced fields (Version, Status, ControllerName, etc.) come from LoadBuiltinData +// and are not part of the config push conversion. +type rawClusterData struct { + name string + heartbeatInterval int + vcpus int + inactiveHeartbeatCount int +} + +// rawClusterToModel maps a rawClusterData to a models.ClusterInfo. +// Config-sourced fields only; built-in fields remain zero. +func rawClusterToModel(r rawClusterData) *models.ClusterInfo { + return &models.ClusterInfo{ + ClusterName: r.name, + HeartbeatInterval: r.heartbeatInterval, + Vcpus: r.vcpus, + InactiveHeartbeatCount: r.inactiveHeartbeatCount, + } +} diff --git a/src/frontend/apiserver/backend/convert_test.go b/src/frontend/apiserver/backend/convert_test.go index 9cb0088..b0bed19 100644 --- a/src/frontend/apiserver/backend/convert_test.go +++ b/src/frontend/apiserver/backend/convert_test.go @@ -1,6 +1,7 @@ package backend import ( + "apiserver/models" "testing" "time" ) @@ -387,3 +388,200 @@ func TestIsArrayJob(t *testing.T) { t.Errorf("array child with idx=0 should still be an array job") } } + +func TestRawHostConfigToModel(t *testing.T) { + tests := []struct { + name string + raw rawHostConfigData + want models.HostInfo + }{ + { + name: "normal values", + raw: rawHostConfigData{ + name: "node01", + hostmodel: "x86_64", + hosttype: "LINUX64", + mxj: 8, + }, + want: models.HostInfo{ + HostName: "node01", + HostModel: "x86_64", + HostType: "LINUX64", + MaxSlot: 8, + }, + }, + { + name: "zero values for integer fields", + raw: rawHostConfigData{ + name: "node02", + hostmodel: "modelB", + hosttype: "linux", + mxj: 0, + }, + want: models.HostInfo{ + HostName: "node02", + HostModel: "modelB", + HostType: "linux", + MaxSlot: 0, + }, + }, + { + name: "empty strings for name/hostmodel/hosttype", + raw: rawHostConfigData{ + name: "", + hostmodel: "", + hosttype: "", + mxj: 4, + }, + want: models.HostInfo{ + HostName: "", + HostModel: "", + HostType: "", + MaxSlot: 4, + }, + }, + { + name: "worker-sourced fields remain zero", + raw: rawHostConfigData{ + name: "node03", + hostmodel: "modelC", + hosttype: "CENTOS7", + mxj: 16, + }, + want: models.HostInfo{ + HostName: "node03", + HostModel: "modelC", + HostType: "CENTOS7", + MaxSlot: 16, + HostStatus: 0, + CPUFactor: 0, + MaxMem: 0, + NumRun: 0, + }, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got := rawHostConfigToModel(tt.raw) + if got.HostName != tt.want.HostName { + t.Errorf("HostName: got %q, want %q", got.HostName, tt.want.HostName) + } + if got.HostModel != tt.want.HostModel { + t.Errorf("HostModel: got %q, want %q", got.HostModel, tt.want.HostModel) + } + if got.HostType != tt.want.HostType { + t.Errorf("HostType: got %q, want %q", got.HostType, tt.want.HostType) + } + if got.MaxSlot != tt.want.MaxSlot { + t.Errorf("MaxSlot: got %d, want %d", got.MaxSlot, tt.want.MaxSlot) + } + // Verify worker-sourced fields stay zero + if got.HostStatus != tt.want.HostStatus { + t.Errorf("HostStatus: got %d, want %d (should be zero, worker-sourced)", got.HostStatus, tt.want.HostStatus) + } + if got.CPUFactor != tt.want.CPUFactor { + t.Errorf("CPUFactor: got %f, want %f (should be zero, worker-sourced)", got.CPUFactor, tt.want.CPUFactor) + } + if got.MaxMem != tt.want.MaxMem { + t.Errorf("MaxMem: got %d, want %d (should be zero, worker-sourced)", got.MaxMem, tt.want.MaxMem) + } + if got.NumRun != tt.want.NumRun { + t.Errorf("NumRun: got %d, want %d (should be zero, worker-sourced)", got.NumRun, tt.want.NumRun) + } + // DisptWindow is deferred — must be empty string, not accidentally set + if got.DisptWindow != "" { + t.Errorf("DisptWindow: got %q, want empty string (deferred field)", got.DisptWindow) + } + }) + } +} + +func TestRawClusterToModel(t *testing.T) { + tests := []struct { + name string + raw rawClusterData + want models.ClusterInfo + }{ + { + name: "normal values", + raw: rawClusterData{ + name: "mycluster", + heartbeatInterval: 30, + vcpus: 16, + inactiveHeartbeatCount: 3, + }, + want: models.ClusterInfo{ + ClusterName: "mycluster", + HeartbeatInterval: 30, + Vcpus: 16, + InactiveHeartbeatCount: 3, + }, + }, + { + name: "empty cluster name maps through unchanged (guard is in CGO layer)", + raw: rawClusterData{ + name: "", + heartbeatInterval: 60, + vcpus: 8, + inactiveHeartbeatCount: 5, + }, + want: models.ClusterInfo{ + ClusterName: "", + HeartbeatInterval: 60, + Vcpus: 8, + InactiveHeartbeatCount: 5, + }, + }, + { + name: "built-in fields remain zero", + raw: rawClusterData{ + name: "test", + heartbeatInterval: 45, + vcpus: 32, + inactiveHeartbeatCount: 2, + }, + want: models.ClusterInfo{ + ClusterName: "test", + HeartbeatInterval: 45, + Vcpus: 32, + InactiveHeartbeatCount: 2, + Version: "", + Status: 0, + ControllerName: "", + NumServers: 0, + }, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got := rawClusterToModel(tt.raw) + if got.ClusterName != tt.want.ClusterName { + t.Errorf("ClusterName: got %q, want %q", got.ClusterName, tt.want.ClusterName) + } + if got.HeartbeatInterval != tt.want.HeartbeatInterval { + t.Errorf("HeartbeatInterval: got %d, want %d", got.HeartbeatInterval, tt.want.HeartbeatInterval) + } + if got.Vcpus != tt.want.Vcpus { + t.Errorf("Vcpus: got %d, want %d", got.Vcpus, tt.want.Vcpus) + } + if got.InactiveHeartbeatCount != tt.want.InactiveHeartbeatCount { + t.Errorf("InactiveHeartbeatCount: got %d, want %d", got.InactiveHeartbeatCount, tt.want.InactiveHeartbeatCount) + } + // Verify built-in/sourced fields stay zero (not overwritten by config push) + if got.Version != tt.want.Version { + t.Errorf("Version: got %q, want %q (should be empty, built-in-sourced)", got.Version, tt.want.Version) + } + if got.Status != tt.want.Status { + t.Errorf("Status: got %d, want %d (should be zero, built-in-sourced)", got.Status, tt.want.Status) + } + if got.ControllerName != tt.want.ControllerName { + t.Errorf("ControllerName: got %q, want %q (should be empty, built-in-sourced)", got.ControllerName, tt.want.ControllerName) + } + if got.NumServers != tt.want.NumServers { + t.Errorf("NumServers: got %d, want %d (should be zero, built-in-sourced)", got.NumServers, tt.want.NumServers) + } + }) + } +} diff --git a/src/frontend/apiserver/backend/handler.go b/src/frontend/apiserver/backend/handler.go index bcc82b8..c3964c4 100644 --- a/src/frontend/apiserver/backend/handler.go +++ b/src/frontend/apiserver/backend/handler.go @@ -24,8 +24,66 @@ func (h *DefaultHandler) OnQueueUpdated(queue *models.QueueInfo) { store.Queues.Put(queue.QueueName, *queue) } -func (h *DefaultHandler) OnHostUpdated(host *models.HostInfo) { - store.Hosts.Put(host.HostName, *host) +// OnHostUpdated merges worker-sourced fields into an existing host entry. +// Config-sourced fields (HostModel, HostType, MaxSlot, DisptWindow) are +// preserved from the prior entry if one exists. +func (h *DefaultHandler) OnHostUpdated(incoming *models.HostInfo) { + merged := *incoming + if existing, ok := store.Hosts.Get(merged.HostName); ok { + merged.HostModel = existing.HostModel + merged.HostType = existing.HostType + merged.MaxSlot = existing.MaxSlot + merged.DisptWindow = existing.DisptWindow + } + store.Hosts.Put(merged.HostName, merged) +} + +// OnHostConfigUpdated merges config-sourced fields into an existing host entry. +// Worker-sourced fields (HostStatus, CPUFactor, NumCPUs, MaxMem, MaxSwap, +// MaxTmp, NumDisk, NumRun, NumSSusp, NumUSusp, NumDone) are preserved from +// the prior entry if one exists. +// NumCores, NumThreads, Topology, SchedStop are preserved here but currently +// have no write path in rawWorkerData — they remain zero until a future worker +// push source populates them (analogous to the DisptWindow deferral). +func (h *DefaultHandler) OnHostConfigUpdated(incoming *models.HostInfo) { + merged := *incoming + if existing, ok := store.Hosts.Get(merged.HostName); ok { + merged.HostStatus = existing.HostStatus + merged.CPUFactor = existing.CPUFactor + merged.NumCPUs = existing.NumCPUs + merged.NumCores = existing.NumCores + merged.NumThreads = existing.NumThreads + merged.MaxMem = existing.MaxMem + merged.MaxSwap = existing.MaxSwap + merged.MaxTmp = existing.MaxTmp + merged.NumDisk = existing.NumDisk + merged.NumRun = existing.NumRun + merged.NumSSusp = existing.NumSSusp + merged.NumUSusp = existing.NumUSusp + merged.NumDone = existing.NumDone + merged.Topology = existing.Topology + merged.SchedStop = existing.SchedStop + } + store.Hosts.Put(merged.HostName, merged) +} + +// OnClusterUpdated merges config-sourced fields into an existing cluster entry. +// Built-in fields (Version, Status, ControllerName, NumServers, Resources, +// HostTypes, HostModels, ClusterAdmins) are preserved from the prior entry +// if one exists. +func (h *DefaultHandler) OnClusterUpdated(incoming *models.ClusterInfo) { + merged := *incoming + if existing, ok := store.Clusters.Get(merged.ClusterName); ok { + merged.Version = existing.Version + merged.Status = existing.Status + merged.ControllerName = existing.ControllerName + merged.NumServers = existing.NumServers + merged.Resources = existing.Resources + merged.HostTypes = existing.HostTypes + merged.HostModels = existing.HostModels + merged.ClusterAdmins = existing.ClusterAdmins + } + store.Clusters.Put(merged.ClusterName, merged) } func (h *DefaultHandler) OnResourceUpdated(res *models.ResourceDefinitionInfo) { @@ -36,3 +94,25 @@ func (h *DefaultHandler) OnResourceLocationUpdated(res *models.ResourceDistribut key := res.ResourceName + ":" + res.Location store.ResourceLocations.Put(key, *res) } + +// dispatchConfigPush delivers converted config-push data to the handler (CGO-free, for testability). +func dispatchConfigPush( + handler NotificationHandler, + queues []models.QueueInfo, + resources []models.ResourceDefinitionInfo, + hostConfigs []models.HostInfo, + cluster *models.ClusterInfo, +) { + for _, q := range queues { + handler.OnQueueUpdated(&q) + } + for _, r := range resources { + handler.OnResourceUpdated(&r) + } + for _, hc := range hostConfigs { + handler.OnHostConfigUpdated(&hc) + } + if cluster != nil { + handler.OnClusterUpdated(cluster) + } +} diff --git a/src/frontend/apiserver/backend/handler_test.go b/src/frontend/apiserver/backend/handler_test.go new file mode 100644 index 0000000..53f68ae --- /dev/null +++ b/src/frontend/apiserver/backend/handler_test.go @@ -0,0 +1,345 @@ +package backend + +import ( + "apiserver/models" + "apiserver/store" + "testing" +) + +// resetStores reinitializes all global stores for test isolation. +func resetStores() { + store.InitStores() +} + +func TestOnHostConfigUpdated_NewHost(t *testing.T) { + resetStores() + h := DefaultHandler{} + + host := &models.HostInfo{ + HostName: "node01", + HostModel: "x86_64", + HostType: "LINUX64", + MaxSlot: 8, + } + + h.OnHostConfigUpdated(host) + + got, ok := store.Hosts.Get("node01") + if !ok { + t.Fatal("host node01 should exist in store after OnHostConfigUpdated") + } + if got.HostModel != "x86_64" { + t.Errorf("HostModel: got %q, want %q", got.HostModel, "x86_64") + } + if got.HostType != "LINUX64" { + t.Errorf("HostType: got %q, want %q", got.HostType, "LINUX64") + } + if got.MaxSlot != 8 { + t.Errorf("MaxSlot: got %d, want 8", got.MaxSlot) + } +} + +func TestOnHostConfigUpdated_PartialMerge(t *testing.T) { + resetStores() + h := DefaultHandler{} + + // Simulate a prior worker push that set worker-sourced fields. + store.Hosts.Put("node01", models.HostInfo{ + HostName: "node01", + HostStatus: 0x00010000, // UNAVAIL + CPUFactor: 2.5, + NumCPUs: 32, + MaxMem: 131072, + NumRun: 5, + }) + + // Config push arrives — should merge config fields, preserve worker fields. + h.OnHostConfigUpdated(&models.HostInfo{ + HostName: "node01", + HostModel: "modelA", + HostType: "CENTOS7", + MaxSlot: 16, + }) + + got, ok := store.Hosts.Get("node01") + if !ok { + t.Fatal("host node01 should exist after merge") + } + // Config-sourced fields taken from the push. + if got.HostModel != "modelA" { + t.Errorf("HostModel: got %q, want %q", got.HostModel, "modelA") + } + if got.HostType != "CENTOS7" { + t.Errorf("HostType: got %q, want %q", got.HostType, "CENTOS7") + } + if got.MaxSlot != 16 { + t.Errorf("MaxSlot: got %d, want 16", got.MaxSlot) + } + // Worker-sourced fields preserved from the prior entry. + if got.HostStatus != 0x00010000 { + t.Errorf("HostStatus: got 0x%08X, want 0x00010000 (should be preserved)", got.HostStatus) + } + if got.CPUFactor != 2.5 { + t.Errorf("CPUFactor: got %f, want 2.5 (should be preserved)", got.CPUFactor) + } + if got.NumCPUs != 32 { + t.Errorf("NumCPUs: got %d, want 32 (should be preserved)", got.NumCPUs) + } + if got.MaxMem != 131072 { + t.Errorf("MaxMem: got %d, want 131072 (should be preserved)", got.MaxMem) + } + if got.NumRun != 5 { + t.Errorf("NumRun: got %d, want 5 (should be preserved)", got.NumRun) + } +} + +func TestOnClusterUpdated_NewCluster(t *testing.T) { + resetStores() + h := DefaultHandler{} + + cluster := &models.ClusterInfo{ + ClusterName: "mycluster", + HeartbeatInterval: 30, + Vcpus: 16, + InactiveHeartbeatCount: 3, + } + + h.OnClusterUpdated(cluster) + + got, ok := store.Clusters.Get("mycluster") + if !ok { + t.Fatal("cluster mycluster should exist in store after OnClusterUpdated") + } + if got.HeartbeatInterval != 30 { + t.Errorf("HeartbeatInterval: got %d, want 30", got.HeartbeatInterval) + } + if got.Vcpus != 16 { + t.Errorf("Vcpus: got %d, want 16", got.Vcpus) + } + if got.InactiveHeartbeatCount != 3 { + t.Errorf("InactiveHeartbeatCount: got %d, want 3", got.InactiveHeartbeatCount) + } +} + +func TestOnClusterUpdated_PartialMerge(t *testing.T) { + resetStores() + h := DefaultHandler{} + + // Simulate built-in data loaded at startup. + store.Clusters.Put("mycluster", models.ClusterInfo{ + ClusterName: "mycluster", + Version: "10.1.0", + Status: 0x01, // CONNECTED + ControllerName: "ctrl01", + NumServers: 3, + }) + + // Config push arrives — should merge config fields, preserve built-in fields. + h.OnClusterUpdated(&models.ClusterInfo{ + ClusterName: "mycluster", + HeartbeatInterval: 45, + Vcpus: 64, + InactiveHeartbeatCount: 5, + }) + + got, ok := store.Clusters.Get("mycluster") + if !ok { + t.Fatal("cluster mycluster should exist after merge") + } + // Config-sourced fields from the push. + if got.HeartbeatInterval != 45 { + t.Errorf("HeartbeatInterval: got %d, want 45", got.HeartbeatInterval) + } + if got.Vcpus != 64 { + t.Errorf("Vcpus: got %d, want 64", got.Vcpus) + } + if got.InactiveHeartbeatCount != 5 { + t.Errorf("InactiveHeartbeatCount: got %d, want 5", got.InactiveHeartbeatCount) + } + // Built-in fields preserved. + if got.Version != "10.1.0" { + t.Errorf("Version: got %q, want %q (should be preserved)", got.Version, "10.1.0") + } + if got.Status != 0x01 { + t.Errorf("Status: got 0x%02X, want 0x01 (should be preserved)", got.Status) + } + if got.ControllerName != "ctrl01" { + t.Errorf("ControllerName: got %q, want %q (should be preserved)", got.ControllerName, "ctrl01") + } + if got.NumServers != 3 { + t.Errorf("NumServers: got %d, want 3 (should be preserved)", got.NumServers) + } +} + +func TestOnHostUpdated_PartialMerge(t *testing.T) { + resetStores() + h := DefaultHandler{} + + // Simulate a prior config push that set config-sourced fields. + store.Hosts.Put("node01", models.HostInfo{ + HostName: "node01", + HostModel: "x86_64", + HostType: "LINUX64", + MaxSlot: 8, + DisptWindow: "always", + }) + + // Worker push arrives — should merge worker fields, preserve config fields. + h.OnHostUpdated(&models.HostInfo{ + HostName: "node01", + HostStatus: 3, + CPUFactor: 1.8, + NumCPUs: 16, + MaxMem: 65536, + NumRun: 2, + }) + + got, ok := store.Hosts.Get("node01") + if !ok { + t.Fatal("host node01 should exist after merge") + } + // Worker-sourced fields from the push. + if got.HostStatus != 3 { + t.Errorf("HostStatus: got %d, want 3", got.HostStatus) + } + if got.CPUFactor != 1.8 { + t.Errorf("CPUFactor: got %f, want 1.8", got.CPUFactor) + } + if got.NumCPUs != 16 { + t.Errorf("NumCPUs: got %d, want 16", got.NumCPUs) + } + if got.MaxMem != 65536 { + t.Errorf("MaxMem: got %d, want 65536", got.MaxMem) + } + if got.NumRun != 2 { + t.Errorf("NumRun: got %d, want 2", got.NumRun) + } + // Config-sourced fields preserved. + if got.HostModel != "x86_64" { + t.Errorf("HostModel: got %q, want %q (should be preserved)", got.HostModel, "x86_64") + } + if got.HostType != "LINUX64" { + t.Errorf("HostType: got %q, want %q (should be preserved)", got.HostType, "LINUX64") + } + if got.MaxSlot != 8 { + t.Errorf("MaxSlot: got %d, want 8 (should be preserved)", got.MaxSlot) + } + if got.DisptWindow != "always" { + t.Errorf("DisptWindow: got %q, want %q (should be preserved)", got.DisptWindow, "always") + } +} + +func TestOnHostUpdated_NewHost(t *testing.T) { + resetStores() + h := DefaultHandler{} + + // No prior entry in store — first push is from worker. + h.OnHostUpdated(&models.HostInfo{ + HostName: "node05", + HostStatus: 1, + CPUFactor: 3.0, + }) + + got, ok := store.Hosts.Get("node05") + if !ok { + t.Fatal("host node05 should exist after first push") + } + if got.HostStatus != 1 { + t.Errorf("HostStatus: got %d, want 1", got.HostStatus) + } +} + +func TestDispatchConfigPush_CallsOnHostConfigUpdatedForEachHost(t *testing.T) { + resetStores() + h := &DefaultHandler{} + + hosts := []models.HostInfo{ + {HostName: "node01", HostModel: "x86_64", HostType: "LINUX64", MaxSlot: 4}, + {HostName: "node02", HostModel: "arm64", HostType: "LINUX64", MaxSlot: 8}, + } + + dispatchConfigPush(h, nil, nil, hosts, nil) + + got1, ok1 := store.Hosts.Get("node01") + if !ok1 { + t.Fatal("node01 should be in store after dispatchConfigPush") + } + if got1.HostModel != "x86_64" { + t.Errorf("node01 HostModel: got %q, want x86_64", got1.HostModel) + } + if got1.MaxSlot != 4 { + t.Errorf("node01 MaxSlot: got %d, want 4", got1.MaxSlot) + } + + got2, ok2 := store.Hosts.Get("node02") + if !ok2 { + t.Fatal("node02 should be in store after dispatchConfigPush") + } + if got2.HostModel != "arm64" { + t.Errorf("node02 HostModel: got %q, want arm64", got2.HostModel) + } + if got2.MaxSlot != 8 { + t.Errorf("node02 MaxSlot: got %d, want 8", got2.MaxSlot) + } +} + +func TestDispatchConfigPush_CallsOnClusterUpdated(t *testing.T) { + resetStores() + h := &DefaultHandler{} + + cluster := &models.ClusterInfo{ + ClusterName: "mycluster", + HeartbeatInterval: 30, + Vcpus: 64, + InactiveHeartbeatCount: 5, + } + + dispatchConfigPush(h, nil, nil, nil, cluster) + + got, ok := store.Clusters.Get("mycluster") + if !ok { + t.Fatal("mycluster should be in store after dispatchConfigPush") + } + if got.HeartbeatInterval != 30 { + t.Errorf("HeartbeatInterval: got %d, want 30", got.HeartbeatInterval) + } + if got.Vcpus != 64 { + t.Errorf("Vcpus: got %d, want 64", got.Vcpus) + } + if got.InactiveHeartbeatCount != 5 { + t.Errorf("InactiveHeartbeatCount: got %d, want 5", got.InactiveHeartbeatCount) + } +} + +func TestDispatchConfigPush_NilClusterIsSkipped(t *testing.T) { + resetStores() + h := &DefaultHandler{} + + dispatchConfigPush(h, nil, nil, nil, nil) + + if n := store.Clusters.Count(); n != 0 { + t.Errorf("Clusters store should be empty when cluster is nil, got %d entries", n) + } +} + +func TestDispatchConfigPush_EmptySlicesNoOp(t *testing.T) { + resetStores() + h := &DefaultHandler{} + + dispatchConfigPush(h, + []models.QueueInfo{}, + []models.ResourceDefinitionInfo{}, + []models.HostInfo{}, + nil, + ) + + if n := store.Hosts.Count(); n != 0 { + t.Errorf("Hosts store should be empty, got %d entries", n) + } + if n := store.Queues.Count(); n != 0 { + t.Errorf("Queues store should be empty, got %d entries", n) + } + if n := store.Clusters.Count(); n != 0 { + t.Errorf("Clusters store should be empty, got %d entries", n) + } +} \ No newline at end of file diff --git a/src/frontend/apiserver/backend/notification.go b/src/frontend/apiserver/backend/notification.go index eb3eea9..34f3605 100644 --- a/src/frontend/apiserver/backend/notification.go +++ b/src/frontend/apiserver/backend/notification.go @@ -11,6 +11,8 @@ type NotificationHandler interface { OnJobDeleted(jobid uint64) OnQueueUpdated(queue *models.QueueInfo) OnHostUpdated(host *models.HostInfo) + OnHostConfigUpdated(host *models.HostInfo) + OnClusterUpdated(cluster *models.ClusterInfo) OnResourceUpdated(res *models.ResourceDefinitionInfo) OnResourceLocationUpdated(res *models.ResourceDistributionInfo) }