forked from lei.chen/myxce
Rongsong Shen
ac7ac99ced
Start from existing entry and apply this-source's fields, rather than starting from incoming and restoring the other-source's fields. The new pattern makes it immediately clear which fields each push type WRITES, without having to reason about which fields are being "restored". OnHostUpdated keeps the merged := *incoming + if ok guard because the worker push also populates HostModel/HostType from attrs; config-sourced values must only win when a prior config push has already set them. All 37 backend unit tests pass.
98 lines
3.3 KiB
Go
98 lines
3.3 KiB
Go
package backend
|
|
|
|
import (
|
|
"apiserver/models"
|
|
"apiserver/store"
|
|
)
|
|
|
|
// DefaultHandler implements NotificationHandler by writing to stores.
|
|
type DefaultHandler struct{}
|
|
|
|
func (h *DefaultHandler) OnJobCreated(job *models.JobInfo) {
|
|
store.Jobs.Put(uint64(job.Unique_JobID), *job)
|
|
}
|
|
|
|
func (h *DefaultHandler) OnJobUpdated(job *models.JobInfo) {
|
|
store.Jobs.Put(uint64(job.Unique_JobID), *job)
|
|
}
|
|
|
|
func (h *DefaultHandler) OnJobDeleted(jobid uint64) {
|
|
store.Jobs.Delete(jobid)
|
|
}
|
|
|
|
func (h *DefaultHandler) OnQueueUpdated(queue *models.QueueInfo) {
|
|
store.Queues.Put(queue.QueueName, *queue)
|
|
}
|
|
|
|
// 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 applies config-sourced fields onto the existing host entry.
|
|
// All other fields (worker-sourced) are preserved from the prior entry.
|
|
// NumCores, NumThreads, Topology, SchedStop have no current write path in
|
|
// rawWorkerData and remain zero until a future push source populates them
|
|
// (analogous to the DisptWindow deferral).
|
|
func (h *DefaultHandler) OnHostConfigUpdated(incoming *models.HostInfo) {
|
|
existing, _ := store.Hosts.Get(incoming.HostName)
|
|
existing.HostName = incoming.HostName
|
|
existing.HostModel = incoming.HostModel
|
|
existing.HostType = incoming.HostType
|
|
existing.MaxSlot = incoming.MaxSlot
|
|
existing.DisptWindow = incoming.DisptWindow
|
|
store.Hosts.Put(existing.HostName, existing)
|
|
}
|
|
|
|
// OnClusterUpdated applies config-sourced fields onto the existing cluster entry.
|
|
// All other fields (built-in/worker-sourced: Version, Status, ControllerName,
|
|
// NumServers, Resources, HostTypes, HostModels, ClusterAdmins) are preserved
|
|
// from the prior entry.
|
|
func (h *DefaultHandler) OnClusterUpdated(incoming *models.ClusterInfo) {
|
|
existing, _ := store.Clusters.Get(incoming.ClusterName)
|
|
existing.ClusterName = incoming.ClusterName
|
|
existing.HeartbeatInterval = incoming.HeartbeatInterval
|
|
existing.Vcpus = incoming.Vcpus
|
|
existing.InactiveHeartbeatCount = incoming.InactiveHeartbeatCount
|
|
store.Clusters.Put(existing.ClusterName, existing)
|
|
}
|
|
|
|
func (h *DefaultHandler) OnResourceUpdated(res *models.ResourceDefinitionInfo) {
|
|
store.Resources.Put(res.ResourceName, *res)
|
|
}
|
|
|
|
func (h *DefaultHandler) OnResourceLocationUpdated(res *models.ResourceDistributionInfo) {
|
|
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)
|
|
}
|
|
}
|