fix(apiserver): fall back to default HostConfig for unmatched workers (bug-1673, gitea #40)
OnHostConfigUpdated only applied config-sourced fields (MaxSlot, HostModel, HostType, DisptWindow) onto a host whose name exact-matched an entry already in the store. hosts.yaml commonly defines only a wildcard "default" entry, so a dynamically-joined worker never received its own config push and MaxSlot stayed at Go's zero-value forever -- decoupled from the C-side scheduling path, which already resolves the real value via worker.c's wildcard fallback. The "default" pseudo-host's config is now cached and retroactively backfilled onto any registered host with no explicit config of its own, in either arrival order (config-before-worker or worker-before-config). The -1 HOSTCONFIG_MXJ_AUTO_DETECT sentinel (bug-1556) is resolved to the host's real NumCPUs when applied via fallback, mirroring worker.c's own resolution, so the display shows usable capacity rather than a bare -1. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01AgYRU3zDTqNGV5Jk8aiKLg
This commit is contained in:
parent
88c7c05daf
commit
fe87dafc85
3 changed files with 239 additions and 1 deletions
|
|
@ -4,6 +4,7 @@ import (
|
||||||
"apiserver/models"
|
"apiserver/models"
|
||||||
"apiserver/store"
|
"apiserver/store"
|
||||||
logger "apiserver/utils"
|
logger "apiserver/utils"
|
||||||
|
"sync"
|
||||||
|
|
||||||
"go.uber.org/zap"
|
"go.uber.org/zap"
|
||||||
)
|
)
|
||||||
|
|
@ -11,6 +12,66 @@ import (
|
||||||
// DefaultHandler implements NotificationHandler by writing to stores.
|
// DefaultHandler implements NotificationHandler by writing to stores.
|
||||||
type DefaultHandler struct{}
|
type DefaultHandler struct{}
|
||||||
|
|
||||||
|
// defaultHostConfigName is the cluster-wide config sentinel in hosts.yaml
|
||||||
|
// (never a real worker) whose fields worker.c's apply_join_state()/
|
||||||
|
// find_matched_hostconfig() already falls back to on the C side for any
|
||||||
|
// worker with no host-specific hosts.yaml entry of its own.
|
||||||
|
const defaultHostConfigName = "default"
|
||||||
|
|
||||||
|
// bug-1673: a dynamically-joined worker with no hosts.yaml entry under its
|
||||||
|
// own literal hostname never receives its own OnHostConfigUpdated call, so
|
||||||
|
// MaxSlot/HostModel/HostType/DisptWindow stayed at Go's zero-value forever
|
||||||
|
// -- decoupled from the C-side scheduling path, which already resolves the
|
||||||
|
// real value via the "default" wildcard entry. These package-level vars
|
||||||
|
// cache that entry's fields plus which hosts have their OWN explicit
|
||||||
|
// (non-default) config, so OnHostUpdated can apply the same fallback for
|
||||||
|
// display purposes that worker.c already applies for real dispatch.
|
||||||
|
var (
|
||||||
|
hostConfigMu sync.Mutex
|
||||||
|
defaultHostConfig *models.HostInfo
|
||||||
|
hostsWithExplicitConfig = map[string]bool{}
|
||||||
|
)
|
||||||
|
|
||||||
|
// resetHostConfigCache clears the cached "default" pseudo-host config and
|
||||||
|
// explicit-config tracking. Test-only: called from this package's _test.go
|
||||||
|
// store-reset helpers, so bug-1673 fallback state doesn't leak between
|
||||||
|
// tests the way store.InitStores() already prevents for the stores
|
||||||
|
// themselves.
|
||||||
|
func resetHostConfigCache() {
|
||||||
|
hostConfigMu.Lock()
|
||||||
|
defaultHostConfig = nil
|
||||||
|
hostsWithExplicitConfig = map[string]bool{}
|
||||||
|
hostConfigMu.Unlock()
|
||||||
|
}
|
||||||
|
|
||||||
|
// applyDefaultHostConfigFallback copies the cached "default" pseudo-host's
|
||||||
|
// config-sourced fields onto host, if one has been observed yet. No-op
|
||||||
|
// otherwise (nothing to fall back to).
|
||||||
|
//
|
||||||
|
// A negative MaxSlot is HOSTCONFIG_MXJ_AUTO_DETECT (bug-1556): "no explicit
|
||||||
|
// cap, use this host's real CPU count" -- the same resolution worker.c's
|
||||||
|
// apply_join_state()/find_matched_hostconfig() already performs for actual
|
||||||
|
// dispatch ("if (nslots<0) nslots = maxcpus"). Resolving it here too means
|
||||||
|
// this display fallback shows the host's real usable capacity (matching
|
||||||
|
// what dispatch will actually honor) instead of a bare "-1" that would
|
||||||
|
// leave the reporter's original "is ncpus equal slots?" question (gitea
|
||||||
|
// #40) just as unanswered as the "0" it replaces.
|
||||||
|
func applyDefaultHostConfigFallback(host *models.HostInfo) {
|
||||||
|
hostConfigMu.Lock()
|
||||||
|
def := defaultHostConfig
|
||||||
|
hostConfigMu.Unlock()
|
||||||
|
if def == nil {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
host.HostModel = def.HostModel
|
||||||
|
host.HostType = def.HostType
|
||||||
|
host.MaxSlot = def.MaxSlot
|
||||||
|
if host.MaxSlot < 0 && host.NumCPUs > 0 {
|
||||||
|
host.MaxSlot = host.NumCPUs
|
||||||
|
}
|
||||||
|
host.DisptWindow = def.DisptWindow
|
||||||
|
}
|
||||||
|
|
||||||
func (h *DefaultHandler) OnJobCreated(job *models.JobInfo) {
|
func (h *DefaultHandler) OnJobCreated(job *models.JobInfo) {
|
||||||
store.Jobs.Put(uint64(job.Unique_JobID), *job)
|
store.Jobs.Put(uint64(job.Unique_JobID), *job)
|
||||||
}
|
}
|
||||||
|
|
@ -42,7 +103,10 @@ func (h *DefaultHandler) OnQueueUpdated(queue *models.QueueInfo) {
|
||||||
|
|
||||||
// OnHostUpdated merges worker-sourced fields into an existing host entry.
|
// OnHostUpdated merges worker-sourced fields into an existing host entry.
|
||||||
// Config-sourced fields (HostModel, HostType, MaxSlot, DisptWindow) are
|
// Config-sourced fields (HostModel, HostType, MaxSlot, DisptWindow) are
|
||||||
// preserved from the prior entry if one exists.
|
// preserved from the prior entry if one exists. If this is the host's
|
||||||
|
// first-ever registration (no prior entry), fall back to the cached
|
||||||
|
// "default" pseudo-host's config (bug-1673) rather than leaving those
|
||||||
|
// fields at Go's zero-value -- mirrors the C-side wildcard fallback.
|
||||||
func (h *DefaultHandler) OnHostUpdated(incoming *models.HostInfo) {
|
func (h *DefaultHandler) OnHostUpdated(incoming *models.HostInfo) {
|
||||||
merged := *incoming
|
merged := *incoming
|
||||||
if existing, ok := store.Hosts.Get(merged.HostName); ok {
|
if existing, ok := store.Hosts.Get(merged.HostName); ok {
|
||||||
|
|
@ -50,6 +114,8 @@ func (h *DefaultHandler) OnHostUpdated(incoming *models.HostInfo) {
|
||||||
merged.HostType = existing.HostType
|
merged.HostType = existing.HostType
|
||||||
merged.MaxSlot = existing.MaxSlot
|
merged.MaxSlot = existing.MaxSlot
|
||||||
merged.DisptWindow = existing.DisptWindow
|
merged.DisptWindow = existing.DisptWindow
|
||||||
|
} else {
|
||||||
|
applyDefaultHostConfigFallback(&merged)
|
||||||
}
|
}
|
||||||
store.Hosts.Put(merged.HostName, merged)
|
store.Hosts.Put(merged.HostName, merged)
|
||||||
}
|
}
|
||||||
|
|
@ -57,7 +123,35 @@ func (h *DefaultHandler) OnHostUpdated(incoming *models.HostInfo) {
|
||||||
// OnHostConfigUpdated applies config-sourced fields onto an existing worker-registered host.
|
// OnHostConfigUpdated applies config-sourced fields onto an existing worker-registered host.
|
||||||
// Hosts not yet registered by a worker are skipped — config entries like "default" (which sets
|
// Hosts not yet registered by a worker are skipped — config entries like "default" (which sets
|
||||||
// cluster-wide defaults) or hosts whose daemons haven't started must not appear in the host list.
|
// cluster-wide defaults) or hosts whose daemons haven't started must not appear in the host list.
|
||||||
|
//
|
||||||
|
// The "default" pseudo-host entry itself is cached (bug-1673) rather than
|
||||||
|
// simply discarded, then retroactively backfilled onto any already-
|
||||||
|
// registered host with no explicit config of its own -- "default" can
|
||||||
|
// arrive either before or after a given worker's own OnHostUpdated call.
|
||||||
func (h *DefaultHandler) OnHostConfigUpdated(incoming *models.HostInfo) {
|
func (h *DefaultHandler) OnHostConfigUpdated(incoming *models.HostInfo) {
|
||||||
|
if incoming.HostName == defaultHostConfigName {
|
||||||
|
hostConfigMu.Lock()
|
||||||
|
cp := *incoming
|
||||||
|
defaultHostConfig = &cp
|
||||||
|
hostConfigMu.Unlock()
|
||||||
|
|
||||||
|
store.Hosts.Range(func(name string, host models.HostInfo) bool {
|
||||||
|
hostConfigMu.Lock()
|
||||||
|
explicit := hostsWithExplicitConfig[name]
|
||||||
|
hostConfigMu.Unlock()
|
||||||
|
if !explicit {
|
||||||
|
applyDefaultHostConfigFallback(&host)
|
||||||
|
store.Hosts.Put(name, host)
|
||||||
|
}
|
||||||
|
return true
|
||||||
|
})
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
hostConfigMu.Lock()
|
||||||
|
hostsWithExplicitConfig[incoming.HostName] = true
|
||||||
|
hostConfigMu.Unlock()
|
||||||
|
|
||||||
existing, ok := store.Hosts.Get(incoming.HostName)
|
existing, ok := store.Hosts.Get(incoming.HostName)
|
||||||
if !ok {
|
if !ok {
|
||||||
return
|
return
|
||||||
|
|
|
||||||
|
|
@ -14,6 +14,7 @@ import (
|
||||||
// resetStores reinitializes all global stores for test isolation.
|
// resetStores reinitializes all global stores for test isolation.
|
||||||
func resetStores() {
|
func resetStores() {
|
||||||
store.InitStores()
|
store.InitStores()
|
||||||
|
resetHostConfigCache()
|
||||||
}
|
}
|
||||||
|
|
||||||
// captureLog swaps utils.Log for an observer core so a test can assert on
|
// captureLog swaps utils.Log for an observer core so a test can assert on
|
||||||
|
|
@ -117,6 +118,148 @@ func TestOnHostConfigUpdated_PartialMerge(t *testing.T) {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// TestOnHostConfigUpdated_DefaultFallback_ConfigBeforeWorker is a regression
|
||||||
|
// guard for bug-1673 (gitea #40): a dynamically-joined worker (e.g.
|
||||||
|
// "worker-1") with no hosts.yaml entry under its own literal hostname never
|
||||||
|
// receives its own OnHostConfigUpdated call, so MaxSlot/HostModel/HostType
|
||||||
|
// stayed at Go's zero-value forever even though the C-side scheduling path
|
||||||
|
// already resolves the real value via the "default" wildcard entry. This
|
||||||
|
// covers the "default" config arriving BEFORE the worker registers.
|
||||||
|
func TestOnHostConfigUpdated_DefaultFallback_ConfigBeforeWorker(t *testing.T) {
|
||||||
|
resetStores()
|
||||||
|
h := DefaultHandler{}
|
||||||
|
|
||||||
|
// "default" pseudo-host config push arrives first (cluster startup).
|
||||||
|
h.OnHostConfigUpdated(&models.HostInfo{
|
||||||
|
HostName: defaultHostConfigName,
|
||||||
|
HostModel: "generic",
|
||||||
|
HostType: "x86_64-linux-gnu",
|
||||||
|
MaxSlot: -1, // HOSTCONFIG_MXJ_AUTO_DETECT sentinel (bug-1556)
|
||||||
|
})
|
||||||
|
|
||||||
|
// Dynamically-joined worker registers under its own name, never listed
|
||||||
|
// in hosts.yaml -- so no matching OnHostConfigUpdated call ever fires
|
||||||
|
// for "worker-1" itself.
|
||||||
|
h.OnHostUpdated(&models.HostInfo{
|
||||||
|
HostName: "worker-1",
|
||||||
|
NumCPUs: 22,
|
||||||
|
})
|
||||||
|
|
||||||
|
got, ok := store.Hosts.Get("worker-1")
|
||||||
|
if !ok {
|
||||||
|
t.Fatal("worker-1 should exist after OnHostUpdated")
|
||||||
|
}
|
||||||
|
if got.MaxSlot != 22 {
|
||||||
|
t.Errorf("MaxSlot: got %d, want 22 (fallen back to \"default\", -1 auto-detect resolved to NumCPUs, not stuck at 0)", got.MaxSlot)
|
||||||
|
}
|
||||||
|
if got.HostModel != "generic" {
|
||||||
|
t.Errorf("HostModel: got %q, want %q (fallen back to \"default\")", got.HostModel, "generic")
|
||||||
|
}
|
||||||
|
if got.HostType != "x86_64-linux-gnu" {
|
||||||
|
t.Errorf("HostType: got %q, want %q (fallen back to \"default\")", got.HostType, "x86_64-linux-gnu")
|
||||||
|
}
|
||||||
|
// Worker-sourced field preserved throughout.
|
||||||
|
if got.NumCPUs != 22 {
|
||||||
|
t.Errorf("NumCPUs: got %d, want 22 (should be preserved)", got.NumCPUs)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestOnHostConfigUpdated_DefaultFallback_WorkerBeforeConfig covers the
|
||||||
|
// opposite ordering: the worker registers first, and "default" arrives
|
||||||
|
// later. The fallback must be retroactively backfilled onto the
|
||||||
|
// already-registered host.
|
||||||
|
func TestOnHostConfigUpdated_DefaultFallback_WorkerBeforeConfig(t *testing.T) {
|
||||||
|
resetStores()
|
||||||
|
h := DefaultHandler{}
|
||||||
|
|
||||||
|
h.OnHostUpdated(&models.HostInfo{
|
||||||
|
HostName: "worker-1",
|
||||||
|
NumCPUs: 16,
|
||||||
|
})
|
||||||
|
|
||||||
|
got, ok := store.Hosts.Get("worker-1")
|
||||||
|
if !ok || got.MaxSlot != 0 {
|
||||||
|
t.Fatalf("precondition: worker-1 should exist with MaxSlot still 0 before \"default\" arrives, got ok=%v MaxSlot=%d", ok, got.MaxSlot)
|
||||||
|
}
|
||||||
|
|
||||||
|
h.OnHostConfigUpdated(&models.HostInfo{
|
||||||
|
HostName: defaultHostConfigName,
|
||||||
|
HostModel: "generic",
|
||||||
|
HostType: "x86_64-linux-gnu",
|
||||||
|
MaxSlot: -1,
|
||||||
|
})
|
||||||
|
|
||||||
|
got, ok = store.Hosts.Get("worker-1")
|
||||||
|
if !ok {
|
||||||
|
t.Fatal("worker-1 should still exist after \"default\" backfill")
|
||||||
|
}
|
||||||
|
if got.MaxSlot != 16 {
|
||||||
|
t.Errorf("MaxSlot: got %d, want 16 (retroactively backfilled from \"default\", -1 auto-detect resolved to NumCPUs)", got.MaxSlot)
|
||||||
|
}
|
||||||
|
if got.NumCPUs != 16 {
|
||||||
|
t.Errorf("NumCPUs: got %d, want 16 (should be preserved)", got.NumCPUs)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestOnHostConfigUpdated_DefaultFallback_UnresolvableStaysNegative covers
|
||||||
|
// the edge case where NumCPUs isn't known yet (0) -- there's nothing to
|
||||||
|
// resolve the -1 auto-detect sentinel to, so it's left as-is rather than
|
||||||
|
// silently becoming a misleading MaxSlot of 0.
|
||||||
|
func TestOnHostConfigUpdated_DefaultFallback_UnresolvableStaysNegative(t *testing.T) {
|
||||||
|
resetStores()
|
||||||
|
h := DefaultHandler{}
|
||||||
|
|
||||||
|
h.OnHostConfigUpdated(&models.HostInfo{
|
||||||
|
HostName: defaultHostConfigName,
|
||||||
|
MaxSlot: -1,
|
||||||
|
})
|
||||||
|
h.OnHostUpdated(&models.HostInfo{HostName: "worker-1"}) // NumCPUs unset (0)
|
||||||
|
|
||||||
|
got, ok := store.Hosts.Get("worker-1")
|
||||||
|
if !ok {
|
||||||
|
t.Fatal("worker-1 should exist after OnHostUpdated")
|
||||||
|
}
|
||||||
|
if got.MaxSlot != -1 {
|
||||||
|
t.Errorf("MaxSlot: got %d, want -1 (unresolvable with NumCPUs=0, must not become a misleading 0)", got.MaxSlot)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestOnHostConfigUpdated_DefaultFallback_DoesNotOverrideExplicitConfig
|
||||||
|
// proves the "default" fallback never clobbers a host's own real,
|
||||||
|
// host-specific hosts.yaml entry -- whichever order the two config pushes
|
||||||
|
// and the worker registration arrive in.
|
||||||
|
func TestOnHostConfigUpdated_DefaultFallback_DoesNotOverrideExplicitConfig(t *testing.T) {
|
||||||
|
resetStores()
|
||||||
|
h := DefaultHandler{}
|
||||||
|
|
||||||
|
h.OnHostUpdated(&models.HostInfo{HostName: "node01", NumCPUs: 8})
|
||||||
|
h.OnHostConfigUpdated(&models.HostInfo{
|
||||||
|
HostName: "node01",
|
||||||
|
HostModel: "modelA",
|
||||||
|
HostType: "CENTOS7",
|
||||||
|
MaxSlot: 16,
|
||||||
|
})
|
||||||
|
|
||||||
|
// "default" arrives after -- must not overwrite node01's real config.
|
||||||
|
h.OnHostConfigUpdated(&models.HostInfo{
|
||||||
|
HostName: defaultHostConfigName,
|
||||||
|
HostModel: "generic",
|
||||||
|
HostType: "x86_64-linux-gnu",
|
||||||
|
MaxSlot: -1,
|
||||||
|
})
|
||||||
|
|
||||||
|
got, ok := store.Hosts.Get("node01")
|
||||||
|
if !ok {
|
||||||
|
t.Fatal("node01 should exist")
|
||||||
|
}
|
||||||
|
if got.MaxSlot != 16 {
|
||||||
|
t.Errorf("MaxSlot: got %d, want 16 (explicit config must not be overridden by \"default\")", got.MaxSlot)
|
||||||
|
}
|
||||||
|
if got.HostModel != "modelA" {
|
||||||
|
t.Errorf("HostModel: got %q, want %q (explicit config must not be overridden by \"default\")", got.HostModel, "modelA")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
func TestOnClusterUpdated_NewCluster(t *testing.T) {
|
func TestOnClusterUpdated_NewCluster(t *testing.T) {
|
||||||
resetStores()
|
resetStores()
|
||||||
h := DefaultHandler{}
|
h := DefaultHandler{}
|
||||||
|
|
|
||||||
|
|
@ -15,6 +15,7 @@ import (
|
||||||
// resetIntegrationStores reinitializes all global stores for test isolation.
|
// resetIntegrationStores reinitializes all global stores for test isolation.
|
||||||
func resetIntegrationStores() {
|
func resetIntegrationStores() {
|
||||||
store.InitStores()
|
store.InitStores()
|
||||||
|
resetHostConfigCache()
|
||||||
utils.Log = zap.NewNop()
|
utils.Log = zap.NewNop()
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue