diff --git a/docs/admin-api.openapi.json b/docs/admin-api.openapi.json index d791221..3b255bb 100644 --- a/docs/admin-api.openapi.json +++ b/docs/admin-api.openapi.json @@ -41,6 +41,9 @@ { "name": "facility" }, + { + "name": "federation" + }, { "name": "guard" }, @@ -591,6 +594,157 @@ } } }, + "/api/v1/federation": { + "get": { + "operationId": "getFederation", + "summary": "This cluster's id, and its federation peers with their health and models", + "tags": [ + "federation" + ], + "responses": { + "200": { + "description": "OK", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/FederationView" + } + } + } + }, + "401": { + "$ref": "#/components/responses/Unauthorized" + }, + "default": { + "$ref": "#/components/responses/Error" + } + } + }, + "post": { + "operationId": "postFederation", + "summary": "Add another llmgw cluster as a peer, or change one", + "description": "The peer is checked first: its /ready, timed, and its /v1/models with the key. Its models join the catalog as PEER/MODEL.", + "tags": [ + "federation" + ], + "parameters": [ + { + "$ref": "#/components/parameters/Actor" + } + ], + "requestBody": { + "required": true, + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/federation.AddRequest" + } + } + } + }, + "responses": { + "201": { + "description": "OK", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/federation.Status" + } + } + } + }, + "401": { + "$ref": "#/components/responses/Unauthorized" + }, + "default": { + "$ref": "#/components/responses/Error" + } + } + } + }, + "/api/v1/federation/{peer}": { + "delete": { + "operationId": "deleteFederationPeer", + "summary": "Remove a federation peer and its models", + "description": "Refused while a model here falls back to one of the peer's.", + "tags": [ + "federation" + ], + "parameters": [ + { + "name": "peer", + "in": "path", + "required": true, + "description": "The federation peer's name, as `llmgw ctl federation list` lists it.", + "schema": { + "type": "string" + } + }, + { + "$ref": "#/components/parameters/Actor" + } + ], + "responses": { + "200": { + "description": "OK", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/Removed" + } + } + } + }, + "401": { + "$ref": "#/components/responses/Unauthorized" + }, + "default": { + "$ref": "#/components/responses/Error" + } + } + } + }, + "/api/v1/federation/{peer}/sync": { + "post": { + "operationId": "postFederationPeerSync", + "summary": "Check a peer now, rather than at the next interval", + "tags": [ + "federation" + ], + "parameters": [ + { + "name": "peer", + "in": "path", + "required": true, + "description": "The federation peer's name, as `llmgw ctl federation list` lists it.", + "schema": { + "type": "string" + } + }, + { + "$ref": "#/components/parameters/Actor" + } + ], + "responses": { + "200": { + "description": "OK", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/federation.Status" + } + } + } + }, + "401": { + "$ref": "#/components/responses/Unauthorized" + }, + "default": { + "$ref": "#/components/responses/Error" + } + } + } + }, "/api/v1/guard": { "get": { "operationId": "getGuard", @@ -1439,6 +1593,58 @@ } } }, + "/api/v1/models/{name}/fallbacks": { + "put": { + "operationId": "putModelsNameFallbacks", + "summary": "Replace the models a model falls back to when it has no capacity", + "description": "A model added at runtime only. A federation peer's model, PEER/MODEL, sends the overflow to the other cluster.", + "tags": [ + "models" + ], + "parameters": [ + { + "name": "name", + "in": "path", + "required": true, + "description": "The model's name in the catalog.", + "schema": { + "type": "string" + } + }, + { + "$ref": "#/components/parameters/Actor" + } + ], + "requestBody": { + "required": true, + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/FallbacksRequest" + } + } + } + }, + "responses": { + "200": { + "description": "OK", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/FallbacksView" + } + } + } + }, + "401": { + "$ref": "#/components/responses/Unauthorized" + }, + "default": { + "$ref": "#/components/responses/Error" + } + } + } + }, "/api/v1/models/{name}/headroom": { "get": { "operationId": "getModelsNameHeadroom", @@ -3445,6 +3651,49 @@ } } }, + "FallbacksRequest": { + "type": "object", + "description": "FallbacksRequest replaces a model's fallbacks; empty takes them all away.", + "properties": { + "fallbacks": { + "type": "array", + "items": { + "type": "string" + } + } + } + }, + "FallbacksView": { + "type": "object", + "description": "FallbacksView is a model's fallbacks, as they are now.", + "properties": { + "model": { + "type": "string" + }, + "fallbacks": { + "type": "array", + "items": { + "type": "string" + } + } + } + }, + "FederationView": { + "type": "object", + "description": "FederationView is this cluster's id and its peers.", + "properties": { + "cluster": { + "type": "string", + "description": "Cluster is this cluster's own id, which its peers see in the hop header." + }, + "peers": { + "type": "array", + "items": { + "$ref": "#/components/schemas/federation.Status" + } + } + } + }, "FetchView": { "type": "object", "description": "FetchView is a download in flight.", @@ -5363,6 +5612,10 @@ "api_key_file": { "type": "string", "description": "APIKeyFile holds the provider's key, on the control plane. A file rather than the key itself, so the key is in neither the catalog, the store nor the audit log; empty for a provider that asks for none." + }, + "peer": { + "type": "string", + "description": "Peer names the federated llmgw cluster this model is offered by, for a model that federation added (`llmgw ctl federation add`) rather than one written by hand. Such a model is named PEER/MODEL, comes and goes with the peer's own list, and is sent only while the peer is up." } } }, @@ -6027,6 +6280,104 @@ } } }, + "federation.AddRequest": { + "type": "object", + "description": "AddRequest adds a peer.", + "properties": { + "name": { + "type": "string", + "description": "Name is what this cluster calls the peer, and the prefix of its models: NAME/MODEL." + }, + "url": { + "type": "string", + "description": "URL is the peer gateway's client address, without /v1." + }, + "key_file": { + "type": "string", + "description": "KeyFile is a file on the control plane holding a key the peer issued to this cluster (or env:NAME, vault:PATH#FIELD)." + }, + "send_tags": { + "type": "boolean", + "description": "SendTags makes requests sent to the peer carry x-llmgw-tags naming this cluster and the sending project. Off unless asked: the peer learns who in this cluster is asking." + } + } + }, + "federation.ModelStatus": { + "type": "object", + "description": "ModelStatus is one of a peer's models: its name here, the peer's own, and whether the peer has stopped offering it.", + "properties": { + "name": { + "type": "string" + }, + "upstream": { + "type": "string" + }, + "withdrawn": { + "type": "boolean" + } + } + }, + "federation.Status": { + "type": "object", + "description": "Status is a peer as `llmgw ctl federation list` shows it.", + "properties": { + "name": { + "type": "string" + }, + "url": { + "type": "string" + }, + "key_file": { + "type": "string" + }, + "send_tags": { + "type": "boolean" + }, + "cluster": { + "type": "string", + "description": "Cluster is the peer's own cluster id." + }, + "up": { + "type": "boolean" + }, + "checked": { + "type": "string", + "format": "date-time", + "description": "Checked is when; zero before the first check." + }, + "latency_ns": { + "type": "integer", + "format": "int64", + "description": "Latency is how long the peer took to answer /ready: the network to the other site and the peer's front, not a model's speed.\n\nA duration, in nanoseconds." + }, + "error": { + "type": "string", + "description": "Error is why the peer is down, empty while it is up." + }, + "latency_ms": { + "type": "number", + "description": "LatencyMs is Latency in milliseconds, for reading." + }, + "synced_at": { + "type": "string", + "format": "date-time", + "description": "SyncedAt is when its models last changed here." + }, + "models": { + "type": "array", + "items": { + "$ref": "#/components/schemas/federation.ModelStatus" + } + }, + "created_at": { + "type": "string", + "format": "date-time" + }, + "created_by": { + "type": "string" + } + } + }, "guard.Attachment": { "type": "object", "description": "Attachment ties a policy to a part of the tenancy tree: scope is org, team, project, user or key, and ScopeID which one.", diff --git a/docs/install.md b/docs/install.md index e38a0d1..668ec5b 100644 --- a/docs/install.md +++ b/docs/install.md @@ -1196,6 +1196,63 @@ prices and budgets, rate limits, usage records, and key allowlists. whose allowlist leaves the external model out cannot reach it, as a fallback or by name. The cluster page and `ctl models` mark it `external`. +### Federating clusters + +Another llmgw cluster — another site, another team's GPUs — can be a provider +of this one, and llmgw knows it is one. Ask the other cluster's operator for a +key in one of their projects (`llmgw keys create`), put it in a file on this +control plane (on every control plane, with HA), and add the peer: + +``` +llmgw ctl federation add east --url https://llm.east.example --key-file /etc/llmgw/secrets/east.key +llmgw ctl federation list +llmgw ctl model fallbacks qwen-7b east/qwen-7b # overflow goes to east +``` + +- **Its models, named PEER/MODEL.** What the peer's `/v1/models` offers that + key is offered here as `east/qwen-7b` and so on, always with the prefix — + never under the peer's own name, even when nothing here uses it, so + whether a request leaves the cluster never depends on what else is in the + catalog, and every usage record and `x-llmgw-served-model` header says which + site answered. A model here called `east/...` wins over the peer's of the + same name. The peer's aliases, and models the peer has from peers of its + own, are not taken: federation is one hop. +- **Kept in step.** Each peer is checked every `federation.interval` (10s): + its `/ready`, timed — the latency `ctl federation list` shows — and its + `/v1/models`. A model it adds appears here; one it drops goes, unless a + model here still falls back to it, when it is kept *withdrawn* (in the + catalog, sent nothing) until the fallback is changed. `ctl federation sync + east` checks now. +- **Taken out when down.** A peer whose `/ready` fails, which refuses the key, + or which does not answer is sent nothing until it answers again; its models + stay in the catalog. A peer's `/ready` counts only what it serves itself, so + two clusters federating each other are not each ready because of the other. +- **Overflow.** Named in a model's fallbacks (`ctl model fallbacks`, for + models added at runtime), a peer's model is used only when the model has + no capacity here — as for any external model. +- **What stays here, stays here.** A peer's model is an external model: a + project with `keys residency --in-cluster`, or under a guardrails policy with + `no_store`, is refused it by name and never sent to it as a fallback. +- **One key, and who is asking.** The peer sees this cluster's key, so its + usage, limits and budgets are one tenant's. The request's `X-Request-Id` + goes with it, so one id finds the request on both sides. With + `--send-tags` the request also carries `x-llmgw-tags: + federated_from=,org=…,team=…,project=…` (ids), for the + peer's usage records to attribute it; off unless given, because it tells + the other site who here is asking. +- **No loops.** Each cluster has an id (`ctl federation list` shows it), and a + request sent to a peer carries `x-llmgw-hop` with the ids it has passed + through. A cluster refuses a request that has already passed through it + (508, `federation_loop`), and does not send one to a peer it came from — + so two clusters falling back to each other try each once. +- `llmgw_federation_peer_up`, `_latency_seconds` and `_models` are on + `/metrics`; the console has a Federation view. `ctl federation remove east` + takes the peer and its models away, refused while a fallback names one. + +Not yet: models declared in server.yaml cannot name a peer's model as a +fallback (the catalog in the file is checked before the peers are read), and +a peer behind a private CA needs that CA in the system's trust store. + ### Trying new weights on some of the traffic first When a model's repository publishes new weights, `llmgw ctl model update` diff --git a/docs/reference.md b/docs/reference.md index 8d44428..d1e9e58 100644 --- a/docs/reference.md +++ b/docs/reference.md @@ -86,6 +86,7 @@ rather than repeated. | `url` | `string` | [External](#external) | | `model` | `string` | [External](#external) | | `api_key_file` | `string` | [External](#external) | +| `peer` | `string` | [External](#external) | | `budgets` | `[]FacilityBudget` | [Facility](#facility) | | `margin` | `float64` | [Facility](#facility) | | `gpu_watts` | `float64` | [Facility](#facility) | @@ -95,6 +96,8 @@ rather than repeated. | `domain` | `string` | [FacilityBudget](#facilitybudget) | | `cap_watts` | `float64` | [FacilityBudget](#facilitybudget) | | `other_watts` | `float64` | [FacilityBudget](#facilitybudget) | +| `name` | `string` | [Federation](#federation) | +| `interval` | `time.Duration` | [Federation](#federation) | | `format` | `string` | [GuardModel](#guardmodel) | | `categories` | `[]string` | [GuardModel](#guardmodel) | | `model` | `string` | [JudgePolicy](#judgepolicy) | @@ -293,6 +296,7 @@ rather than repeated. | `control` | `Control` | [Server](#server) | | `models` | `[]catalog.Model` | [Server](#server) | | `facility` | `Facility` | [Server](#server) | +| `federation` | `Federation` | [Server](#server) | | `endpoints` | `[]Endpoint` | [Server](#server) | | `bin` | `string` | [SimConfig](#simconfig) | | `python` | `string` | [TRTLLMConfig](#trtllmconfig) | @@ -863,6 +867,15 @@ APIKeyFile holds the provider's key, on the control plane. A file rather than the key itself, so the key is in neither the catalog, the store nor the audit log; empty for a provider that asks for none. +### `peer` + +`string` + +Peer names the federated llmgw cluster this model is offered by, for +a model that federation added (`llmgw ctl federation add`) rather +than one written by hand. Such a model is named PEER/MODEL, comes and +goes with the peer's own list, and is sent only while the peer is up. + ## Facility ### `budgets` @@ -932,6 +945,24 @@ one feed carries alone, derated as the facility derates it (usually OtherWatts is drawn in the domain by what llmgw does not place on: switches, storage, machines outside the cluster. +## Federation + +### `name` + +`string` + +Name is this cluster as a peer reads it, in the federated_from tag +sent with a request to a peer set to receive tags (`ctl federation add +--send-tags`). Empty is this machine's host name. + +### `interval` + +`time.Duration` + +Interval is how often each peer is asked whether it is ready, timed, +and what it offers. A peer that does not answer is taken out of +routing until it does. Zero is 10s. + ## GuardModel ### `format` @@ -2482,6 +2513,16 @@ maintenance is `llmgw ctl facility hold`, not configuration. Takes a [Facility](#facility) block. +### `federation` + +`Federation` + +Federation is how this cluster takes part in federation with other +llmgw clusters. The peers themselves are added at runtime, with +`llmgw ctl federation add`, and kept in the store. + +Takes a [Federation](#federation) block. + ### `endpoints` `[]Endpoint` diff --git a/internal/cli/ctl.go b/internal/cli/ctl.go index f829f17..ebe72cf 100644 --- a/internal/cli/ctl.go +++ b/internal/cli/ctl.go @@ -168,6 +168,7 @@ func newCtlCmd() *cobra.Command { newAuditCmd(client, &asJSON), newNotifyCmd(client, &asJSON), newUsageCmd(client, &asJSON), + newFederationCmd(client, &asJSON), newCertCmd(), ) return cmd diff --git a/internal/cli/federation.go b/internal/cli/federation.go new file mode 100644 index 0000000..3368219 --- /dev/null +++ b/internal/cli/federation.go @@ -0,0 +1,211 @@ +package cli + +import ( + "encoding/json" + "fmt" + "net/url" + "strings" + "text/tabwriter" + "time" + + "github.com/spf13/cobra" + + "gitea.mynodes.xyz/ilker/llmgw/internal/control/adminapi" + "gitea.mynodes.xyz/ilker/llmgw/internal/control/federation" +) + +func newFederationCmd(client func() *ctlClient, asJSON *bool) *cobra.Command { + c := &cobra.Command{ + Use: "federation", + Short: "Offer another llmgw cluster's models here, as PEER/MODEL", + Long: "A peer is another llmgw gateway, reached with a key it issued to this\n" + + "cluster. Its models join this catalog as PEER/MODEL and follow its own\n" + + "list; each is sent requests only while the peer's /ready answers, which\n" + + "is checked every federation.interval (10s) and timed. Name one as a\n" + + "local model's fallback to send overflow to the other site:\n" + + " llmgw ctl model fallbacks qwen-7b east/qwen-7b\n\n" + + "A project whose requests stay in the cluster (keys residency\n" + + "--in-cluster), or a policy that keeps nothing, is never sent to a peer.\n" + + "Requests carry x-llmgw-hop, so two clusters federating each other do\n" + + "not send one back and forth.", + } + c.AddCommand(newFederationAddCmd(client), newFederationListCmd(client, asJSON), + newFederationRemoveCmd(client), newFederationSyncCmd(client, asJSON)) + return c +} + +func newFederationAddCmd(client func() *ctlClient) *cobra.Command { + var req federation.AddRequest + c := &cobra.Command{ + Use: "add NAME --url URL --key-file FILE", + Short: "Add another llmgw cluster as a peer, or change one", + Long: "The peer is checked first — its /ready, and its /v1/models with the\n" + + "key — and refused if either fails. The key file is read on the control\n" + + "plane (or env:NAME, vault:PATH#FIELD), at every check, so a rotated\n" + + "key is used without a change here; every control plane needs it.\n\n" + + "--send-tags makes each request sent to the peer carry x-llmgw-tags\n" + + "federated_from=,org=,team=,project=, so the peer's\n" + + "usage can say who here asked. Off unless given: the peer otherwise sees\n" + + "one key, this cluster's, and nothing about who is behind it.", + Example: " llmgw ctl federation add east --url https://llm.east.example --key-file /etc/llmgw/east.key\n" + + " llmgw ctl federation add east --url https://llm.east.example --key-file /etc/llmgw/east.key --send-tags", + Args: cobra.ExactArgs(1), + RunE: func(cmd *cobra.Command, args []string) error { + req.Name = args[0] + body, _ := json.Marshal(req) + var st federation.Status + if err := client().post(cmd.Context(), "/api/v1/federation", body, &st); err != nil { + return err + } + fmt.Fprintf(cmd.OutOrStdout(), "%s added: %d models, %s\n", st.Name, len(st.Models), healthText(st)) + for _, m := range st.Models { + fmt.Fprintln(cmd.OutOrStdout(), " "+m.Name) + } + return nil + }, + } + f := c.Flags() + f.StringVar(&req.URL, "url", "", "the peer gateway's client address, without /v1") + f.StringVar(&req.KeyFile, "key-file", "", "file on the control plane holding the key the peer issued to this cluster") + f.BoolVar(&req.SendTags, "send-tags", false, "tell the peer which cluster and project each request is from") + _ = c.MarkFlagRequired("url") + _ = c.MarkFlagRequired("key-file") + return c +} + +func newFederationListCmd(client func() *ctlClient, asJSON *bool) *cobra.Command { + return &cobra.Command{ + Use: "list", + Short: "Peers, whether each is up, how far away, and the models it offers", + Args: cobra.NoArgs, + RunE: func(cmd *cobra.Command, _ []string) error { + var v adminapi.FederationView + if err := client().get(cmd.Context(), "/api/v1/federation", &v); err != nil { + return err + } + if *asJSON { + return emitJSON(v) + } + out := cmd.OutOrStdout() + fmt.Fprintf(out, "this cluster: %s\n", v.Cluster) + if len(v.Peers) == 0 { + fmt.Fprintln(out, "no peers: llmgw ctl federation add NAME --url URL --key-file FILE") + return nil + } + w := tabwriter.NewWriter(out, 0, 0, 2, ' ', 0) + fmt.Fprintln(w, "PEER\tURL\tHEALTH\tLATENCY\tTAGS\tMODELS") + for _, p := range v.Peers { + tags := "off" + if p.SendTags { + tags = "on" + } + lat := "-" + if p.Latency > 0 { + lat = fmt.Sprintf("%.1fms", p.LatencyMs) + } + fmt.Fprintf(w, "%s\t%s\t%s\t%s\t%s\t%s\n", p.Name, p.URL, healthText(p), lat, tags, modelList(p)) + } + return w.Flush() + }, + } +} + +func newFederationRemoveCmd(client func() *ctlClient) *cobra.Command { + return &cobra.Command{ + Use: "remove NAME", + Short: "Remove a peer and its models", + Long: "Refused while a model here falls back to one of the peer's: change its\n" + + "fallbacks first (llmgw ctl model fallbacks).", + Args: cobra.ExactArgs(1), + RunE: func(cmd *cobra.Command, args []string) error { + var out adminapi.Removed + if err := client().delete(cmd.Context(), "/api/v1/federation/"+url.PathEscape(args[0]), &out); err != nil { + return err + } + fmt.Fprintf(cmd.OutOrStdout(), "%s removed\n", out.Removed) + return nil + }, + } +} + +func newFederationSyncCmd(client func() *ctlClient, asJSON *bool) *cobra.Command { + return &cobra.Command{ + Use: "sync NAME", + Short: "Check a peer now and take its current models, rather than at the next interval", + Args: cobra.ExactArgs(1), + RunE: func(cmd *cobra.Command, args []string) error { + var st federation.Status + if err := client().post(cmd.Context(), "/api/v1/federation/"+url.PathEscape(args[0])+"/sync", nil, &st); err != nil { + return err + } + if *asJSON { + return emitJSON(st) + } + fmt.Fprintf(cmd.OutOrStdout(), "%s: %s, %s\n", st.Name, healthText(st), modelList(st)) + return nil + }, + } +} + +// healthText is "up" or "down: why", with how long ago it was checked. +func healthText(st federation.Status) string { + ago := "" + if !st.Checked.IsZero() { + ago = fmt.Sprintf(" (%s ago)", time.Since(st.Checked).Round(time.Second)) + } + if st.Up { + return "up" + ago + } + return "down" + ago + ": " + st.Error +} + +// modelList is the peer's models by their names here, a withdrawn one +// marked as such. +func modelList(st federation.Status) string { + if len(st.Models) == 0 { + return "-" + } + names := make([]string, 0, len(st.Models)) + for _, m := range st.Models { + n := m.Name + if m.Withdrawn { + n += " (withdrawn)" + } + names = append(names, n) + } + return strings.Join(names, ", ") +} + +func newModelFallbacksCmd(client func() *ctlClient) *cobra.Command { + var clear bool + c := &cobra.Command{ + Use: "fallbacks MODEL [FALLBACK...]", + Short: "Set the models a model falls back to when it has no capacity", + Long: "Replaces the model's fallbacks, tried in order when nothing serving it\n" + + "has room. A federation peer's model, PEER/MODEL, sends the overflow to\n" + + "the other cluster. A model added at runtime only; one declared in\n" + + "server.yaml has its fallbacks there. --clear takes them all away.", + Example: " llmgw ctl model fallbacks qwen-7b east/qwen-7b\n" + + " llmgw ctl model fallbacks qwen-7b qwen-3b east/qwen-7b\n" + + " llmgw ctl model fallbacks qwen-7b --clear", + Args: cobra.MinimumNArgs(1), + RunE: func(cmd *cobra.Command, args []string) error { + if len(args) == 1 && !clear { + return fmt.Errorf("name the fallbacks, or give --clear to take them away") + } + body, _ := json.Marshal(adminapi.FallbacksRequest{Fallbacks: args[1:]}) + var v adminapi.FallbacksView + if err := client().put(cmd.Context(), "/api/v1/models/"+args[0]+"/fallbacks", body, &v); err != nil { + return err + } + if len(v.Fallbacks) == 0 { + fmt.Fprintf(cmd.OutOrStdout(), "%s has no fallbacks\n", v.Model) + return nil + } + fmt.Fprintf(cmd.OutOrStdout(), "%s falls back to %s\n", v.Model, strings.Join(v.Fallbacks, ", then ")) + return nil + }, + } + c.Flags().BoolVar(&clear, "clear", false, "take every fallback away") + return c +} diff --git a/internal/cli/model.go b/internal/cli/model.go index e80be9b..29ad603 100644 --- a/internal/cli/model.go +++ b/internal/cli/model.go @@ -38,7 +38,7 @@ func newModelCmd(client func() *ctlClient, asJSON *bool) *cobra.Command { newModelServeCmd(client), newModelUpdatesCmd(client, asJSON), newModelUpdateCmd(client), newModelRollbackCmd(client), newModelOriginCmd(client), newModelCanaryCmd(client), newModelPromoteCmd(client), newModelCanariesCmd(client, asJSON), - newModelGateCmd(client)) + newModelGateCmd(client), newModelFallbacksCmd(client)) return c } diff --git a/internal/cli/server.go b/internal/cli/server.go index 7fd6475..821e5f8 100644 --- a/internal/cli/server.go +++ b/internal/cli/server.go @@ -2,7 +2,9 @@ package cli import ( "context" + "crypto/rand" "crypto/tls" + "encoding/hex" "errors" "fmt" "log/slog" @@ -41,6 +43,7 @@ import ( "gitea.mynodes.xyz/ilker/llmgw/internal/control/agentsvc" "gitea.mynodes.xyz/ilker/llmgw/internal/control/catalog" "gitea.mynodes.xyz/ilker/llmgw/internal/control/fabrictest" + "gitea.mynodes.xyz/ilker/llmgw/internal/control/federation" "gitea.mynodes.xyz/ilker/llmgw/internal/control/fleetmetrics" "gitea.mynodes.xyz/ilker/llmgw/internal/control/modelsvc" "gitea.mynodes.xyz/ilker/llmgw/internal/control/reconciler" @@ -176,6 +179,21 @@ func newServerCmd() *cobra.Command { rt.Start(ctx) defer rt.Stop() + // Which cluster this is, to federated peers: the id in the hop + // header that stops a request going round two clusters that + // federate each other, and the name in the tags a peer may be + // sent. The id is the store's, so every control plane of one + // cluster says the same. + clusterID, err := federationID(ctx, rt.Store()) + if err != nil { + return err + } + clusterName := cfg.Federation.Name + if clusterName == "" { + clusterName = host + } + rt.SetFederationIdentity(clusterID, clusterName) + // Webhooks for what needs a person, when server.yaml names any. notifier, err := notify.New(ctx, cfg.Notify, log) if err != nil { @@ -333,6 +351,21 @@ func newServerCmd() *cobra.Command { if st := rt.Store(); st != nil { svc := modelsvc.New(cfg.Models, st, admin, log) svc.SetCheck(servableCheck(cfg.WantedEngines()["llamacpp"])) + // Other llmgw clusters' models, offered as PEER/MODEL: + // read before the catalog is built, so a model that falls + // back to one is whole at startup. + fed := federation.New(federation.Config{Self: clusterID, Interval: cfg.Federation.Interval}, + st, svc, admin.Catalog, log) + if err := fed.Load(ctx); err != nil { + log.Warn("could not read the federation peers; continuing without them", "err", err) + } + svc.SetFederated(fed.Models) + admin.SetPeers(fed) + api.SetFederation(fed) + if err := rt.RegisterMetrics(fed); err != nil { + return err + } + go fed.Run(ctx) // Rebuild the catalog from configuration plus what was // stored, and put that in force — otherwise models added // before the last restart would be silently absent until @@ -965,3 +998,25 @@ func withLimits(s *http.Server) *http.Server { s.MaxHeaderBytes = 1 << 20 return s } + +// federationID is this cluster's id among federated peers: kept in the store, +// so every control plane of the cluster and every restart says the same. A +// gateway with no store makes one for its lifetime, which is all a router +// with no peers of its own needs to refuse a request that comes back to it. +func federationID(ctx context.Context, st *storepkg.DB) (string, error) { + if st == nil { + b := make([]byte, 8) + if _, err := rand.Read(b); err != nil { + return "", err + } + return hex.EncodeToString(b), nil + } + secret, err := st.ClusterSecret(ctx, "cluster-id") + if err != nil { + return "", fmt.Errorf("this cluster's id for federation: %w", err) + } + // A prefix is plenty to tell clusters apart, and the id is sent to + // peers: nothing is derived from it, but there is no reason to send + // more of a secret than needed. + return secret[:16], nil +} diff --git a/internal/cli/tui_console.go b/internal/cli/tui_console.go index 02bd929..78c5023 100644 --- a/internal/cli/tui_console.go +++ b/internal/cli/tui_console.go @@ -42,6 +42,7 @@ const ( cReplicas cUsage cQuotas + cFederation cMirror cJobs cLogs @@ -61,7 +62,7 @@ const ( cCount ) -var consoleLabels = [cCount]string{"Overview", "Monitor", "Nodes", "GPUs", "Models", "Replicas", "Usage", "Quotas", "Mirror", "Jobs", "Logs", "Accounts", "API keys", "Organizations", "Audit", "Manual", +var consoleLabels = [cCount]string{"Overview", "Monitor", "Nodes", "GPUs", "Models", "Replicas", "Usage", "Quotas", "Federation", "Mirror", "Jobs", "Logs", "Accounts", "API keys", "Organizations", "Audit", "Manual", "server.yaml", "Add a node", "Replicate the store", "Serve a model", "Upgrade the cluster", "Restart all replicas", "Back up now", "Leave"} @@ -79,7 +80,10 @@ type consoleData struct { replicas []adminapi.ReplicaView jobs []modelserve.Job audit []store.AuditEvent - err error + // federation is the other llmgw clusters whose models are offered + // here; empty on a control plane without a store. + federation adminapi.FederationView + err error } type ( @@ -238,6 +242,7 @@ func (m consoleModel) fetch() tea.Cmd { _ = c.get(ctx, "/api/v1/replicas", &d.replicas) _ = c.get(ctx, "/api/v1/jobs", &d.jobs) _ = c.get(ctx, "/api/v1/audit?since=7d&limit=300", &d.audit) + _ = c.get(ctx, "/api/v1/federation", &d.federation) return consoleDataMsg(d) } } @@ -494,6 +499,9 @@ func (m consoleModel) key(k tea.KeyMsg) (tea.Model, tea.Cmd) { case cAudit: m.dialog = "error" m.dialogTitle, m.dialogText = auditDialog(m.data.audit[m.row]) + case cFederation: + m.dialog = "error" + m.dialogTitle, m.dialogText = peerDialog(m.data.federation.Peers[m.row]) case cServe: return m, m.fetchRepo(m.hf.models[m.row].ID) case cAccounts: @@ -908,6 +916,8 @@ func (m consoleModel) rowCount() int { return len(m.data.jobs) case cAudit: return len(m.data.audit) + case cFederation: + return len(m.data.federation.Peers) case cServe: return len(m.hf.models) case cAccounts: @@ -1149,6 +1159,8 @@ func (m consoleModel) panel() (title, body, keys string) { return m.jobsPanel(pw) case cAudit: return m.auditPanel(pw) + case cFederation: + return m.federationPanel(pw) case cMonitor: return m.monitorPanel(pw) case cServe: diff --git a/internal/cli/tui_federation.go b/internal/cli/tui_federation.go new file mode 100644 index 0000000..b38417b --- /dev/null +++ b/internal/cli/tui_federation.go @@ -0,0 +1,80 @@ +package cli + +import ( + "fmt" + "strings" + "time" + + "gitea.mynodes.xyz/ilker/llmgw/internal/control/federation" +) + +// The console's federation view: the other llmgw clusters whose models are +// offered here, whether each is up and how far away, as `llmgw ctl +// federation list` shows them. Peers are added and removed with ctl, which +// needs a key file on this machine that a form here could not check. + +func (m consoleModel) federationPanel(pw int) (string, string, string) { + keys := "→ or Enter open · ↑↓ choose · Q leave" + if m.inPanel { + keys = "↑↓ peer · Enter its models · ← back · Q leave" + } + v := m.data.federation + head := "\n " + sDim.Render("Other llmgw clusters whose models are offered here as PEER/MODEL.") + if v.Cluster != "" { + head += "\n " + sDim.Render("This cluster's id: "+v.Cluster) + } + head += "\n\n" + if len(v.Peers) == 0 { + return "Federation", head + " " + sDim.Render("No peers. Add one with: llmgw ctl federation add NAME --url URL --key-file FILE"), keys + } + var rows [][]string + for _, p := range v.Peers { + health := sOK.Render("up") + if !p.Up { + health = sBad.Render("down") + } + lat := "-" + if p.Latency > 0 { + lat = fmt.Sprintf("%.1fms", p.LatencyMs) + } + offered := 0 + for _, md := range p.Models { + if !md.Withdrawn { + offered++ + } + } + rows = append(rows, []string{p.Name, health, lat, fmt.Sprint(offered), p.URL}) + } + return "Federation", head + table([]string{"PEER", "HEALTH", "LATENCY", "MODELS", "URL"}, rows, m.row, pw, m.inPanel), keys +} + +// peerDialog is one peer whole: how it was last found, and its models. +func peerDialog(p federation.Status) (string, string) { + var b strings.Builder + fmt.Fprintf(&b, "%s\n", p.URL) + if p.Up { + fmt.Fprintf(&b, "%s, /ready in %.1fms", sOK.Render("up"), p.LatencyMs) + } else { + fmt.Fprintf(&b, "%s: %s", sBad.Render("down"), wrap(p.Error, 80, " ")) + } + if !p.Checked.IsZero() { + fmt.Fprintf(&b, ", checked %s ago", time.Since(p.Checked).Round(time.Second)) + } + tags := "not told who asks" + if p.SendTags { + tags = "told this cluster and the project of each request (x-llmgw-tags)" + } + fmt.Fprintf(&b, "\ncluster %s, key %s; %s\n\n", p.Cluster, p.KeyFile, tags) + if len(p.Models) == 0 { + b.WriteString(sDim.Render("offers this cluster's key no models") + "\n") + } + for _, md := range p.Models { + line := md.Name + if md.Withdrawn { + line += sDim.Render(" withdrawn by the peer; kept while a fallback here names it") + } + b.WriteString(" " + line + "\n") + } + fmt.Fprintf(&b, "\n%s", sDim.Render("llmgw ctl federation sync|remove "+p.Name+" · any key")) + return p.Name, b.String() +} diff --git a/internal/cli/tui_federation_test.go b/internal/cli/tui_federation_test.go new file mode 100644 index 0000000..e09beb9 --- /dev/null +++ b/internal/cli/tui_federation_test.go @@ -0,0 +1,37 @@ +package cli + +import ( + "strings" + "testing" + "time" + + tea "github.com/charmbracelet/bubbletea" + + "gitea.mynodes.xyz/ilker/llmgw/internal/control/adminapi" + "gitea.mynodes.xyz/ilker/llmgw/internal/control/federation" +) + +func TestTheConsoleShowsEachPeerAndOpensItsModels(t *testing.T) { + m, _ := consoleWith(t, cFederation) + m.data.federation = adminapi.FederationView{Cluster: "c-west", Peers: []federation.Status{ + {Name: "east", URL: "https://east.example", Cluster: "c-east", + Health: federation.Health{Up: true, Latency: 12 * time.Millisecond, Checked: time.Now()}, + LatencyMs: 12, Models: []federation.ModelStatus{{Name: "east/qwen", Upstream: "qwen"}, + {Name: "east/old", Upstream: "old", Withdrawn: true}}}, + {Name: "north", URL: "https://north.example", Health: federation.Health{Error: "the peer did not answer"}}, + }} + var tm tea.Model = m + view := tm.View() + for _, want := range []string{"east", "12.0ms", "north", "down", "c-west"} { + if !strings.Contains(view, want) { + t.Errorf("the federation view has no %q:\n%s", want, view) + } + } + tm = press(tm, "enter") + view = tm.View() + for _, want := range []string{"east/qwen", "east/old", "withdrawn", "c-east"} { + if !strings.Contains(view, want) { + t.Errorf("the peer's dialog has no %q:\n%s", want, view) + } + } +} diff --git a/internal/config/config.go b/internal/config/config.go index d444152..7d81804 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -186,6 +186,11 @@ type Server struct { // maintenance is `llmgw ctl facility hold`, not configuration. Facility Facility `koanf:"facility"` + // Federation is how this cluster takes part in federation with other + // llmgw clusters. The peers themselves are added at runtime, with + // `llmgw ctl federation add`, and kept in the store. + Federation Federation `koanf:"federation"` + // Endpoints is the phase-1 static replica list. It is only consulted when // no model declares a profile, which is what makes a router-only smoke test // against sim-engines possible without a control plane. @@ -269,6 +274,18 @@ func (f Facility) Policy() (facility.Policy, error) { return p, nil } +// Federation is this cluster's side of federation with other llmgw clusters. +type Federation struct { + // Name is this cluster as a peer reads it, in the federated_from tag + // sent with a request to a peer set to receive tags (`ctl federation add + // --send-tags`). Empty is this machine's host name. + Name string `koanf:"name"` + // Interval is how often each peer is asked whether it is ready, timed, + // and what it offers. A peer that does not answer is taken out of + // routing until it does. Zero is 10s. + Interval time.Duration `koanf:"interval"` +} + // Monitoring points the dashboard at a Prometheus holding this cluster's // history. // diff --git a/internal/control/adminapi/api.go b/internal/control/adminapi/api.go index 50807b3..a328e8c 100644 --- a/internal/control/adminapi/api.go +++ b/internal/control/adminapi/api.go @@ -33,6 +33,7 @@ import ( "gitea.mynodes.xyz/ilker/llmgw/internal/control/catalog" "gitea.mynodes.xyz/ilker/llmgw/internal/control/fabrics" "gitea.mynodes.xyz/ilker/llmgw/internal/control/facility" + "gitea.mynodes.xyz/ilker/llmgw/internal/control/federation" "gitea.mynodes.xyz/ilker/llmgw/internal/control/inventory" "gitea.mynodes.xyz/ilker/llmgw/internal/control/nvlink" "gitea.mynodes.xyz/ilker/llmgw/internal/control/reconciler" @@ -59,6 +60,9 @@ type API struct { history History // audit reads the audit log. audit AuditReader + // federation is the other llmgw clusters whose models are offered here; + // nil without a store. + federation *federation.Service // guard is the guardrail policies' library and attachments. guard *GuardAdmin // revoke records a removed node, so its certificate is refused. @@ -177,6 +181,7 @@ func (a *API) Register(mux Mux) { mux.HandleFunc("POST /api/v1/models/{name}/rollback", a.rollbackModel) a.registerCanary(mux) a.registerEval(mux) + a.registerFederation(mux) mux.HandleFunc("POST /api/v1/models/{name}/origin", a.setOrigin) mux.HandleFunc("GET /api/v1/plan", a.planModel) mux.HandleFunc("POST /api/v1/metrics/range", a.metricsBatch) @@ -1839,6 +1844,9 @@ func externalLabel(m catalog.Model) string { if !m.IsExternal() { return "" } + if m.IsPeer() { + return "peer " + m.External.Peer + " · " + m.External.Model + } host := m.External.URL if u, err := url.Parse(host); err == nil && u.Host != "" { host = u.Host diff --git a/internal/control/adminapi/federation.go b/internal/control/adminapi/federation.go new file mode 100644 index 0000000..480ea57 --- /dev/null +++ b/internal/control/adminapi/federation.go @@ -0,0 +1,123 @@ +package adminapi + +import ( + "context" + "encoding/json" + "io" + "net/http" + "strings" + + "gitea.mynodes.xyz/ilker/llmgw/internal/control/catalog" + "gitea.mynodes.xyz/ilker/llmgw/internal/control/federation" +) + +// Federation: other llmgw clusters whose models this one offers, as +// PEER/MODEL, and the fallbacks that send overflow to them. + +// SetFederation gives the API the peers to list and change. +func (a *API) SetFederation(f *federation.Service) { a.federation = f } + +// FederationView is this cluster's id and its peers. +type FederationView struct { + // Cluster is this cluster's own id, which its peers see in the hop + // header. + Cluster string `json:"cluster"` + Peers []federation.Status `json:"peers"` +} + +// FallbacksRequest replaces a model's fallbacks; empty takes them all away. +type FallbacksRequest struct { + Fallbacks []string `json:"fallbacks"` +} + +// FallbacksView is a model's fallbacks, as they are now. +type FallbacksView struct { + Model string `json:"model"` + Fallbacks []string `json:"fallbacks"` +} + +type fallbackSetter interface { + SetFallbacks(ctx context.Context, name string, fallbacks []string, by string) (catalog.Model, error) +} + +func (a *API) registerFederation(mux Mux) { + need := func(w http.ResponseWriter) bool { + if a.federation == nil { + writeError(w, http.StatusNotImplemented, "this control plane has no store to keep federation peers in") + return false + } + return true + } + mux.HandleFunc("GET /api/v1/federation", func(w http.ResponseWriter, _ *http.Request) { + if !need(w) { + return + } + writeJSON(w, http.StatusOK, FederationView{Cluster: a.federation.Self(), Peers: a.federation.List()}) + }) + mux.HandleFunc("POST /api/v1/federation", func(w http.ResponseWriter, r *http.Request) { + if !need(w) { + return + } + var req federation.AddRequest + if err := json.NewDecoder(io.LimitReader(r.Body, 1<<14)).Decode(&req); err != nil { + writeError(w, http.StatusBadRequest, "the request is not valid JSON: "+err.Error()) + return + } + st, err := a.federation.Add(r.Context(), req, actorOf(r)) + if err != nil { + writeError(w, http.StatusBadRequest, err.Error()) + return + } + writeJSON(w, http.StatusCreated, st) + }) + mux.HandleFunc("DELETE /api/v1/federation/{peer}", func(w http.ResponseWriter, r *http.Request) { + if !need(w) { + return + } + name := r.PathValue("peer") + if err := a.federation.Remove(r.Context(), name); err != nil { + writeError(w, http.StatusBadRequest, err.Error()) + return + } + writeJSON(w, http.StatusOK, Removed{Removed: name}) + }) + mux.HandleFunc("POST /api/v1/federation/{peer}/sync", func(w http.ResponseWriter, r *http.Request) { + if !need(w) { + return + } + st, err := a.federation.Sync(r.Context(), r.PathValue("peer")) + if err != nil { + writeError(w, http.StatusNotFound, err.Error()) + return + } + writeJSON(w, http.StatusOK, st) + }) + mux.HandleFunc("PUT /api/v1/models/{name}/fallbacks", func(w http.ResponseWriter, r *http.Request) { + setter, ok := a.models.(fallbackSetter) + if !ok { + writeError(w, http.StatusNotImplemented, "this control plane has no store to keep the setting in") + return + } + var req FallbacksRequest + if err := json.NewDecoder(io.LimitReader(r.Body, 1<<14)).Decode(&req); err != nil { + writeError(w, http.StatusBadRequest, "the request is not valid JSON: "+err.Error()) + return + } + var fbs []string + for _, f := range req.Fallbacks { + if f = strings.TrimSpace(f); f != "" { + fbs = append(fbs, f) + } + } + m, err := setter.SetFallbacks(r.Context(), r.PathValue("name"), fbs, actorOf(r)) + if err != nil { + writeError(w, http.StatusBadRequest, err.Error()) + return + } + out := m.Fallbacks + if out == nil { + out = []string{} + } + writeJSON(w, http.StatusOK, FallbacksView{Model: m.Name, Fallbacks: out}) + }) +} diff --git a/internal/control/adminapi/routes.go b/internal/control/adminapi/routes.go index b199ef5..cdb9152 100644 --- a/internal/control/adminapi/routes.go +++ b/internal/control/adminapi/routes.go @@ -8,6 +8,7 @@ import ( "gitea.mynodes.xyz/ilker/llmgw/internal/bench/eval" "gitea.mynodes.xyz/ilker/llmgw/internal/control/catalog" "gitea.mynodes.xyz/ilker/llmgw/internal/control/fabrics" + "gitea.mynodes.xyz/ilker/llmgw/internal/control/federation" "gitea.mynodes.xyz/ilker/llmgw/internal/control/modelserve" "gitea.mynodes.xyz/ilker/llmgw/internal/control/nvlink" "gitea.mynodes.xyz/ilker/llmgw/internal/guard" @@ -56,14 +57,15 @@ type Param struct { // PathParams describes the parameters in Routes' paths, by the segment // before them and their name. var PathParams = map[string]string{ - "models/{name}": "The model's name in the catalog.", - "nodes/{id}": "The node's ID, as `llmgw ctl nodes` lists it.", - "jobs/{id}": "The job's ID, as starting it answered.", - "runs/{id}": "The quality run's ID, as starting it answered.", - "requests/{id}": "The request's ID: the x-request-id a client was answered with.", - "fabrics/{id}": "The physical fabric's ID, as GET /api/v1/fabrics lists it.", - "holds/{domain}": "The domain as level:name, rack:r12-07, as `llmgw ctl facility` lists it.", - "quotas/{scope}": "The scope: org:NAME, team:ORG/TEAM, project:ORG/TEAM/PROJECT, or their IDs after the colon.", + "models/{name}": "The model's name in the catalog.", + "nodes/{id}": "The node's ID, as `llmgw ctl nodes` lists it.", + "jobs/{id}": "The job's ID, as starting it answered.", + "runs/{id}": "The quality run's ID, as starting it answered.", + "requests/{id}": "The request's ID: the x-request-id a client was answered with.", + "fabrics/{id}": "The physical fabric's ID, as GET /api/v1/fabrics lists it.", + "quotas/{scope}": "The scope: org:NAME, team:ORG/TEAM, project:ORG/TEAM/PROJECT, or their IDs after the colon.", + "federation/{peer}": "The federation peer's name, as `llmgw ctl federation list` lists it.", + "holds/{domain}": "The domain as level:name, rack:r12-07, as `llmgw ctl facility` lists it.", } // Routes is every endpoint Register mounts. @@ -222,6 +224,21 @@ var Routes = []Route{ {Method: "DELETE", Path: "/api/v1/models/{name}/eval-gate", Summary: "Stop holding a model's updates to a suite", Response: EvalGateView{}}, + // Federation: other llmgw clusters as providers. + {Method: "GET", Path: "/api/v1/federation", Summary: "This cluster's id, and its federation peers with their health and models", + Response: FederationView{}}, + {Method: "POST", Path: "/api/v1/federation", Summary: "Add another llmgw cluster as a peer, or change one", + Description: "The peer is checked first: its /ready, timed, and its /v1/models with the key. Its models join the catalog as PEER/MODEL.", + Request: federation.AddRequest{}, Response: federation.Status{}, Status: http.StatusCreated}, + {Method: "DELETE", Path: "/api/v1/federation/{peer}", Summary: "Remove a federation peer and its models", + Description: "Refused while a model here falls back to one of the peer's.", + Response: Removed{}}, + {Method: "POST", Path: "/api/v1/federation/{peer}/sync", Summary: "Check a peer now, rather than at the next interval", + Response: federation.Status{}}, + {Method: "PUT", Path: "/api/v1/models/{name}/fallbacks", Summary: "Replace the models a model falls back to when it has no capacity", + Description: "A model added at runtime only. A federation peer's model, PEER/MODEL, sends the overflow to the other cluster.", + Request: FallbacksRequest{}, Response: FallbacksView{}}, + // Tenancy: quotas and guardrails. {Method: "GET", Path: "/api/v1/quotas", Summary: "Every capacity quota, with what its scope holds now", Response: []QuotaView{}}, diff --git a/internal/control/catalog/catalog.go b/internal/control/catalog/catalog.go index 1bc2c45..aab717c 100644 --- a/internal/control/catalog/catalog.go +++ b/internal/control/catalog/catalog.go @@ -1083,8 +1083,16 @@ type External struct { // rather than the key itself, so the key is in neither the catalog, the // store nor the audit log; empty for a provider that asks for none. APIKeyFile string `json:"api_key_file,omitempty" koanf:"api_key_file"` + // Peer names the federated llmgw cluster this model is offered by, for + // a model that federation added (`llmgw ctl federation add`) rather + // than one written by hand. Such a model is named PEER/MODEL, comes and + // goes with the peer's own list, and is sent only while the peer is up. + Peer string `json:"peer,omitempty" koanf:"peer"` } +// IsPeer reports whether a model is offered by a federated cluster. +func (m Model) IsPeer() bool { return m.External != nil && m.External.Peer != "" } + func (e *External) validate(model string, profiles int) error { if profiles > 0 { return fmt.Errorf("model %q: an external model is served elsewhere and has no profiles", model) diff --git a/internal/control/control.go b/internal/control/control.go index 7fcb9b9..bf13c85 100644 --- a/internal/control/control.go +++ b/internal/control/control.go @@ -89,6 +89,7 @@ type Controller struct { disp reconciler.Dispatcher sink EndpointSink usage UsageSource + peers Peers mu sync.Mutex last string // fingerprint of the last published list @@ -148,6 +149,27 @@ func (c *Controller) collectUsage() { } } +// Peers says which federated clusters' models may be sent requests now. +// +// Federation keeps it: whether each peer answered its last check, and +// whether it still offers the model. A peer model is in the catalog either +// way — a local model may name it as a fallback — and is published only +// while this says yes, which is what takes a peer that is down out of +// routing within a publish. +type Peers interface { + // PeerRoute is the peer's own cluster id and whether requests sent to + // it carry tags, and false when the model is not to be sent anything. + PeerRoute(peer, upstreamModel string) (cluster string, sendTags, ok bool) +} + +// SetPeers supplies the federation's view of its peers. Without one, no +// peer model is published. +func (c *Controller) SetPeers(p Peers) { + c.mu.Lock() + c.peers = p + c.mu.Unlock() +} + // LoadSource reports each model's replicas' load, keyed by catalog name. type LoadSource interface { ModelLoad() map[string]reconciler.Load @@ -352,11 +374,25 @@ func (c *Controller) endpoints() []endpoint.Info { // one each, with the provider's key read from its file every time, so a key // rotated on disk is used from the next publish on. func (c *Controller) externalEndpoints() []endpoint.Info { + c.mu.Lock() + peers := c.peers + c.mu.Unlock() var out []endpoint.Info for _, m := range c.cat.Load().Models() { if !m.IsExternal() { continue } + var cluster string + var tags bool + if m.IsPeer() { + ok := false + if peers != nil { + cluster, tags, ok = peers.PeerRoute(m.External.Peer, m.External.Model) + } + if !ok { + continue + } + } key := "" if f := m.External.APIKeyFile; f != "" { b, err := secrets.Read(context.Background(), f) @@ -366,11 +402,18 @@ func (c *Controller) externalEndpoints() []endpoint.Info { } key = strings.TrimSpace(string(b)) } + // A peer's endpoint says so in its id, which is what usage records + // as the replica that answered. + id := "external:" + m.Name + if m.IsPeer() { + id = "peer:" + m.Name + } out = append(out, endpoint.Info{ - ID: "external:" + m.Name, URL: strings.TrimRight(m.External.URL, "/"), + ID: id, URL: strings.TrimRight(m.External.URL, "/"), Models: append([]string{m.Name}, m.Aliases...), APIKey: key, Engine: endpoint.EngineExternal, Pool: "external", Weight: 1, UpstreamModel: m.External.Model, + Peer: m.External.Peer, PeerCluster: cluster, PeerTags: tags, }) } return out @@ -380,8 +423,8 @@ func (c *Controller) externalEndpoints() []endpoint.Info { func fingerprint(infos []endpoint.Info) string { var b strings.Builder for _, i := range infos { - fmt.Fprintf(&b, "%s|%s|%s|%s|%v|%g|%d|%s|%x;", i.ID, i.URL, i.Engine, i.Pool, i.Models, i.Weight, i.Priority, - i.UpstreamModel, sha256.Sum256([]byte(i.APIKey))) + fmt.Fprintf(&b, "%s|%s|%s|%s|%v|%g|%d|%s|%s|%s|%t|%x;", i.ID, i.URL, i.Engine, i.Pool, i.Models, i.Weight, + i.Priority, i.UpstreamModel, i.Peer, i.PeerCluster, i.PeerTags, sha256.Sum256([]byte(i.APIKey))) } return b.String() } diff --git a/internal/control/federation/federation.go b/internal/control/federation/federation.go new file mode 100644 index 0000000..193dc13 --- /dev/null +++ b/internal/control/federation/federation.go @@ -0,0 +1,661 @@ +// Package federation makes other llmgw clusters providers of this one. +// +// A peer is another llmgw gateway, reached with a key it issued to this +// cluster. Its models are offered here as external models named PEER/MODEL, +// kept in step with its own /v1/models, and sent requests only while its +// /ready answers. A local model names one as a fallback to send overflow to +// the other site. +// +// Always PEER/MODEL, never the peer's own name for a model, even when nothing +// here uses that name. A name that left the cluster or stayed depending on +// what the catalog happened to hold would move traffic between sites when +// somebody added an unrelated model; the prefix also says, on every usage row +// and every x-llmgw-served-model header, which site answered. +// +// Everything that keeps a request inside the cluster already applies: a peer +// model is an external model, so a project whose requests stay in the +// cluster, or a policy that keeps nothing, is never sent to one. +package federation + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "io" + "log/slog" + "net/http" + "net/url" + "regexp" + "slices" + "strings" + "sync" + "time" + + "gitea.mynodes.xyz/ilker/llmgw/internal/control/catalog" + "gitea.mynodes.xyz/ilker/llmgw/internal/secrets" + "gitea.mynodes.xyz/ilker/llmgw/internal/store" +) + +// ClusterHeader is the response header in which a gateway says its own +// cluster id, on /v1/models. Federation reads a peer's from it, to know the +// peer in a request's hop header. +const ClusterHeader = "x-llmgw-cluster" + +// Store is the persistence federation needs. +type Store interface { + PutFederationPeer(ctx context.Context, p store.FederationPeer) error + SetFederationPeerModels(ctx context.Context, name, cluster string, models json.RawMessage, at time.Time) error + DeleteFederationPeer(ctx context.Context, name string) error + FederationPeers(ctx context.Context) ([]store.FederationPeer, error) +} + +// Catalog is what puts a changed peer list in force: the model service. +type Catalog interface { + Refresh(ctx context.Context) error +} + +// Config tunes federation. +type Config struct { + // Self is this cluster's id. A peer that says the same is this cluster, + // reached by another address, and is refused. + Self string + // Interval is how often each peer is checked; zero is 10s. + Interval time.Duration + // Timeout bounds one check of one peer; zero is 5s. + Timeout time.Duration + // Client makes the checks; nil is a plain one. + Client *http.Client +} + +// PeerModel is what a peer says about one model it offers. +type PeerModel struct { + // ID is the peer's name for it, and what it is asked for. + ID string `json:"id"` + Description string `json:"description,omitempty"` + ContextLength int `json:"context_length,omitempty"` + MaxOutput int `json:"max_output,omitempty"` + Capabilities catalog.Capabilities `json:"capabilities,omitempty"` + // Withdrawn is a model the peer no longer offers, kept in the catalog + // because a model here still falls back to it: removing it would make + // that model invalid. Nothing is sent to it, and it goes when nothing + // names it any more. + Withdrawn bool `json:"withdrawn,omitempty"` +} + +// Service keeps the peers, their health and their models. +type Service struct { + cfg Config + store Store + catalog Catalog + // current is the catalog in force, read to tell which of a peer's + // models a local model names. + current func() *catalog.Catalog + log *slog.Logger + + // op serialises what changes the peer set or a peer's models, so two + // rebuilds cannot land in the wrong order. + op sync.Mutex + + mu sync.Mutex + peers map[string]*peer +} + +type peer struct { + rec store.FederationPeer + models []PeerModel + health Health +} + +// Health is how a peer answered its last check. +type Health struct { + Up bool `json:"up"` + // Checked is when; zero before the first check. + Checked time.Time `json:"checked,omitempty"` + // Latency is how long the peer took to answer /ready: the network to + // the other site and the peer's front, not a model's speed. + Latency time.Duration `json:"latency_ns,omitempty"` + // Error is why the peer is down, empty while it is up. + Error string `json:"error,omitempty"` +} + +// New builds the service. Load reads the peers kept in the store. +func New(cfg Config, st Store, cat Catalog, current func() *catalog.Catalog, log *slog.Logger) *Service { + if cfg.Interval <= 0 { + cfg.Interval = 10 * time.Second + } + if cfg.Timeout <= 0 { + cfg.Timeout = 5 * time.Second + } + if cfg.Client == nil { + cfg.Client = &http.Client{} + } + if log == nil { + log = slog.Default() + } + return &Service{cfg: cfg, store: st, catalog: cat, current: current, log: log, peers: map[string]*peer{}} +} + +// Self is this cluster's id. +func (s *Service) Self() string { return s.cfg.Self } + +// Load reads the peers and the models they last offered from the store, so +// the catalog built at startup has them before any peer is asked again. +// Every peer reads as down until its first check. +func (s *Service) Load(ctx context.Context) error { + recs, err := s.store.FederationPeers(ctx) + if err != nil { + return err + } + s.mu.Lock() + defer s.mu.Unlock() + for _, r := range recs { + var models []PeerModel + if err := json.Unmarshal(r.Models, &models); err != nil { + s.log.Warn("a federation peer's models could not be read; they come back at its next check", + "peer", r.Name, "err", err) + } + s.peers[r.Name] = &peer{rec: r, models: models, health: Health{Error: "not checked yet"}} + } + return nil +} + +// Models are the peers' models as catalog entries, withdrawn ones included: +// what the model service adds to the catalog. +func (s *Service) Models() []catalog.Model { + s.mu.Lock() + defer s.mu.Unlock() + var out []catalog.Model + for _, name := range s.namesLocked() { + p := s.peers[name] + for _, m := range p.models { + out = append(out, catalogModel(p.rec, m)) + } + } + return out +} + +func (s *Service) namesLocked() []string { + names := make([]string, 0, len(s.peers)) + for n := range s.peers { + names = append(names, n) + } + slices.Sort(names) + return names +} + +// catalogModel is a peer's model as an entry of this catalog. +func catalogModel(rec store.FederationPeer, m PeerModel) catalog.Model { + desc := "on " + rec.Name + if m.Description != "" { + desc += ": " + m.Description + } + return catalog.Model{ + Name: rec.Name + "/" + m.ID, Source: "peer:" + rec.Name, Description: desc, + ContextLength: m.ContextLength, MaxOutput: m.MaxOutput, Capabilities: m.Capabilities, + External: &catalog.External{URL: rec.URL, Model: m.ID, APIKeyFile: rec.KeyFile, Peer: rec.Name}, + } +} + +// PeerRoute says whether a peer's model may be sent requests now: the peer +// answered its last check and still offers it. The controller publishes the +// model's endpoint only while it does. +func (s *Service) PeerRoute(name, model string) (cluster string, sendTags, ok bool) { + s.mu.Lock() + defer s.mu.Unlock() + p, found := s.peers[name] + if !found || !p.health.Up { + return "", false, false + } + for _, m := range p.models { + if m.ID == model { + return p.rec.Cluster, p.rec.SendTags, !m.Withdrawn + } + } + return "", false, false +} + +// peerName is what a peer may be called: the prefix of its models' names, +// and a tag value, so letters, digits and a few separators, and no slash. +var peerName = regexp.MustCompile(`^[a-z0-9][a-z0-9_.-]{0,31}$`) + +// AddRequest adds a peer. +type AddRequest struct { + // Name is what this cluster calls the peer, and the prefix of its + // models: NAME/MODEL. + Name string `json:"name"` + // URL is the peer gateway's client address, without /v1. + URL string `json:"url"` + // KeyFile is a file on the control plane holding a key the peer issued + // to this cluster (or env:NAME, vault:PATH#FIELD). + KeyFile string `json:"key_file"` + // SendTags makes requests sent to the peer carry x-llmgw-tags naming + // this cluster and the sending project. Off unless asked: the peer + // learns who in this cluster is asking. + SendTags bool `json:"send_tags,omitempty"` +} + +// Add checks a peer and keeps it, its models joining the catalog. +// +// Checked before anything is kept: a mistyped address or a key the peer +// refuses is said now, rather than found as a peer that is always down. +func (s *Service) Add(ctx context.Context, req AddRequest, by string) (Status, error) { + req.Name = strings.TrimSpace(req.Name) + req.URL = strings.TrimRight(strings.TrimSpace(req.URL), "/") + req.URL = strings.TrimSuffix(req.URL, "/v1") + if !peerName.MatchString(req.Name) { + return Status{}, fmt.Errorf("peer name %q: 1 to 32 lower-case letters, digits, '_', '.' or '-', starting with a letter or digit", req.Name) + } + u, err := url.Parse(req.URL) + if err != nil || (u.Scheme != "https" && u.Scheme != "http") || u.Host == "" { + return Status{}, fmt.Errorf("peer address %q is not an http(s) address", req.URL) + } + if req.KeyFile == "" { + return Status{}, errors.New("give the file holding the key the peer issued to this cluster (--key-file)") + } + if cur := s.current(); cur != nil { + for _, m := range cur.Models() { + if !m.IsPeer() && strings.HasPrefix(m.Name, req.Name+"/") { + return Status{}, fmt.Errorf("model %q of this cluster already starts with %s/; call the peer something else", + m.Name, req.Name) + } + } + } + + s.op.Lock() + defer s.op.Unlock() + + rec := store.FederationPeer{Name: req.Name, URL: req.URL, KeyFile: req.KeyFile, SendTags: req.SendTags, + CreatedAt: time.Now(), CreatedBy: by} + s.mu.Lock() + old, existed := s.peers[req.Name] + if existed { + rec.CreatedAt, rec.CreatedBy = old.rec.CreatedAt, old.rec.CreatedBy + } + s.mu.Unlock() + + res := s.check(ctx, rec) + if res.err != nil { + return Status{}, fmt.Errorf("peer %s: %w", req.Name, res.err) + } + if res.cluster != "" && res.cluster == s.cfg.Self { + return Status{}, fmt.Errorf("%s is this cluster: its gateway answers with this cluster's own id", req.URL) + } + rec.Cluster, rec.SyncedAt = res.cluster, time.Now() + models := res.models + if existed { + models = s.keepReferenced(old, res.models) + } + b, _ := json.Marshal(models) + rec.Models = b + if err := s.store.PutFederationPeer(ctx, rec); err != nil { + return Status{}, err + } + s.mu.Lock() + s.peers[req.Name] = &peer{rec: rec, models: models, health: res.health} + s.mu.Unlock() + if err := s.catalog.Refresh(ctx); err != nil { + return Status{}, fmt.Errorf("the peer is kept, but the catalog could not be rebuilt: %w", err) + } + s.log.Info("federation peer added", "peer", req.Name, "url", req.URL, "models", len(models), "by", by) + st, _ := s.Status(req.Name) + return st, nil +} + +// Remove takes a peer and its models away. +// +// Refused while a model here falls back to one of them, or has one as its +// canary, for the reason removing a model is: a fallback pointing at +// nothing is found during the outage it was meant to cover. +func (s *Service) Remove(ctx context.Context, name string) error { + s.op.Lock() + defer s.op.Unlock() + + s.mu.Lock() + p, ok := s.peers[name] + s.mu.Unlock() + if !ok { + return fmt.Errorf("no federation peer called %q: `llmgw ctl federation list` lists them", name) + } + var users []string + if cur := s.current(); cur != nil { + for _, m := range cur.Models() { + if m.IsPeer() { + continue + } + for _, ref := range references(m) { + if strings.HasPrefix(ref, name+"/") { + users = append(users, m.Name+" → "+ref) + } + } + } + } + if len(users) > 0 { + return fmt.Errorf("models here still use %s's: %s; change their fallbacks first (`llmgw ctl model fallbacks`)", + name, strings.Join(users, ", ")) + } + if err := s.store.DeleteFederationPeer(ctx, name); err != nil && !errors.Is(err, store.ErrNotFound) { + return err + } + s.mu.Lock() + delete(s.peers, name) + s.mu.Unlock() + if err := s.catalog.Refresh(ctx); err != nil { + return fmt.Errorf("the peer is removed, but the catalog could not be rebuilt: %w", err) + } + s.log.Info("federation peer removed", "peer", name, "models", len(p.models)) + return nil +} + +// references are the other models a model names: its fallbacks and its +// canary. +func references(m catalog.Model) []string { + out := slices.Clone(m.Fallbacks) + if m.Canary != nil { + out = append(out, m.Canary.Model) + } + return out +} + +// keepReferenced is what a peer offers now, plus those it offered before and +// no longer does that a model here still names, kept withdrawn. +func (s *Service) keepReferenced(old *peer, offered []PeerModel) []PeerModel { + out := slices.Clone(offered) + named := map[string]bool{} + if cur := s.current(); cur != nil { + for _, m := range cur.Models() { + if m.IsPeer() { + continue + } + for _, ref := range references(m) { + named[ref] = true + } + } + } + for _, m := range old.models { + if slices.ContainsFunc(offered, func(o PeerModel) bool { return o.ID == m.ID }) { + continue + } + if named[old.rec.Name+"/"+m.ID] { + m.Withdrawn = true + out = append(out, m) + } + } + return out +} + +// Sync checks one peer now, rather than at the next interval. +func (s *Service) Sync(ctx context.Context, name string) (Status, error) { + s.mu.Lock() + p, ok := s.peers[name] + var rec store.FederationPeer + if ok { + rec = p.rec + } + s.mu.Unlock() + if !ok { + return Status{}, fmt.Errorf("no federation peer called %q: `llmgw ctl federation list` lists them", name) + } + res := s.check(ctx, rec) + s.apply(ctx, map[string]checkResult{name: res}) + st, _ := s.Status(name) + return st, nil +} + +// SyncAll checks every peer at once, and puts what changed in force. +func (s *Service) SyncAll(ctx context.Context) { + s.mu.Lock() + recs := make([]store.FederationPeer, 0, len(s.peers)) + for _, p := range s.peers { + recs = append(recs, p.rec) + } + s.mu.Unlock() + + results := make(map[string]checkResult, len(recs)) + var mu sync.Mutex + var wg sync.WaitGroup + for _, rec := range recs { + wg.Go(func() { + res := s.check(ctx, rec) + mu.Lock() + results[rec.Name] = res + mu.Unlock() + }) + } + wg.Wait() + s.apply(ctx, results) +} + +// apply records checks' results: each peer's health always, and its models +// when it said them, rebuilding the catalog once if any changed. +func (s *Service) apply(ctx context.Context, results map[string]checkResult) { + s.op.Lock() + defer s.op.Unlock() + + type change struct { + name, cluster string + models []PeerModel + } + var changes []change + s.mu.Lock() + for name, res := range results { + p, ok := s.peers[name] + if !ok { + continue // removed while it was being checked + } + if p.health.Up != res.health.Up { + if res.health.Up { + s.log.Info("federation peer is up", "peer", name, "latency", res.health.Latency) + } else { + s.log.Warn("federation peer is down; its models are not sent requests", "peer", name, "err", res.health.Error) + } + } + p.health = res.health + if res.err != nil { + continue // its list is kept as it was: down is not empty + } + next := s.keepReferenced(p, res.models) + if res.cluster != p.rec.Cluster || !slices.Equal(next, p.models) { + p.models, p.rec.Cluster, p.rec.SyncedAt = next, res.cluster, time.Now() + changes = append(changes, change{name, res.cluster, next}) + } + } + s.mu.Unlock() + + if len(changes) == 0 { + return + } + for _, c := range changes { + b, _ := json.Marshal(c.models) + if err := s.store.SetFederationPeerModels(ctx, c.name, c.cluster, b, time.Now()); err != nil { + s.log.Warn("could not keep a federation peer's models", "peer", c.name, "err", err) + } + s.log.Info("federation peer's models changed", "peer", c.name, "models", len(c.models)) + } + if err := s.catalog.Refresh(ctx); err != nil { + s.log.Warn("the catalog could not be rebuilt after a peer's models changed", "err", err) + } +} + +// checkResult is one check of one peer. +type checkResult struct { + health Health + cluster string + models []PeerModel + // err is set when the models could not be read; health says the peer + // is down then too. + err error +} + +// check asks a peer whether it is ready, timed, and what it offers this +// cluster's key. +func (s *Service) check(ctx context.Context, rec store.FederationPeer) checkResult { + ctx, cancel := context.WithTimeout(ctx, s.cfg.Timeout) + defer cancel() + res := checkResult{health: Health{Checked: time.Now()}} + fail := func(err error) checkResult { + res.health.Up, res.health.Error, res.err = false, err.Error(), err + return res + } + + // /ready first, and timed: it is open, cheap, and says whether the peer + // has anything of its own to serve with. + req, err := http.NewRequestWithContext(ctx, http.MethodGet, rec.URL+"/ready", nil) + if err != nil { + return fail(err) + } + start := time.Now() + resp, err := s.cfg.Client.Do(req) + if err != nil { + return fail(fmt.Errorf("the peer did not answer: %w", err)) + } + _, _ = io.Copy(io.Discard, io.LimitReader(resp.Body, 1<<12)) + resp.Body.Close() + res.health.Latency = time.Since(start) + if resp.StatusCode != http.StatusOK { + return fail(fmt.Errorf("the peer is not ready (%s): it has nothing of its own to serve with", resp.Status)) + } + + key := "" + if rec.KeyFile != "" { + b, err := secrets.Read(ctx, rec.KeyFile) + if err != nil { + return fail(fmt.Errorf("the key for the peer could not be read: %w", err)) + } + key = strings.TrimSpace(string(b)) + } + req, err = http.NewRequestWithContext(ctx, http.MethodGet, rec.URL+"/v1/models", nil) + if err != nil { + return fail(err) + } + if key != "" { + req.Header.Set("Authorization", "Bearer "+key) + } + resp, err = s.cfg.Client.Do(req) + if err != nil { + return fail(fmt.Errorf("the peer's models could not be read: %w", err)) + } + defer resp.Body.Close() + switch { + case resp.StatusCode == http.StatusUnauthorized || resp.StatusCode == http.StatusForbidden: + return fail(fmt.Errorf("the peer refused this cluster's key (%s): check the key file", resp.Status)) + case resp.StatusCode != http.StatusOK: + return fail(fmt.Errorf("the peer's models could not be read: %s", resp.Status)) + } + cluster := resp.Header.Get(ClusterHeader) + if cluster == "" { + return fail(errors.New("the peer does not say its cluster id: is it an llmgw gateway of this version or later?")) + } + var list struct { + Data []struct { + ID string `json:"id"` + Description string `json:"description"` + ContextLength int `json:"context_length"` + MaxOutput int `json:"max_output_tokens"` + Capabilities catalog.Capabilities `json:"capabilities"` + AliasOf string `json:"alias_of"` + Peer string `json:"peer"` + } `json:"data"` + } + if err := json.NewDecoder(io.LimitReader(resp.Body, 4<<20)).Decode(&list); err != nil { + return fail(fmt.Errorf("the peer's model list is not what an llmgw gateway answers: %w", err)) + } + for _, m := range list.Data { + // Not an alias, which is a second name for a model already here, + // nor a model the peer itself has from a peer of its own: federating + // those would grow a name a hop, A/B/A/..., each time two clusters + // read each other's lists. + if m.AliasOf != "" || m.Peer != "" || !validModelID(m.ID) { + continue + } + res.models = append(res.models, PeerModel{ID: m.ID, Description: m.Description, + ContextLength: m.ContextLength, MaxOutput: m.MaxOutput, Capabilities: m.Capabilities}) + } + slices.SortFunc(res.models, func(a, b PeerModel) int { return strings.Compare(a.ID, b.ID) }) + res.cluster = cluster + res.health.Up = true + return res +} + +// validModelID is a model name this cluster can carry: printable, no spaces, +// and short enough to be a name. +func validModelID(id string) bool { + if id == "" || len(id) > 200 { + return false + } + return !strings.ContainsFunc(id, func(r rune) bool { return r <= ' ' || r == 0x7f }) +} + +// Run checks every peer at once each interval until ctx ends, the first +// time straight away. +func (s *Service) Run(ctx context.Context) { + t := time.NewTicker(s.cfg.Interval) + defer t.Stop() + for { + s.SyncAll(ctx) + select { + case <-ctx.Done(): + return + case <-t.C: + } + } +} + +// Status is a peer as `llmgw ctl federation list` shows it. +type Status struct { + Name string `json:"name"` + URL string `json:"url"` + KeyFile string `json:"key_file,omitempty"` + SendTags bool `json:"send_tags"` + // Cluster is the peer's own cluster id. + Cluster string `json:"cluster,omitempty"` + Health + // LatencyMs is Latency in milliseconds, for reading. + LatencyMs float64 `json:"latency_ms"` + // SyncedAt is when its models last changed here. + SyncedAt time.Time `json:"synced_at,omitempty"` + Models []ModelStatus `json:"models"` + CreatedAt time.Time `json:"created_at"` + CreatedBy string `json:"created_by,omitempty"` +} + +// ModelStatus is one of a peer's models: its name here, the peer's own, +// and whether the peer has stopped offering it. +type ModelStatus struct { + Name string `json:"name"` + Upstream string `json:"upstream"` + Withdrawn bool `json:"withdrawn,omitempty"` +} + +// Status is one peer; false when there is none of the name. +func (s *Service) Status(name string) (Status, bool) { + s.mu.Lock() + defer s.mu.Unlock() + p, ok := s.peers[name] + if !ok { + return Status{}, false + } + return statusOf(p), true +} + +// List is every peer, by name. +func (s *Service) List() []Status { + s.mu.Lock() + defer s.mu.Unlock() + out := make([]Status, 0, len(s.peers)) + for _, n := range s.namesLocked() { + out = append(out, statusOf(s.peers[n])) + } + return out +} + +func statusOf(p *peer) Status { + st := Status{Name: p.rec.Name, URL: p.rec.URL, KeyFile: p.rec.KeyFile, SendTags: p.rec.SendTags, + Cluster: p.rec.Cluster, Health: p.health, LatencyMs: float64(p.health.Latency.Microseconds()) / 1000, + SyncedAt: p.rec.SyncedAt, CreatedAt: p.rec.CreatedAt, CreatedBy: p.rec.CreatedBy, + Models: make([]ModelStatus, 0, len(p.models))} + for _, m := range p.models { + st.Models = append(st.Models, ModelStatus{Name: p.rec.Name + "/" + m.ID, Upstream: m.ID, Withdrawn: m.Withdrawn}) + } + return st +} diff --git a/internal/control/federation/metrics.go b/internal/control/federation/metrics.go new file mode 100644 index 0000000..ffa1257 --- /dev/null +++ b/internal/control/federation/metrics.go @@ -0,0 +1,42 @@ +package federation + +import "github.com/prometheus/client_golang/prometheus" + +var ( + upDesc = prometheus.NewDesc("llmgw_federation_peer_up", + "Whether a federated peer answered its last check: 1 up, 0 down and not sent requests.", + []string{"peer"}, nil) + latencyDesc = prometheus.NewDesc("llmgw_federation_peer_latency_seconds", + "How long a federated peer took to answer /ready at its last check.", + []string{"peer"}, nil) + modelsDesc = prometheus.NewDesc("llmgw_federation_peer_models", + "How many models a federated peer offers this cluster.", + []string{"peer"}, nil) +) + +// Describe and Collect publish each peer's health beside the traffic it +// takes, so "overflow went to the other site" and "the other site was slow" +// are on one chart. +func (s *Service) Describe(ch chan<- *prometheus.Desc) { + ch <- upDesc + ch <- latencyDesc + ch <- modelsDesc +} + +func (s *Service) Collect(ch chan<- prometheus.Metric) { + for _, st := range s.List() { + up := 0.0 + if st.Up { + up = 1 + } + offered := 0 + for _, m := range st.Models { + if !m.Withdrawn { + offered++ + } + } + ch <- prometheus.MustNewConstMetric(upDesc, prometheus.GaugeValue, up, st.Name) + ch <- prometheus.MustNewConstMetric(latencyDesc, prometheus.GaugeValue, st.Latency.Seconds(), st.Name) + ch <- prometheus.MustNewConstMetric(modelsDesc, prometheus.GaugeValue, float64(offered), st.Name) + } +} diff --git a/internal/control/modelsvc/modelsvc.go b/internal/control/modelsvc/modelsvc.go index 392b67d..f93e944 100644 --- a/internal/control/modelsvc/modelsvc.go +++ b/internal/control/modelsvc/modelsvc.go @@ -51,6 +51,9 @@ type Service struct { // SetCheck. check func(context.Context, catalog.Model) error + // federated are the models federated clusters offer; see SetFederated. + federated func() []catalog.Model + // mu serialises rebuilds. They are rare — a person adding a model — and two // at once could apply catalogs in either order, leaving the one in force // disagreeing with the database that outlives it. @@ -128,7 +131,7 @@ func (s *Service) Add(ctx context.Context, m catalog.Model, by string) error { if err != nil { return err } - next := append([]catalog.Model{}, s.declared...) + next := s.base(stored) for _, e := range stored { if e.Name == m.Name { continue // this call replaces it @@ -174,7 +177,7 @@ func (s *Service) Remove(ctx context.Context, name string) error { if err != nil { return err } - next := append([]catalog.Model{}, s.declared...) + next := s.base(stored) found := false for _, e := range stored { if e.Name == name { @@ -371,7 +374,7 @@ func (s *Service) update(ctx context.Context, name, by, what string, change func if err != nil { return catalog.Model{}, err } - next := append([]catalog.Model{}, s.declared...) + next := s.base(stored) found := false for _, e := range stored { if e.Name == name { @@ -416,12 +419,19 @@ func (s *Service) Load(ctx context.Context) (*catalog.Catalog, error) { s.mu.Lock() defer s.mu.Unlock() + return s.load(ctx) +} + +// load is Load, with the lock held. +func (s *Service) load(ctx context.Context) (*catalog.Catalog, error) { stored, err := s.storedModels(ctx) if err != nil { return nil, err } - next := append([]catalog.Model{}, s.declared...) + // The federated models before the stored ones, so a stored model that + // falls back to a peer's is checked against a catalog that has it. + next := s.base(stored) for _, m := range stored { // Checked one at a time and against what is already accepted, so one // bad model cannot take a good one with it. @@ -459,6 +469,82 @@ func (s *Service) storedModels(ctx context.Context) ([]catalog.Model, error) { return out, nil } +// SetFederated supplies the models federated clusters offer, which join the +// catalog after the declared ones and before those added at runtime. +// +// A function rather than a list, read at every rebuild, because the list +// changes whenever a peer's does; Refresh is how federation says it has. +func (s *Service) SetFederated(f func() []catalog.Model) { + s.mu.Lock() + defer s.mu.Unlock() + s.federated = f +} + +// Refresh rebuilds the catalog and puts it in force: after a peer's models +// changed. Built as at startup, so a stored model that no longer makes a +// valid catalog is skipped and said rather than taking the rest with it. +func (s *Service) Refresh(ctx context.Context) error { + s.mu.Lock() + defer s.mu.Unlock() + built, err := s.load(ctx) + if err != nil { + return err + } + s.apply.ReplaceCatalog(ctx, built) + return nil +} + +// base is the declared models, then the federated ones whose names nothing +// declared or added at runtime already uses. +// +// A peer's model is named PEER/MODEL, which a local model could also be +// called — Hugging Face ids look just like that. The local one wins, for the +// reason configuration wins over the database: a request for a model this +// cluster runs must not leave it because a peer started offering a name. +func (s *Service) base(stored []catalog.Model) []catalog.Model { + next := append([]catalog.Model{}, s.declared...) + if s.federated == nil { + return next + } + taken := map[string]bool{} + for _, list := range [][]catalog.Model{s.declared, stored} { + for _, m := range list { + taken[m.Name] = true + for _, a := range m.Aliases { + taken[a] = true + } + } + } + for _, f := range s.federated() { + if taken[f.Name] { + s.log.Warn("a federated model has the name of a model of this cluster's; not offered", + "model", f.Name) + continue + } + next = append(next, f) + } + return next +} + +// SetFallbacks replaces the models a model added at runtime falls back to +// when it has no capacity: other models of the catalog, a federated peer's +// among them. +func (s *Service) SetFallbacks(ctx context.Context, name string, fallbacks []string, by string) (catalog.Model, error) { + final, err := s.update(ctx, name, by, "change its fallbacks", func(m *catalog.Model) error { + for _, f := range fallbacks { + if f == m.Name { + return fmt.Errorf("model %q cannot fall back to itself", name) + } + } + m.Fallbacks = fallbacks + return nil + }) + if err == nil { + s.log.Info("model fallbacks set", "model", name, "fallbacks", fallbacks, "by", by) + } + return final, err +} + // SetCheck runs before a model is added, and refuses it by returning an error. // Used to catch weights the engine cannot load, which otherwise are placed and // then fail on every node they are tried on. diff --git a/internal/docgen/reference_gen.go b/internal/docgen/reference_gen.go index 2f35025..6b4c73e 100644 --- a/internal/docgen/reference_gen.go +++ b/internal/docgen/reference_gen.go @@ -98,6 +98,7 @@ var Reference = []Block{ {Key: "url", Type: "string", Doc: "URL is the provider's base address, without /v1: https://api.openai.com.", Nested: ""}, {Key: "model", Type: "string", Doc: "Model is what the provider calls it; requests are rewritten to it.", Nested: ""}, {Key: "api_key_file", Type: "string", Doc: "APIKeyFile holds the provider's key, on the control plane. A file\nrather than the key itself, so the key is in neither the catalog, the\nstore nor the audit log; empty for a provider that asks for none.", Nested: ""}, + {Key: "peer", Type: "string", Doc: "Peer names the federated llmgw cluster this model is offered by, for\na model that federation added (`llmgw ctl federation add`) rather\nthan one written by hand. Such a model is named PEER/MODEL, comes and\ngoes with the peer's own list, and is sent only while the peer is up.", Nested: ""}, }}, {Package: "config", Name: "Facility", Title: "Facility", Doc: "", Fields: []Field{ {Key: "budgets", Type: "[]FacilityBudget", Doc: "Budgets are the domains with a power cap.", Nested: ""}, @@ -112,6 +113,10 @@ var Reference = []Block{ {Key: "cap_watts", Type: "float64", Doc: "CapWatts is what the domain may draw. For a dual-fed rack it is what\none feed carries alone, derated as the facility derates it (usually\n80%), since that is what the rack has to survive.", Nested: ""}, {Key: "other_watts", Type: "float64", Doc: "OtherWatts is drawn in the domain by what llmgw does not place on:\nswitches, storage, machines outside the cluster.", Nested: ""}, }}, + {Package: "config", Name: "Federation", Title: "Federation", Doc: "", Fields: []Field{ + {Key: "name", Type: "string", Doc: "Name is this cluster as a peer reads it, in the federated_from tag\nsent with a request to a peer set to receive tags (`ctl federation add\n--send-tags`). Empty is this machine's host name.", Nested: ""}, + {Key: "interval", Type: "time.Duration", Doc: "Interval is how often each peer is asked whether it is ready, timed,\nand what it offers. A peer that does not answer is taken out of\nrouting until it does. Zero is 10s.", Nested: ""}, + }}, {Package: "catalog", Name: "GuardModel", Title: "GuardModel", Doc: "", Fields: []Field{ {Key: "format", Type: "string", Doc: "Format is llamaguard3 (Meta's Llama Guard 3: \"safe\", or \"unsafe\" and\nthe categories), shieldgemma (Google's ShieldGemma: Yes or No), or\njson ({\"safe\": false, \"categories\": [\"...\"]}).", Nested: ""}, {Key: "categories", Type: "[]string", Doc: "Categories the model is asked about, for shieldgemma — each a policy\nsentence — and the names reported for llamaguard3's S1, S2, ….", Nested: ""}, @@ -367,6 +372,7 @@ var Reference = []Block{ {Key: "control", Type: "Control", Doc: "", Nested: "config.Control"}, {Key: "models", Type: "[]catalog.Model", Doc: "", Nested: ""}, {Key: "facility", Type: "Facility", Doc: "Facility is the power caps of racks, PDUs and other domains, and how\na replica's draw is estimated against them. Where a node is comes from\nits llmgw.io/site, room, row, rack and pdu labels; holding a domain for\nmaintenance is `llmgw ctl facility hold`, not configuration.", Nested: "config.Facility"}, + {Key: "federation", Type: "Federation", Doc: "Federation is how this cluster takes part in federation with other\nllmgw clusters. The peers themselves are added at runtime, with\n`llmgw ctl federation add`, and kept in the store.", Nested: "config.Federation"}, {Key: "endpoints", Type: "[]Endpoint", Doc: "Endpoints is the phase-1 static replica list. It is only consulted when\nno model declares a profile, which is what makes a router-only smoke test\nagainst sim-engines possible without a control plane.", Nested: ""}, }}, {Package: "engines", Name: "SimConfig", Title: "SimConfig", Doc: "", Fields: []Field{ diff --git a/internal/router/endpoint/endpoint.go b/internal/router/endpoint/endpoint.go index b2a49d4..bfdb397 100644 --- a/internal/router/endpoint/endpoint.go +++ b/internal/router/endpoint/endpoint.go @@ -41,6 +41,13 @@ type Info struct { // from the catalog's name: an external provider's. Requests are rewritten // to it. Empty sends the name as the client did. UpstreamModel string + // Peer names the federated llmgw cluster behind an external endpoint, + // empty for every other. PeerCluster is that cluster's own id, which a + // request that has already passed through it carries in its hop header, + // and PeerTags whether requests sent there name who sent them. + Peer string + PeerCluster string + PeerTags bool } // EngineExternal marks an endpoint that is a provider outside the cluster. diff --git a/internal/router/federation.go b/internal/router/federation.go new file mode 100644 index 0000000..3d29bfc --- /dev/null +++ b/internal/router/federation.go @@ -0,0 +1,155 @@ +package router + +import ( + "context" + "net/http" + "slices" + "strings" + + "gitea.mynodes.xyz/ilker/llmgw/internal/auth" + "gitea.mynodes.xyz/ilker/llmgw/internal/router/endpoint" +) + +// Federation, as the router sees it: requests sent to another llmgw cluster +// say which clusters they have passed through, and a request that comes back +// to one of them is refused rather than sent round again. + +const ( + // hopHeader lists the cluster ids a request has passed through, oldest + // first: "3f9a…, 81c0…". Each gateway adds its own when it sends the + // request to a peer. + hopHeader = "x-llmgw-hop" + // clusterHeader says this gateway's cluster id, on /v1/models, which is + // where a peer learns it. + clusterHeader = "x-llmgw-cluster" + // maxHops bounds a chain of peers whatever their ids say: a backstop + // for two clusters given the same id by a restore of one's store into + // the other, where the loop check cannot tell them apart. + maxHops = 4 +) + +// federationID is this cluster as its peers know it: its id, in the hop +// header, and its name, in the tags a peer may be sent. +type federationID struct { + id, name string +} + +// SetFederationIdentity says which cluster this gateway is. Without it, +// requests are never refused as loops and peers are sent no hop of this one. +func (r *Router) SetFederationIdentity(id, name string) { + r.federation.Store(&federationID{id: id, name: name}) +} + +func (r *Router) clusterID() federationID { + if f := r.federation.Load(); f != nil { + return *f + } + return federationID{} +} + +// hops reads the clusters a request has passed through. +func hops(h http.Header) []string { + var out []string + for _, v := range h.Values(hopHeader) { + for _, p := range strings.Split(v, ",") { + if p = strings.TrimSpace(p); p != "" { + out = append(out, p) + } + } + } + return out +} + +// refuseLoops answers 508 to a request that has already passed through this +// cluster, or through more clusters than any sensible chain of peers has. +// +// Before anything else, authentication included: such a request was sent by +// a peer that had no capacity, and answering it at once is what lets that +// peer move on to its next fallback instead of waiting out a queue here. +func (r *Router) refuseLoops(next http.Handler) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) { + if req.Header.Get(hopHeader) == "" || !strings.HasPrefix(req.URL.Path, "/v1/") { + next.ServeHTTP(w, req) + return + } + seen := hops(req.Header) + self := r.clusterID().id + switch { + case self != "" && slices.Contains(seen, self): + writeErrorCode(w, http.StatusLoopDetected, "invalid_request_error", "federation_loop", + "this request has already passed through this cluster; it is not sent round again") + case len(seen) >= maxHops: + writeErrorCode(w, http.StatusLoopDetected, "invalid_request_error", "federation_loop", + "this request has passed through too many federated clusters") + default: + next.ServeHTTP(w, req) + return + } + r.m.observeRequest("", "", http.StatusLoopDetected, 0, nil, 0) + }) +} + +// withoutVisited drops the endpoints of peers a request has already passed +// through: sending it there would only be refused as a loop, after a round +// trip to the other site. +func withoutVisited(eps []endpoint.Snapshot, h http.Header) []endpoint.Snapshot { + seen := hops(h) + if len(seen) == 0 { + return eps + } + return slices.DeleteFunc(slices.Clone(eps), func(e endpoint.Snapshot) bool { + return e.Peer != "" && e.PeerCluster != "" && slices.Contains(seen, e.PeerCluster) + }) +} + +// peerHeaders are what a request sent to a peer carries beyond what any +// upstream is sent: the hop header with this cluster added, and, when the +// peer is set to receive them, tags naming this cluster and the project. +// +// The peer sees one key, this cluster's, so without the tags every request +// from here is one tenant to it. They are off by default because they tell +// another site who in this one is asking. +func (r *Router) peerHeaders(ctx context.Context, ep endpoint.Snapshot, h http.Header) http.Header { + out := http.Header{} + self := r.clusterID() + chain := hops(h) + if self.id != "" { + chain = append(chain, self.id) + } + if len(chain) > 0 { + out.Set(hopHeader, strings.Join(chain, ", ")) + } + if ep.PeerTags { + tags := []string{"federated_from=" + tagValue(self.name)} + if p, ok := auth.FromContext(ctx); ok { + for _, kv := range [][2]string{{"org", p.OrgID}, {"team", p.TeamID}, {"project", p.ProjectID}} { + if kv[1] != "" { + tags = append(tags, kv[0]+"="+tagValue(kv[1])) + } + } + } + out.Set(tagsHeader, strings.Join(tags, ",")) + } + return out +} + +// tagValue keeps a value inside one tag: x-llmgw-tags is comma-separated. +func tagValue(v string) string { + v = strings.NewReplacer(",", "_", "=", "_").Replace(v) + if len(v) > maxTagValue { + v = v[:maxTagValue] + } + return v +} + +// localEndpoints counts the endpoints that are not a federated peer's: what +// this cluster can serve with itself. +func (r *Router) localEndpoints() int { + n := 0 + for _, s := range r.reg.Snapshots("") { + if s.Peer == "" { + n++ + } + } + return n +} diff --git a/internal/router/federation_test.go b/internal/router/federation_test.go new file mode 100644 index 0000000..83fe84a --- /dev/null +++ b/internal/router/federation_test.go @@ -0,0 +1,594 @@ +package router + +import ( + "context" + "crypto/sha256" + "encoding/json" + "fmt" + "log/slog" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "slices" + "strings" + "sync" + "testing" + "time" + + "gitea.mynodes.xyz/ilker/llmgw/internal/control" + "gitea.mynodes.xyz/ilker/llmgw/internal/control/catalog" + "gitea.mynodes.xyz/ilker/llmgw/internal/control/federation" + "gitea.mynodes.xyz/ilker/llmgw/internal/control/modelsvc" + "gitea.mynodes.xyz/ilker/llmgw/internal/control/reconciler" + "gitea.mynodes.xyz/ilker/llmgw/internal/router/endpoint" + "gitea.mynodes.xyz/ilker/llmgw/internal/store" +) + +// fedCluster is one llmgw cluster, whole but for its nodes: a router serving +// on a loopback port, the control plane that publishes to it, the model +// service and federation. Its one model is served by a fake engine standing +// in for the replica a node would run. +type fedCluster struct { + name, id string + model string + rt *Router + srv *httptest.Server + ctl *control.Controller + models *modelsvc.Service + fed *federation.Service + engine *fakeEngine + // key is a client's key here, in project. + key, project string + keys int +} + +type fedNoDispatch struct{} + +func (fedNoDispatch) StartShard(context.Context, reconciler.ShardCommand) error { return nil } +func (fedNoDispatch) StopShard(context.Context, string, string) error { return nil } +func (fedNoDispatch) DrainShard(context.Context, string, string) error { return nil } + +// withLocal is the sink between a control plane with no nodes and its +// router: what the control plane publishes, and the cluster's own engine. +type withLocal struct { + rt *Router + local endpoint.Info +} + +func (s withLocal) SetEndpoints(ctx context.Context, infos []endpoint.Info) { + s.rt.SetEndpoints(ctx, append(slices.Clone(infos), s.local)) +} + +func (s withLocal) SetCatalog(c *catalog.Catalog) { s.rt.SetCatalog(c) } + +func newFedCluster(t *testing.T, name, model string) *fedCluster { + t.Helper() + c := &fedCluster{name: name, id: name + "-cluster-id", model: model} + c.engine = newFakeEngine(t, model) + + cfg := baseConfig() + cfg.Auth.Enabled = true + cfg.Auth.Store = filepath.Join(t.TempDir(), "store.db") + cfg.Auth.FlushInterval = 20 * time.Millisecond + cfg.Auth.BatchSize = 1 + cfg.Router.Retries = 0 + c.rt = newTestRouter(t, cfg) + c.rt.SetFederationIdentity(c.id, name) + + empty, err := catalog.New(nil) + if err != nil { + t.Fatal(err) + } + log := slog.New(slog.DiscardHandler) + sink := withLocal{rt: c.rt, local: endpoint.Info{ID: name + "-replica", URL: c.engine.srv.URL, Models: []string{model}}} + c.ctl = control.New(control.DefaultConfig(), catalog.NewLive(empty), fedNoDispatch{}, sink, log) + c.models = modelsvc.New(nil, c.rt.store, c.ctl, log) + c.fed = federation.New(federation.Config{Self: c.id, Interval: time.Hour}, c.rt.store, c.models, c.ctl.Catalog, log) + if err := c.fed.Load(t.Context()); err != nil { + t.Fatal(err) + } + c.models.SetFederated(c.fed.Models) + c.ctl.SetPeers(c.fed) + c.srv = httptest.NewServer(c.rt.Handler()) + t.Cleanup(c.srv.Close) + + if err := c.models.Add(t.Context(), catModel(model, nil), "test"); err != nil { + t.Fatal(err) + } + // The control plane has published nothing yet — an empty list is no + // change — so the engine is put in by hand the first time. + sink.SetEndpoints(t.Context(), nil) + + org, err := c.rt.store.CreateOrg(t.Context(), "acme-"+name) + if err != nil { + t.Fatal(err) + } + team, err := c.rt.store.CreateTeam(t.Context(), org, "platform") + if err != nil { + t.Fatal(err) + } + if c.project, err = c.rt.store.CreateProject(t.Context(), team, "gateway"); err != nil { + t.Fatal(err) + } + c.key = c.issueKey(t) + return c +} + +// issueKey makes a key in the cluster's project, as `llmgw keys create` +// would, and returns its secret. +func (c *fedCluster) issueKey(t *testing.T) string { + t.Helper() + c.keys++ + secret := store.KeyPrefix + strings.Repeat(string(rune('a'+c.keys)), 43) + sum := sha256.Sum256([]byte(secret)) + if err := c.rt.store.CreateKey(t.Context(), store.Key{ + ID: fmt.Sprintf("%s-key-%d", c.name, c.keys), Hash: sum[:], Prefix: "pfx" + c.name, Name: "test", + ProjectID: c.project, CreatedAt: time.Now(), + }); err != nil { + t.Fatal(err) + } + return secret +} + +// federate makes other a peer of c, with a key other issued to c. +func (c *fedCluster) federate(t *testing.T, other *fedCluster, sendTags bool) federation.Status { + t.Helper() + keyFile := filepath.Join(t.TempDir(), other.name+".key") + if err := os.WriteFile(keyFile, []byte(other.issueKey(t)+"\n"), 0o600); err != nil { + t.Fatal(err) + } + st, err := c.fed.Add(t.Context(), federation.AddRequest{Name: other.name, URL: other.srv.URL, + KeyFile: keyFile, SendTags: sendTags}, "test") + if err != nil { + t.Fatal(err) + } + c.ctl.Publish(t.Context()) + return st +} + +// ask sends a chat completion to the cluster as its client. +func (c *fedCluster) ask(t *testing.T, model string, hdr map[string]string) *httptest.ResponseRecorder { + t.Helper() + h := map[string]string{"Authorization": "Bearer " + c.key} + for k, v := range hdr { + h[k] = v + } + return post(t, c.rt.Handler(), map[string]any{"model": model, + "messages": []any{map[string]any{"role": "user", "content": "hello"}}}, h) +} + +// usageOf waits for the cluster's record of a request. +func (c *fedCluster) usageOf(t *testing.T, requestID string) store.Usage { + t.Helper() + deadline := time.Now().Add(5 * time.Second) + for time.Now().Before(deadline) { + u, ok, err := c.rt.store.UsageByRequest(t.Context(), requestID) + if err != nil { + t.Fatal(err) + } + if ok { + return u + } + time.Sleep(20 * time.Millisecond) + } + t.Fatalf("%s has no record of request %s", c.name, requestID) + return store.Usage{} +} + +func listedModels(t *testing.T, c *fedCluster) (map[string]map[string]any, http.Header) { + t.Helper() + req := httptest.NewRequest(http.MethodGet, "/v1/models", nil) + req.Header.Set("Authorization", "Bearer "+c.key) + rec := httptest.NewRecorder() + c.rt.Handler().ServeHTTP(rec, req) + var list struct { + Data []map[string]any `json:"data"` + } + if err := json.Unmarshal(rec.Body.Bytes(), &list); err != nil { + t.Fatalf("%d %s", rec.Code, rec.Body.String()) + } + out := map[string]map[string]any{} + for _, m := range list.Data { + out[fmt.Sprint(m["id"])] = m + } + return out, rec.Header() +} + +func TestAPeersModelsAreOfferedHereUnderItsNameAndFollowItsList(t *testing.T) { + west := newFedCluster(t, "west", "qwen-west") + east := newFedCluster(t, "east", "qwen-east") + + st := west.federate(t, east, false) + if !st.Up || st.Cluster != east.id || st.Latency <= 0 { + t.Errorf("east after adding: %+v", st) + } + listed, hdr := listedModels(t, west) + if listed["east/qwen-east"] == nil || listed["east/qwen-east"]["peer"] != "east" { + t.Fatalf("west offers %v; east's model should be there as east/qwen-east", slices.Collect(mapsKeys(listed))) + } + if hdr.Get(clusterHeader) != west.id { + t.Errorf("west's /v1/models says cluster %q", hdr.Get(clusterHeader)) + } + if m, ok := west.ctl.Catalog().Resolve("east/qwen-east"); !ok || !m.IsPeer() || m.External.Model != "qwen-east" { + t.Errorf("west's catalog has %+v", m) + } + + // Asked for by its name here, answered by east's engine under east's name. + rec := west.ask(t, "east/qwen-east", nil) + if rec.Code != http.StatusOK || east.engine.calls.Load() != 1 { + t.Fatalf("%d %s; east's engine was called %d times", rec.Code, rec.Body.String(), east.engine.calls.Load()) + } + var sent map[string]any + _ = json.Unmarshal(east.engine.last.Load().([]byte), &sent) + if sent["model"] != "qwen-east" { + t.Errorf("east's engine was asked for %v", sent["model"]) + } + + // East adds a model, then drops its first: west follows at its next check. + if err := east.models.Add(t.Context(), catModel("qwen-east-2", nil), "test"); err != nil { + t.Fatal(err) + } + if err := east.models.Remove(t.Context(), "qwen-east"); err != nil { + t.Fatal(err) + } + east.rt.SetEndpoints(t.Context(), []endpoint.Info{{ID: "e2", URL: east.engine.srv.URL, Models: []string{"qwen-east-2"}}}) + if _, err := west.fed.Sync(t.Context(), "east"); err != nil { + t.Fatal(err) + } + west.ctl.Publish(t.Context()) + if _, ok := west.ctl.Catalog().Resolve("east/qwen-east-2"); !ok { + t.Error("a model east added is not offered here") + } + if _, ok := west.ctl.Catalog().Resolve("east/qwen-east"); ok { + t.Error("a model east dropped, which nothing here names, is still offered") + } +} + +func mapsKeys(m map[string]map[string]any) func(func(string) bool) { + return func(yield func(string) bool) { + for k := range m { + if !yield(k) { + return + } + } + } +} + +func TestOverflowGoesToThePeerWhichSeesOneKeyAndTheRequestID(t *testing.T) { + west := newFedCluster(t, "west", "qwen-west") + east := newFedCluster(t, "east", "qwen-east") + west.federate(t, east, false) + if _, err := west.models.SetFallbacks(t.Context(), "qwen-west", []string{"east/qwen-east"}, "test"); err != nil { + t.Fatal(err) + } + west.ctl.Publish(t.Context()) + + // West has capacity: east is not asked. + if rec := west.ask(t, "qwen-west", nil); rec.Code != http.StatusOK || east.engine.calls.Load() != 0 { + t.Fatalf("%d; east called %d times", rec.Code, east.engine.calls.Load()) + } + + // West's replica is full: the overflow goes east. + west.engine.status.Store(http.StatusServiceUnavailable) + rec := west.ask(t, "qwen-west", map[string]string{"X-Request-Id": "overflow-1"}) + if rec.Code != http.StatusOK { + t.Fatalf("%d %s", rec.Code, rec.Body.String()) + } + if got := rec.Header().Values("x-llmgw-served-model"); !slices.Equal(got, []string{"east/qwen-east"}) { + t.Errorf("served by %q", got) + } + // East's own answer headers do not reach the client beside west's. + if got := rec.Header().Values("X-Request-Id"); len(got) != 1 { + t.Errorf("the client was given request ids %q", got) + } + if east.engine.calls.Load() != 1 { + t.Errorf("east's engine was called %d times", east.engine.calls.Load()) + } + + // West records which peer served; east, the one key west uses there, + // the request's id, and no tags, which are off unless asked for. + here := west.usageOf(t, "overflow-1") + if here.ServedModel != "east/qwen-east" || here.Replica != "peer:east/qwen-east" { + t.Errorf("west's record: served %q by %q", here.ServedModel, here.Replica) + } + there := east.usageOf(t, "overflow-1") + if there.KeyID != "east-key-2" || there.Model != "qwen-east" { + t.Errorf("east's record: key %q, model %q", there.KeyID, there.Model) + } + if len(there.Tags) != 0 { + t.Errorf("east was told %v without send-tags", there.Tags) + } + + // With tags on, east can say which cluster and project asked. + west.federate(t, east, true) + rec = west.ask(t, "qwen-west", map[string]string{"X-Request-Id": "overflow-2"}) + if rec.Code != http.StatusOK { + t.Fatalf("%d %s", rec.Code, rec.Body.String()) + } + there = east.usageOf(t, "overflow-2") + if there.Tags["federated_from"] != "west" || there.Tags["project"] != west.project { + t.Errorf("east's record carries tags %v", there.Tags) + } +} + +func TestTwoClustersFederatingEachOtherDoNotSendARequestRoundAgain(t *testing.T) { + west := newFedCluster(t, "west", "qwen-west") + east := newFedCluster(t, "east", "qwen-east") + west.federate(t, east, false) + east.federate(t, west, false) + + // Neither offers the other's peer models back: no east/west/... names. + if _, ok := west.ctl.Catalog().Resolve("east/west/qwen-west"); ok { + t.Error("west took back its own model through east") + } + if _, ok := east.ctl.Catalog().Resolve("west/east/qwen-east"); ok { + t.Error("east took back its own model through west") + } + + // Each falls back to the other, and both are full. + if _, err := west.models.SetFallbacks(t.Context(), "qwen-west", []string{"east/qwen-east"}, "test"); err != nil { + t.Fatal(err) + } + if _, err := east.models.SetFallbacks(t.Context(), "qwen-east", []string{"west/qwen-west"}, "test"); err != nil { + t.Fatal(err) + } + west.ctl.Publish(t.Context()) + east.ctl.Publish(t.Context()) + west.engine.status.Store(http.StatusServiceUnavailable) + east.engine.status.Store(http.StatusServiceUnavailable) + + rec := west.ask(t, "qwen-west", nil) + if rec.Code == http.StatusOK { + t.Fatalf("served: %s", rec.Body.String()) + } + // West's replica once, east's once; east did not send it back west. + if w, e := west.engine.calls.Load(), east.engine.calls.Load(); w != 1 || e != 1 { + t.Errorf("west's engine was called %d times, east's %d; each should be once", w, e) + } + + // And a request that says it has passed through a cluster already is + // refused there, before anything else. + rec = west.ask(t, "qwen-west", map[string]string{hopHeader: east.id + ", " + west.id}) + if rec.Code != http.StatusLoopDetected { + t.Errorf("a request that came back: %d %s", rec.Code, rec.Body.String()) + } +} + +func TestAProjectThatStaysInTheClusterIsNeverSentToAPeer(t *testing.T) { + west := newFedCluster(t, "west", "qwen-west") + east := newFedCluster(t, "east", "qwen-east") + west.federate(t, east, false) + if _, err := west.models.SetFallbacks(t.Context(), "qwen-west", []string{"east/qwen-east"}, "test"); err != nil { + t.Fatal(err) + } + west.ctl.Publish(t.Context()) + west.engine.status.Store(http.StatusServiceUnavailable) + + if err := west.rt.store.SetInCluster(t.Context(), west.project, true); err != nil { + t.Fatal(err) + } + if rec := west.ask(t, "east/qwen-east", nil); rec.Code != http.StatusForbidden { + t.Errorf("a peer's model, asked for by a project that stays in the cluster: %d %s", rec.Code, rec.Body.String()) + } + if rec := west.ask(t, "qwen-west", nil); rec.Code == http.StatusOK { + t.Errorf("served by %q", rec.Header().Get("x-llmgw-served-model")) + } + if east.engine.calls.Load() != 0 { + t.Errorf("east was sent %d requests", east.engine.calls.Load()) + } +} + +func TestAPolicyThatKeepsNothingIsNeverSentToAPeer(t *testing.T) { + west := newFedCluster(t, "west", "qwen-west") + east := newFedCluster(t, "east", "qwen-east") + west.federate(t, east, false) + if _, err := west.models.SetFallbacks(t.Context(), "qwen-west", []string{"east/qwen-east"}, "test"); err != nil { + t.Fatal(err) + } + west.ctl.Publish(t.Context()) + west.engine.status.Store(http.StatusServiceUnavailable) + g := &switchGuard{} + g.noStore.Store(true) + west.rt.SetGuard(g) + + if rec := west.ask(t, "east/qwen-east", nil); rec.Code != http.StatusBadRequest { + t.Errorf("a peer's model under no_store: %d %s", rec.Code, rec.Body.String()) + } + if rec := west.ask(t, "qwen-west", nil); rec.Code == http.StatusOK { + t.Errorf("served by %q", rec.Header().Get("x-llmgw-served-model")) + } + if east.engine.calls.Load() != 0 { + t.Errorf("east was sent %d requests", east.engine.calls.Load()) + } +} + +func TestAPeerThatIsDownIsTakenOutOfRoutingAndBackWhenItAnswers(t *testing.T) { + west := newFedCluster(t, "west", "qwen-west") + east := newFedCluster(t, "east", "qwen-east") + west.federate(t, east, false) + if len(west.rt.reg.Snapshots("east/qwen-east")) != 1 { + t.Fatal("east's model is not routed to while east is up") + } + + // East has nothing of its own to serve with: its /ready says so. + east.rt.SetEndpoints(t.Context(), nil) + if st, _ := west.fed.Sync(t.Context(), "east"); st.Up || st.Error == "" { + t.Errorf("east with nothing to serve: %+v", st) + } + west.ctl.Publish(t.Context()) + if n := len(west.rt.reg.Snapshots("east/qwen-east")); n != 0 { + t.Errorf("east is down and still routed to: %d endpoints", n) + } + if _, ok := west.ctl.Catalog().Resolve("east/qwen-east"); !ok { + t.Error("east's model left the catalog when east went down; down is not gone") + } + if rec := west.ask(t, "east/qwen-east", nil); rec.Code == http.StatusOK { + t.Error("served by a peer that is down") + } + + // Back. + east.ctl.Publish(t.Context()) + east.rt.SetEndpoints(t.Context(), []endpoint.Info{{ID: "e", URL: east.engine.srv.URL, Models: []string{"qwen-east"}}}) + if st, _ := west.fed.Sync(t.Context(), "east"); !st.Up { + t.Errorf("east is back: %+v", st) + } + west.ctl.Publish(t.Context()) + if rec := west.ask(t, "east/qwen-east", nil); rec.Code != http.StatusOK { + t.Errorf("east is back: %d %s", rec.Code, rec.Body.String()) + } + + // Gone altogether. + east.srv.Close() + if st, _ := west.fed.Sync(t.Context(), "east"); st.Up { + t.Errorf("east is gone: %+v", st) + } + west.ctl.Publish(t.Context()) + if n := len(west.rt.reg.Snapshots("east/qwen-east")); n != 0 { + t.Errorf("east is gone and still routed to: %d endpoints", n) + } +} + +func TestAPeerModelAFallbackNamesIsKeptWithdrawnAndThePeerCannotBeRemoved(t *testing.T) { + west := newFedCluster(t, "west", "qwen-west") + east := newFedCluster(t, "east", "qwen-east") + west.federate(t, east, false) + if _, err := west.models.SetFallbacks(t.Context(), "qwen-west", []string{"east/qwen-east"}, "test"); err != nil { + t.Fatal(err) + } + + if err := west.fed.Remove(t.Context(), "east"); err == nil || !strings.Contains(err.Error(), "qwen-west") { + t.Errorf("removing a peer a fallback names: %v", err) + } + + // East stops offering it: kept, withdrawn, and not routed to. + if err := east.models.Remove(t.Context(), "qwen-east"); err != nil { + t.Fatal(err) + } + // Its replica goes too; east still serves something, so it is up. + east.rt.SetEndpoints(t.Context(), []endpoint.Info{{ID: "other", URL: east.engine.srv.URL, Models: []string{"other"}}}) + if _, err := west.fed.Sync(t.Context(), "east"); err != nil { + t.Fatal(err) + } + west.ctl.Publish(t.Context()) + if _, ok := west.ctl.Catalog().Resolve("qwen-west"); !ok { + t.Fatal("the model that falls back to east's lost its place in the catalog") + } + st, _ := west.fed.Status("east") + if !slices.ContainsFunc(st.Models, func(m federation.ModelStatus) bool { return m.Upstream == "qwen-east" && m.Withdrawn }) { + t.Errorf("east's models: %+v", st.Models) + } + if n := len(west.rt.reg.Snapshots("east/qwen-east")); n != 0 { + t.Errorf("a withdrawn model is routed to: %d endpoints", n) + } + + // Once nothing names it, it goes, and so may the peer. + if _, err := west.models.SetFallbacks(t.Context(), "qwen-west", nil, "test"); err != nil { + t.Fatal(err) + } + if _, err := west.fed.Sync(t.Context(), "east"); err != nil { + t.Fatal(err) + } + if _, ok := west.ctl.Catalog().Resolve("east/qwen-east"); ok { + t.Error("a withdrawn model nothing names is still in the catalog") + } + if err := west.fed.Remove(t.Context(), "east"); err != nil { + t.Fatal(err) + } + if len(west.fed.List()) != 0 { + t.Error("east is still a peer") + } +} + +func TestPeersAndTheirModelsSurviveARestart(t *testing.T) { + west := newFedCluster(t, "west", "qwen-west") + east := newFedCluster(t, "east", "qwen-east") + west.federate(t, east, false) + if _, err := west.models.SetFallbacks(t.Context(), "qwen-west", []string{"east/qwen-east"}, "test"); err != nil { + t.Fatal(err) + } + + // A new control plane against the same store, before east is asked + // anything: the model falling back to east's is whole. + log := slog.New(slog.DiscardHandler) + empty, _ := catalog.New(nil) + ctl := control.New(control.DefaultConfig(), catalog.NewLive(empty), fedNoDispatch{}, withLocal{rt: west.rt}, log) + models := modelsvc.New(nil, west.rt.store, ctl, log) + fed := federation.New(federation.Config{Self: west.id}, west.rt.store, models, ctl.Catalog, log) + if err := fed.Load(t.Context()); err != nil { + t.Fatal(err) + } + models.SetFederated(fed.Models) + built, err := models.Load(t.Context()) + if err != nil { + t.Fatal(err) + } + m, ok := built.Resolve("qwen-west") + if !ok || !slices.Equal(m.Fallbacks, []string{"east/qwen-east"}) { + t.Errorf("after a restart: %+v", m) + } + if _, ok := built.Resolve("east/qwen-east"); !ok { + t.Error("east's model is not in the catalog after a restart") + } + // Down until checked. + if _, _, ok := fed.PeerRoute("east", "qwen-east"); ok { + t.Error("a peer not yet checked is routed to") + } +} + +func TestAClusterCannotBeItsOwnPeer(t *testing.T) { + west := newFedCluster(t, "west", "qwen-west") + keyFile := filepath.Join(t.TempDir(), "self.key") + if err := os.WriteFile(keyFile, []byte(west.issueKey(t)), 0o600); err != nil { + t.Fatal(err) + } + _, err := west.fed.Add(t.Context(), federation.AddRequest{Name: "self", URL: west.srv.URL, KeyFile: keyFile}, "test") + if err == nil || !strings.Contains(err.Error(), "this cluster") { + t.Errorf("adding itself: %v", err) + } +} + +func TestAPeerIsCheckedBeforeItIsKept(t *testing.T) { + west := newFedCluster(t, "west", "qwen-west") + east := newFedCluster(t, "east", "qwen-east") + keyFile := filepath.Join(t.TempDir(), "wrong.key") + if err := os.WriteFile(keyFile, []byte(store.KeyPrefix+strings.Repeat("z", 43)), 0o600); err != nil { + t.Fatal(err) + } + for name, req := range map[string]federation.AddRequest{ + "a key the peer refuses": {Name: "east", URL: east.srv.URL, KeyFile: keyFile}, + "nothing at the address": {Name: "east", URL: "http://127.0.0.1:1", KeyFile: keyFile}, + "a name with a slash": {Name: "ea/st", URL: east.srv.URL, KeyFile: keyFile}, + "no key file": {Name: "east", URL: east.srv.URL}, + } { + if _, err := west.fed.Add(t.Context(), req, "test"); err == nil { + t.Errorf("%s: accepted", name) + } + } + if len(west.fed.List()) != 0 { + t.Errorf("kept: %+v", west.fed.List()) + } +} + +// The federation's checks run beside requests and catalog rebuilds. +func TestFederationChecksAreSafeBesideTraffic(t *testing.T) { + west := newFedCluster(t, "west", "qwen-west") + east := newFedCluster(t, "east", "qwen-east") + west.federate(t, east, false) + var wg sync.WaitGroup + for range 4 { + wg.Go(func() { + for range 5 { + west.fed.SyncAll(t.Context()) + west.ctl.Publish(t.Context()) + } + }) + wg.Go(func() { + for range 5 { + west.ask(t, "east/qwen-east", nil) + _ = west.fed.List() + } + }) + } + wg.Wait() +} diff --git a/internal/router/handlers.go b/internal/router/handlers.go index f1ed618..bedf77d 100644 --- a/internal/router/handlers.go +++ b/internal/router/handlers.go @@ -139,8 +139,13 @@ func (r *Router) Handler() http.Handler { // Open, like /health: a load balancer does not have a key, and this leaks // one bit — whether anything is servable — which the next request would // reveal anyway. + // + // Only this cluster's own endpoints count, not a federated peer's: a + // peer reads this to decide whether to send here, and two clusters + // federating each other would otherwise each be ready because the other + // is, with nothing of their own to serve with. mux.HandleFunc("GET /ready", func(w http.ResponseWriter, _ *http.Request) { - if len(r.reg.IDs()) == 0 { + if r.localEndpoints() == 0 { http.Error(w, "no endpoints are available to serve", http.StatusServiceUnavailable) return } @@ -298,7 +303,7 @@ func (r *Router) Handler() http.Handler { } if r.auth == nil { - return mux + return r.refuseLoops(mux) } exempt := []string{"/health", "/ready", "/leader", "/", "/about", "/_static/", "/_lang/", "/claim/", "/signup", "/login", "/login/", "/logout", "/profile", "/profile/", "/scim/", @@ -312,7 +317,7 @@ func (r *Router) Handler() http.Handler { // Wrapping the whole mux rather than checking inside each handler: adding a // route must not be able to add an unauthenticated one by omission. Health // stays open because a load balancer has no key. - return auth.Middleware(r.auth, r.log, exempt...)(mux) + return r.refuseLoops(auth.Middleware(r.auth, r.log, exempt...)(mux)) } // parsedRequest is what the router needs from a request. Everything else in the @@ -1017,6 +1022,15 @@ func (r *Router) handleModels(w http.ResponseWriter, req *http.Request) { entry := map[string]any{ "id": n, "object": "model", "created": time.Now().Unix(), "owned_by": owner, } + // Not OpenAI's either: what a federated peer reading this list needs + // to leave out — a second name for a model already listed, and a + // model this cluster has from a peer of its own. + if m := r.modelMap()[n]; m.Name != "" && m.Name != n { + entry["alias_of"] = m.Name + } + if m := r.modelMap()[n]; m.IsPeer() { + entry["peer"] = m.External.Peer + } // Only when known. A context_length of 0 would read as "no context", // which is a worse answer than saying nothing. m := r.modelMap()[n] @@ -1038,6 +1052,9 @@ func (r *Router) handleModels(w http.ResponseWriter, req *http.Request) { } data = append(data, entry) } + if id := r.clusterID().id; id != "" { + w.Header().Set(clusterHeader, id) + } writeJSON(w, http.StatusOK, map[string]any{"object": "list", "data": data}) } diff --git a/internal/router/proxy/proxy.go b/internal/router/proxy/proxy.go index 8c4d164..9afbf85 100644 --- a/internal/router/proxy/proxy.go +++ b/internal/router/proxy/proxy.go @@ -149,6 +149,10 @@ type Request struct { Body []byte // Header carries headers to forward (authorization is deliberately not). Header http.Header + // Extra are headers set on the upstream request beyond those forwarded + // from the client: what the router itself says to this upstream, such + // as the hop header a federated cluster is sent. + Extra http.Header // APIKey is the engine's own key, sent in place of the client's. APIKey string // Stream says whether an SSE response is expected. @@ -255,6 +259,9 @@ func (p *Proxy) Do(ctx context.Context, w http.ResponseWriter, r Request) (Outco return Outcome{Retryable: false}, err } req.Header = cloneForUpstream(r.Header) + for k, vs := range r.Extra { + req.Header[http.CanonicalHeaderKey(k)] = vs + } ct := r.ContentType if ct == "" { ct = "application/json" @@ -625,9 +632,15 @@ var hopByHop = map[string]bool{ "Upgrade": true, } +// copyHeaders gives the client the upstream's response headers. +// +// A header the gateway has already set is its own answer and is kept: an +// upstream that is itself an llmgw — a federated peer — sends X-Request-Id +// and x-llmgw-served-model of its own, and the client would otherwise be +// given both. func copyHeaders(dst, src http.Header) { for k, vs := range src { - if hopByHop[http.CanonicalHeaderKey(k)] { + if hopByHop[http.CanonicalHeaderKey(k)] || len(dst.Values(k)) > 0 { continue } for _, v := range vs { diff --git a/internal/router/router.go b/internal/router/router.go index 11962f1..26498dc 100644 --- a/internal/router/router.go +++ b/internal/router/router.go @@ -102,7 +102,9 @@ type Router struct { usageRetention time.Duration // isLeader says whether this process leads; nil is always. isLeader atomic.Pointer[func() bool] - acct *accounting.Accountant + // federation is which cluster this is, to its peers; nil until said. + federation atomic.Pointer[federationID] + acct *accounting.Accountant // guard judges requests and answers; nil judges nothing. guardLibrary // is the policies it was built from, for the pages that show them. guard guard.Guard @@ -929,6 +931,13 @@ var errLoading = errors.New("model loading") func (r *Router) dispatchOne(ctx context.Context, w http.ResponseWriter, req *parsedRequest, model string, entry catalog.Model) dispatchResult { candidates := r.reg.Snapshots(model) + if visited := withoutVisited(candidates, req.Header); len(visited) < len(candidates) { + if len(visited) == 0 { + return dispatchResult{err: fmt.Errorf("%w: %s is served by a federated cluster this request has already passed through", + errNoCapacity, model)} + } + candidates = visited + } if len(candidates) == 0 { // Nothing is serving it. That may be because nothing ever will, or // because it loads when asked and nobody has asked yet — and the second @@ -1081,6 +1090,9 @@ func (r *Router) dispatchOne(ctx context.Context, w http.ResponseWriter, req *pa // restart. Not relayed; loaded again and tried again. preq.RetryStatus = http.StatusNotFound } + if dec.Endpoint.Peer != "" { + preq.Extra = r.peerHeaders(ctx, dec.Endpoint, req.Header) + } if req.Stream { gaps := r.m.token.gap.WithLabelValues(r.metricModel(model)) preq.OnEvent = func(d time.Duration) { gaps.Observe(d.Seconds()) } diff --git a/internal/store/copy.go b/internal/store/copy.go index 81ce820..a296b87 100644 --- a/internal/store/copy.go +++ b/internal/store/copy.go @@ -68,6 +68,7 @@ var copyOrder = []string{ // Runs before their cases, which point at them. "eval_runs", "eval_cases", + "federation_peers", } // notCopied are tables deliberately left behind. diff --git a/internal/store/federation.go b/internal/store/federation.go new file mode 100644 index 0000000..987bf26 --- /dev/null +++ b/internal/store/federation.go @@ -0,0 +1,131 @@ +package store + +import ( + "context" + "database/sql" + "encoding/json" + "fmt" + "time" +) + +// FederationPeer is another llmgw cluster whose models this one offers. +type FederationPeer struct { + Name string + URL string + // KeyFile holds the key this cluster uses at the peer, on the control + // plane: a path, never the key, for the reason an external model's is. + KeyFile string + // SendTags makes requests sent to the peer carry x-llmgw-tags naming + // this cluster and the sending project, so the peer can attribute them. + SendTags bool + // Cluster is the peer's own cluster id, as it last said it: what a + // request that has passed through the peer carries in its hop header. + Cluster string + // Models is what the peer last offered, as JSON. + Models json.RawMessage + SyncedAt time.Time + CreatedAt time.Time + CreatedBy string +} + +// PutFederationPeer adds a peer, or replaces one whole: its address, key +// file, tags setting and what it offers. Who added it, and when, stay as +// they were first written. +func (s *DB) PutFederationPeer(ctx context.Context, p FederationPeer) error { + if p.Name == "" || p.URL == "" { + return fmt.Errorf("a federation peer needs a name and an address") + } + if p.CreatedAt.IsZero() { + p.CreatedAt = time.Now() + } + models := string(p.Models) + if models == "" { + models = "[]" + } + if !json.Valid([]byte(models)) { + return fmt.Errorf("peer %q: its models are not valid JSON", p.Name) + } + _, err := s.db.ExecContext(ctx, + `INSERT INTO federation_peers (name, url, key_file, send_tags, cluster, models, synced_at, created_at, created_by) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT (name) DO UPDATE SET url = excluded.url, key_file = excluded.key_file, + send_tags = excluded.send_tags, cluster = excluded.cluster, models = excluded.models, + synced_at = excluded.synced_at`, + p.Name, p.URL, p.KeyFile, boolInt(p.SendTags), p.Cluster, models, syncedMillis(p.SyncedAt), + millis(p.CreatedAt), p.CreatedBy) + return err +} + +// SetFederationPeerModels records what a peer offers now, and its cluster +// id, without touching how it is reached. ErrNotFound when the peer was +// removed meanwhile: a sync that finishes after a removal must not bring +// the peer back. +func (s *DB) SetFederationPeerModels(ctx context.Context, name, cluster string, models json.RawMessage, at time.Time) error { + if !json.Valid(models) { + return fmt.Errorf("peer %q: its models are not valid JSON", name) + } + res, err := s.db.ExecContext(ctx, + `UPDATE federation_peers SET cluster = ?, models = ?, synced_at = ? WHERE name = ?`, + cluster, string(models), syncedMillis(at), name) + if err != nil { + return err + } + if n, _ := res.RowsAffected(); n == 0 { + return ErrNotFound + } + return nil +} + +// DeleteFederationPeer removes a peer; removing one that is not there is +// ErrNotFound, so a mistyped name is said rather than silently accepted. +func (s *DB) DeleteFederationPeer(ctx context.Context, name string) error { + res, err := s.db.ExecContext(ctx, `DELETE FROM federation_peers WHERE name = ?`, name) + if err != nil { + return err + } + if n, _ := res.RowsAffected(); n == 0 { + return ErrNotFound + } + return nil +} + +// FederationPeers lists the peers by name. +func (s *DB) FederationPeers(ctx context.Context) ([]FederationPeer, error) { + return queryAll(ctx, s.db, `SELECT name, url, key_file, send_tags, cluster, models, synced_at, created_at, created_by + FROM federation_peers ORDER BY name`, nil, func(rows *sql.Rows) (FederationPeer, error) { + var p FederationPeer + var tags int + var models string + var synced, created int64 + err := rows.Scan(&p.Name, &p.URL, &p.KeyFile, &tags, &p.Cluster, &models, &synced, &created, &p.CreatedBy) + p.SendTags, p.Models = tags == 1, json.RawMessage(models) + p.CreatedAt = fromMillis(created) + if synced > 0 { + p.SyncedAt = fromMillis(synced) + } + return p, err + }) +} + +// FederationPeer is one peer by name; ErrNotFound when there is none. +func (s *DB) FederationPeer(ctx context.Context, name string) (FederationPeer, error) { + peers, err := s.FederationPeers(ctx) + if err != nil { + return FederationPeer{}, err + } + for _, p := range peers { + if p.Name == name { + return p, nil + } + } + return FederationPeer{}, ErrNotFound +} + +// syncedMillis is 0 for a peer never synced, which reads back as no time +// rather than as a date in 1754. +func syncedMillis(t time.Time) int64 { + if t.IsZero() { + return 0 + } + return t.UnixMilli() +} diff --git a/internal/store/federation_test.go b/internal/store/federation_test.go new file mode 100644 index 0000000..4d80c95 --- /dev/null +++ b/internal/store/federation_test.go @@ -0,0 +1,50 @@ +package store + +import ( + "encoding/json" + "errors" + "path/filepath" + "testing" + "time" +) + +func TestAFederationPeerIsKeptWithWhatItLastOffered(t *testing.T) { + s, err := OpenSQLite(t.Context(), filepath.Join(t.TempDir(), "s.db")) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = s.Close() }) + ctx := t.Context() + + if err := s.PutFederationPeer(ctx, FederationPeer{Name: "east", URL: "https://east", KeyFile: "/k", + SendTags: true, CreatedBy: "ops"}); err != nil { + t.Fatal(err) + } + peers, err := s.FederationPeers(ctx) + if err != nil || len(peers) != 1 { + t.Fatalf("%v %+v", err, peers) + } + if p := peers[0]; !p.SendTags || string(p.Models) != "[]" || !p.SyncedAt.IsZero() || p.CreatedBy != "ops" { + t.Errorf("as kept: %+v", p) + } + + at := time.Now() + if err := s.SetFederationPeerModels(ctx, "east", "c1", json.RawMessage(`[{"id":"m"}]`), at); err != nil { + t.Fatal(err) + } + p, err := s.FederationPeer(ctx, "east") + if err != nil || p.Cluster != "c1" || string(p.Models) != `[{"id":"m"}]` || p.SyncedAt.UnixMilli() != at.UnixMilli() { + t.Errorf("after a sync: %v %+v", err, p) + } + + // A sync finishing after the peer was removed does not bring it back. + if err := s.DeleteFederationPeer(ctx, "east"); err != nil { + t.Fatal(err) + } + if err := s.SetFederationPeerModels(ctx, "east", "c1", json.RawMessage(`[]`), at); !errors.Is(err, ErrNotFound) { + t.Errorf("syncing a removed peer: %v", err) + } + if err := s.DeleteFederationPeer(ctx, "east"); !errors.Is(err, ErrNotFound) { + t.Errorf("removing it twice: %v", err) + } +} diff --git a/internal/store/migrations/postgres/0029_federation_peers.sql b/internal/store/migrations/postgres/0029_federation_peers.sql new file mode 100644 index 0000000..6f169fa --- /dev/null +++ b/internal/store/migrations/postgres/0029_federation_peers.sql @@ -0,0 +1,23 @@ +-- Ported from the SQLite migration of the same number. +-- Other llmgw clusters this one federates with (`llmgw ctl federation add`): +-- each peer's address, the file on the control plane holding the key this +-- cluster uses there, whether requests sent to it carry tags naming who sent +-- them, the peer's own cluster id, and the models it last offered. +-- +-- The models are kept, not only fetched, so a catalog that names one of them +-- as a fallback is whole at startup, before the peer has been asked again: +-- a local model falling back to a peer's would otherwise be skipped as +-- naming a model that is not there. Stored as JSON for the same reason a +-- runtime model's definition is: a list of what the peer says about each, +-- which grows. +CREATE TABLE federation_peers ( + name TEXT PRIMARY KEY, + url TEXT NOT NULL, + key_file TEXT NOT NULL DEFAULT '', + send_tags BIGINT NOT NULL DEFAULT 0, + cluster TEXT NOT NULL DEFAULT '', + models TEXT NOT NULL DEFAULT '[]', + synced_at BIGINT NOT NULL DEFAULT 0, + created_at BIGINT NOT NULL, + created_by TEXT NOT NULL DEFAULT '' +); diff --git a/internal/store/migrations/sqlite/0029_federation_peers.sql b/internal/store/migrations/sqlite/0029_federation_peers.sql new file mode 100644 index 0000000..94935cf --- /dev/null +++ b/internal/store/migrations/sqlite/0029_federation_peers.sql @@ -0,0 +1,22 @@ +-- Other llmgw clusters this one federates with (`llmgw ctl federation add`): +-- each peer's address, the file on the control plane holding the key this +-- cluster uses there, whether requests sent to it carry tags naming who sent +-- them, the peer's own cluster id, and the models it last offered. +-- +-- The models are kept, not only fetched, so a catalog that names one of them +-- as a fallback is whole at startup, before the peer has been asked again: +-- a local model falling back to a peer's would otherwise be skipped as +-- naming a model that is not there. Stored as JSON for the same reason a +-- runtime model's definition is: a list of what the peer says about each, +-- which grows. +CREATE TABLE federation_peers ( + name TEXT PRIMARY KEY, + url TEXT NOT NULL, + key_file TEXT NOT NULL DEFAULT '', + send_tags INTEGER NOT NULL DEFAULT 0, + cluster TEXT NOT NULL DEFAULT '', + models TEXT NOT NULL DEFAULT '[]', + synced_at INTEGER NOT NULL DEFAULT 0, + created_at INTEGER NOT NULL, + created_by TEXT NOT NULL DEFAULT '' +); diff --git a/internal/store/postgres_test.go b/internal/store/postgres_test.go index 94e5980..f1884f8 100644 --- a/internal/store/postgres_test.go +++ b/internal/store/postgres_test.go @@ -44,7 +44,7 @@ func openPG(t *testing.T) *DB { `TRUNCATE thread_messages, threads, memory_notes, usage, usage_5m, usage_1h, usage_1d, api_keys, memberships, key_claims, limits, models, sessions, users, projects, teams, organizations, leadership, control_planes, files, batches, batch_results, request_bodies, - eval_runs, eval_cases, facility_holds, node_labels RESTART IDENTITY CASCADE`) + eval_runs, eval_cases, facility_holds, node_labels, federation_peers RESTART IDENTITY CASCADE`) if err != nil { t.Fatalf("clearing the test database: %v", err) } diff --git a/notes/c24-federation.md b/notes/c24-federation.md new file mode 100644 index 0000000..deb7d27 --- /dev/null +++ b/notes/c24-federation.md @@ -0,0 +1,40 @@ +## CHANGELOG + +### Federating clusters: another llmgw cluster as a provider + +- `llmgw ctl federation add NAME --url URL --key-file FILE [--send-tags]` + makes another llmgw gateway a peer. It is checked first (its `/ready`, + timed, and its `/v1/models` with the key) and kept in the store (migration + 0029, `federation_peers`) with the models it offers. +- A peer's models join the catalog as `NAME/MODEL`, always prefixed, and + follow the peer's own list every `federation.interval` (10s). A model the + peer drops goes, unless a model here still falls back to it: then it is + kept withdrawn (in the catalog, sent nothing) until the fallback changes. + The peer's aliases and its own peers' models are not taken. +- A peer whose `/ready` fails, which refuses the key, or which does not answer + is taken out of routing until it answers again. `/ready` now counts only a + gateway's own endpoints, not its peers'. +- `llmgw ctl model fallbacks MODEL [FALLBACK...]` (and `PUT + /api/v1/models/{name}/fallbacks`) sets a runtime model's fallbacks, so + overflow can go to a peer's model. +- Requests to a peer carry `x-llmgw-hop` with the ids of the clusters they + have passed through (each cluster's id is kept in the store). A request that + comes back to a cluster is refused with 508 `federation_loop`, and a peer a + request came from is not sent it. +- The peer sees one key, this cluster's, and the request's `X-Request-Id`. + With `--send-tags`, also `x-llmgw-tags: federated_from=, + org=,team=,project=` (ids); off by default. +- In-cluster residency and `no_store` keep requests off peers, by name and as + fallbacks. Usage records the peer's model as served, and `peer:NAME/MODEL` + as the replica. +- `ctl federation list|remove|sync`, a Federation view in the console, + `llmgw_federation_peer_up`, `_latency_seconds` and `_models` metrics, admin + API `/api/v1/federation`, and a "Federating clusters" section in + docs/install.md. New `federation:` block in server.yaml (`name`, + `interval`). +- A response header the gateway already set (X-Request-Id, + x-llmgw-served-model) is no longer repeated from the upstream's. + +## BACKLOG + +- [x] 24. Multi-cluster federation — built: peers via `ctl federation` kept in the store, their models as PEER/MODEL synced from /v1/models (withdrawn-while-referenced), /ready health with latency and routing withdrawal, `ctl model fallbacks` for overflow, hop header loop refusal, residency/no_store hold, optional attribution tags, console view, metrics, docs. Left: server.yaml-declared models cannot name a peer model as a fallback (config catalog is validated before peers are read); per-peer CA / mTLS for peers behind a private CA; transitive federation (A→B→C models) deliberately not offered; latency is /ready RTT only (no per-request peer TTFT used in routing or weighting); the peer's adapters are not imported (listed as aliases); no console form to add/remove peers (CLI only); key file must exist on every control plane.