Federation: another llmgw cluster as a provider, its models as PEER/MODEL

ctl federation add/list/remove/sync keeps peers in the store (migration
0029); their /v1/models are offered here as external models named
PEER/MODEL and kept in step, a model still named by a fallback kept
withdrawn. Peers are checked at /ready, timed, and taken out of routing
while down. Requests to a peer carry x-llmgw-hop and are refused (508) when
they come back to a cluster they passed through; residency and no_store
keep requests off peers; X-Request-Id and, with --send-tags, attribution
tags go along. ctl model fallbacks sets overflow to a peer. Console view,
metrics, admin API routes and docs.

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:21:30 +02:00
1 parent d41104a452
commit 40e9fd2ab1
33 files changed
+2944 -23

No files matched your search

+8
View File
@@ -32,6 +32,7 @@ import (
"gitea.mynodes.xyz/ilker/llmgw/internal/control"
"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/inventory"
"gitea.mynodes.xyz/ilker/llmgw/internal/control/nvlink"
"gitea.mynodes.xyz/ilker/llmgw/internal/control/reconciler"
@@ -58,6 +59,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.
@@ -172,6 +176,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)
@@ -1810,6 +1815,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})
})
}
+24 -7
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,13 +57,14 @@ 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.",
"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.",
}
// Routes is every endpoint Register mounts.
@@ -210,6 +212,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{}},