1013 lines
No EOL
72 KiB
Markdown
1013 lines
No EOL
72 KiB
Markdown
# API Server Backend Enhancement Plan
|
||
|
||
## Context
|
||
|
||
The Go API server at `src/frontend/apiserver/` serves as the REST frontend for the XCE cluster management system. Currently it operates in a push-based model: the C controller sends updates (jobs, workers, queues, resources) and the API server stores them in-memory. Write/control operations (job submit, job control, queue control) are forwarded to the controller via CGO.
|
||
|
||
Analysis of the C test tool (`src/controller/testsuite/test.c`) and the `client.h` API revealed several backend operations that the API server does not yet support. Based on user feedback:
|
||
|
||
- **Push-based, not pull**: Resource values and job info come via push, not synchronous queries
|
||
- **Remove from scope**: Resource Value Update (worker concern), Job Info sync query, Run Tasks (worker concern), Set Pending Reason (scheduler concern)
|
||
- **Keep at low priority**: Job Dispatch (future `brun`-like feature), Shutdown (admin tool, separate)
|
||
|
||
The remaining actionable items are:
|
||
|
||
1. **Cluster Config push handling** (P0) — Without this, host configs and cluster settings from the controller are silently dropped; operators cannot see configured hosts via the API
|
||
2. **Host Control** (P0) — Without this, operators cannot open or close hosts through the API; current workaround requires direct controller access
|
||
3. **Resource Value push handling** (P1) — Without this, live resource values from workers never reach the API; /resources/distribution returns stale config-sourced data only
|
||
4. **Host Resource Requirement Select** (P1) — Without this, users cannot query which hosts match their job's resource requirements; they cannot understand why a job isn't running
|
||
5. **Heartbeat keep-alive** (P1) — Without this, the API server's controller connection silently drops after timeout; push subscriptions stop without any error indication
|
||
|
||
### Design Principles
|
||
|
||
- **In-memory, no persistence**: the API server does not persist state by design. On restart, all data rebuilds from controller pushes. This is intentional — the controller is the source of truth.
|
||
- **RegisterToBackend is synchronous**: the call blocks until the controller connection is established and the initial configuration push is received. Startup sequencing depends on this.
|
||
- **RBAC enforcement**: all new endpoints must be registered under the existing JWT + RBAC middleware chain in `routers.go`. Host control (`host:control`) is restricted to the `cluster_admin` role; host select (`host:view`) is available to all authenticated users who need to query why their jobs can't run.
|
||
|
||
---
|
||
|
||
## P0-1: Handle Cluster Config Push
|
||
|
||
### Problem
|
||
`handleConfigPush()` in `clib.go` only processes `clusterconfig` pushes for queues and resource definitions. The `ClusterConfig` struct also contains host configs, user groups, host groups, and cluster settings that are silently dropped.
|
||
|
||
### Changes
|
||
|
||
**`src/frontend/apiserver/backend/clib.go`** — Expand `handleConfigPush`:
|
||
- The existing `handleConfigPush` already has a nil guard at line 222: `if conf == nil { return 0 }`. All new host config and cluster settings code must be inserted **after** this guard — do not remove or reposition it.
|
||
- After processing queues and resources, add `cClusterConfHostConfigs()` in `convert.go` using `conf.nhconfigs` (count) and `conf.hconfigs` (array) — same `unsafe.Slice(conf.hconfigs, n)` pattern as `cClusterConfQueues`. Call it in `handleConfigPush` the same way queues are iterated.
|
||
- Convert `C.xce_hostconfig_t` to Go model via new `cHostConfigToModel()` in `convert.go`
|
||
- Call `gHandler.OnHostConfigUpdated()` for each host config
|
||
- Extract cluster-level settings (cluster name, heartbeat interval, etc.) and call `gHandler.OnClusterUpdated()`
|
||
- **User groups and host groups**: `ClusterConfig` also contains `ugroups` and `hgroups` (flagged as silently dropped in the Problem section). **Deferred from this phase**: the API server has no `/xce/v1/usergroups` or `/xce/v1/hostgroups` endpoints yet, and the models for these types don't exist. Adding group handling in this phase would require defining new models, new store entries, new handler methods, and new REST routes — that is a separate feature, not a prerequisite for host config and cluster settings. For now, `ugroups` and `hgroups` fields in `ClusterConfig` continue to be ignored (not silently — this plan explicitly acknowledges the omission). A follow-on task should add group handling before any feature that depends on group membership checks.
|
||
|
||
**`src/frontend/apiserver/backend/convert_pure.go`** — Follow the established `rawWorkerData` / `rawWorkerToModel` pattern exactly:
|
||
- Add `rawHostConfigData` struct with only the C fields that map to `models.HostInfo`: `name` (→HostName), `hostmodel` (→HostModel), `hosttype` (→HostType), `mxj` (→MaxSlot). Do NOT add `resources` (`xce_config_t**` — no `HostInfo` field) or `thresholds` (no `HostInfo` field). `dispatchwindow` is deferred (→`DisptWindow: ""`; see dispatchwindow note below).
|
||
- Add `rawHostConfigToModel(r rawHostConfigData) *models.HostInfo` — pure Go conversion, no CGO, testable without a live build. **TDD prerequisite**: before implementing, add table-driven tests in `backend/convert_test.go` covering: zero-value integer fields (`MaxSlot=0`), empty strings for `name`/`hostmodel`/`hosttype`, and normal values. Follow the existing `TestRawJobToModel` / `TestRawWorkerToModel` patterns in that file.
|
||
- Add `rawClusterData` struct and `rawClusterToModel(r rawClusterData) *models.ClusterInfo` — pure Go conversion. **TDD prerequisite**: add tests in `convert_test.go` covering: zero values for `heartbeatInterval`/`vcpus`/`inactiveHeartbeatCount`, and normal values.
|
||
```go
|
||
func rawHostConfigToModel(r rawHostConfigData) *models.HostInfo {
|
||
return &models.HostInfo{
|
||
HostName: r.name,
|
||
HostModel: r.hostmodel,
|
||
HostType: r.hosttype,
|
||
MaxSlot: r.mxj,
|
||
// DisptWindow: "" — deferred, see dispatchwindow note
|
||
}
|
||
}
|
||
```
|
||
|
||
**`src/frontend/apiserver/backend/convert.go`**:
|
||
- Add `cHostConfigToModel()` — CGO extraction layer (thin; just reads C fields, calls `rawHostConfigToModel`):
|
||
```go
|
||
func cHostConfigToModel(hc *C.xce_hostconfig_t) *models.HostInfo {
|
||
return rawHostConfigToModel(rawHostConfigData{
|
||
name: C.GoString(hc.name),
|
||
hostmodel: C.GoString(hc.hostmodel),
|
||
hosttype: C.GoString(hc.hosttype),
|
||
mxj: int(hc.mxj),
|
||
})
|
||
}
|
||
```
|
||
- **`dispatchwindow` conversion**: `xce_hostconfig_t.dispatchwindow` is `xce_timewindow_config_t*` — a complex nested struct (`nsegs`, `segs[]` each with `open`/`close` time fields). No existing Go conversion utility exists, and `cWorkerToModel` also leaves `DisptWindow` empty. For this phase, **set `DisptWindow: ""`** (leave as empty string) and add this comment so the deferral is visible in code review:
|
||
```go
|
||
// 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.
|
||
DisptWindow: "",
|
||
```
|
||
Implementing the full timewindow → JSON serialization is deferred — it requires building a Go mirror struct for `xce_timewindow_config_t` and calling `json.Marshal`.
|
||
- The ClusterConfig struct from `controller.capnp` line 636 includes: `clusterName`, `heartbeatInterval`, `inactiveHeartbeatCount`, `fetchJobInterval`, `netioTimeout`, `vcpus`, plus `resources`, `queues`, `ugroups`, `hgroups`, `hostConfigs`, `limits`
|
||
- **C struct field names differ from Cap'n Proto names.** CGO accesses the C struct `cluster_config_t` (from `clusterconfig.h`) directly — use snake_case field names: `conf.clustername`, `conf.heartbeat_interval`, `conf.inactive_heartbeat_count`, `conf.fetch_job_interval`, `conf.netio_timeout`, `conf.vcpus`. The Cap'n Proto names above are reference only; using them in Go code is a compile error.
|
||
|
||
**`src/frontend/apiserver/backend/notification.go`** — Add to `NotificationHandler`:
|
||
- `OnClusterUpdated(cluster *models.ClusterInfo)`
|
||
- `OnHostConfigUpdated(host *models.HostInfo)`
|
||
|
||
**`src/frontend/apiserver/backend/handler.go`** — Implement in `DefaultHandler`:
|
||
- `OnClusterUpdated`: **partial merge** — do NOT overwrite the entire entry. The config push only owns `HeartbeatInterval`, `Vcpus`, `InactiveHeartbeatCount` (and `ClusterName` as the key). Existing fields (`Version`, `Status`, `ControllerName`, `NumServers`, `Resources`, `HostTypes`, `HostModels`, `ClusterAdmins`) come from the startup built-in data load and must be preserved. Read the existing entry, update only config-owned fields, then write back:
|
||
```go
|
||
existing, ok := store.Clusters.Get(cluster.ClusterName)
|
||
if ok {
|
||
existing.HeartbeatInterval = cluster.HeartbeatInterval
|
||
existing.Vcpus = cluster.Vcpus
|
||
existing.InactiveHeartbeatCount = cluster.InactiveHeartbeatCount
|
||
store.Clusters.Put(cluster.ClusterName, existing)
|
||
} else {
|
||
store.Clusters.Put(cluster.ClusterName, *cluster)
|
||
}
|
||
```
|
||
**ClusterInfo field ownership**: config-sourced fields (`HeartbeatInterval`, `Vcpus`, `InactiveHeartbeatCount`) are always overwritten by the push; built-in/sourced fields (`Version`, `Status`, `ControllerName`, `NumServers`, `Resources`, `HostTypes`, `HostModels`, `ClusterAdmins`) are always preserved from the existing entry. If no entry exists, the full ClusterInfo is inserted. When new fields are added to `ClusterInfo`, each must be classified as config-sourced or built-in/sourced — ambiguous fields default to preserved (do not overwrite existing data on a config push).
|
||
- `OnHostConfigUpdated`: **partial merge** — do NOT overwrite the entire entry. Read the existing entry, update only config-sourced fields (`HostModel`, `HostType`, `MaxSlot`=mxj, `DisptWindow`), then write back. Worker-sourced fields (`HostStatus`, `CPUFactor`, `MaxMem`, `MaxSwap`, `NumRun`, `NumSSusp`, etc.) must be preserved. If no entry exists, create with config fields only (worker fields remain zero). Overwriting the full entry on every config push would erase live worker data. `store.Hosts.Get()` returns `(models.HostInfo, bool)` — use the two-value form:
|
||
```go
|
||
existing, ok := store.Hosts.Get(host.HostName)
|
||
if ok {
|
||
existing.HostModel = host.HostModel
|
||
existing.HostType = host.HostType
|
||
existing.MaxSlot = host.MaxSlot
|
||
existing.DisptWindow = host.DisptWindow
|
||
store.Hosts.Put(host.HostName, existing)
|
||
} else {
|
||
store.Hosts.Put(host.HostName, *host)
|
||
}
|
||
```
|
||
|
||
**Partial merge semantics**: the config push is a full cluster snapshot (all currently configured hosts appear in each push). `OnHostConfigUpdated` is called once per host per push. Config-sourced fields (`HostModel`, `HostType`, `MaxSlot`, `DisptWindow`) are always overwritten with the push values; **worker-sourced fields** (`HostStatus`, `CPUFactor`, `MaxMem`, `MaxSwap`, `NumRun`, `NumSSusp`, `NumUSusp`, `NumRsv`, `NumLIM`, `NumLIMR`, `CPUTime`, `MaxCPUTime`) are always preserved from the existing store entry. This explicit field ownership list must be maintained when new fields are added to `HostInfo` — each new field must be classified as config-sourced or worker-sourced. If a field's source is ambiguous, preserve it (worker data is live and should never be overwritten by a config snapshot). This is always a field-level partial update — never a full entry replacement. **Host removal**: if a host disappears from a subsequent config push, no `OnHostConfigUpdated` call fires for it, so its entry remains in the store with stale config fields. Host deletion from the store is explicitly deferred — a future implementation will add a `LastConfigSeenAt time.Time` field to `HostInfo` (updated by `OnHostConfigUpdated`) and a `LastWorkerSeenAt time.Time` field (updated by `OnHostUpdated`). These timestamp fields enable future TTL-based cleanup without requiring a model change later. For now they default to `time.Time{}` (the zero value), which is clearly distinguishable from a real timestamp.
|
||
|
||
**First-startup display**: after config push but before any worker check-in, host entries have zero-valued worker-sourced fields. `HostStateStr(0)` currently returns `"OK"` (at `models/hosts.go:32`), which would incorrectly show config-only hosts as healthy. Change `HostStateStr` so that `state == 0` (the zero value) maps to `"UNKNOWN"` instead of `"OK"`: `if state == 0 { return "UNKNOWN" }` before the existing flag checks. This correctly signals that the host status has not yet been reported by a worker.
|
||
|
||
- `OnHostUpdated`: **partial merge** — the existing method in `handler.go:27` currently does `store.Hosts.Put(host.HostName, *host)` (full overwrite). This overwrites config-sourced fields (`HostModel`, `HostType`, `MaxSlot`, `DisptWindow`) with zero/empty values from the worker push, destroying data set by `OnHostConfigUpdated`. Fix: read the existing entry, update only worker-sourced fields, preserve config-sourced fields. **TDD prerequisite**: before implementing any partial-merge handler, write `backend/handler_test.go` with table-driven tests that (1) prove `OnHostUpdated` currently does a full overwrite (RED test for the new behavior), (2) verify `OnHostUpdated` partial merge preserves config fields, (3) verify `OnHostConfigUpdated` partial merge preserves worker fields, (4) verify `OnClusterUpdated` partial merge preserves built-in fields. Use `store.InitStores()` for store reset between test cases.
|
||
```go
|
||
func (h *DefaultHandler) OnHostUpdated(host *models.HostInfo) {
|
||
existing, ok := store.Hosts.Get(host.HostName)
|
||
if ok {
|
||
// Overwrite only worker-sourced fields; preserve config-sourced fields.
|
||
existing.HostStatus = host.HostStatus
|
||
existing.CPUFactor = host.CPUFactor
|
||
existing.MaxMem = host.MaxMem
|
||
existing.MaxSwap = host.MaxSwap
|
||
existing.NumRun = host.NumRun
|
||
existing.NumSSusp = host.NumSSusp
|
||
existing.NumUSusp = host.NumUSusp
|
||
existing.NumRsv = host.NumRsv
|
||
existing.NumLIM = host.NumLIM
|
||
existing.NumLIMR = host.NumLIMR
|
||
existing.CPUTime = host.CPUTime
|
||
existing.MaxCPUTime = host.MaxCPUTime
|
||
store.Hosts.Put(host.HostName, existing)
|
||
} else {
|
||
store.Hosts.Put(host.HostName, *host)
|
||
}
|
||
}
|
||
```
|
||
This is the mirror image of `OnHostConfigUpdated`: the worker push updates its own fields and preserves config fields; the config push updates its own fields and preserves worker fields. Both paths are read-then-write. There is no concurrent push scenario for a single cluster: all pushes (config and worker updates) originate from one controller and are delivered via a single `go_push_handler` callback thread, so only one push is in-flight at a time. `store.Hosts.Put` is a thread-safe map write for safety against any read access from HTTP handlers. Future multi-cluster support will maintain a separate in-memory state copy per cluster — each cluster's pushes are isolated to their own store, so cross-cluster contention is not a concern.
|
||
|
||
**`src/frontend/apiserver/models/`** — **Add** missing fields to the **existing** `ClusterInfo` struct. Do NOT replace the struct — the existing fields (`Version`, `Status`, `ControllerName`, `NumServers`, `Resources`, `HostTypes`, `HostModels`, `ClusterAdmins`) at `models/clusters.go:16-26` are seeded at startup by `store.LoadBuiltinData()` and must be preserved. Only the three new fields below are added. The C struct (`cluster_config_t`) exposes `heartbeat_interval` (int), `vcpus` (int), `inactive_heartbeat_count` (int), `fetch_job_interval` (int), and `netio_timeout` (int). At minimum, `HeartbeatInterval` and `Vcpus` are required by other units in this plan (P0-1's `OnClusterUpdated` wiring and P1-3's `StartHeartbeat`). Add:
|
||
```go
|
||
// In models/clusters.go — ADD these fields to the existing ClusterInfo struct.
|
||
// Do NOT replace the struct; existing fields (Version, Status, ControllerName, etc.)
|
||
// are seeded by store.LoadBuiltinData() and must be preserved.
|
||
type ClusterInfo struct {
|
||
ClusterName string `json:"cluster_name"` // existing
|
||
Version string `json:"version"` // existing
|
||
Status int `json:"status"` // existing
|
||
ControllerName string `json:"controller_name"` // existing
|
||
NumServers int `json:"num_servers"` // existing
|
||
Resources string `json:"resources"` // existing
|
||
HostTypes string `json:"host_types"` // existing
|
||
HostModels string `json:"host_models"` // existing
|
||
ClusterAdmins string `json:"cluster_admins"` // existing
|
||
HeartbeatInterval int `json:"heartbeat_interval"` // NEW — seconds; used by StartHeartbeat
|
||
Vcpus int `json:"vcpus"` // NEW
|
||
InactiveHeartbeatCount int `json:"inactive_heartbeat_count"` // NEW — controller connection timeout = HeartbeatInterval × InactiveHeartbeatCount
|
||
// FetchJobInterval and NetioTimeout are available but not currently consumed by any API endpoint; add if needed
|
||
}
|
||
```
|
||
The raw conversion follows the same `rawClusterData` / `rawClusterToModel` pattern as `rawWorkerData`:
|
||
```go
|
||
// In convert_pure.go
|
||
type rawClusterData struct {
|
||
name string
|
||
heartbeatInterval int
|
||
vcpus int
|
||
inactiveHeartbeatCount int
|
||
}
|
||
func rawClusterToModel(r rawClusterData) *models.ClusterInfo {
|
||
return &models.ClusterInfo{
|
||
ClusterName: r.name,
|
||
HeartbeatInterval: r.heartbeatInterval,
|
||
Vcpus: r.vcpus,
|
||
InactiveHeartbeatCount: r.inactiveHeartbeatCount,
|
||
}
|
||
}
|
||
```
|
||
```go
|
||
// In convert.go — CGO extraction layer calling rawClusterToModel
|
||
func cClusterConfToClusterInfo(conf *C.cluster_config_t) *models.ClusterInfo {
|
||
return rawClusterToModel(rawClusterData{
|
||
name: C.GoString(conf.clustername),
|
||
heartbeatInterval: int(conf.heartbeat_interval),
|
||
vcpus: int(conf.vcpus),
|
||
inactiveHeartbeatCount: int(conf.inactive_heartbeat_count),
|
||
})
|
||
}
|
||
```
|
||
Also add **`cClusterConfResources()`** and **`cResdefToModel()`** to `convert.go` (these are added in P0-1, not P1-1 — P1-1 does not reuse these functions):
|
||
- `cClusterConfResources(conf *C.cluster_config_t) []*models.ResourceInfo` — iterates `conf.nres` / `conf.resources` array using `unsafe.Slice` pattern, calls `cResdefToModel()` for each
|
||
- `cResdefToModel(r *C.xce_resdef_t) *models.ResourceInfo` — converts `xce_resdef_t` fields to `models.ResourceInfo`. **Named `cResdefToModel` (not `cResourceToModel`) to avoid collision with the existing `cResourceToModel(res *C.xce_resource_t) *models.ResourceDefinitionInfo` at `convert.go:96`**, which serves the same cluster config push path but converts a different C struct type (`xce_resource_t`, not `xce_resdef_t`) to a different Go model (`ResourceDefinitionInfo`, not `ResourceInfo`).
|
||
- `cClusterConfHostConfigs(conf *C.cluster_config_t) []models.HostInfo` — iterates `conf.nhconfigs` / `conf.hconfigs` array using `unsafe.Slice` pattern, calls `cHostConfigToModel()` for each:
|
||
```go
|
||
// 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
|
||
}
|
||
```
|
||
|
||
These replace the existing placeholder comment in the Changes list. Without them, `handleConfigPush` has nowhere to store the resource definitions that arrive in the initial cluster config push.
|
||
|
||
---
|
||
|
||
## P0-2: Host Control via Backend
|
||
|
||
### Problem
|
||
`ControlHost()` in `backend/stub.go` returns a static error `"host control not yet supported via C library"`. The C side has `xce_hostctrl_t` (action + hostname) and `worker_ctrl()` in the controller, but:
|
||
1. `xcecli_hostctrl()` does NOT exist in `client.h` — no C client function for it
|
||
2. The Cap'n Proto Request schema has `OPCode hostcontrol @14` but **no corresponding body field** for host control requests
|
||
|
||
This means we need to **add the C-side plumbing first** before the Go server can use it.
|
||
|
||
### Changes
|
||
|
||
**C Side (new):**
|
||
|
||
**`src/include/client.h`** — Add:
|
||
```c
|
||
extern int xcecli_hostctrl(xcecli_ctx_t ctx, xce_hostctrl_t* ctrl);
|
||
```
|
||
|
||
**`src/lib/client.c`** — Implement `xcecli_hostctrl()` following the exact pattern of `xcecli_queuectrl()`:
|
||
```
|
||
xcecli_hostctrl() (client.c)
|
||
→ xce_hostctrl() (xceapi.c)
|
||
→ xce_hostctrl_req() (request.c) // builds xce_request_t, calls xce_request()
|
||
→ cbuf_get(&ctx->wbuf, &status) // blocks until controller sends reply
|
||
```
|
||
|
||
**`src/lib/client.c`** — mirror `xcecli_queuectrl()` exactly:
|
||
```c
|
||
int xcecli_hostctrl(xcecli_ctx_t c, xce_hostctrl_t* ctrlinfo)
|
||
{
|
||
api_ctx_t* ctx = (api_ctx_t*)c;
|
||
int rc;
|
||
if ((ctx == NULL) || (ctrlinfo == NULL)) return -EINVAL;
|
||
rc = xce_hostctrl(&(ctx->ctx), ctrlinfo);
|
||
return rc;
|
||
}
|
||
```
|
||
|
||
**`src/lib/xceapi.c`** — Add `xce_hostctrl()` mirroring `xce_queuectrl()`:
|
||
```c
|
||
int xce_hostctrl(xce_ctx_t* ctx, xce_hostctrl_t* ctrl)
|
||
{
|
||
int rc, status;
|
||
if (!ctx || ctx->state != XCE_CTX_RUN) return -EINVAL;
|
||
rc = xce_hostctrl_req(ctx->ep, ctrl);
|
||
if (rc < 0) return rc;
|
||
rc = cbuf_get(&(ctx->wbuf), &status);
|
||
if (rc < 0) return rc;
|
||
return status;
|
||
}
|
||
```
|
||
|
||
**`src/lib/request.c`** — Add `xce_hostctrl_req()` mirroring `xce_queuectrl_req()`:
|
||
```c
|
||
int xce_hostctrl_req(endpoint_t* ep, xce_hostctrl_t* ctrlinfo)
|
||
{
|
||
xce_request_t request;
|
||
if (!ep || !ctrlinfo || !ctrlinfo->hostname || !*ctrlinfo->hostname)
|
||
return -EINVAL;
|
||
/* Reject unsupported actions before the round-trip.
|
||
* worker_ctrl() only handles HOST_OPEN and HOST_CLOSE; all other values
|
||
* (including HOST_DO_NONE=0, HOST_TAG, HOST_UNTAG) hit its default branch
|
||
* and return -1, so reject them here for a clearer error. */
|
||
if (ctrlinfo->action != HOST_OPEN && ctrlinfo->action != HOST_CLOSE)
|
||
return -EINVAL;
|
||
request.opcode = OPCode_hostcontrol;
|
||
request.body_which = Request_body_hostctrlmsg;
|
||
request.body.hostctrlmsg = ctrlinfo;
|
||
return xce_request(ep, &request);
|
||
}
|
||
```
|
||
|
||
**`src/include/request.h`** — Add declaration: `extern int xce_hostctrl_req(endpoint_t*, xce_hostctrl_t*);`
|
||
|
||
**`src/include/xceapi.h`** — Add declaration: `extern int xce_hostctrl(xce_ctx_t*, xce_hostctrl_t*);`
|
||
|
||
**`src/schema/controller.capnp`** — Two additions:
|
||
|
||
1. Add `HostControl` struct (after `QueueControl` ~line 516), mapping to `xce_hostctrl_t` (types.h:630-633):
|
||
```capnp
|
||
struct HostControl $C.mapname("xce_hostctrl_t") {
|
||
action @0: UInt32 $C.mapname("action");
|
||
hostname @1: Text $C.mapname("hostname");
|
||
}
|
||
```
|
||
|
||
2. Add `hostctrlmsg` to Request body union (after `queuectrlmsg @17`):
|
||
```capnp
|
||
hostctrlmsg @21: HostControl;
|
||
```
|
||
(Use @21 — the next available field number after `clrpreasonmsg @20`. Note: Cap'n Proto field numbers in the Request body union and in the OPCode enum are **independent namespaces** — the OPCode enum also has a `@21` value for `clrpreason`, but that is a different struct/enum and there is no conflict.)
|
||
|
||
After regenerating with `ninja -C build`, `controller.capnp.h` gains `Request_body_hostctrlmsg = 18` and the generated encode/decode functions for `HostControl`. (Do not hardcode the integer `18` in code — always use the generated symbolic constant `Request_body_hostctrlmsg`. Regenerate and check that the constant is present before committing.)
|
||
|
||
**`src/include/typemap.h`** — Add `xce_hostctrl_t* hostctrlmsg;` to the `xce_request_t` body union (line ~196, after `queuectrlmsg`). This handwritten C struct must be kept in sync with `controller.capnp` manually — regenerating the capnp schema updates `struct Request` in `controller.capnp.h` but does NOT update `typemap.h`:
|
||
```c
|
||
xce_queuectrl_t* queuectrlmsg;
|
||
xce_hostctrl_t* hostctrlmsg; /* NEW — must match @21 in controller.capnp */
|
||
xce_taskrun_t* runtaskmsg;
|
||
```
|
||
|
||
**Drift detection** (deferred): `typemap.h` and `controller.capnp` can silently diverge — new body union variants added to the capnp schema won't appear in `typemap.h` automatically. A CI step should count declared union arms in `typemap.h.body` and compare against `Request_body_*` constants in `controller.capnp.h`. A count mismatch means a field was added to one but not the other. Concretely: `grep -c 'msg;' typemap.h` vs `grep -c 'Request_body_' controller.capnp.h` should match after subtracting 1 for `Request_body_nodata`. This CI check is **deferred to a follow-on task** per project decision — for now, manual verification after regenerating the capnp schema is sufficient.
|
||
|
||
**`src/lib/helper.c`** — Add `Request_body_hostctrlmsg` case to both switches (`xce_request_body()` and `xce_request_msg_name()`). Note: `OPCode_hostcontrol` already exists in the opcode name switch at `helper.c:132` — only the `Request_body_*` case is new:
|
||
```c
|
||
case Request_body_hostctrlmsg:
|
||
return req->body.hostctrlmsg; // in xce_request_body()
|
||
```
|
||
Without this, `xce_request_body()` returns NULL for host control requests (breaks tracing/logging).
|
||
|
||
**`src/controller/cluster.c`** — Add `OPCode_hostcontrol` case (after `OPCode_queuecontrol`):
|
||
```c
|
||
case OPCode_hostcontrol:
|
||
rc = do_hostcontrol_op(c, ep, &request);
|
||
break;
|
||
```
|
||
|
||
**`src/controller/worker.c`** — Add `do_hostcontrol_op()` mirroring `do_queuecontrol_op()` in `queue.c`. `worker_ctrl()` already exists at `worker.c:579` and is declared in `controller/include/cluster.h:30`:
|
||
```c
|
||
int do_hostcontrol_op(cluster_t* c, endpoint_t* ep, xce_request_t* request)
|
||
{
|
||
int rc = 0;
|
||
xce_reply_t reply;
|
||
init_reply_by_req(&reply, request);
|
||
rc = worker_ctrl(c, request->body.hostctrlmsg);
|
||
reply.status = rc;
|
||
reply.data_which = RequestReply_data_nodata;
|
||
xce_reply(ep, c->myhost, &reply);
|
||
return rc;
|
||
}
|
||
```
|
||
|
||
Declare `do_hostcontrol_op` in `controller/include/cluster.h`.
|
||
|
||
**Go Side:**
|
||
|
||
**`src/frontend/apiserver/controllers/hosts.go`** — **Fix pre-existing data race**: `var allFlag bool` (line 22) is a package-level mutable variable shared across concurrent goroutines. Two concurrent `POST /hosts/control` requests can interleave: Request A sets `allFlag = false`, Request B sets `allFlag = true`, then A reads `true` and operates on all hosts instead of one. Eliminate the package-level variable. Return `allFlag` as a local `bool` from `preCheckHostControlRequest` and pass it as a parameter to `ControlHost`. The same fix applies to `queues.go` which has the identical pattern at lines 534-536 / 465.
|
||
|
||
**`src/frontend/apiserver/router/routers.go`** — Register both new routes under the existing JWT + RBAC middleware chain. Assign RBAC actions:
|
||
- `POST /xce/v1/hosts/control` → RBAC action `host:control`. Permitted roles: `cluster_admin` role. Only `cluster_admin` can execute host open/close operations — this is a cluster-altering action that affects availability.
|
||
- `POST /xce/v1/hosts/select` → RBAC action `host:view`. Permitted roles: all authenticated users. Normal users need to query host resource information to understand why their jobs cannot run, so this must be accessible beyond just admins. The action name is `host:view` (not `host:select`) to match the existing route registration in `routers.go:111` which already uses `RequirePermission("host:view")`.
|
||
|
||
Add both actions to `config/rbac.yaml` under the appropriate roles. The existing JWT authentication middleware applies to all `/xce/v1/` routes; no additional auth changes are needed beyond the RBAC entries.
|
||
|
||
**`src/frontend/apiserver/backend/clib.go`** — **Replace** the existing `ControlHost()` stub body (lines 310–312) with the real implementation below. Do NOT add a second definition — the stub already exists and must be replaced in-place:
|
||
```go
|
||
func ControlHost(hostname string, action int) error {
|
||
// Only HOST_OPEN and HOST_CLOSE are handled by worker_ctrl() in the controller.
|
||
// HOST_TAG/HOST_UNTAG will return an error from the controller side.
|
||
cHostname := C.CString(hostname)
|
||
defer C.free(unsafe.Pointer(cHostname))
|
||
ctrl := C.xce_hostctrl_t{
|
||
action: C.int(action),
|
||
hostname: cHostname,
|
||
}
|
||
// Note: C.xce_hostctrl_t has no union fields so direct struct literal is safe.
|
||
rc := C.xcecli_hostctrl(gCtx.ctx, &ctrl)
|
||
if rc < 0 {
|
||
return fmt.Errorf("xcecli_hostctrl failed: %d", int(rc))
|
||
}
|
||
return nil
|
||
}
|
||
```
|
||
|
||
Note on `OPCode_reply` routing: `xcecli_hostctrl()` blocks in `cbuf_get()` waiting for the controller reply. This is safe — confirmed by reading `xceapi.c`: every `xce_*` function calls the request function then immediately calls `cbuf_get(&ctx->wbuf, &status)`. The C network thread delivers `OPCode_reply` messages directly into `ctx->wbuf`; they never reach `go_push_handler`. No change needed to the Go dispatch switch.
|
||
|
||
**Blocking CGO call risk**: `xcecli_hostctrl()` blocks the calling goroutine's OS thread for the duration of the `cbuf_get()` wait. Under normal operation the controller responds within milliseconds. However, if the controller is unresponsive, no timeout bounds the wait. The existing `queuectrl` call has the same characteristic and has operated in production without issue — the controller is on the same machine (UNIX domain socket), so network latency is zero and controller response time is bounded. If a timeout becomes necessary in the future, the fix is to wrap the CGO call in a goroutine with `context.WithTimeout` and return an error on deadline — but this is not needed for the initial implementation given the local-socket deployment model.
|
||
|
||
**`src/frontend/apiserver/backend/stub.go`** — Four stub changes required for `//go:build !cgo` (xcecli) builds:
|
||
1. `ControlHost`: keep existing stub function; change body to return `errNoCGO` (do NOT remove it).
|
||
2. `StartHeartbeat`: add new stub — `func StartHeartbeat(ctx context.Context) {}` (no-op is correct; xcecli has no heartbeat).
|
||
3. `SetHeartbeatInterval`: add new stub — `func SetHeartbeatInterval(secs int) {}` (no-op; xcecli never calls this but the symbol must exist for link).
|
||
4. `EvalSelectForHost`: add new stub — `func EvalSelectForHost(expr string, host models.HostInfo, resourceVals map[string]float64) (bool, error) { return false, errNoCGO }`.
|
||
5. `Cleanup`: add new stub — `func Cleanup() {}`. `clib.go` defines `Cleanup()` at line 142; without a `!cgo` counterpart, any `xcecli` shutdown path that calls `backend.Cleanup()` (e.g., a defer in `main.go`) will fail at link time with "undefined: backend.Cleanup".
|
||
|
||
Also add `"context"` to the `stub.go` import block — `context.Context` in `StartHeartbeat`'s signature requires the `context` package. The current stub file only imports `"apiserver/models"` and `"errors"`; without `"context"` the stub will not compile.
|
||
|
||
Without stubs 2 and 3, `xcecli` (built with `CGO_ENABLED=0`) fails to link with undefined symbol errors.
|
||
|
||
---
|
||
|
||
## P1-1: Handle Resource Value Push Notifications
|
||
|
||
### Investigation: Resource Value Delivery Path
|
||
|
||
There is **one active path** through which resource value updates reach the API server:
|
||
|
||
> **Path A (dataset push) and Path C (OPCode_update) are NOT used for the apiserver.**
|
||
>
|
||
> - `Data_data_resources` resource definitions arrive via the cluster config push (`OPCode_push` / `configpushmsg`), handled by P0-1. The controller never sends `Data_data_resources` or `Data_data_resvals` as a separate `pushmsg` to the apiserver.
|
||
> - `OPCode_update` / `Update_data_resvals` is a **worker → controller** message (sent via `xce_update_resvals()` in xceapi.c, handled by `do_update_resvals()` in `opupd.c`). The controller converts it to an `OPCode_notify` message before forwarding to the apiserver via `apiserver_resource_hook_fun`. The apiserver never receives `OPCode_update` directly.
|
||
|
||
#### Active Path — `OPCode_notify` / `Request_body_notifymsg` (NotifyData)
|
||
|
||
`go_push_handler` does not yet handle `OPCode_notify` (distinct opcode from `OPCode_push`; must be added at the top-level switch, not inside `handlePush()`).
|
||
The body is `xce_notify_data_t` (retrieve via `xce_request_body(req)`, then cast).
|
||
Relevant case:
|
||
- `NotifyData_data_schedUpdateResvals` (event == 8): `data.sched_update_resvals` is `xce_resval_list_t*` (`typemap.h:140`), which is `{int count, xce_resval_t** vals}` (`typemap.h:100-103`)
|
||
|
||
### Changes
|
||
|
||
**`src/frontend/apiserver/backend/clib.go`**:
|
||
|
||
1. Add `OPCode_notify` case at the **top-level opcode switch** in `go_push_handler()` (NOT inside `handlePush()`):
|
||
|
||
Notify messages are sent with `OPCode_notify` (enum value 10), which is **distinct from** `OPCode_push` (value 6). Placing the handler inside `handlePush()` would make it dead code — `handlePush()` is only reached when `req.opcode == OPCode_push`.
|
||
```go
|
||
case C.OPCode_notify:
|
||
// Guard body_which before casting — xce_request_body() returns NULL for any
|
||
// body_which it doesn't recognise, and dereferencing a nil *xce_notify_data_t
|
||
// panics. Mirror the pattern used in handlePush().
|
||
if req.body_which != C.Request_body_notifymsg {
|
||
return 0
|
||
}
|
||
nd := (*C.xce_notify_data_t)(unsafe.Pointer(C.xce_request_body(req)))
|
||
if nd == nil { return 0 } // defensive: mirror handleConfigPush nil check
|
||
return handleNotifyData(nd)
|
||
```
|
||
|
||
2. Add `handleNotifyData()`:
|
||
|
||
> **C struct naming note**: `xce_notify_data_t` uses `int event` as its union discriminant field — NOT `data_which` (which is the discriminant name used by `xce_configdata_t`, `xce_reply_t`, and others). When adding new `NotifyData` cases to `helper.c`, switch on `d->event`, not `d->data_which`. The Go code below is correct; this note is for C-side additions.
|
||
|
||
```go
|
||
func handleNotifyData(nd *C.xce_notify_data_t) C.int {
|
||
if nd.event == C.NotifyData_data_schedUpdateResvals {
|
||
// xce_notify_data_ptr() returns nd->data.sched_update_resvals (the pointer
|
||
// stored in the union field), confirmed in helper.c.
|
||
// Origin: worker sends OPCode_update → controller's do_update_resvals fires
|
||
// XCE_CONTROLLER_HOOK_RESOURCE → apiserver_resource_hook_fun forwards via
|
||
// OPCode_notify / NotifyData_data_schedUpdateResvals. These are
|
||
// worker-measured values (despite the "sched_" prefix in the event name).
|
||
rvl := (*C.xce_resval_list_t)(C.xce_notify_data_ptr(nd))
|
||
if rvl == nil { return 0 }
|
||
n := int(rvl.count)
|
||
if n == 0 || rvl.vals == nil { return 0 }
|
||
vals := unsafe.Slice(rvl.vals, n)
|
||
for _, vp := range vals {
|
||
if vp == nil { continue }
|
||
rd := cResvalToDistribution(vp)
|
||
if rd != nil {
|
||
gHandler.OnResourceLocationUpdated(rd)
|
||
}
|
||
}
|
||
}
|
||
return 0
|
||
}
|
||
```
|
||
|
||
The call site above uses a helper `cResvalToDistribution(vp *C.xce_resval_t) *models.ResourceDistributionInfo` defined in `convert.go` (see below).
|
||
|
||
**`src/frontend/apiserver/backend/convert.go`** — Add `cResvalToDistribution()`:
|
||
- `xce_val_t` (`types.h:25-36`) has discriminant field `int which` (NOT `int type`). The union members are: `ivalue32` (int32), `uvalue32` (uint32), `ivalue64` (int64), `uvalue64` (uint64), `svalue` (char*), **`fvalue`** (double), `bvalue` (int). Switch on `val.which` using the **`Value_u_*` constants** from `controller.capnp.h`. **Do NOT use the `XCE_VAL_*` macros** (`XCE_VAL_STRING=0`, `XCE_VAL_NUMERIC=1`, `XCE_VAL_BOOLEAN=2`) — they are an unrelated classification and their integer values do not correspond to the `which` field. **Do NOT combine `Value_u_ivalue32`/`Value_u_ivalue64` (or the u32/u64 pair) into a single case** — each needs its own typed helper since reading `ivalue64` when only `ivalue32` is set gives garbage upper 32 bits (confirmed by union layout in `types.h:28-31`).
|
||
- CGO cannot access the `xce_val_t` union directly; add these C helpers in the `clib.go` CGO preamble:
|
||
```c
|
||
static int xce_val_which(xce_val_t *v) { return v->which; }
|
||
static int32_t xce_val_i32(xce_val_t *v) { return v->u.ivalue32; }
|
||
static int64_t xce_val_i64(xce_val_t *v) { return v->u.ivalue64; }
|
||
static uint32_t xce_val_u32(xce_val_t *v) { return v->u.uvalue32; }
|
||
static uint64_t xce_val_u64(xce_val_t *v) { return v->u.uvalue64; }
|
||
static double xce_val_fval(xce_val_t *v) { return v->u.fvalue; } /* union field is named 'fvalue', not 'dvalue' — double-typed but float-named */
|
||
static char* xce_val_str(xce_val_t *v) { return v->u.svalue; }
|
||
static int xce_val_bool(xce_val_t *v) { return v->u.bvalue; }
|
||
/* NOTE: Do NOT combine ivalue32/ivalue64 or uvalue32/uvalue64 into one case.
|
||
* xce_val_t is a C union — ivalue32 (4 bytes) and ivalue64 (8 bytes) occupy
|
||
* the same address. Reading ivalue64 when which==ivalue32 gives garbage upper
|
||
* 32 bits. Each accessor must match the exact which value. */
|
||
```
|
||
- Add `cResvalToDistribution()` in `convert.go`. Called from `handleNotifyData` — returns `nil` on invalid input (vp.val == nil), caller must check before passing to handler:
|
||
```go
|
||
// cResvalToDistribution converts a C xce_resval_t into a Go ResourceDistributionInfo.
|
||
// vp.val is *C.xce_val_t (a pointer field in xce_resval_t) — pass directly, no &.
|
||
func cResvalToDistribution(vp *C.xce_resval_t) *models.ResourceDistributionInfo {
|
||
rd := &models.ResourceDistributionInfo{
|
||
ResourceName: C.GoString(vp.name),
|
||
Location: C.GoString(vp.host),
|
||
IsDynamic: true,
|
||
}
|
||
// Guard: vp.val is a pointer field in xce_resval_t; it may be NULL if the
|
||
// controller sends a resval with no value set. Dereferencing NULL in
|
||
// xce_val_which() would segfault in C. Log the error and skip this entry
|
||
// rather than returning partial data or crashing.
|
||
if vp.val == nil {
|
||
logger.Warn("resval with nil value, skipping", zap.String("resource", rd.ResourceName), zap.String("host", rd.Location))
|
||
return nil
|
||
}
|
||
switch C.xce_val_which(vp.val) {
|
||
case C.Value_u_ivalue32:
|
||
rd.ValueType = int(models.NUMERIC) // models.NUMERIC=1; XCE_VAL_NUMERIC is a C macro, not a Go symbol
|
||
rd.NumericValue = float64(C.xce_val_i32(vp.val))
|
||
case C.Value_u_ivalue64:
|
||
rd.ValueType = int(models.NUMERIC)
|
||
rd.NumericValue = float64(C.xce_val_i64(vp.val))
|
||
case C.Value_u_uvalue32:
|
||
rd.ValueType = int(models.NUMERIC)
|
||
rd.NumericValue = float64(C.xce_val_u32(vp.val))
|
||
case C.Value_u_uvalue64:
|
||
rd.ValueType = int(models.NUMERIC)
|
||
rd.NumericValue = float64(C.xce_val_u64(vp.val))
|
||
case C.Value_u_fvalue:
|
||
rd.ValueType = int(models.NUMERIC)
|
||
rd.NumericValue = float64(C.xce_val_fval(vp.val))
|
||
case C.Value_u_svalue:
|
||
rd.ValueType = int(models.STRING) // models.STRING=0; XCE_VAL_STRING is a C macro, not a Go symbol
|
||
rd.StringValue = C.GoString(C.xce_val_str(vp.val))
|
||
case C.Value_u_bvalue:
|
||
// ResourceDistributionInfo has a dedicated BooleanValue bool field (models.BOOLEAN=2).
|
||
rd.ValueType = int(models.BOOLEAN)
|
||
rd.BooleanValue = C.xce_val_bool(vp.val) != 0
|
||
default:
|
||
// Unknown union discriminant — log and skip rather than returning corrupted data.
|
||
// Future protocol versions may add new value types; this guard prevents silent
|
||
// zero-initialization of fields that would appear as empty strings or zeros.
|
||
logger.Warn("unknown xce_val_t discriminant, skipping", zap.Int("which", int(C.xce_val_which(vp.val))), zap.String("resource", rd.ResourceName), zap.String("host", rd.Location))
|
||
return nil
|
||
}
|
||
return rd
|
||
}
|
||
```
|
||
|
||
**`cResdefToModel()` note** — P0-1 adds `cResdefToModel(r *C.xce_resdef_t) *models.ResourceInfo` (note: `resdef` not `resource`) to avoid colliding with the existing `cResourceToModel(res *C.xce_resource_t) *models.ResourceDefinitionInfo` at `convert.go:96`. These are different C structs (`xce_resdef_t` vs `xce_resource_t`) mapping to different Go models (`ResourceInfo` vs `ResourceDefinitionInfo`). P1-1 does NOT add either function — it reuses the existing `cResourceToModel` for the resource definition path and `cResvalToDistribution` for the resval push path.
|
||
|
||
**`src/frontend/apiserver/backend/notification.go`** — No change needed. The existing `OnResourceLocationUpdated(res *models.ResourceDistributionInfo)` already writes to `store.ResourceLocations` keyed by `(resName, host)`. Reuse it via `cResvalToDistribution`. **Do NOT add** `OnResourceValUpdated` — that would create an interface split whose `interface{}` → `ResourceDistributionInfo` field-mapping is unspecified.
|
||
|
||
**`src/frontend/apiserver/store/stores.go`** — The existing `ResourceLocations` store maps `(resName, host)` → `ResourceDistributionInfo`. Live values from push merge into that store via `OnResourceLocationUpdated`, which already performs this write.
|
||
|
||
---
|
||
|
||
## P1-2: Host Resource Requirement Select (FilterHostByResReqSelect)
|
||
|
||
### Problem
|
||
`FilterHostByResReqSelect()` in `controllers/hosts.go:376` returns empty results — it's a stub. This needs to evaluate resource requirement expressions against current host data.
|
||
|
||
### Investigation: Existing PEG Infrastructure
|
||
|
||
The codebase already has three PEG grammar files compiled via packcc into C parsers:
|
||
|
||
| File | Prefix | What it parses |
|
||
|------|--------|----------------|
|
||
| `lib/section.peg` | `xce_section` | Full resource section: `select[…] rusage[…] span[…] same[…] affinity[…]` |
|
||
| `lib/rusage.peg` | `xce_rusage` | The rusage clause: `res=value[:duration=N][:decay=0/1], …` |
|
||
| `lib/xce_resreq.peg` | `xce_resreq` | Compound/alternate resource spec: `{primary} + {primary}` |
|
||
|
||
`section.peg` extracts the `select[…]` content into `xce_resreq_data_t.selectString` (a raw string). However, `xce_resreq_select_` is currently **an empty struct** (`resreq.h:67`). The select string is stored but never parsed further.
|
||
|
||
**The select expression** (e.g. `"type == any && mem > 100"`) is a boolean filter expression over host attributes and dynamic resource values. It is used by both the API server (`/xce/v1/hosts/select`) and the scheduler (host selection for job dispatch) — so the parser belongs in `lib/resreq.c` where all the other parsers live.
|
||
|
||
### Design: New `xce_select.peg` Grammar
|
||
|
||
Add `lib/xce_select.peg` with prefix `xce_select`. The grammar parses a boolean expression tree:
|
||
|
||
```peg
|
||
%prefix "xce_select"
|
||
%auxil "select_context_t *"
|
||
%common { #include "resreq.h" ... }
|
||
|
||
statement <- _ expr _ EOF
|
||
|
||
expr <- or_expr
|
||
|
||
or_expr <- left:and_expr (_ "||" _ right:and_expr)* { /* build OR node */ }
|
||
|
||
and_expr <- left:cmp_expr (_ "&&" _ right:cmp_expr)* { /* build AND node */ }
|
||
|
||
cmp_expr <- '(' _ expr _ ')'
|
||
/ attr:identifier _ op:cmp_op _ val:cmp_value
|
||
{ /* build COMPARE node: attr, op, val */ }
|
||
|
||
cmp_op <- ">=" / "<=" / "!=" / "==" / ">" / "<"
|
||
{ auxil->op = parse_cmp_op($0); }
|
||
|
||
cmp_value <- number { auxil->val_type = SEL_VAL_NUM; auxil->val_num = atof($0); }
|
||
/ "any" { auxil->val_type = SEL_VAL_ANY; }
|
||
/ identifier { auxil->val_type = SEL_VAL_STR; auxil->val_str = strdup($0); }
|
||
|
||
identifier <- [a-zA-Z_][a-zA-Z_0-9]* { auxil->attr = strdup($0); }
|
||
number <- [0-9]+('.'[0-9]*)? { auxil->tmp_num = atof($0); }
|
||
_ <- [ \t]*
|
||
EOF <- !.
|
||
```
|
||
|
||
### Data Structures and Declarations (in `resreq.h`)
|
||
|
||
Replace the empty `xce_resreq_select_` with a proper expression tree, and add `extern` declarations for the three new functions (required so CGO can call them when `#include "resreq.h"` is in `clib.go`'s preamble):
|
||
|
||
```c
|
||
/* Add alongside existing extern declarations in resreq.h */
|
||
extern int parse_select(xce_select_t** sel, char* expr, char** errmsg);
|
||
extern int eval_select(xce_select_t* sel,
|
||
double (*get_attr)(char*, uintptr_t), uintptr_t ctx);
|
||
extern void free_select(xce_select_t* sel);
|
||
```
|
||
|
||
```c
|
||
typedef enum {
|
||
XCE_SEL_AND = 0,
|
||
XCE_SEL_OR,
|
||
XCE_SEL_CMP,
|
||
} xce_sel_op_t;
|
||
|
||
typedef enum {
|
||
XCE_SEL_VAL_NUM = 0,
|
||
XCE_SEL_VAL_STR,
|
||
XCE_SEL_VAL_ANY, /* "any" keyword — matches any value */
|
||
} xce_sel_val_type_t;
|
||
|
||
typedef enum {
|
||
XCE_SEL_EQ = 0, XCE_SEL_NE,
|
||
XCE_SEL_GT, XCE_SEL_GE,
|
||
XCE_SEL_LT, XCE_SEL_LE,
|
||
} xce_sel_cmp_t;
|
||
|
||
typedef struct xce_sel_node_ xce_sel_node_t;
|
||
|
||
struct xce_sel_node_ {
|
||
xce_sel_op_t op;
|
||
union {
|
||
struct { xce_sel_node_t* left; xce_sel_node_t* right; } logical;
|
||
struct {
|
||
char* attr; /* e.g. "type", "mem", "r1m" */
|
||
xce_sel_cmp_t cmp;
|
||
xce_sel_val_type_t val_type;
|
||
union { double num; char* str; } val;
|
||
} cmp;
|
||
} u;
|
||
};
|
||
|
||
struct xce_resreq_select_ {
|
||
xce_sel_node_t* root;
|
||
};
|
||
```
|
||
|
||
### C API in `lib/resreq.c`
|
||
|
||
Add three functions:
|
||
|
||
```c
|
||
/* Parse a select expression string into *sel. Returns 0 on success.
|
||
* Note: expr is char* (not const) to match the existing parse_* pattern in
|
||
* resreq.c where context structs store a char* input field — passing const
|
||
* would require a const-discard cast and produce -Werror warnings. */
|
||
int parse_select(xce_select_t** sel, char* expr, char** errmsg);
|
||
|
||
/* Evaluate *sel against a host attribute lookup function.
|
||
* get_attr(name, ctx) returns the attribute value (float) or NAN if unknown.
|
||
* Returns 1 if host matches, 0 if not. */
|
||
int eval_select(xce_select_t* sel,
|
||
double (*get_attr)(char*, uintptr_t), uintptr_t ctx);
|
||
|
||
void free_select(xce_select_t* sel);
|
||
```
|
||
|
||
The `get_attr` callback is intentionally generic — it can be backed by:
|
||
- A `xce_worker_t*` for scheduler use (pure C, no CGO needed)
|
||
- A flat attribute table for the Go API server path (see Go integration below)
|
||
|
||
### Note on N-ary Operators
|
||
|
||
**The grammar rules in the Design section above must be rewritten** — the `*` (zero-or-more) quantifier form shown there is not directly supported with accumulator state in packcc action bodies. The `while (next_right)` pseudo-code is conceptual; packcc does not expose loop state across repetitions in a user action.
|
||
|
||
**Correct packcc approach — right-recursive alternation:**
|
||
|
||
```peg
|
||
or_expr <- l:and_expr _ "||" _ r:or_expr
|
||
{ $$ = build_logical_node(XCE_SEL_OR, l, r); }
|
||
/ and_expr
|
||
|
||
and_expr <- l:cmp_expr _ "&&" _ r:and_expr
|
||
{ $$ = build_logical_node(XCE_SEL_AND, l, r); }
|
||
/ cmp_expr
|
||
```
|
||
|
||
PEG parsers cannot handle left-recursive rules (they would loop infinitely), so left-associativity via the `*` pattern requires explicit accumulation in the grammar's semantic actions. Right-recursive alternation (as above) is the standard packcc solution. The resulting tree is right-associative (`a || b || c` → `or(a, or(b, c))`), which is semantically equivalent for boolean operators. If left-associativity is required for evaluation order, use an auxiliary list accumulated in `auxil` during parsing:
|
||
|
||
```c
|
||
/* In auxil struct: */
|
||
struct { xce_sel_node_t** items; int count; int cap; } pending_or;
|
||
|
||
/* or_tail rule collects each right-hand operand: */
|
||
or_tail <- _ "||" _ r:and_expr { append_to_pending(auxil, r); }
|
||
or_expr <- head:and_expr or_tail* {
|
||
/* build left-associative chain from head + pending list */
|
||
$$ = build_left_chain(XCE_SEL_OR, head, auxil->pending_or);
|
||
clear_pending(auxil);
|
||
}
|
||
```
|
||
|
||
The right-recursive approach is simpler and sufficient — use it unless left-associativity is explicitly required by the select expression spec.
|
||
|
||
Also update **`kill_primary_resreq()`** in `resreq.c` to call `free_select(resreq->select)` before `xlist_fini`. This is the correct location — `kill_primary_resreq()` (resreq.c:236) owns the `xce_primary_resreq_t` and frees its members; `kill_resreq_data()` only frees string fields and `kill_resreq()` delegates to it. Adding `free_select()` to either of those would be in the wrong function. Without this fix, every parsed select expression leaks its AST.
|
||
|
||
**NULL safety and struct freeing at the call site are both mandatory.**
|
||
|
||
`xce_primary_resreq_t.select` is NULL for every resreq that has no select[] clause (virtually all existing job resource requirements). The codebase convention (confirmed in `kill_compound_resreq`: `kill_primary_resreq(primary); FREEUP(primary);`) is: kill_ functions free the interior, the **caller** frees the container. `free_select` must follow the same convention — free the node tree only, not `sel` itself. Therefore `kill_primary_resreq` must both call `free_select` AND free the struct:
|
||
|
||
```c
|
||
if (resreq->select) {
|
||
free_select(resreq->select); /* frees node tree (interior) */
|
||
FREEUP(resreq->select); /* frees xce_select_t struct (container) */
|
||
}
|
||
```
|
||
|
||
**Do NOT use `resreq->select = NULL` without `FREEUP`** — the `= NULL` assignment does not call `free()` and the 8-byte `xce_select_t` struct leaks. Without the `if` guard, calling `free_select(NULL)` on any existing job (whose `select` is NULL) will crash; `free_select` should start with `if (!sel) return;` but the guard at the call site is also required per the codebase defensive pattern.
|
||
|
||
### Go Integration
|
||
|
||
**`src/frontend/apiserver/backend/clib.go`** — Export `EvalSelectForHost()`:
|
||
|
||
Three additions to `clib.go` are required before the code will compile:
|
||
1. Add `#include "resreq.h"` to the CGO preamble block (alongside `client.h`, `netio.h`, etc.) — without it `xce_select_t`, `parse_select`, `eval_select`, and `free_select` are all undeclared.
|
||
2. Add `extern` declaration for the Go-exported callback plus a destroy helper in the CGO preamble (same pattern as `go_push_handler`). Without the extern, `C.xce_select_get_attr` is undefined:
|
||
```c
|
||
extern double xce_select_get_attr(char* name, uintptr_t ctx);
|
||
/* xce_select_destroy: free tree nodes then the struct in one call.
|
||
* free_select() frees only the interior (node tree); parse_select()
|
||
* malloc'd the xce_select_t struct, so the caller must free it too.
|
||
* A single C helper avoids Go defer LIFO ordering issues (free before
|
||
* free_select would be use-after-free). */
|
||
static void xce_select_destroy(xce_select_t* sel) {
|
||
if (sel) { free_select(sel); free(sel); }
|
||
}
|
||
```
|
||
3. Add `"runtime/cgo"` and `"math"` to the Go import block — `runtime/cgo` for `cgo.NewHandle`/`h.Delete()`, `math` for `math.NaN()` (the NaN return value from the callback; `C.NAN` is a macro and cannot be called from CGO).
|
||
|
||
The `void* ctx` parameter of `eval_select` carries per-call state. Go maps cannot be passed as `void*` (they contain GC-managed pointers, violating CGO pointer rules). Use a **`cgo.Handle`** instead:
|
||
|
||
```go
|
||
// hostSelectCtx holds per-evaluation attribute data accessible from the C callback.
|
||
type hostSelectCtx struct {
|
||
host models.HostInfo
|
||
resourceVals map[string]float64
|
||
}
|
||
|
||
//export xce_select_get_attr
|
||
func xce_select_get_attr(name *C.char, ctx C.uintptr_t) C.double {
|
||
h := cgo.Handle(ctx) // C.uintptr_t is an integer, not a pointer — no go vet warning
|
||
sc := h.Value().(*hostSelectCtx)
|
||
attrName := C.GoString(name)
|
||
// look up attrName in sc.host fields and sc.resourceVals
|
||
// return NaN if unknown: use C.double(math.NaN()) — NAN is a C macro,
|
||
// not a callable function, so C.NAN / C.math_nan() do not compile in CGO.
|
||
// Requires adding "math" to Go imports in clib.go.
|
||
...
|
||
}
|
||
|
||
func EvalSelectForHost(expr string, host models.HostInfo, resourceVals map[string]float64) (bool, error) {
|
||
cExpr := C.CString(expr)
|
||
defer C.free(unsafe.Pointer(cExpr))
|
||
|
||
var sel *C.xce_select_t
|
||
var errmsg *C.char
|
||
if rc := C.parse_select(&sel, cExpr, &errmsg); rc < 0 {
|
||
// Guard against NULL errmsg: parse_select may return an error
|
||
// (e.g. -EINVAL for a NULL/empty input) without allocating an error
|
||
// message. C.GoString(nil) has undefined behaviour in CGO.
|
||
var msg string
|
||
if errmsg != nil {
|
||
msg = C.GoString(errmsg)
|
||
C.free(unsafe.Pointer(errmsg))
|
||
} else {
|
||
msg = "unknown parse error"
|
||
}
|
||
return false, fmt.Errorf("parse_select: %s", msg)
|
||
}
|
||
defer C.xce_select_destroy(sel) // frees tree nodes then the struct itself
|
||
|
||
// Register Go context with a handle — safe to pass uintptr into C
|
||
sc := &hostSelectCtx{host: host, resourceVals: resourceVals}
|
||
h := cgo.NewHandle(sc)
|
||
defer h.Delete()
|
||
|
||
result := C.eval_select(sel,
|
||
(*[0]byte)(C.xce_select_get_attr), // //export function pointer
|
||
C.uintptr_t(h)) // cgo.Handle is a uintptr — integer cast, no unsafe.Pointer needed; avoids go vet "unsafe.Pointer(uintptr(...))" warning
|
||
return result != 0, nil
|
||
}
|
||
```
|
||
|
||
**`src/frontend/apiserver/controllers/hosts.go`** — `FilterHostByResReqSelect`:
|
||
- Iterate `store.Hosts`
|
||
- For each host, collect live resource vals from `store.ResourceLocations`
|
||
- Call `backend.EvalSelectForHost(selectExpr, host, resourceVals)`
|
||
- Include host in result if it returns `true`
|
||
|
||
**Performance**: this path is O(H × R) per request where H = number of hosts and R = number of resource types per host. For each host, collecting resource vals from `store.ResourceLocations` requires one map lookup per resource type that might appear in the expression. For clusters with hundreds of hosts and dozens of resource types, a single `/select` call may involve thousands of CGO round-trips (one `eval_select` call per host). **Acceptable for now**: `/xce/v1/hosts/select` is a user-facing query endpoint, not a hot scheduling path. If profiling shows this is a bottleneck, the optimization is to parse the select expression once (call `parse_select` before the host loop, pass the `xce_select_t*` directly) — `EvalSelectForHost` currently parses on every call. A future `EvalSelectForHostParsed(sel *C.xce_select_t, ...)` variant would amortize parsing cost. Do not over-engineer this in the initial implementation.
|
||
|
||
**Input safety**: the select expression is user-supplied input passed through a CGO boundary to a C PEG parser. The following safety measures are required:
|
||
1. **Expression length limit**: cap the select expression string to 1024 bytes at the Go handler level before passing to `parse_select`. Select expressions in job submissions are short by nature (e.g., `type == LINUX64 && r1m > 1.5`); 1024 bytes is generous and prevents allocating unbounded C buffers. Reject with HTTP 400 if the limit is exceeded. **Update `preCheckHostRequest()` in `controllers/hosts.go`** to change the existing 4094-character limit to 1024 bytes — the old limit was set before CGO-based evaluation was designed and is too permissive for the PEG parser.
|
||
2. **CGO crash containment**: `recover()` in Go only catches Go panics. A C SIGSEGV (null pointer dereference in `parse_select`/`eval_select`) terminates the process immediately — Go defers and `recover()` are bypassed. The PEG parser must be hardened with internal null/bounds checks in C. Add a comment in the handler: `// Note: recover() does NOT catch C SIGSEGV. C-level null dereference in parse_select/eval_select terminates the process. Rely on C-level validation, not Go recovery.` For defense-in-depth, still wrap `EvalSelectForHost` in `defer func() { if r := recover(); r != nil { ... } }()` — this catches Go-level panics (e.g., nil interface dereference in the Go callback) but cannot prevent C-level crashes. Additionally, C-level hardening is required: `parse_select` must validate input before parsing (empty string, NULL pointer, length check) and `eval_select` must null-check `sel != NULL` before traversing the tree.
|
||
3. **Hostname sanitization**: the `host_name` field in `/xce/v1/hosts/control` requests must be validated against RFC 952/1123 hostname rules (alphanumeric, hyphens, dots; max 253 chars) before passing to CGO. The list-length cap for the `host_name` array is 256 entries per request.
|
||
|
||
### Build Integration
|
||
|
||
`src/lib/meson.build` lists `.peg` sources **explicitly** (confirmed: `section.peg`, `rusage.peg`, `xce_resreq.peg` are each listed individually — no glob). Add `xce_select.peg` to that explicit list.
|
||
|
||
### Scope Note
|
||
|
||
This parser lives in `libxce` (not in the Go layer). **Rationale**: the `select[]` expression is parsed by the existing `xce_section.peg` infrastructure in `lib/resreq.c` — the select string is already extracted there into `xce_resreq_data_t.selectString`. Placing `parse_select` + `eval_select` in the same library keeps the data flow coherent (parse → the same struct that holds the raw string) and allows a future scheduler path to call `eval_select` directly against `xce_worker_t*` attributes without going through CGO on the hot scheduling path.
|
||
|
||
**Scheduler integration is explicitly deferred**: this plan does not add scheduler calls to `eval_select`. The C API is designed to be scheduler-compatible (the `get_attr` callback abstraction allows any attribute source), but wiring it into the scheduler's host selection loop is a separate task. If the decision is made to keep `select` evaluation in the Go layer only (e.g., because scheduler rewrite is planned), `parse_select`/`eval_select` can be moved to a Go PEG implementation at that point without breaking the API server path. The C placement is a forward-compatible choice, not a commitment to the scheduler design.
|
||
|
||
**Grammar validation requirement**: before finalizing `xce_select.peg`, validate it against a corpus of real select expressions from existing job submissions and documentation examples. The grammar above is derived from the documented select syntax but has not been tested against production workloads. Known expression forms to cover: `type == any`, `type == LINUX64`, `mem > 100`, `r1m < 1.5 && ncpus >= 2`, `(mem > 256 || swap > 512) && type != WINDOWS`. Without this validation, the grammar may fail on valid expressions that users already rely on.
|
||
|
||
---
|
||
|
||
## P1-3: Heartbeat Keep-Alive
|
||
|
||
### Problem
|
||
The API server joins the controller as `HostRole_apiserver` but never sends heartbeat keep-alives. If the controller times out the connection, the API server loses its push subscription silently.
|
||
|
||
### Changes
|
||
|
||
**`src/frontend/apiserver/backend/clib.go`** — Add `"sync/atomic"`, `"context"`, and `"time"` to the Go import block. Add a package-level interval store and `StartHeartbeat()`:
|
||
|
||
```go
|
||
// heartbeatIntervalSecs is the current heartbeat interval in whole seconds.
|
||
// Initialised to 30; updated atomically by SetHeartbeatInterval when
|
||
// OnClusterUpdated fires with the controller-supplied value.
|
||
var heartbeatIntervalSecs int32 = 30
|
||
|
||
// heartbeatFn is the variable holding the heartbeat C call, injectable for testing.
|
||
// In production it wraps C.xcecli_heartbeat; tests replace it with a counting stub.
|
||
var heartbeatFn func(secs int) int = func(secs int) int {
|
||
return int(C.xcecli_heartbeat(gCtx.ctx, C.int(secs)))
|
||
}
|
||
|
||
// SetHeartbeatInterval stores a new heartbeat interval (in seconds).
|
||
// Called from OnClusterUpdated in handler.go when a cluster config push arrives.
|
||
// The running StartHeartbeat goroutine reads this value on each tick so it picks
|
||
// up the controller-specified interval without a restart.
|
||
// Range: clamped to [1, 3600] (1 second minimum to prevent spin; 3600 seconds = 1 hour max).
|
||
func SetHeartbeatInterval(secs int) {
|
||
if secs < 1 {
|
||
secs = 1
|
||
}
|
||
if secs > 3600 {
|
||
secs = 3600
|
||
}
|
||
atomic.StoreInt32(&heartbeatIntervalSecs, int32(secs))
|
||
}
|
||
|
||
func StartHeartbeat(ctx context.Context) {
|
||
go func() {
|
||
// Read the current interval at first tick; re-read on every tick so
|
||
// an OnClusterUpdated call mid-flight takes effect without a restart.
|
||
secs := int(atomic.LoadInt32(&heartbeatIntervalSecs))
|
||
if secs < 1 { secs = 1 }
|
||
ticker := time.NewTicker(time.Duration(secs) * time.Second)
|
||
defer ticker.Stop()
|
||
for {
|
||
select {
|
||
case <-ctx.Done():
|
||
return // clean shutdown — no goroutine leak on server exit
|
||
case <-ticker.C:
|
||
newSecs := int(atomic.LoadInt32(&heartbeatIntervalSecs))
|
||
if newSecs >= 1 && newSecs != secs {
|
||
secs = newSecs
|
||
ticker.Reset(time.Duration(secs) * time.Second)
|
||
}
|
||
rc := heartbeatFn(secs)
|
||
if rc < 0 {
|
||
logger.Warn("Heartbeat failed", zap.Int("rc", int(rc)))
|
||
// Single failure is logged but not escalated. Connection
|
||
// recovery is handled by the push subscription layer.
|
||
}
|
||
}
|
||
}
|
||
}()
|
||
}
|
||
```
|
||
|
||
Note: `xce_heartbeat()` in `xceapi.c` already suppresses sends when recent traffic exists (`now - ts_last_out < hb_interval`), so the goroutine is a safe periodic trigger, not a rate-multiplier.
|
||
|
||
**`src/frontend/apiserver/backend/handler.go`** — In `OnClusterUpdated`, call `SetHeartbeatInterval` after partial-merging to the cluster store:
|
||
```go
|
||
func (h *DefaultHandler) OnClusterUpdated(cluster *models.ClusterInfo) {
|
||
// Partial merge: preserve existing built-in/sourced fields, overwrite only config-sourced fields.
|
||
existing, ok := store.Clusters.Get(cluster.ClusterName)
|
||
if ok {
|
||
existing.HeartbeatInterval = cluster.HeartbeatInterval
|
||
existing.Vcpus = cluster.Vcpus
|
||
existing.InactiveHeartbeatCount = cluster.InactiveHeartbeatCount
|
||
store.Clusters.Put(cluster.ClusterName, existing)
|
||
} else {
|
||
store.Clusters.Put(cluster.ClusterName, *cluster)
|
||
}
|
||
if cluster.HeartbeatInterval > 0 {
|
||
SetHeartbeatInterval(cluster.HeartbeatInterval) // wire interval from config push into heartbeat goroutine
|
||
}
|
||
}
|
||
```
|
||
|
||
**`src/frontend/apiserver/cmd/main.go`** — Call `backend.StartHeartbeat(ctx)` after `backend.RegisterToBackend()`, passing the server's root context so the goroutine exits when the server shuts down. `RegisterToBackend` is a synchronous operation — it blocks until the controller connection is established and the initial configuration push is received. This means the first `OnClusterUpdated` call (which sets `heartbeatIntervalSecs` via `SetHeartbeatInterval`) fires **before** `StartHeartbeat` is called. The heartbeat therefore starts with the correct interval from the controller, not the 30s default:
|
||
```go
|
||
ctx, cancel := context.WithCancel(context.Background())
|
||
defer cancel()
|
||
// ...
|
||
backend.RegisterToBackend(...) // synchronous: blocks until connected + first config push received
|
||
backend.StartHeartbeat(ctx) // interval already set by OnClusterUpdated from config push
|
||
// The 30s default is only used if StartHeartbeat is called before the first
|
||
// config push arrives; in normal startup this does not happen because
|
||
// RegisterToBackend is synchronous.
|
||
```
|
||
|
||
---
|
||
|
||
## Files to Modify
|
||
|
||
| Priority | File | Change |
|
||
|----------|------|--------|
|
||
| P0-1 | `backend/clib.go` | Expand `handleConfigPush` for host configs + cluster settings; call `cClusterConfToClusterInfo()` then `gHandler.OnClusterUpdated()` |
|
||
| P0-1 | `backend/convert.go` | Add `cHostConfigToModel()`, `cClusterConfHostConfigs()` (iterates `conf.nhconfigs`/`conf.hconfigs` via `unsafe.Slice` — guards both `n == 0` and `conf.hconfigs == nil`), `cClusterConfToClusterInfo()`, `cClusterConfResources()`, `cResdefToModel()` (named to avoid collision with existing `cResourceToModel`) |
|
||
| P0-1 | `backend/convert_pure.go` | Add `rawHostConfigData`+`rawHostConfigToModel()` and `rawClusterData`+`rawClusterToModel()` |
|
||
| P0-1 | `backend/notification.go` | Add `OnClusterUpdated`, `OnHostConfigUpdated` |
|
||
| P0-1 | `backend/handler.go` | Implement `OnClusterUpdated` (partial merge: overwrite only config-sourced fields + calls `SetHeartbeatInterval`), `OnHostConfigUpdated` (partial merge), `OnHostUpdated` (partial merge: overwrite only worker-sourced fields, preserve config-sourced fields) |
|
||
| P0-1 | `models/` | Add `HeartbeatInterval`, `Vcpus`, `InactiveHeartbeatCount` fields to `ClusterInfo` |
|
||
| P0-1 | `models/hosts.go` | Change `HostStateStr(0)` from returning `"OK"` to returning `"UNKNOWN"` — a host that has not yet reported its status via worker push should not appear healthy |
|
||
| P0-2 | `schema/controller.capnp` | Add `HostControl` struct + `hostctrlmsg @21` to Request body union |
|
||
| P0-2 | `include/typemap.h` | Add `xce_hostctrl_t* hostctrlmsg;` to `xce_request_t.body` union (handwritten — not auto-generated) |
|
||
| P0-2 | `include/client.h` | Add `xcecli_hostctrl()` declaration |
|
||
| P0-2 | `include/xceapi.h` | Add `xce_hostctrl()` declaration |
|
||
| P0-2 | `include/request.h` | Add `xce_hostctrl_req()` declaration |
|
||
| P0-2 | `lib/client.c` | Implement `xcecli_hostctrl()` |
|
||
| P0-2 | `lib/xceapi.c` | Implement `xce_hostctrl()` with `cbuf_get()` wait |
|
||
| P0-2 | `lib/request.c` | Implement `xce_hostctrl_req()` |
|
||
| P0-2 | `lib/helper.c` | Add `Request_body_hostctrlmsg` case to `xce_request_body()` + `xce_request_msg_name()` |
|
||
| P0-2 | `controller/cluster.c` | Add `OPCode_hostcontrol` case → `do_hostcontrol_op()` |
|
||
| P0-2 | `controller/worker.c` | Add `do_hostcontrol_op()` calling `worker_ctrl()` |
|
||
| P0-2 | `controller/include/cluster.h` | Declare `do_hostcontrol_op()` |
|
||
| P0-2 | `backend/clib.go` | Implement real `ControlHost()` via `xcecli_hostctrl()` |
|
||
| P0-2 | `controllers/hosts.go` | **Fix pre-existing data race**: eliminate package-level `var allFlag bool` (line 22); return `allFlag` as a local `bool` from `preCheckHostControlRequest` and pass as parameter to `ControlHost`. Same fix in `controllers/queues.go` lines 534-536 / 465. |
|
||
| P0-2 | `backend/stub.go` | Keep `ControlHost` stub; change body to `errNoCGO`. Add stubs: `StartHeartbeat(ctx context.Context) {}`, `SetHeartbeatInterval(secs int) {}`, `EvalSelectForHost(...) (bool, error)` returning `errNoCGO`, `Cleanup() {}`. Add `"context"` to import block (for `context.Context` in `StartHeartbeat`). |
|
||
| P0-2 | `router/routers.go` | Register `POST /xce/v1/hosts/control` under JWT + RBAC middleware with action `host:control` (`cluster_admin` role), and `POST /xce/v1/hosts/select` under JWT + RBAC middleware with action `host:view` (all authenticated users). Both routes already exist at lines 111-112 — this row confirms their RBAC actions are correct. Both routes inherit the existing RateLimit middleware. |
|
||
| P0-2 | `config/rbac.yaml` | Add `host:control` and `host:view` action entries with role assignments |
|
||
| P1-1 | `backend/clib.go` | Add CGO val helpers (`xce_val_which`, `xce_val_i32`, `xce_val_i64`, `xce_val_u32`, `xce_val_u64`, `xce_val_fval`, `xce_val_str`, `xce_val_bool`) to preamble; add `OPCode_notify` case in `go_push_handler` + `handleNotifyData()` for `NotifyData_data_schedUpdateResvals`; iterate resvals with nil guard on `cResvalToDistribution(vp)` — skip entries that return nil (logs warning for nil vp.val). Do NOT add `OPCode_update` handler — that is a worker→controller message; the apiserver never receives it. |
|
||
| P1-1 | `backend/convert.go` | Add `cResvalToDistribution()` — constructs `*models.ResourceDistributionInfo` from `*C.xce_resval_t`; returns `nil` (with warning log) if `vp.val == nil` instead of returning partial struct. Switch on `vp.val` (already `*C.xce_val_t`, no `&`) using `Value_u_*` constants; handle all 7 cases separately: `Value_u_ivalue32` → `NUMERIC` via `xce_val_i32` (NOT i64; ivalue32 is 4 bytes), `Value_u_ivalue64` → `NUMERIC` via `xce_val_i64`, `Value_u_uvalue32` → `NUMERIC` via `xce_val_u32` (NOT u64), `Value_u_uvalue64` → `NUMERIC` via `xce_val_u64`, `Value_u_fvalue` → `NUMERIC` via `xce_val_fval`, `Value_u_svalue` → `STRING` via `xce_val_str`, `Value_u_bvalue` → `BOOLEAN` via `xce_val_bool`. Do NOT combine i32/i64 or u32/u64 — each must use its own typed accessor. **`cResdefToModel()` already exists** (added in P0-1 for `xce_resdef_t` → `ResourceInfo`) — do NOT add a second definition. The existing `cResourceToModel(res *C.xce_resource_t) *models.ResourceDefinitionInfo` (at `convert.go:96`) serves a different purpose (worker push path for resource definitions) and is not related to P1-1. |
|
||
| P1-1 | `backend/notification.go` | No change — reuse existing `OnResourceLocationUpdated(*models.ResourceDistributionInfo)`; do NOT add `OnResourceValUpdated` |
|
||
| P1-1 | `backend/handler.go` | No change — existing `DefaultHandler.OnResourceLocationUpdated` already writes to `store.ResourceLocations` |
|
||
| P1-2 | `include/resreq.h` | Replace empty `xce_resreq_select_` with expression tree types; add enum types; add `extern` declarations for `parse_select`, `eval_select`, `free_select` |
|
||
| P1-2 | `lib/xce_select.peg` | New PEG grammar for select boolean expressions (with N-ary chain accumulation) |
|
||
| P1-2 | `lib/resreq.c` | Add `parse_select()`, `eval_select()`, `free_select()` (free_select frees node tree only — caller frees container); update `kill_primary_resreq()` to add `if (resreq->select) { free_select(resreq->select); FREEUP(resreq->select); }` before `xlist_fini` — both calls required: free_select for the tree, FREEUP for the xce_select_t struct |
|
||
| P1-2 | `lib/meson.build` | Verify glob vs explicit; add `xce_select.peg` if explicit |
|
||
| P1-2 | `backend/clib.go` | Add `EvalSelectForHost()` using `cgo.Handle` for callback context |
|
||
| P1-2 | `controllers/hosts.go` | Implement `FilterHostByResReqSelect` using `EvalSelectForHost`; update `preCheckHostRequest()` to reduce select expression limit from 4094 chars to 1024 bytes (see Input safety section) |
|
||
| P1-3 | `backend/clib.go` | Add `heartbeatIntervalSecs int32` atomic (default 30), `SetHeartbeatInterval(secs int)`, `heartbeatFn` var for test injection, `StartHeartbeat(ctx context.Context)` with context cancellation and per-tick interval re-read |
|
||
| P1-3 | `backend/handler.go` | `OnClusterUpdated` calls `SetHeartbeatInterval(cluster.HeartbeatInterval)` after storing to Clusters |
|
||
| P1-3 | `cmd/main.go` | Create root `context.WithCancel`; call `backend.StartHeartbeat(ctx)` after `RegisterToBackend` |
|
||
|
||
## Verification
|
||
|
||
1. **P0-1**: After implementing, restart API server → verify stores are populated from initial config push (check `/xce/v1/hosts`, `/xce/v1/queues`, `/xce/v1/resources/definition` endpoints return data immediately after startup, not just after change events)
|
||
|
||
2. **P0-2**: `POST /xce/v1/hosts/control` with `{"action": "close", "host_name": ["hostname"]}` should forward to controller and return success (not the current stub error). Test both `open` and `close` actions.
|
||
|
||
3. **P1-1**: After a worker updates its resource values, verify the API server receives the push and updates `ResourceLocations` store entries. Check via `GET /xce/v1/resources/distribution`.
|
||
|
||
4. **P1-2**: `POST /xce/v1/hosts/select` with `{"select": "type == LINUX64 && max_mem > 64"}` should return matching hosts from the store.
|
||
|
||
5. **P1-3**: Write a unit test for `StartHeartbeat` that replaces the C call with a counter. The test must not depend on real time — use a fake ticker or inject the tick channel:
|
||
```go
|
||
func TestStartHeartbeat_TicksAndStops(t *testing.T) {
|
||
// Replace the C heartbeat call with an atomic counter (inject via function var)
|
||
var callCount int32
|
||
origFn := heartbeatFn
|
||
heartbeatFn = func(secs int) int { atomic.AddInt32(&callCount, 1); return 0 }
|
||
defer func() { heartbeatFn = origFn }()
|
||
|
||
ctx, cancel := context.WithCancel(context.Background())
|
||
atomic.StoreInt32(&heartbeatIntervalSecs, 1) // 1s for test speed
|
||
StartHeartbeat(ctx)
|
||
|
||
time.Sleep(3500 * time.Millisecond)
|
||
cancel()
|
||
time.Sleep(100 * time.Millisecond) // let goroutine exit
|
||
|
||
n := atomic.LoadInt32(&callCount)
|
||
if n < 2 || n > 5 {
|
||
t.Errorf("expected 2-5 heartbeat calls in 3.5s, got %d", n)
|
||
}
|
||
}
|
||
```
|
||
To support injection, declare `var heartbeatFn func(secs int) int` in `clib.go` and have `StartHeartbeat` call it instead of `C.xcecli_heartbeat` directly. The test verifies: (a) the goroutine fires, (b) `cancel()` stops it with no further calls after context cancellation.
|
||
|
||
6. **Build verification**: `make test` in `src/frontend/` passes. C components build with `ninja -C build`.
|
||
|
||
7. **RBAC verification**: Confirm both new routes are registered under the JWT + RBAC middleware chain in `routers.go`. Verify that `POST /xce/v1/hosts/control` returns 403 for non-admin users and `POST /xce/v1/hosts/select` returns results for any authenticated user. Verify that `config/rbac.yaml` contains `host:control` and `host:view` action entries.
|
||
|
||
8. **Input safety verification**: Test `POST /xce/v1/hosts/select` with a select expression exceeding 1024 bytes — must return HTTP 400. Test with an invalid expression (e.g., `"` unclosed quote) — must return a parse error, not crash the server. Test `POST /xce/v1/hosts/control` with `host_name` list exceeding 256 entries — must return HTTP 400.
|
||
|
||
**Rate limiting**: both `/xce/v1/hosts/control` and `/xce/v1/hosts/select` must be registered under the existing RateLimit middleware group in `routers.go`. The select endpoint is particularly sensitive: the PEG parse cost can be exploited for CPU exhaustion if rate-limited too loosely. Confirm in the route registration that both endpoints inherit the global rate limiter.
|
||
|
||
[feedback] Follow tdd process |