Merge C24 federation: another llmgw cluster as a provider

Conflicts with facility-aware placement (config, control, admin API routes,
store test tables) all keep both; the generated reference and OpenAPI
document regenerated.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Ilker ManapandClaude Opus 5.5 committed 2026-10-05 01:29:35 +02:00
commit 33334c8f33
33 files changed
+2945 -24

No files matched your search

+351
View File
@@ -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.",
+57
View File
@@ -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=<federation.name>,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`
+41
View File
@@ -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`
+1
View File
@@ -168,6 +168,7 @@ func newCtlCmd() *cobra.Command {
newAuditCmd(client, &asJSON),
newNotifyCmd(client, &asJSON),
newUsageCmd(client, &asJSON),
newFederationCmd(client, &asJSON),
newCertCmd(),
)
return cmd
+211
View File
@@ -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=<this cluster>,org=,team=,project=<ids>, 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
}
+1 -1
View File
@@ -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
}
+55
View File
@@ -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
}
+14 -2
View File
@@ -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:
+80
View File
@@ -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()
}
+37
View File
@@ -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)
}
}
}
+17
View File
@@ -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.
//
+8
View File
@@ -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
+123
View File
@@ -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})
})
}
+25 -8
View File
@@ -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{}},
+8
View File
@@ -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)
+46 -3
View File
@@ -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()
}
+661
View File
@@ -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
}
+42
View File
@@ -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)
}
}
+90 -4
View File
@@ -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.
+6
View File
@@ -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{
+7
View File
@@ -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.
+155
View File
@@ -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
}
+594
View File
@@ -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()
}
+20 -3
View File
@@ -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})
}
+14 -1
View File
@@ -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 {
+13 -1
View File
@@ -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()) }
+1
View File
@@ -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.
+131
View File
@@ -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()
}
+50
View File
@@ -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)
}
}
@@ -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 ''
);
@@ -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 ''
);
+1 -1
View File
@@ -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)
}
+40
View File
@@ -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=<federation.name>,
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.