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)
This commit is contained in:
Rongsong Shen 2026-04-16 11:25:03 +00:00
parent 00c4472d6c
commit 893812e7aa
8 changed files with 850 additions and 11 deletions

View file

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