mirror of
https://github.com/mudler/LocalAI.git
synced 2026-11-06 04:23:15 -05:00
fix(nodes): back off failed replica scale-ups per model
When a scale-up failed, the reconciler tried again on every busy or pressure tick, which logged a warning and sent a new backend install each time. The "attempt" field in that warning was the loop index, so it always read 1 and hid the repeats. A model that no request has loaded yet logged "no load info" at warn level on every tick. Track consecutive scale-up failures per model in memory. After a failure, wait 30s, doubling on each further failure up to 10 minutes, before the next try. A successful scale-up resets the count. Log the failure count and the wait. Mark the missing load record with ErrNoLoadInfo and log it at debug level, because only a first request can fix it. Assisted-by: Claude Code:claude-sonnet-5-5 Signed-off-by: Ettore Di Giacinto <mudler@localai.io>
This commit is contained in:
3 files changed
+152
-5
No files matched your search
@@ -157,6 +157,41 @@ type ReplicaReconciler struct {
|
||||
// pinnedResolver exempts pinned models from idle scale-down (see
|
||||
// ReplicaReconcilerOptions.PinnedResolver). nil disables the exemption.
|
||||
pinnedResolver PinnedModelResolver
|
||||
// scaleUpFailures tracks CONSECUTIVE failed scale-up attempts per model so a
|
||||
// model whose replicas keep failing to load is retried with exponential
|
||||
// backoff instead of on every busy or pressure tick. In memory only: a
|
||||
// restart or a leader change starts from a clean slate, which is safe.
|
||||
scaleUpMu sync.Mutex
|
||||
scaleUpFailures map[string]scaleUpFailure
|
||||
// now is the clock used for backoff windows. Tests replace it.
|
||||
now func() time.Time
|
||||
}
|
||||
|
||||
// scaleUpFailure is the backoff state of one model.
|
||||
type scaleUpFailure struct {
|
||||
count int
|
||||
until time.Time
|
||||
}
|
||||
|
||||
const (
|
||||
// scaleUpBackoffBase is the wait after the first failed scale-up of a model.
|
||||
scaleUpBackoffBase = 30 * time.Second
|
||||
// scaleUpBackoffMax caps the wait, so a repaired node is picked up within
|
||||
// minutes without operator action.
|
||||
scaleUpBackoffMax = 10 * time.Minute
|
||||
)
|
||||
|
||||
// scaleUpBackoffDelay returns the wait after the given number of consecutive
|
||||
// failures: the base delay, doubled for each further failure, up to the cap.
|
||||
func scaleUpBackoffDelay(failures int) time.Duration {
|
||||
if failures < 1 {
|
||||
return 0
|
||||
}
|
||||
// 30s << 5 already exceeds the cap; stop shifting before it can overflow.
|
||||
if failures > 6 {
|
||||
return scaleUpBackoffMax
|
||||
}
|
||||
return min(scaleUpBackoffBase<<(failures-1), scaleUpBackoffMax)
|
||||
}
|
||||
|
||||
// ModelScheduler abstracts the scheduling logic needed by the reconciler.
|
||||
@@ -244,6 +279,8 @@ func NewReplicaReconciler(opts ReplicaReconcilerOptions) *ReplicaReconciler {
|
||||
pressure: opts.Pressure,
|
||||
pressureThreshold: pressureThreshold,
|
||||
pinnedResolver: opts.PinnedResolver,
|
||||
scaleUpFailures: map[string]scaleUpFailure{},
|
||||
now: time.Now,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1100,16 +1137,44 @@ func (rc *ReplicaReconciler) scaleUp(ctx context.Context, cfg ModelSchedulingCon
|
||||
return false
|
||||
}
|
||||
|
||||
model := cfg.Target()
|
||||
now := rc.now()
|
||||
rc.scaleUpMu.Lock()
|
||||
state := rc.scaleUpFailures[model]
|
||||
rc.scaleUpMu.Unlock()
|
||||
if now.Before(state.until) {
|
||||
xlog.Debug("Reconciler: scale-up backing off after a failure", "model", model,
|
||||
"failures", state.count, "retry_after", state.until.Format(time.RFC3339))
|
||||
return false
|
||||
}
|
||||
|
||||
scheduled := 0
|
||||
for i := 0; i < count; i++ {
|
||||
node, err := rc.scheduler.ScheduleAndLoadModel(ctx, cfg.Target(), candidateNodeIDs)
|
||||
node, err := rc.scheduler.ScheduleAndLoadModel(ctx, model, candidateNodeIDs)
|
||||
if err != nil {
|
||||
xlog.Warn("Reconciler: failed to scale up replica", "model", cfg.Target(),
|
||||
"attempt", i+1, "error", err)
|
||||
state.count++
|
||||
delay := scaleUpBackoffDelay(state.count)
|
||||
state.until = now.Add(delay)
|
||||
rc.scaleUpMu.Lock()
|
||||
rc.scaleUpFailures[model] = state
|
||||
rc.scaleUpMu.Unlock()
|
||||
// A model nobody has served yet is not a fault of any node: it
|
||||
// repeats until the first request, so it stays at debug level.
|
||||
level := xlog.Warn
|
||||
if errors.Is(err, ErrNoLoadInfo) {
|
||||
level = xlog.Debug
|
||||
}
|
||||
level("Reconciler: failed to scale up replica", "model", model,
|
||||
"consecutive_failures", state.count, "retry_in", delay, "error", err)
|
||||
break // stop trying on first failure
|
||||
}
|
||||
scheduled++
|
||||
xlog.Info("Reconciler: scaled up replica", "model", cfg.Target(), "node", node.Name)
|
||||
xlog.Info("Reconciler: scaled up replica", "model", model, "node", node.Name)
|
||||
}
|
||||
if scheduled > 0 {
|
||||
rc.scaleUpMu.Lock()
|
||||
delete(rc.scaleUpFailures, model)
|
||||
rc.scaleUpMu.Unlock()
|
||||
}
|
||||
return scheduled > 0
|
||||
}
|
||||
|
||||
@@ -0,0 +1,78 @@
|
||||
package nodes
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
. "github.com/onsi/ginkgo/v2"
|
||||
. "github.com/onsi/gomega"
|
||||
)
|
||||
|
||||
var _ = Describe("ReplicaReconciler scale-up backoff", func() {
|
||||
var (
|
||||
now time.Time
|
||||
scheduler *fakeScheduler
|
||||
rc *ReplicaReconciler
|
||||
cfg ModelSchedulingConfig
|
||||
)
|
||||
|
||||
BeforeEach(func() {
|
||||
now = time.Unix(1_700_000_000, 0)
|
||||
scheduler = &fakeScheduler{scheduleErr: errors.New("load boom")}
|
||||
rc = NewReplicaReconciler(ReplicaReconcilerOptions{Scheduler: scheduler})
|
||||
rc.now = func() time.Time { return now }
|
||||
cfg = ModelSchedulingConfig{ModelName: "m"}
|
||||
})
|
||||
|
||||
It("skips the scheduler while a failed model is backing off", func() {
|
||||
Expect(rc.scaleUp(context.Background(), cfg, 1)).To(BeFalse())
|
||||
Expect(scheduler.scheduleCalls).To(HaveLen(1))
|
||||
|
||||
now = now.Add(scaleUpBackoffBase - time.Second)
|
||||
Expect(rc.scaleUp(context.Background(), cfg, 1)).To(BeFalse())
|
||||
Expect(scheduler.scheduleCalls).To(HaveLen(1), "no schedule call inside the backoff window")
|
||||
|
||||
now = now.Add(2 * time.Second)
|
||||
Expect(rc.scaleUp(context.Background(), cfg, 1)).To(BeFalse())
|
||||
Expect(scheduler.scheduleCalls).To(HaveLen(2), "retry once the window has passed")
|
||||
})
|
||||
|
||||
It("doubles the delay on each failure up to the cap", func() {
|
||||
Expect(scaleUpBackoffDelay(1)).To(Equal(scaleUpBackoffBase))
|
||||
Expect(scaleUpBackoffDelay(2)).To(Equal(2 * scaleUpBackoffBase))
|
||||
Expect(scaleUpBackoffDelay(3)).To(Equal(4 * scaleUpBackoffBase))
|
||||
Expect(scaleUpBackoffDelay(50)).To(Equal(scaleUpBackoffMax))
|
||||
Expect(scaleUpBackoffDelay(1 << 30)).To(Equal(scaleUpBackoffMax))
|
||||
})
|
||||
|
||||
It("forgets the failures after a successful scale-up", func() {
|
||||
Expect(rc.scaleUp(context.Background(), cfg, 1)).To(BeFalse())
|
||||
now = now.Add(scaleUpBackoffMax)
|
||||
scheduler.scheduleErr = nil
|
||||
scheduler.scheduleNode = &BackendNode{Name: "n"}
|
||||
Expect(rc.scaleUp(context.Background(), cfg, 1)).To(BeTrue())
|
||||
|
||||
scheduler.scheduleErr = errors.New("load boom")
|
||||
scheduler.scheduleNode = nil
|
||||
Expect(rc.scaleUp(context.Background(), cfg, 1)).To(BeFalse())
|
||||
// The failure counter restarted, so the window is the base delay.
|
||||
now = now.Add(scaleUpBackoffBase + time.Second)
|
||||
before := len(scheduler.scheduleCalls)
|
||||
rc.scaleUp(context.Background(), cfg, 1)
|
||||
Expect(scheduler.scheduleCalls).To(HaveLen(before + 1))
|
||||
})
|
||||
|
||||
It("backs off each model on its own", func() {
|
||||
Expect(rc.scaleUp(context.Background(), cfg, 1)).To(BeFalse())
|
||||
other := ModelSchedulingConfig{ModelName: "other"}
|
||||
Expect(rc.scaleUp(context.Background(), other, 1)).To(BeFalse())
|
||||
Expect(scheduler.scheduleCalls).To(HaveLen(2))
|
||||
})
|
||||
|
||||
It("recognizes a missing load record through the sentinel", func() {
|
||||
err := fmt.Errorf("no load info for model m: %w (cause: %w)", ErrNoLoadInfo, errors.New("record not found"))
|
||||
Expect(errors.Is(err, ErrNoLoadInfo)).To(BeTrue())
|
||||
})
|
||||
})
|
||||
@@ -607,7 +607,7 @@ func (r *SmartRouter) ScheduleAndLoadModel(ctx context.Context, modelName string
|
||||
// the worker rejected ("backend name is empty"). Skip cleanly: the
|
||||
// model needs to be served at least once via Route() so its load
|
||||
// info is stored — then the reconciler can replicate it.
|
||||
return nil, fmt.Errorf("no load info for model %s: serve at least one request for it before the reconciler can replicate (cause: %w)", modelName, err)
|
||||
return nil, fmt.Errorf("no load info for model %s: serve at least one request for it before the reconciler can replicate: %w (cause: %w)", modelName, ErrNoLoadInfo, err)
|
||||
}
|
||||
|
||||
// Deserialize the stored model options
|
||||
@@ -2142,6 +2142,10 @@ func (r *SmartRouter) EvictLRU(ctx context.Context, nodeID string) (string, erro
|
||||
// and none can be evicted to make room.
|
||||
var ErrEvictionBusy = errors.New("all models busy, cannot evict")
|
||||
|
||||
// ErrNoLoadInfo is returned by ScheduleAndLoadModel when no replica of the model
|
||||
// has ever been loaded, so there are no stored load options to replicate.
|
||||
var ErrNoLoadInfo = errors.New("no load info")
|
||||
|
||||
// ErrNoAvailableNodes is returned when the scheduler cannot find any healthy
|
||||
// node to serve a model — all nodes are full and eviction cannot free a slot,
|
||||
// or a node selector excludes every candidate. The HTTP layer maps this to
|
||||
|
||||
Reference in new issue
Block a user