diff --git a/CHANGELOG.md b/CHANGELOG.md index a08a344..b80abf3 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -23,20 +23,6 @@ APIs change between minors without a deprecation period. ## Unreleased -### Added - -- **Guardrails in the console**: a view of the policies and what each checks, - whether their judge models are serving, where they are attached (attach - and detach from it), and what they did in a window. `GET - /api/v1/guard/events` answers the same, and `GET /api/v1/guard` now says - what each policy does. - -### Fixed - -- The `CONTROL` line named no floating address. - -## v0.11.3-rc.1 — 2026-10-05 - Two control planes, run in anger on the reference cluster. ### Added @@ -56,6 +42,16 @@ Two control planes, run in anger on the reference cluster. come back stops it. The other control plane is upgraded first, over ssh. - `server add` says when the mirror's `--api-allow` would refuse the new control plane. +- **Guardrails in the console**: a view of the policies and what each checks, + whether their judge models are serving, where they are attached (attach + and detach from it), and what they did in a window. `GET + /api/v1/guard/events` answers the same, and `GET /api/v1/guard` now says + what each policy does. + +- **A standby's portal is the leader's.** Its client port passes everything + but `/health`, `/ready` and `/leader` to the leader, and its admin port the + dashboard as well as the API; a proxy left pointing at a standby showed an + empty cluster. The leader trusts the control planes' `X-Forwarded-For`. ### Fixed @@ -67,6 +63,7 @@ Two control planes, run in anger on the reference cluster. - An agent's reconnect backoff carried over from one outage to the next, so each failover made the next one slower; it starts again after a session that worked. +- The `CONTROL` line named no floating address. ## v0.11.2 — 2026-10-05 diff --git a/docs/install.md b/docs/install.md index 3916b4e..27f1e30 100644 --- a/docs/install.md +++ b/docs/install.md @@ -1940,8 +1940,12 @@ Without it, put a load balancer in front that checks `/ready` — not `CONTROL` line: which leads, which stand by, one that does not answer, and whether the floating address is on the leader. A standby with nothing to take over from is the thing to see before it is needed. -- **Either machine's console.** A standby's admin API passes everything under - `/api/` to the leader, over the port control planes already reach each +- **Either machine, the leader's cluster.** A standby passes its client port + — the portal, its admin pages, the API — to the leader's, keeping only + `/health`, `/ready` and `/leader` to describe itself, so a proxy pointed at + the wrong machine still shows the cluster; the leader trusts a control + plane's `X-Forwarded-For`, so clients keep their own address. Its admin + API passes everything but `/debug/` and `/metrics` to the leader, over the port control planes already reach each other on and with the cluster's certificates both ways, so `sudo llmgw` and `llmgw ctl` show the same cluster on both — the leader's — and a change made on the standby is made by the leader. A node's certificate is refused diff --git a/internal/cli/leaderforward.go b/internal/cli/leaderforward.go index 46065cb..b4d4136 100644 --- a/internal/cli/leaderforward.go +++ b/internal/cli/leaderforward.go @@ -7,6 +7,7 @@ import ( "errors" "fmt" "log/slog" + "net" "net/http" "net/http/httputil" "net/url" @@ -16,6 +17,7 @@ import ( "golang.org/x/net/http2" + "gitea.mynodes.xyz/ilker/llmgw/internal/clientip" "gitea.mynodes.xyz/ilker/llmgw/internal/control" "gitea.mynodes.xyz/ilker/llmgw/internal/store" "gitea.mynodes.xyz/ilker/llmgw/internal/tlsconf" @@ -53,7 +55,7 @@ type leaderForward struct { // process is not it. func (f *leaderForward) standby(next http.Handler) http.Handler { return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - if !strings.HasPrefix(r.URL.Path, "/api/") || f.lead.IsLeader() { + if adminStaysHere(r.URL.Path) || f.lead.IsLeader() { next.ServeHTTP(w, r) return } @@ -90,6 +92,89 @@ func (f *leaderForward) standby(next http.Handler) http.Handler { }) } +// adminStaysHere are the admin paths about this process rather than the +// cluster: its profile, its own metrics, its own health. Never forwarded, and +// never answered for another — pprof dumps the heap, keys and all. +func adminStaysHere(path string) bool { + return strings.HasPrefix(path, "/debug/") || path == "/metrics" || path == "/health" +} + +// publicStaysHere are the client port's paths that must describe this +// process: a load balancer and keepalived ask them which machine to send +// traffic to, and a standby saying the leader is ready would draw it. +func publicStaysHere(path string) bool { + return path == "/health" || path == "/ready" || path == "/leader" || path == "/metrics" +} + +// public wraps the client-facing handler: on a standby, everything but its +// own health goes to the leader's client port — the portal, its admin pages +// and the API, which on a standby would show an empty cluster and serve +// nothing. Plain HTTP, as the port is; the leader trusts a control plane's +// X-Forwarded-For, so a client is still known by its own address. +func (f *leaderForward) public(next http.Handler, port string) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + // Once only: a request another control plane forwarded is answered + // here, leading or not, rather than sent on while the lease moves. + if publicStaysHere(r.URL.Path) || f.lead.IsLeader() || r.Header.Get(hdrForwardedBy) != "" { + next.ServeHTTP(w, r) + return + } + target, _, err := f.leader(r.Context()) + if err != nil || target.Hostname() == "" { + next.ServeHTTP(w, r) + return + } + host := net.JoinHostPort(target.Hostname(), port) + p := &httputil.ReverseProxy{ + Rewrite: func(pr *httputil.ProxyRequest) { + pr.Out.URL.Scheme, pr.Out.URL.Host = "http", host + // The name the client used, for the portal's links and cookies. + pr.Out.Host = pr.In.Host + peer, _, _ := net.SplitHostPort(pr.In.RemoteAddr) + xff := pr.In.Header.Get("X-Forwarded-For") + if xff != "" { + xff += ", " + } + pr.Out.Header.Set("X-Forwarded-For", xff+peer) + for _, h := range []string{"X-Forwarded-Proto", "X-Forwarded-Host"} { + if v := pr.In.Header.Get(h); v != "" { + pr.Out.Header.Set(h, v) + } + } + pr.Out.Header.Set(hdrForwardedBy, f.self) + }, + FlushInterval: -1, + ErrorHandler: func(w http.ResponseWriter, _ *http.Request, err error) { + writeForwardError(w, http.StatusBadGateway, "this control plane is standing by, and the leader did not answer: "+err.Error()) + }, + } + p.ServeHTTP(w, r) + }) +} + +// trustControlPlanes keeps the client port trusting the other control +// planes' X-Forwarded-For, besides router.trusted_proxies: a standby passes +// requests on, and the client is the address it was given, not the standby. +// Read again every half minute, for one added since. +func (f *leaderForward) trustControlPlanes(ctx context.Context, configured []string) { + for { + list := slices.Clone(configured) + if cps, err := f.st.ControlPlanes(ctx); err == nil { + for _, cp := range cps { + if u, err := url.Parse(cp.URL); err == nil && u.Hostname() != "" { + list = append(list, u.Hostname()) + } + } + } + clientip.SetTrusted(list) + select { + case <-ctx.Done(): + return + case <-time.After(30 * time.Second): + } + } +} + // leader is where the lease's holder is reached, and its name. func (f *leaderForward) leader(ctx context.Context) (*url.URL, string, error) { ctx, cancel := context.WithTimeout(ctx, 3*time.Second) @@ -99,6 +184,10 @@ func (f *leaderForward) leader(ctx context.Context) (*url.URL, string, error) { return nil, "", errors.New("no control plane holds the lease right now") } name := control.HolderName(lease.Owner) + if name == f.self { + // This machine's previous process, whose lease has not run out. + return nil, name, errors.New("the lease is still held by this machine's previous run; it runs out within seconds") + } cps, err := f.st.ControlPlanes(ctx) if err != nil { return nil, name, err @@ -117,7 +206,7 @@ func (f *leaderForward) leader(ctx context.Context) (*url.URL, string, error) { func (f *leaderForward) leaderSide(admin http.Handler) http.Handler { return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { path := strings.TrimPrefix(r.URL.Path, forwardPrefix) - if !strings.HasPrefix(path, "/api/") { + if adminStaysHere(path) { http.NotFound(w, r) return } diff --git a/internal/cli/leaderforward_test.go b/internal/cli/leaderforward_test.go index 1ccef1f..5f2fdc4 100644 --- a/internal/cli/leaderforward_test.go +++ b/internal/cli/leaderforward_test.go @@ -4,6 +4,7 @@ import ( "encoding/json" "io" "log/slog" + "net" "net/http" "net/http/httptest" "path/filepath" @@ -140,9 +141,15 @@ func TestAStandbysAdminAPIIsTheLeaders(t *testing.T) { if got["actor"] != "root@ctl-b" { t.Errorf("who asked was lost: %q", got["actor"]) } - // What is not the API stays on the machine: the dashboard, pprof. - if _, got := c.ask(t, "/debug/pprof/"); got["path"] != "local" { - t.Errorf("a non-API path was forwarded: %v", got) + // The dashboard's pages are the leader's too. + if _, got := c.ask(t, "/nodes/cuda01"); got["path"] != "/nodes/cuda01" { + t.Errorf("a dashboard page was answered here: %v", got) + } + // What is about this process stays on it: pprof, its metrics. + for _, p := range []string{"/debug/pprof/", "/metrics"} { + if _, got := c.ask(t, p); got["path"] != "local" { + t.Errorf("%s was forwarded: %v", p, got) + } } } @@ -186,3 +193,52 @@ func TestAStandbyWithNoLeaderSaysSo(t *testing.T) { t.Errorf("%d %v", resp.StatusCode, got) } } + +func TestAStandbysPortalIsTheLeaders(t *testing.T) { + c := newForwardCluster(t) + // The leader's client port, on the address the store lists for it. + var got http.Header + var gotHost string + leader := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + got, gotHost = r.Header.Clone(), r.Host + _, _ = w.Write([]byte("leader " + r.URL.Path)) + })) + defer leader.Close() + _, port, _ := net.SplitHostPort(strings.TrimPrefix(leader.URL, "http://")) + + local := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { _, _ = w.Write([]byte("standby " + r.URL.Path)) }) + srv := httptest.NewServer(c.standby.public(local, port)) + defer srv.Close() + get := func(path string, hdr map[string]string) string { + req, _ := http.NewRequestWithContext(t.Context(), http.MethodGet, srv.URL+path, nil) + req.Host = "llmgw.example" + for k, v := range hdr { + req.Header.Set(k, v) + } + resp, err := http.DefaultClient.Do(req) + if err != nil { + t.Fatal(err) + } + defer resp.Body.Close() + b, _ := io.ReadAll(resp.Body) + return string(b) + } + + if body := get("/admin/replicas", map[string]string{"X-Forwarded-For": "203.0.113.7", "X-Forwarded-Proto": "https"}); body != "leader /admin/replicas" { + t.Fatalf("the portal was answered by %q", body) + } + if gotHost != "llmgw.example" || got.Get("X-Forwarded-Proto") != "https" || + !strings.HasPrefix(got.Get("X-Forwarded-For"), "203.0.113.7, ") || got.Get(hdrForwardedBy) != "ctl-b" { + t.Errorf("host %q, headers %v", gotHost, got) + } + // What a load balancer asks is about this machine. + for _, p := range []string{"/ready", "/health", "/leader"} { + if body := get(p, nil); body != "standby "+p { + t.Errorf("%s answered by %q", p, body) + } + } + // Forwarded once only. + if body := get("/v1/models", map[string]string{hdrForwardedBy: "ctl-c"}); body != "standby /v1/models" { + t.Errorf("a forwarded request was forwarded again: %q", body) + } +} diff --git a/internal/cli/server.go b/internal/cli/server.go index b1fd48d..a247a30 100644 --- a/internal/cli/server.go +++ b/internal/cli/server.go @@ -513,6 +513,9 @@ func newServerCmd() *cobra.Command { served := adminHandler if forward != nil { served = forward.standby(adminHandler) + _, port, _ := net.SplitHostPort(cfg.Listen) + apiSrv.Handler = forward.public(apiSrv.Handler, port) + go forward.trustControlPlanes(ctx, cfg.Router.TrustedProxies) } adminSrv := withLimits(&http.Server{ Handler: router.AdminGuard(router.AdminAuth(adminToken, cfg.DebugPprof, served)), diff --git a/scripts/gitea/__pycache__/release_notes.cpython-314.pyc b/scripts/gitea/__pycache__/release_notes.cpython-314.pyc new file mode 100644 index 0000000..c48c915 Binary files /dev/null and b/scripts/gitea/__pycache__/release_notes.cpython-314.pyc differ diff --git a/test/e2e/phase3_test.go b/test/e2e/phase3_test.go index 63d6466..0745041 100644 --- a/test/e2e/phase3_test.go +++ b/test/e2e/phase3_test.go @@ -261,7 +261,7 @@ func TestControlPlaneFeedsRouter(t *testing.T) { }() defer close(stop) - h.waitForEndpoints(t, 2, 15*time.Second) + h.waitForEndpoints(t, 2, 30*time.Second) ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) defer cancel() @@ -304,7 +304,7 @@ func TestRouterKeepsServingWhileNodeIsLost(t *testing.T) { }() defer close(stop) - h.waitForEndpoints(t, 2, 15*time.Second) + h.waitForEndpoints(t, 2, 30*time.Second) // Continuous traffic for the whole experiment. ctx, cancel := context.WithCancel(context.Background())