mirror of
https://github.com/mudler/LocalAI.git
synced 2026-09-12 22:33:54 -04:00
fix(distributed): stop flapping agent nodes on backend listing
Only backend workers subscribe to backend.list. ListBackends asked every node that was not pending, offline or draining, so an agent worker could only answer "no responders", which the error handling reads as a node that has gone away. Every poll of the backends view therefore marked each agent node unhealthy, and its next heartbeat marked it healthy again. While unhealthy the node is not schedulable, so this also cost agent capacity for as long as each flap lasted. Skip non-backend workers, as the backend-op fan-out already does for the same reason. A backend worker that does not answer is still marked unhealthy: that one really is gone. Signed-off-by: Ettore Di Giacinto <mudler@localai.io> Assisted-by: Claude Code:claude-opus-5 [golangci-lint]
This commit is contained in:
1 parent
df1a40f9c0
commit
bebd812e7d
2 files changed
+95
-2
No files matched your search
@@ -0,0 +1,84 @@
|
||||
package nodes
|
||||
|
||||
import (
|
||||
"context"
|
||||
"runtime"
|
||||
"time"
|
||||
|
||||
. "github.com/onsi/ginkgo/v2"
|
||||
. "github.com/onsi/gomega"
|
||||
"gorm.io/gorm"
|
||||
|
||||
"github.com/mudler/LocalAI/core/services/messaging"
|
||||
"github.com/mudler/LocalAI/core/services/testutil"
|
||||
)
|
||||
|
||||
// Agent workers do not subscribe to the backend.* subjects, so asking one to
|
||||
// list its backends can only answer "no responders". ListBackends read that as
|
||||
// a node that had gone away and marked it unhealthy; the node's next heartbeat
|
||||
// marked it healthy again. Every poll of the backends view therefore flapped
|
||||
// every agent node in the cluster, and while it was unhealthy the router would
|
||||
// not schedule onto it.
|
||||
var _ = Describe("Backend listing across mixed node types", func() {
|
||||
var (
|
||||
db *gorm.DB
|
||||
registry *NodeRegistry
|
||||
mc *scriptedMessagingClient
|
||||
mgr *DistributedBackendManager
|
||||
ctx context.Context
|
||||
)
|
||||
|
||||
BeforeEach(func() {
|
||||
if runtime.GOOS == "darwin" {
|
||||
Skip("testcontainers requires Docker, not available on macOS CI")
|
||||
}
|
||||
db = testutil.SetupTestDB()
|
||||
var err error
|
||||
registry, err = NewNodeRegistry(db)
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
mc = newScriptedMessagingClient()
|
||||
mgr = &DistributedBackendManager{
|
||||
local: stubLocalBackendManager{},
|
||||
adapter: NewRemoteUnloaderAdapter(nil, mc, 3*time.Minute, 15*time.Minute),
|
||||
registry: registry,
|
||||
}
|
||||
ctx = context.Background()
|
||||
})
|
||||
|
||||
register := func(name, nodeType string) *BackendNode {
|
||||
node := &BackendNode{Name: name, NodeType: nodeType, Address: name + ":50051"}
|
||||
Expect(registry.Register(ctx, node, true)).To(Succeed())
|
||||
fetched, err := registry.GetByName(ctx, name)
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
Expect(fetched.Status).To(Equal(StatusHealthy))
|
||||
return fetched
|
||||
}
|
||||
|
||||
statusOf := func(id string) string {
|
||||
n, err := registry.Get(ctx, id)
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
return n.Status
|
||||
}
|
||||
|
||||
It("leaves an agent node healthy instead of flapping it", func() {
|
||||
agent := register("agent-worker-1", NodeTypeAgent)
|
||||
mc.scriptNoResponders(messaging.SubjectNodeBackendList(agent.ID))
|
||||
|
||||
_, err := mgr.ListBackends()
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
|
||||
Expect(statusOf(agent.ID)).To(Equal(StatusHealthy),
|
||||
"an agent node cannot answer backend.list and must not be judged on it")
|
||||
})
|
||||
|
||||
It("still marks a backend node unhealthy when it does not answer", func() {
|
||||
backendNode := register("worker-a", NodeTypeBackend)
|
||||
mc.scriptNoResponders(messaging.SubjectNodeBackendList(backendNode.ID))
|
||||
|
||||
_, err := mgr.ListBackends()
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
|
||||
Expect(statusOf(backendNode.ID)).To(Equal(StatusUnhealthy),
|
||||
"a backend worker that does not answer is genuinely gone")
|
||||
})
|
||||
})
|
||||
@@ -331,8 +331,9 @@ func (d *DistributedBackendManager) DeleteBackendDetailed(ctx context.Context, n
|
||||
// populated from the first node seen so single-node-minded callers still work.
|
||||
//
|
||||
// Pending/offline/draining nodes are skipped because they aren't expected to
|
||||
// answer NATS requests; unhealthy nodes are still queried — ErrNoResponders
|
||||
// then marks them unhealthy and the loop continues.
|
||||
// answer NATS requests, and so are non-backend workers, which do not subscribe
|
||||
// to backend.list at all; unhealthy backend nodes are still queried —
|
||||
// ErrNoResponders then marks them unhealthy and the loop continues.
|
||||
func (d *DistributedBackendManager) ListBackends() (gallery.SystemBackends, error) {
|
||||
result := make(gallery.SystemBackends)
|
||||
allNodes, err := d.registry.List(context.Background())
|
||||
@@ -344,6 +345,14 @@ func (d *DistributedBackendManager) ListBackends() (gallery.SystemBackends, erro
|
||||
if node.Status == StatusPending || node.Status == StatusOffline || node.Status == StatusDraining {
|
||||
continue
|
||||
}
|
||||
// Only backend workers subscribe to backend.list. Asking an agent
|
||||
// worker can only answer "no responders", which the error handling
|
||||
// below reads as a node that has gone away, so every poll of this view
|
||||
// marked every agent node unhealthy and its next heartbeat marked it
|
||||
// healthy again. The backend-op fan-out skips them for the same reason.
|
||||
if node.NodeType != "" && node.NodeType != NodeTypeBackend {
|
||||
continue
|
||||
}
|
||||
reply, err := d.adapter.ListBackends(node.ID)
|
||||
if err != nil {
|
||||
if errors.Is(err, nats.ErrNoResponders) {
|
||||
|
||||
Reference in new issue
Block a user