Skip to content

Commit 3fccaa7

Browse files
localai-botmudler
authored and
richiejp
committed
fix(distributed): make per-node backend upgrade actually upgrade (#10838)
* test(core/http): make the suite's HTTP port overridable app_test.go and openresponses_test.go hardcoded 127.0.0.1:9090. When another service already listens on 9090 the suite does not fail fast: the server goroutine logs the bind error and the specs then poll whatever is squatting the port until Eventually times out. On machines where 9090 is permanently taken this makes the pre-commit coverage gate impossible to pass. Introduce testHTTPAddr, defaulting to 127.0.0.1:9090 (what CI has always used) and overridable via LOCALAI_TEST_HTTP_PORT for local runs. Assisted-by: Claude:claude-fable-5 golangci-lint Signed-off-by: Ettore Di Giacinto <mudler@localai.io> * fix(distributed): make per-node backend upgrade actually upgrade The node detail page's Upgrade button reused the node-scoped install path (POST /api/nodes/:id/backends/install). That fires NATS backend.install with force=false, and the worker's install handler is deliberately "ensure installed": when the backend binary already exists on disk it short-circuits without touching the gallery. Since only an installed backend can be upgraded, the whole chain was a guaranteed successful no-op - the UI then toasted "backend upgraded" without even waiting for the async job. Route upgrades through the real force-reinstall path instead: - BackendManager.UpgradeBackend now receives the ManagementOp (like InstallBackend already did) so implementations can honor op.TargetNodeID. - DistributedBackendManager.UpgradeBackend scopes the backend.upgrade fan-out to op.TargetNodeID when set, and errors when the target node does not report the backend as installed. - New POST /api/nodes/:id/backends/upgrade endpoint enqueues an Upgrade=true node-scoped op (async 202 + jobID, mirroring install). - NodeDetail UI calls the new endpoint and reports the dispatch ("Upgrading ... on this node...") instead of claiming success; the Operations panel tracks the actual job. Verified against a live local cluster (NATS + Postgres + two workers): the target worker stops the running process, force-reinstalls from the gallery and re-downloads the OCI image; the second worker receives no backend.upgrade event; upgrading a backend missing from the target node fails the job with a clear error. Assisted-by: Claude:claude-fable-5 golangci-lint Signed-off-by: Ettore Di Giacinto <mudler@localai.io> --------- Signed-off-by: Ettore Di Giacinto <mudler@localai.io> Co-authored-by: Ettore Di Giacinto <mudler@localai.io> Signed-off-by: Richard Palethorpe <io@richiejp.com>
1 parent b326678 commit 3fccaa7

16 files changed

Lines changed: 316 additions & 88 deletions

File tree

‎core/application/upgrade_checker.go‎

Lines changed: 5 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -215,8 +215,11 @@ func (uc *UpgradeChecker) runCheck(ctx context.Context) {
215215
var err error
216216
if bm != nil {
217217
// Background auto-upgrade: no live admin watching a progress bar,
218-
// so opID is empty and the distributed path skips progress streaming.
219-
err = bm.UpgradeBackend(ctx, "", name, nil)
218+
// so op.ID is empty and the distributed path skips progress streaming.
219+
err = bm.UpgradeBackend(ctx, &galleryop.ManagementOp[gallery.GalleryBackend, any]{
220+
GalleryElementName: name,
221+
Upgrade: true,
222+
}, nil)
220223
} else {
221224
err = gallery.UpgradeBackend(ctx, uc.systemState, uc.modelLoader,
222225
uc.galleries, name, nil, uc.appConfig.RequireBackendIntegrity)

‎core/http/app_test.go‎

Lines changed: 40 additions & 40 deletions
Original file line numberDiff line numberDiff line change
@@ -383,13 +383,13 @@ var _ = Describe("API test", func() {
383383
Expect(err).ToNot(HaveOccurred())
384384

385385
go func() {
386-
if err := app.Start("127.0.0.1:9090"); err != nil && err != http.ErrServerClosed {
386+
if err := app.Start(testHTTPAddr); err != nil && err != http.ErrServerClosed {
387387
xlog.Error("server error", "error", err)
388388
}
389389
}()
390390

391391
defaultConfig := openai.DefaultConfig(apiKey)
392-
defaultConfig.BaseURL = "http://127.0.0.1:9090/v1"
392+
defaultConfig.BaseURL = "http://" + testHTTPAddr + "/v1"
393393

394394
client2 = openaigo.NewClient("")
395395
client2.BaseURL = defaultConfig.BaseURL
@@ -418,7 +418,7 @@ var _ = Describe("API test", func() {
418418

419419
Context("Auth Tests", func() {
420420
It("Should fail if the api key is missing", func() {
421-
err, sc := postInvalidRequest("http://127.0.0.1:9090/models/available")
421+
err, sc := postInvalidRequest("http://" + testHTTPAddr + "/models/available")
422422
Expect(err).ToNot(BeNil())
423423
Expect(sc).To(Equal(401))
424424
})
@@ -427,7 +427,7 @@ var _ = Describe("API test", func() {
427427
Context("URL routing Tests", func() {
428428
It("Should support reverse-proxy when unauthenticated", func() {
429429

430-
err, sc, body := getRequest("http://127.0.0.1:9090/myprefix/", http.Header{
430+
err, sc, body := getRequest("http://"+testHTTPAddr+"/myprefix/", http.Header{
431431
"X-Forwarded-Proto": {"https"},
432432
"X-Forwarded-Host": {"example.org"},
433433
"X-Forwarded-Prefix": {"/myprefix/"},
@@ -441,7 +441,7 @@ var _ = Describe("API test", func() {
441441

442442
It("Should support reverse-proxy when authenticated", func() {
443443

444-
err, sc, body := getRequest("http://127.0.0.1:9090/myprefix/", http.Header{
444+
err, sc, body := getRequest("http://"+testHTTPAddr+"/myprefix/", http.Header{
445445
"Authorization": {bearerKey},
446446
"X-Forwarded-Proto": {"https"},
447447
"X-Forwarded-Host": {"example.org"},
@@ -459,7 +459,7 @@ var _ = Describe("API test", func() {
459459
// requests them through the proxy.
460460
It("Should support reverse-proxy when prefix is stripped by the proxy", func() {
461461

462-
err, sc, body := getRequest("http://127.0.0.1:9090/app", http.Header{
462+
err, sc, body := getRequest("http://"+testHTTPAddr+"/app", http.Header{
463463
"X-Forwarded-Proto": {"https"},
464464
"X-Forwarded-Host": {"example.org"},
465465
"X-Forwarded-Prefix": {"/myprefix"},
@@ -477,7 +477,7 @@ var _ = Describe("API test", func() {
477477
// from a foreign origin. BasePathPrefix must reject these via
478478
// SafeForwardedPrefix and fall back to "/".
479479
It("Should ignore an unsafe X-Forwarded-Prefix and not poison asset URLs", func() {
480-
err, sc, body := getRequest("http://127.0.0.1:9090/app", http.Header{
480+
err, sc, body := getRequest("http://"+testHTTPAddr+"/app", http.Header{
481481
"X-Forwarded-Proto": {"https"},
482482
"X-Forwarded-Host": {"example.org"},
483483
"X-Forwarded-Prefix": {"//evil.com"},
@@ -492,13 +492,13 @@ var _ = Describe("API test", func() {
492492
Context("Applying models", func() {
493493

494494
It("applies models from a gallery", func() {
495-
models, err := getModels("http://127.0.0.1:9090/models/available")
495+
models, err := getModels("http://" + testHTTPAddr + "/models/available")
496496
Expect(err).To(BeNil())
497497
Expect(len(models)).To(Equal(2), fmt.Sprint(models))
498498
Expect(models[0].Installed).To(BeFalse(), fmt.Sprint(models))
499499
Expect(models[1].Installed).To(BeFalse(), fmt.Sprint(models))
500500

501-
response := postModelApplyRequest("http://127.0.0.1:9090/models/apply", modelApplyRequest{
501+
response := postModelApplyRequest("http://"+testHTTPAddr+"/models/apply", modelApplyRequest{
502502
ID: "test@bert2",
503503
})
504504

@@ -507,7 +507,7 @@ var _ = Describe("API test", func() {
507507
uuid := response["uuid"].(string)
508508
resp := map[string]any{}
509509
Eventually(func() bool {
510-
response := getModelStatus("http://127.0.0.1:9090/models/jobs/" + uuid)
510+
response := getModelStatus("http://" + testHTTPAddr + "/models/jobs/" + uuid)
511511
fmt.Println(response)
512512
resp = response
513513
return response["processed"].(bool)
@@ -526,7 +526,7 @@ var _ = Describe("API test", func() {
526526
Expect(content["usage"]).To(ContainSubstring("You can test this model with curl like this"))
527527
Expect(content["foo"]).To(Equal("bar"))
528528

529-
models, err = getModels("http://127.0.0.1:9090/models/available")
529+
models, err = getModels("http://" + testHTTPAddr + "/models/available")
530530
Expect(err).To(BeNil())
531531
Expect(len(models)).To(Equal(2), fmt.Sprint(models))
532532
Expect(models[0].Name).To(Or(Equal("bert"), Equal("bert2")))
@@ -541,7 +541,7 @@ var _ = Describe("API test", func() {
541541
})
542542
It("overrides models", func() {
543543

544-
response := postModelApplyRequest("http://127.0.0.1:9090/models/apply", modelApplyRequest{
544+
response := postModelApplyRequest("http://"+testHTTPAddr+"/models/apply", modelApplyRequest{
545545
URL: bertEmbeddingsURL,
546546
Name: "bert",
547547
Overrides: map[string]any{
@@ -554,7 +554,7 @@ var _ = Describe("API test", func() {
554554
uuid := response["uuid"].(string)
555555

556556
Eventually(func() bool {
557-
response := getModelStatus("http://127.0.0.1:9090/models/jobs/" + uuid)
557+
response := getModelStatus("http://" + testHTTPAddr + "/models/jobs/" + uuid)
558558
return response["processed"].(bool)
559559
}, "360s", "10s").Should(Equal(true))
560560

@@ -567,7 +567,7 @@ var _ = Describe("API test", func() {
567567
Expect(content["backend"]).To(Equal("llama"))
568568
})
569569
It("apply models without overrides", func() {
570-
response := postModelApplyRequest("http://127.0.0.1:9090/models/apply", modelApplyRequest{
570+
response := postModelApplyRequest("http://"+testHTTPAddr+"/models/apply", modelApplyRequest{
571571
URL: bertEmbeddingsURL,
572572
Name: "bert",
573573
Overrides: map[string]any{},
@@ -578,7 +578,7 @@ var _ = Describe("API test", func() {
578578
uuid := response["uuid"].(string)
579579

580580
Eventually(func() bool {
581-
response := getModelStatus("http://127.0.0.1:9090/models/jobs/" + uuid)
581+
response := getModelStatus("http://" + testHTTPAddr + "/models/jobs/" + uuid)
582582
return response["processed"].(bool)
583583
}, "360s", "10s").Should(Equal(true))
584584

@@ -622,14 +622,14 @@ parameters:
622622
}
623623

624624
var response schema.GalleryResponse
625-
err := postRequestResponseJSON("http://127.0.0.1:9090/models/import-uri", &importReq, &response)
625+
err := postRequestResponseJSON("http://"+testHTTPAddr+"/models/import-uri", &importReq, &response)
626626
Expect(err).ToNot(HaveOccurred())
627627
Expect(response.ID).ToNot(BeEmpty())
628628

629629
uuid := response.ID
630630
resp := map[string]any{}
631631
Eventually(func() bool {
632-
response := getModelStatus("http://127.0.0.1:9090/models/jobs/" + uuid)
632+
response := getModelStatus("http://" + testHTTPAddr + "/models/jobs/" + uuid)
633633
resp = response
634634
return response["processed"].(bool)
635635
}, "360s", "10s").Should(Equal(true))
@@ -657,7 +657,7 @@ parameters:
657657
}
658658

659659
var response schema.GalleryResponse
660-
err := postRequestResponseJSON("http://127.0.0.1:9090/models/import-uri", &importReq, &response)
660+
err := postRequestResponseJSON("http://"+testHTTPAddr+"/models/import-uri", &importReq, &response)
661661
// The endpoint should return an error immediately
662662
Expect(err).To(HaveOccurred())
663663
Expect(err.Error()).To(ContainSubstring("failed to discover model config"))
@@ -693,14 +693,14 @@ parameters:
693693
}
694694

695695
var response schema.GalleryResponse
696-
err := postRequestResponseJSON("http://127.0.0.1:9090/models/import-uri", &importReq, &response)
696+
err := postRequestResponseJSON("http://"+testHTTPAddr+"/models/import-uri", &importReq, &response)
697697
Expect(err).ToNot(HaveOccurred())
698698
Expect(response.ID).ToNot(BeEmpty())
699699

700700
uuid := response.ID
701701
resp := map[string]any{}
702702
Eventually(func() bool {
703-
response := getModelStatus("http://127.0.0.1:9090/models/jobs/" + uuid)
703+
response := getModelStatus("http://" + testHTTPAddr + "/models/jobs/" + uuid)
704704
resp = response
705705
return response["processed"].(bool)
706706
}, "360s", "10s").Should(Equal(true))
@@ -763,13 +763,13 @@ chat_template_kwargs:
763763
app, err = API(localAIApp)
764764
Expect(err).ToNot(HaveOccurred())
765765
go func() {
766-
if err := app.Start("127.0.0.1:9090"); err != nil && err != http.ErrServerClosed {
766+
if err := app.Start(testHTTPAddr); err != nil && err != http.ErrServerClosed {
767767
xlog.Error("server error", "error", err)
768768
}
769769
}()
770770

771771
defaultConfig := openai.DefaultConfig("")
772-
defaultConfig.BaseURL = "http://127.0.0.1:9090/v1"
772+
defaultConfig.BaseURL = "http://" + testHTTPAddr + "/v1"
773773

774774
client2 = openaigo.NewClient("")
775775
client2.BaseURL = defaultConfig.BaseURL
@@ -813,7 +813,7 @@ chat_template_kwargs:
813813
// Mock-backend is registered via SetExternalBackend so it appears
814814
// alongside any built-in entries; verifying that string proves the
815815
// endpoint is wired up regardless of which real backends exist.
816-
resp, err := http.Get("http://127.0.0.1:9090/system")
816+
resp, err := http.Get("http://" + testHTTPAddr + "/system")
817817
Expect(err).ToNot(HaveOccurred())
818818
Expect(resp.StatusCode).To(Equal(200))
819819
dat, err := io.ReadAll(resp.Body)
@@ -851,7 +851,7 @@ chat_template_kwargs:
851851
} `json:"message"`
852852
} `json:"choices"`
853853
}
854-
err := postRequestResponseJSON("http://127.0.0.1:9090/v1/chat/completions", &reqBody, &chatResp)
854+
err := postRequestResponseJSON("http://"+testHTTPAddr+"/v1/chat/completions", &reqBody, &chatResp)
855855
Expect(err).ToNot(HaveOccurred())
856856
Expect(chatResp.Choices).ToNot(BeEmpty())
857857

@@ -889,22 +889,22 @@ chat_template_kwargs:
889889
}
890890

891891
var createResp map[string]any
892-
err := postRequestResponseJSON("http://127.0.0.1:9090/api/agent/tasks", &taskBody, &createResp)
892+
err := postRequestResponseJSON("http://"+testHTTPAddr+"/api/agent/tasks", &taskBody, &createResp)
893893
Expect(err).ToNot(HaveOccurred())
894894
Expect(createResp["id"]).ToNot(BeEmpty())
895895
taskID := createResp["id"].(string)
896896

897897
// Get the task
898898
var task schema.Task
899-
resp, err := http.Get("http://127.0.0.1:9090/api/agent/tasks/" + taskID)
899+
resp, err := http.Get("http://" + testHTTPAddr + "/api/agent/tasks/" + taskID)
900900
Expect(err).ToNot(HaveOccurred())
901901
Expect(resp.StatusCode).To(Equal(200))
902902
body, _ := io.ReadAll(resp.Body)
903903
json.Unmarshal(body, &task)
904904
Expect(task.Name).To(Equal("Test Task"))
905905

906906
// List tasks
907-
resp, err = http.Get("http://127.0.0.1:9090/api/agent/tasks")
907+
resp, err = http.Get("http://" + testHTTPAddr + "/api/agent/tasks")
908908
Expect(err).ToNot(HaveOccurred())
909909
Expect(resp.StatusCode).To(Equal(200))
910910
var tasks []schema.Task
@@ -914,18 +914,18 @@ chat_template_kwargs:
914914

915915
// Update task
916916
taskBody["name"] = "Updated Task"
917-
err = putRequestJSON("http://127.0.0.1:9090/api/agent/tasks/"+taskID, &taskBody)
917+
err = putRequestJSON("http://"+testHTTPAddr+"/api/agent/tasks/"+taskID, &taskBody)
918918
Expect(err).ToNot(HaveOccurred())
919919

920920
// Verify update
921-
resp, err = http.Get("http://127.0.0.1:9090/api/agent/tasks/" + taskID)
921+
resp, err = http.Get("http://" + testHTTPAddr + "/api/agent/tasks/" + taskID)
922922
Expect(err).ToNot(HaveOccurred())
923923
body, _ = io.ReadAll(resp.Body)
924924
json.Unmarshal(body, &task)
925925
Expect(task.Name).To(Equal("Updated Task"))
926926

927927
// Delete task
928-
req, _ := http.NewRequest("DELETE", "http://127.0.0.1:9090/api/agent/tasks/"+taskID, nil)
928+
req, _ := http.NewRequest("DELETE", "http://"+testHTTPAddr+"/api/agent/tasks/"+taskID, nil)
929929
req.Header.Set("Authorization", bearerKey)
930930
resp, err = http.DefaultClient.Do(req)
931931
Expect(err).ToNot(HaveOccurred())
@@ -942,7 +942,7 @@ chat_template_kwargs:
942942
}
943943

944944
var createResp map[string]any
945-
err := postRequestResponseJSON("http://127.0.0.1:9090/api/agent/tasks", &taskBody, &createResp)
945+
err := postRequestResponseJSON("http://"+testHTTPAddr+"/api/agent/tasks", &taskBody, &createResp)
946946
Expect(err).ToNot(HaveOccurred())
947947
taskID := createResp["id"].(string)
948948

@@ -953,14 +953,14 @@ chat_template_kwargs:
953953
}
954954

955955
var jobResp schema.JobExecutionResponse
956-
err = postRequestResponseJSON("http://127.0.0.1:9090/api/agent/jobs/execute", &jobBody, &jobResp)
956+
err = postRequestResponseJSON("http://"+testHTTPAddr+"/api/agent/jobs/execute", &jobBody, &jobResp)
957957
Expect(err).ToNot(HaveOccurred())
958958
Expect(jobResp.JobID).ToNot(BeEmpty())
959959
jobID := jobResp.JobID
960960

961961
// Get job status
962962
var job schema.Job
963-
resp, err := http.Get("http://127.0.0.1:9090/api/agent/jobs/" + jobID)
963+
resp, err := http.Get("http://" + testHTTPAddr + "/api/agent/jobs/" + jobID)
964964
Expect(err).ToNot(HaveOccurred())
965965
Expect(resp.StatusCode).To(Equal(200))
966966
body, _ := io.ReadAll(resp.Body)
@@ -969,7 +969,7 @@ chat_template_kwargs:
969969
Expect(job.TaskID).To(Equal(taskID))
970970

971971
// List jobs
972-
resp, err = http.Get("http://127.0.0.1:9090/api/agent/jobs")
972+
resp, err = http.Get("http://" + testHTTPAddr + "/api/agent/jobs")
973973
Expect(err).ToNot(HaveOccurred())
974974
Expect(resp.StatusCode).To(Equal(200))
975975
var jobs []schema.Job
@@ -979,13 +979,13 @@ chat_template_kwargs:
979979

980980
// Cancel job (if still pending/running)
981981
if job.Status == schema.JobStatusPending || job.Status == schema.JobStatusRunning {
982-
req, _ := http.NewRequest("POST", "http://127.0.0.1:9090/api/agent/jobs/"+jobID+"/cancel", nil)
982+
req, _ := http.NewRequest("POST", "http://"+testHTTPAddr+"/api/agent/jobs/"+jobID+"/cancel", nil)
983983
req.Header.Set("Authorization", bearerKey)
984984
resp, err = http.DefaultClient.Do(req)
985985
Expect(err).ToNot(HaveOccurred())
986986
if resp.StatusCode == http.StatusBadRequest {
987987
// The worker can finish between the status read and cancellation request.
988-
resp, err = http.Get("http://127.0.0.1:9090/api/agent/jobs/" + jobID)
988+
resp, err = http.Get("http://" + testHTTPAddr + "/api/agent/jobs/" + jobID)
989989
Expect(err).ToNot(HaveOccurred())
990990
Expect(resp.StatusCode).To(Equal(http.StatusOK))
991991
body, _ = io.ReadAll(resp.Body)
@@ -1012,13 +1012,13 @@ chat_template_kwargs:
10121012
}
10131013

10141014
var createResp map[string]any
1015-
err := postRequestResponseJSON("http://127.0.0.1:9090/api/agent/tasks", &taskBody, &createResp)
1015+
err := postRequestResponseJSON("http://"+testHTTPAddr+"/api/agent/tasks", &taskBody, &createResp)
10161016
Expect(err).ToNot(HaveOccurred())
10171017

10181018
// Execute by name
10191019
paramsBody := map[string]string{"param1": "value1"}
10201020
var jobResp schema.JobExecutionResponse
1021-
err = postRequestResponseJSON("http://127.0.0.1:9090/api/agent/tasks/Named Task/execute", &paramsBody, &jobResp)
1021+
err = postRequestResponseJSON("http://"+testHTTPAddr+"/api/agent/tasks/Named Task/execute", &paramsBody, &jobResp)
10221022
Expect(err).ToNot(HaveOccurred())
10231023
Expect(jobResp.JobID).ToNot(BeEmpty())
10241024
})
@@ -1078,13 +1078,13 @@ chat_template_kwargs:
10781078
Expect(err).ToNot(HaveOccurred())
10791079

10801080
go func() {
1081-
if err := app.Start("127.0.0.1:9090"); err != nil && err != http.ErrServerClosed {
1081+
if err := app.Start(testHTTPAddr); err != nil && err != http.ErrServerClosed {
10821082
xlog.Error("server error", "error", err)
10831083
}
10841084
}()
10851085

10861086
defaultConfig := openai.DefaultConfig("")
1087-
defaultConfig.BaseURL = "http://127.0.0.1:9090/v1"
1087+
defaultConfig.BaseURL = "http://" + testHTTPAddr + "/v1"
10881088
client2 = openaigo.NewClient("")
10891089
client2.BaseURL = defaultConfig.BaseURL
10901090
// Wait for API to be ready

0 commit comments

Comments
 (0)