From 80f4da42adc49f79bb2da0447227e77a0cd2d602 Mon Sep 17 00:00:00 2001 From: Ettore Di Giacinto Date: Sun, 6 Sep 2026 05:06:09 +0000 Subject: [PATCH] fix(distributed): bound the health monitor's miss streaks to live rows HealthMonitor.misses holds one consecutive-failed-probe count per (node, model, replica) and nothing ever removed an entry whose row had gone. It is the only per-node state in a frontend that grows on model churn rather than on fleet size, so a deployment that loads and unloads models for months accumulates an integer per tuple it ever probed and gives none back. There are four ways a row stops being visible to the pass, not one. A node departs and the pass skips its probes; a node goes offline or unhealthy on a stale heartbeat and the pass skips it entirely; an operator sets a node draining; or the row is removed by an unload, a scale-down or an eviction, and nothing tells this monitor. So the bound is the pass itself, and not a subscription on the departure notifier. The notifier evicts the caches a DEPARTURE invalidates and it keeps that one meaning; this reads a different fact, that there is no longer a row to count misses against, and covers all four cases with one rule. A row the pass could not probe is marked seen before the probe, so an unreachable worker still leaves its streak exactly as it was rather than having it forgiven; a pass that could not list the fleet prunes nothing, since it observed nothing. Forgetting only ever delays a reap by up to the miss threshold and can never cause one. Assisted-by: Claude Opus 5 [claude-code] Signed-off-by: Ettore Di Giacinto --- core/services/nodes/health.go | 76 ++++++++- .../services/nodes/health_miss_bounds_test.go | 152 ++++++++++++++++++ core/services/nodes/health_mock_test.go | 18 +++ 3 files changed, 242 insertions(+), 4 deletions(-) create mode 100644 core/services/nodes/health_miss_bounds_test.go diff --git a/core/services/nodes/health.go b/core/services/nodes/health.go index 3f3156293..84f38d8a1 100644 --- a/core/services/nodes/health.go +++ b/core/services/nodes/health.go @@ -51,9 +51,13 @@ type HealthMonitor struct { // See DepartureNotifier: this monitor is its only caller. departures *DepartureNotifier missesMu sync.Mutex - misses map[modelKey]int // consecutive failed-probe counts; reset on success or model removal - cancel context.CancelFunc - cancelMu sync.Mutex + // misses holds one consecutive-failed-probe count per (node, model, + // replica). It is bounded to the rows the LAST completed pass actually + // walked; see forgetUnseenMisses for why that, and not a departure + // subscription, is what bounds it. + misses map[modelKey]int + cancel context.CancelFunc + cancelMu sync.Mutex } // NewHealthMonitor creates a new HealthMonitor. @@ -224,9 +228,15 @@ func (hm *HealthMonitor) doCheckAll(ctx context.Context) { nodes, err := hm.registry.List(ctx) if err != nil { xlog.Error("Health monitor: failed to list nodes", "error", err) + // No prune. A pass that could not read the fleet observed nothing, and + // pruning against nothing would wipe every streak in progress. return } + // Every model row this pass walked, whether or not it managed to probe it. + // See forgetUnseenMisses. + seen := make(map[modelKey]struct{}) + for _, node := range nodes { if node.Status == StatusDraining { continue @@ -326,6 +336,14 @@ func (hm *HealthMonitor) doCheckAll(ctx context.Context) { if m.WorkerLocalAddress == "" { continue } + // Marked seen BEFORE the probe, so that a row this pass could + // not reach keeps the streak it already had. The unreachable + // branch below leaves that streak exactly as it was, and a + // prune that only counted PROBED rows would clear it instead, + // forgiving a backend that really has died every time a peer + // link blipped. + key := modelKey{NodeID: node.ID, ModelName: m.ModelName, ReplicaIndex: m.ReplicaIndex} + seen[key] = struct{}{} // Through the node's tunnel, never a direct dial to m.WorkerLocalAddress: // that address is a port inside the worker. A worker this // replica cannot reach is not evidence that its backend died, @@ -359,7 +377,6 @@ func (hm *HealthMonitor) doCheckAll(ctx context.Context) { continue } - key := modelKey{NodeID: node.ID, ModelName: m.ModelName, ReplicaIndex: m.ReplicaIndex} hm.missesMu.Lock() if ok { // Probe succeeded — wipe any previous miss streak. @@ -393,4 +410,55 @@ func (hm *HealthMonitor) doCheckAll(ctx context.Context) { } } } + + hm.forgetUnseenMisses(seen) +} + +// forgetUnseenMisses drops every miss streak whose model row this pass did not +// walk, so the map is bounded by the fleet's live rows rather than by every +// (node, model, replica) the process has ever seen. +// +// It is level triggered and it is deliberately NOT a subscriber on the +// departure notifier, though the notifier is where the rest of the per-node +// caches are evicted from and it would have been the shorter change. +// +// A departure subscription would fix ONE of the four ways a streak is +// stranded. The others are a node marked offline or unhealthy on a stale +// heartbeat, a node the operator set draining, and a model row removed by +// anything other than this loop (an unload, a scale-down, an eviction). Each +// leaves a counter with no row behind it, and each would need its own hook. +// What all four have in common is exactly the thing this reads: the row was +// not there when the pass looked. So the notifier keeps its one meaning, "no +// live replica holds this node's tunnel", and this keeps its own, "there is no +// longer a row to count misses against". +// +// Forgetting is safe in one direction only, and it is the safe one: a cleared +// streak DELAYS a reap by up to perModelMissThreshold passes and can never +// cause one. That is what makes it sound to prune on a pass whose +// GetNodeModels call failed, which is otherwise indistinguishable here from a +// node with no models. +func (hm *HealthMonitor) forgetUnseenMisses(seen map[modelKey]struct{}) { + hm.missesMu.Lock() + defer hm.missesMu.Unlock() + for key := range hm.misses { + if _, ok := seen[key]; !ok { + delete(hm.misses, key) + } + } +} + +// missCount reports the streak recorded for one model row. It exists for the +// specs that assert this map is bounded, which is otherwise a property with no +// observable effect until the process runs out of memory. +func (hm *HealthMonitor) missCount(nodeID, modelName string, replicaIndex int) int { + hm.missesMu.Lock() + defer hm.missesMu.Unlock() + return hm.misses[modelKey{NodeID: nodeID, ModelName: modelName, ReplicaIndex: replicaIndex}] +} + +// trackedMisses reports how many model rows currently hold a miss streak. +func (hm *HealthMonitor) trackedMisses() int { + hm.missesMu.Lock() + defer hm.missesMu.Unlock() + return len(hm.misses) } diff --git a/core/services/nodes/health_miss_bounds_test.go b/core/services/nodes/health_miss_bounds_test.go new file mode 100644 index 000000000..89eba8eb9 --- /dev/null +++ b/core/services/nodes/health_miss_bounds_test.go @@ -0,0 +1,152 @@ +package nodes + +import ( + "context" + "fmt" + "time" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" + + "github.com/mudler/LocalAI/core/services/cluster" +) + +// The miss map counts consecutive failed probes per (node, model, replica). It +// is the only per-node state in this process that grows on model churn rather +// than on fleet size, so a deployment that loads and unloads models for months +// accumulates one integer per tuple it ever probed and never gives one back. +// +// The bound is the pass itself: a streak survives only while the pass can still +// see the row it counts against. Every spec below is one way a row stops being +// visible, and the last two are the two ways a row must NOT be forgotten. +var _ = Describe("HealthMonitor miss-streak bounds", func() { + const staleThreshold = 30 * time.Second + + var ( + ctx context.Context + store *fakeNodeHealthStore + factory *fakeBackendClientFactory + hm *HealthMonitor + ) + + // deadModel is one model row whose backend answers every probe with a + // failure, so a single pass leaves a streak of exactly one. + deadModel := func(nodeID, modelName, addr string) NodeModel { + factory.setClient(addr, &fakeBackendClient{healthy: false, err: fmt.Errorf("connection refused")}) + return NodeModel{NodeID: nodeID, ModelName: modelName, WorkerLocalAddress: addr} + } + + BeforeEach(func() { + ctx = context.Background() + store = newFakeNodeHealthStore() + factory = newFakeBackendClientFactory() + hm = newTestHealthMonitor(store, factory, true, staleThreshold) + hm.perModelHealthCheck = true + }) + + It("forgets the streak of a model row that is no longer registered", func() { + // An unload, a scale-down or an LRU eviction removes the row without + // telling this monitor. Nothing else in the loop ever revisits the key, + // so the counter is stranded for the life of the process. + store.addNode(makeTestNode("node-a", "worker-a", "10.0.0.1:50051", StatusHealthy, freshTime())) + store.setNodeModels("node-a", deadModel("node-a", "m", "10.0.0.1:50053")) + + hm.doCheckAll(ctx) + Expect(hm.missCount("node-a", "m", 0)).To(Equal(1)) + + store.setNodeModels("node-a") + hm.doCheckAll(ctx) + + Expect(hm.trackedMisses()).To(BeZero()) + }) + + It("forgets the streaks of a node whose heartbeat went stale", func() { + // The pass skips the node entirely from this branch, so every streak it + // held is unreachable from here on. autoOffline deletes its rows too, + // which is what makes the counters count against nothing. + store.addNode(makeTestNode("node-b", "worker-b", "10.0.0.2:50051", StatusHealthy, freshTime())) + store.setNodeModels("node-b", deadModel("node-b", "m", "10.0.0.2:50053")) + + hm.doCheckAll(ctx) + Expect(hm.missCount("node-b", "m", 0)).To(Equal(1)) + + store.setHeartbeat("node-b", staleTime(staleThreshold)) + hm.doCheckAll(ctx) + + Expect(hm.trackedMisses()).To(BeZero()) + }) + + It("forgets the streaks of a node the operator set draining", func() { + store.addNode(makeTestNode("node-c", "worker-c", "10.0.0.3:50051", StatusHealthy, freshTime())) + store.setNodeModels("node-c", deadModel("node-c", "m", "10.0.0.3:50053")) + + hm.doCheckAll(ctx) + Expect(hm.missCount("node-c", "m", 0)).To(Equal(1)) + + store.getNode("node-c").Status = StatusDraining + hm.doCheckAll(ctx) + + Expect(hm.trackedMisses()).To(BeZero()) + }) + + It("forgets the streaks of a node whose tunnel departed", func() { + // The departure branch skips the probes deliberately: there is no route + // to dial. It also announces the departure, and every other per-node + // cache is dropped from that announcement. This one is not, because a + // departure is only one of the four ways a row stops being visible. + store.addNode(makeTestNode("node-d", "worker-d", "10.0.0.4:50051", StatusHealthy, freshTime())) + store.setNodeModels("node-d", deadModel("node-d", "m", "10.0.0.4:50053")) + + hm.doCheckAll(ctx) + Expect(hm.missCount("node-d", "m", 0)).To(Equal(1)) + + hm.presence = &stubPresence{answer: cluster.PresenceGone} + hm.reconnectGrace = time.Minute + hm.departures = NewDepartureNotifier() + hm.doCheckAll(ctx) + + Expect(hm.trackedMisses()).To(BeZero()) + }) + + It("keeps the streak of a row the pass could not probe", func() { + // An unreachable worker is not evidence about its backends. The streak + // is neither advanced nor cleared, so a row that was two misses from + // removal before the peer link blipped is still two misses from removal + // after it. A prune that counted only PROBED rows would forgive it. + store.addNode(makeTestNode("node-e", "worker-e", "10.0.0.5:50051", StatusHealthy, freshTime())) + store.setNodeModels("node-e", deadModel("node-e", "m", "10.0.0.5:50053")) + + hm.doCheckAll(ctx) + Expect(hm.missCount("node-e", "m", 0)).To(Equal(1)) + + factory.refuseForNode = fmt.Errorf("no tunnel for you") + hm.doCheckAll(ctx) + + Expect(hm.missCount("node-e", "m", 0)).To(Equal(1)) + }) + + It("keeps every streak when the pass could not read the fleet", func() { + // A pass that failed to list nodes observed nothing at all, and pruning + // against nothing would clear every streak in the deployment on one + // database hiccup. + store.addNode(makeTestNode("node-f", "worker-f", "10.0.0.6:50051", StatusHealthy, freshTime())) + store.setNodeModels("node-f", deadModel("node-f", "m", "10.0.0.6:50053")) + + hm.doCheckAll(ctx) + Expect(hm.missCount("node-f", "m", 0)).To(Equal(1)) + + hm.registry = &listFailsStore{NodeHealthStore: store, err: fmt.Errorf("connection reset")} + hm.doCheckAll(ctx) + + Expect(hm.missCount("node-f", "m", 0)).To(Equal(1)) + }) +}) + +// listFailsStore is the fake store with its fleet listing broken, which a real +// registry cannot be made to do on demand. +type listFailsStore struct { + NodeHealthStore + err error +} + +func (s *listFailsStore) List(context.Context) ([]BackendNode, error) { return nil, s.err } diff --git a/core/services/nodes/health_mock_test.go b/core/services/nodes/health_mock_test.go index 592f30d2e..b01591156 100644 --- a/core/services/nodes/health_mock_test.go +++ b/core/services/nodes/health_mock_test.go @@ -53,6 +53,24 @@ func (f *fakeNodeHealthStore) addNodeModel(nodeID string, nm NodeModel) { f.models[nodeID] = append(f.models[nodeID], nm) } +// setNodeModels replaces a node's model rows, so a spec can make a row +// disappear the way an unload, a scale-down or an eviction does. +func (f *fakeNodeHealthStore) setNodeModels(nodeID string, models ...NodeModel) { + f.mu.Lock() + defer f.mu.Unlock() + f.models[nodeID] = models +} + +// setHeartbeat backdates a node's heartbeat so a spec can drive the stale +// branch without waiting. +func (f *fakeNodeHealthStore) setHeartbeat(nodeID string, at time.Time) { + f.mu.Lock() + defer f.mu.Unlock() + if n, ok := f.nodes[nodeID]; ok { + n.LastHeartbeat = at + } +} + func (f *fakeNodeHealthStore) getNode(id string) *BackendNode { f.mu.Lock() defer f.mu.Unlock()