From 74775a820bacbe2349d35f1b3a4ae699aa8272c2 Mon Sep 17 00:00:00 2001 From: Ettore Di Giacinto Date: Sat, 26 Sep 2026 17:50:54 +0000 Subject: [PATCH] test(failover): cover retry per endpoint family and remote fail-back Assisted-by: Claude:claude-opus-5-5 Signed-off-by: Ettore Di Giacinto --- tests/e2e/cloud_proxy_helpers_test.go | 24 ++++ tests/e2e/e2e_failover_test.go | 172 ++++++++++++++++++++++++++ tests/e2e/e2e_suite_test.go | 33 +++++ tests/e2e/mock-backend/main.go | 4 + 4 files changed, 233 insertions(+) create mode 100644 tests/e2e/e2e_failover_test.go diff --git a/tests/e2e/cloud_proxy_helpers_test.go b/tests/e2e/cloud_proxy_helpers_test.go index 819d9aa08..f98049b4c 100644 --- a/tests/e2e/cloud_proxy_helpers_test.go +++ b/tests/e2e/cloud_proxy_helpers_test.go @@ -5,6 +5,7 @@ import ( "io" "net/http" "net/http/httptest" + "slices" "strings" "sync" "sync/atomic" @@ -46,6 +47,9 @@ type fakeOpenAIUpstreamServer struct { mu sync.Mutex script func(req []byte) (status int, body string, contentType string) + // models is what GET /v1/models lists: failover liveness probes check + // that the upstream still serves the target's model. + models []string } func newFakeOpenAIUpstream() *fakeOpenAIUpstreamServer { @@ -59,6 +63,20 @@ func newFakeOpenAIUpstream() *fakeOpenAIUpstreamServer { } func (f *fakeOpenAIUpstreamServer) serve(w http.ResponseWriter, r *http.Request) { + // Answered before recording: periodic probes must not overwrite the + // request a spec is about to assert on. + if r.Method == http.MethodGet && r.URL.Path == "/v1/models" { + f.mu.Lock() + ids := slices.Clone(f.models) + f.mu.Unlock() + data := make([]map[string]string, 0, len(ids)) + for _, id := range ids { + data = append(data, map[string]string{"id": id}) + } + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(map[string]any{"data": data}) + return + } atomic.AddInt32(&f.recorder.RequestHits, 1) body, _ := io.ReadAll(r.Body) f.recorder.mu.Lock() @@ -80,6 +98,12 @@ func (f *fakeOpenAIUpstreamServer) serve(w http.ResponseWriter, r *http.Request) func (f *fakeOpenAIUpstreamServer) URL() string { return f.srv.URL } func (f *fakeOpenAIUpstreamServer) Close() { f.srv.Close() } +func (f *fakeOpenAIUpstreamServer) SetModels(ids ...string) { + f.mu.Lock() + defer f.mu.Unlock() + f.models = ids +} + func (f *fakeOpenAIUpstreamServer) SetScript(script func(req []byte) (status int, body string, contentType string)) { f.mu.Lock() defer f.mu.Unlock() diff --git a/tests/e2e/e2e_failover_test.go b/tests/e2e/e2e_failover_test.go new file mode 100644 index 000000000..540ea73c5 --- /dev/null +++ b/tests/e2e/e2e_failover_test.go @@ -0,0 +1,172 @@ +package e2e_test + +import ( + "bytes" + "encoding/json" + "io" + "mime/multipart" + "net/http" + "os" + "path/filepath" + "time" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" + "gopkg.in/yaml.v3" +) + +var _ = Describe("Failover chains", Label("failover"), func() { + postJSON := func(path string, body map[string]any) *http.Response { + b, err := json.Marshal(body) + Expect(err).ToNot(HaveOccurred()) + resp, err := http.Post(apiURL+path, "application/json", bytes.NewReader(b)) + Expect(err).ToNot(HaveOccurred()) + return resp + } + expectServedByFallback := func(resp *http.Response) { + defer func() { _ = resp.Body.Close() }() + body, _ := io.ReadAll(resp.Body) + Expect(resp.StatusCode).To(BeNumerically("<", 300), "%s\nheaders: %v", body, resp.Header) + Expect(resp.Header.Get("X-LocalAI-Served-Model")).To(Equal("failover-fallback")) + Expect(resp.Header.Get("X-LocalAI-Failover")).To(Equal("fallback")) + } + + // The entry name is the chain suffix: chain- is written by the suite. + DescribeTable("retries every endpoint family on the next target", + func(path string, body func(model string) map[string]any) { + expectServedByFallback(postJSON(path, body("chain-"+CurrentSpecReport().LeafNodeText))) + }, + Entry("chat", "/chat/completions", func(m string) map[string]any { + return map[string]any{"model": m, "messages": []map[string]string{{"role": "user", "content": "hi"}}} + }), + Entry("completion", "/completions", func(m string) map[string]any { + return map[string]any{"model": m, "prompt": "hi"} + }), + Entry("embeddings", "/embeddings", func(m string) map[string]any { + return map[string]any{"model": m, "input": "hi"} + }), + Entry("tts", "/audio/speech", func(m string) map[string]any { + return map[string]any{"model": m, "input": "hi", "voice": "default"} + }), + Entry("image", "/images/generations", func(m string) map[string]any { + return map[string]any{"model": m, "prompt": "a cat", "size": "256x256"} + }), + Entry("rerank", "/rerank", func(m string) map[string]any { + return map[string]any{"model": m, "query": "q", "documents": []string{"a", "b"}} + }), + Entry("vad", "/vad", func(m string) map[string]any { + return map[string]any{"model": m, "audio": []float32{0, 0, 0, 0}} + }), + ) + + It("retries transcription with the multipart body", func() { + var body bytes.Buffer + mw := multipart.NewWriter(&body) + Expect(mw.WriteField("model", "chain-transcription")).To(Succeed()) + fw, err := mw.CreateFormFile("file", "a.wav") + Expect(err).ToNot(HaveOccurred()) + // 200 ms of 16 kHz mono silence. + _, err = fw.Write(wavFromPCM(make([]byte, 6400), 16000)) + Expect(err).ToNot(HaveOccurred()) + Expect(mw.Close()).To(Succeed()) + resp, err := http.Post(apiURL+"/audio/transcriptions", mw.FormDataContentType(), &body) + Expect(err).ToNot(HaveOccurred()) + expectServedByFallback(resp) + }) + + Describe("remote targets", Ordered, func() { + var up1, up2 *fakeOpenAIUpstreamServer + + chatReply := func([]byte) (int, string, string) { + return 200, `{"id":"x","object":"chat.completion","choices":[{"index":0,"message":{"role":"assistant","content":"hi"},"finish_reason":"stop"}]}`, "application/json" + } + + BeforeAll(func() { + if cloudProxyPath == "" { + Skip("cloud-proxy backend binary not built (make build-cloud-proxy-backend)") + } + up1, up2 = newFakeOpenAIUpstream(), newFakeOpenAIUpstream() + DeferCleanup(up1.Close) + DeferCleanup(up2.Close) + up1.SetModels("up-1") + up2.SetModels("up-2") + registerFailoverRemoteModels(up1.URL(), up2.URL()) + }) + + It("fails over when the primary upstream errors and fails back when it recovers", func() { + up1.SetScript(func([]byte) (int, string, string) { + return 503, `{"error":"no healthy nodes"}`, "application/json" + }) + up2.SetScript(chatReply) + + resp := postJSON("/chat/completions", map[string]any{"model": "chain-remote", "messages": []map[string]string{{"role": "user", "content": "hi"}}}) + defer func() { _ = resp.Body.Close() }() + body, _ := io.ReadAll(resp.Body) + Expect(resp.StatusCode).To(Equal(200), string(body)) + Expect(resp.Header.Get("X-LocalAI-Served-Model")).To(Equal("up-2")) + Expect(resp.Header.Get("X-LocalAI-Failover")).To(Equal("fallback")) + Expect(chainActive("chain-remote")).To(Equal("up-2")) + + up1.SetScript(chatReply) + Eventually(func() string { return chainActive("chain-remote") }, 30*time.Second, 500*time.Millisecond). + Should(Equal("up-1")) + + resp2 := postJSON("/chat/completions", map[string]any{"model": "chain-remote", "messages": []map[string]string{{"role": "user", "content": "hi"}}}) + defer func() { _ = resp2.Body.Close() }() + Expect(resp2.StatusCode).To(Equal(200)) + Expect(resp2.Header.Get("X-LocalAI-Served-Model")).To(Equal("up-1")) + Expect(resp2.Header.Get("X-LocalAI-Failover")).To(BeEmpty()) + }) + }) +}) + +// chainActive returns the active target of a chain as the REST status reports +// it, or "" when the status cannot be read. +func chainActive(chain string) string { + r, err := http.Get(anthropicBaseURL + "/api/failover/" + chain) + if err != nil { + return "" + } + defer func() { _ = r.Body.Close() }() + var st struct { + Active string `json:"active"` + } + _ = json.NewDecoder(r.Body).Decode(&st) + return st.Active +} + +// registerFailoverRemoteModels registers two cloud-proxy passthrough models +// (up-1, up-2) and a chain over them. The upstream URLs exist only at runtime, +// so the YAMLs are written after startup and the loader re-reads the models +// directory; the failover manager picks the chain up on its next tick. +func registerFailoverRemoteModels(url1, url2 string) { + proxyModel := func(name, upstream string) map[string]any { + return map[string]any{ + "name": name, + "backend": "cloud-proxy", + "parameters": map[string]any{"model": name + ".bin"}, + "proxy": map[string]any{ + "mode": "passthrough", + "provider": "openai", + "upstream_url": upstream + "/v1/chat/completions", + "api_key_env": "CLOUD_PROXY_E2E_OPENAI_KEY", + }, + } + } + chain := map[string]any{ + "name": "chain-remote", + "failover": map[string]any{ + "targets": []map[string]any{{"model": "up-1"}, {"model": "up-2"}}, + "probe": map[string]any{"interval": "1s"}, + "recovery": map[string]any{"probes": 2, "min_dwell": "2s"}, + }, + } + for _, cfg := range []map[string]any{proxyModel("up-1", url1), proxyModel("up-2", url2), chain} { + data, err := yaml.Marshal(cfg) + Expect(err).ToNot(HaveOccurred()) + Expect(os.WriteFile(filepath.Join(modelsPath, cfg["name"].(string)+".yaml"), data, 0644)).To(Succeed()) + } + Expect(localAIApp.ModelConfigLoader().LoadModelConfigsFromPath(modelsPath)).To(Succeed()) + Eventually(func() string { return chainActive("chain-remote") }, 10*time.Second, 200*time.Millisecond). + Should(Equal("up-1")) +} diff --git a/tests/e2e/e2e_suite_test.go b/tests/e2e/e2e_suite_test.go index 31d8d0d9c..b27d25f64 100644 --- a/tests/e2e/e2e_suite_test.go +++ b/tests/e2e/e2e_suite_test.go @@ -118,6 +118,39 @@ var _ = BeforeSuite(func() { Expect(err).ToNot(HaveOccurred()) Expect(os.WriteFile(configPath, configYAML, 0644)).To(Succeed()) + // Failover chains, one per endpoint family: target 0 is a mock model whose + // load always fails (the mock rejects models named fail-load*), so every + // request exercises the retry onto failover-fallback. The fallback's model + // file must exist: the failover manager's liveness probe marks a local + // target with a missing file down, and mock-model.bin is never created. + Expect(os.WriteFile(filepath.Join(modelsPath, "failover-fallback.bin"), nil, 0644)).To(Succeed()) + fallbackData, err := yaml.Marshal(map[string]any{ + "name": "failover-fallback", + "backend": "mock-backend", + "parameters": map[string]any{"model": "failover-fallback.bin"}, + }) + Expect(err).ToNot(HaveOccurred()) + Expect(os.WriteFile(filepath.Join(modelsPath, "failover-fallback.yaml"), fallbackData, 0644)).To(Succeed()) + for _, family := range []string{"chat", "completion", "embeddings", "transcription", "tts", "image", "rerank", "vad"} { + for _, cfg := range []map[string]any{ + { + "name": "fail-" + family, + "backend": "mock-backend", + "parameters": map[string]any{"model": "fail-load-" + family}, + }, + { + "name": "chain-" + family, + "failover": map[string]any{ + "targets": []map[string]any{{"model": "fail-" + family}, {"model": "failover-fallback"}}, + }, + }, + } { + data, err := yaml.Marshal(cfg) + Expect(err).ToNot(HaveOccurred()) + Expect(os.WriteFile(filepath.Join(modelsPath, cfg["name"].(string)+".yaml"), data, 0644)).To(Succeed()) + } + } + // Create model config for autoparser tests (NoGrammar so tool calls // are driven entirely by the backend's ChatDeltas, not grammar enforcement) autoparserConfig := map[string]any{ diff --git a/tests/e2e/mock-backend/main.go b/tests/e2e/mock-backend/main.go index 36520cb55..0d842b42e 100644 --- a/tests/e2e/mock-backend/main.go +++ b/tests/e2e/mock-backend/main.go @@ -93,6 +93,10 @@ func (m *MockBackend) LoadModel(ctx context.Context, in *pb.ModelOptions) (*pb.R "draft_model", in.DraftModel, "mmproj", in.MMProj) recordLoadParams(in) + // Lets e2e specs build a failover target whose backend cannot load. + if strings.HasPrefix(in.Model, "fail-load") { + return &pb.Result{Message: "mock: load failure", Success: false}, nil + } return &pb.Result{ Message: "Model loaded successfully (mocked)", Success: true,