mirror of
https://github.com/mudler/LocalAI.git
synced 2026-09-29 17:44:30 -04:00
fix(failover): count a chain switch once per cluster
Every frontend recorded localai_failover_switches_total for the switches it adopted from the leader, so a cluster of N frontends counted each switch N times. Only the leader, which decides the switch, records it. Assisted-by: Claude:claude-opus-5-5 Signed-off-by: Ettore Di Giacinto <mudler@localai.io>
This commit is contained in:
1 parent
4f801dfc99
commit
5ecaf8f792
2 files changed
+85
-1
No files matched your search
@@ -713,7 +713,12 @@ func (m *Manager) Subscribe(buffer int) (<-chan Event, func()) {
|
||||
}
|
||||
|
||||
func (m *Manager) emitLocked(ev Event) {
|
||||
if ev.Type == EventChainSwitched {
|
||||
// Every frontend emits the switch to its own subscribers, but only the
|
||||
// one that decided it counts it, or the cluster-wide total would be N
|
||||
// times the real one. Followers adopt the leader's switches (ApplyChain)
|
||||
// and apply pins the leader applies too. Without a sync layer each
|
||||
// frontend decides for itself, so each counts.
|
||||
if ev.Type == EventChainSwitched && (m.leader || m.sync == nil) {
|
||||
recordSwitch(ev)
|
||||
}
|
||||
for _, c := range m.subs {
|
||||
|
||||
@@ -0,0 +1,79 @@
|
||||
package failover
|
||||
|
||||
import (
|
||||
"context"
|
||||
"sync"
|
||||
|
||||
. "github.com/onsi/ginkgo/v2"
|
||||
. "github.com/onsi/gomega"
|
||||
"go.opentelemetry.io/otel"
|
||||
sdkmetric "go.opentelemetry.io/otel/sdk/metric"
|
||||
"go.opentelemetry.io/otel/sdk/metric/metricdata"
|
||||
)
|
||||
|
||||
var (
|
||||
switchReaderOnce sync.Once
|
||||
switchReader *sdkmetric.ManualReader
|
||||
)
|
||||
|
||||
// switchCount reads localai_failover_switches_total. The global provider can
|
||||
// only be installed once per process, so every spec shares one reader and
|
||||
// compares counts before and after.
|
||||
func switchCount() int64 {
|
||||
switchReaderOnce.Do(func() {
|
||||
switchReader = sdkmetric.NewManualReader()
|
||||
otel.SetMeterProvider(sdkmetric.NewMeterProvider(sdkmetric.WithReader(switchReader)))
|
||||
})
|
||||
var rm metricdata.ResourceMetrics
|
||||
Expect(switchReader.Collect(context.Background(), &rm)).To(Succeed())
|
||||
var n int64
|
||||
for _, sm := range rm.ScopeMetrics {
|
||||
for _, mt := range sm.Metrics {
|
||||
if mt.Name != "localai_failover_switches_total" {
|
||||
continue
|
||||
}
|
||||
if sum, ok := mt.Data.(metricdata.Sum[int64]); ok {
|
||||
for _, dp := range sum.DataPoints {
|
||||
n += dp.Value
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
return n
|
||||
}
|
||||
|
||||
var _ = Describe("switch metric", func() {
|
||||
It("counts a switch once across the cluster, on the frontend that decided it", func() {
|
||||
clock := newFakeClock()
|
||||
src := newFakeSource(remote("x"), local("y"), chainCfg("chain", nil, t("x"), t("y")))
|
||||
bus := &loopSync{pins: map[string]string{}}
|
||||
leaderIsA := true
|
||||
gateFor := func(isA bool) LeaderGate {
|
||||
return func(_ context.Context, fn func()) bool {
|
||||
if isA != leaderIsA {
|
||||
return false
|
||||
}
|
||||
fn()
|
||||
return true
|
||||
}
|
||||
}
|
||||
a := New(src, WithClock(clock), WithLeaderGate(gateFor(true)))
|
||||
b := New(src, WithClock(clock), WithLeaderGate(gateFor(false)))
|
||||
bus.add(a)
|
||||
bus.add(b)
|
||||
a.SetStateSync(bus)
|
||||
b.SetStateSync(bus)
|
||||
a.Tick(context.Background())
|
||||
b.Tick(context.Background())
|
||||
|
||||
before := switchCount()
|
||||
a.ReportFailure("x", errBoom) // the leader switches, the follower adopts
|
||||
st, _ := b.ChainStatus("chain")
|
||||
Expect(st.Active).To(Equal("y"))
|
||||
Expect(switchCount() - before).To(Equal(int64(1)))
|
||||
|
||||
before = switchCount()
|
||||
Expect(b.Pin("chain", "x")).To(Succeed()) // applied on both frontends
|
||||
Expect(switchCount() - before).To(Equal(int64(1)))
|
||||
})
|
||||
})
|
||||
Reference in new issue
Block a user