ilker/llmgw
2
0 Fork 1
Code Issues Pull Requests Actions Packages Projects Releases 95 Wiki Activity

The cluster's settings live in the store, and every change takes effect

Browse Source 
Everything in server.yaml that is not one machine's own is now a versioned
document in the store (cluster_settings), the same on every control plane.
The first start against a store imports the file's cluster keys, with the
baseline guard policy when there is none; after that those keys in a file
are ignored, with a warning.

Each control plane polls the document's version and its own server.yaml's
hash. A change is validated and applied: guard live (Gate.Replace), anything
else by a graceful restart that systemd brings back, standbys first and the
leader 30s later. A version that does not load on a machine is not applied;
at start the newest one that loads is used, so a bad edit cannot keep every
control plane down.

llmgw ctl config show/get/set/unset/edit/history/rollback, the admin API
under /api/v1/settings, and the console's Settings screen, which layers the
store's document over the machine's file, write it. Commands beside the
control plane that read cluster keys (init, node add, server add, backup,
upgrade) read the layered configuration; init --bundle writes offline to
the store once it holds the settings.

A standby that becomes the leader re-reads models, engine choices and
facility holds from the store.

Retires appendGuardIfMissing: the import does it.

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 15:25:23 +02:00
1 parent cbbfce235e
commit 6289b01ed6
57 files changed
+2579 -224
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

BACKLOG.md
+50
View File
@@ -761,6 +761,56 @@ portal is not done — the page shows; the console and ctl change.
---
## E. Every setting takes effect when it is changed (2026-10-05)
Asked for after a site visit: an upgrade left server.yaml without guardrails,
a node on another subnet stayed lost until firewall_networks was edited by
hand, and nothing in server.yaml, agent.yaml or a generated unit applies until
somebody restarts the right service on every machine. The rule from here on:
**a file holds only what one machine needs to start; everything else is the
cluster's, kept in the store, changed through ctl, the console or the web,
audited, and in force on every control plane within seconds.**
### E1. Cluster settings in the store (L) — in progress
- Table `cluster_settings` (versioned YAML document, who and why). The
cluster-wide part of server.yaml lives there; server.yaml keeps the
per-machine keys only (listeners, TLS paths, store DSN, secrets, tracing,
monitoring, mirror.api/tls, backup_dir/key, endpoints).
- First start against a store with no document imports the file's cluster
keys (and the baseline guard policy when there is none); after that those
keys in a file are ignored, with a warning.
- Each control plane polls the document's version and its own server.yaml's
hash. A change is validated, then applied: `guard` live, the rest by a
graceful restart of the process (standbys first, the leader last), which
systemd brings back. No "restart now?" anywhere.
- `llmgw ctl config show|history|get|set|unset|edit|rollback`; the console's
settings screen edits the store for cluster keys; admin API
`GET/PUT /api/v1/settings`, `/history`, `/rollback`.
- A standby that becomes leader re-reads the catalog, engine choices and
facility state from the store (it kept what it read at start).
- Retires `appendGuardIfMissing` (server init): the import does it.
### E2. Generated files from settings (L)
Every machine runs the agent (root) with roles — node, control plane, mirror,
store member — and the agent renders and applies what was written once by
init: the llmgw nft table, the mirror's unit flags, keepalived, etcd and
Patroni configuration. The firewall's allowed set is the cluster's membership
(every node, control plane, mirror and store member) plus
`firewall.extra` from the store; 7443 stays open (mTLS). A node added or
removed changes every machine's rules. Retires `firewall.Admit` and
`firewall_networks` in server.yaml (imported into the store).
### E3. agent.yaml down to bootstrap (M)
agent.yaml keeps the control planes' addresses, its certificate and its id.
Labels, fabrics, engines, heartbeat and supervisor settings move to the
store per node and reach the agent over its stream; a change to the file is
noticed and the agent restarts itself.
### E4. Web settings page (S)
The same document, edited on /admin/settings, for whoever has no terminal.
---
## D. Facts worth knowing before starting
- Admin API: `127.0.0.1:9090`, no auth, `ctlClient` in `internal/cli/ctl.go`.
CHANGELOG.md
+21 -16
View File
@@ -23,8 +23,29 @@ APIs change between minors without a deprecation period.
## Unreleased
### Changed
- **The cluster's settings live in the store, and every change is in force
within seconds.** Everything in server.yaml that is not one machine's own
(guardrails, limits and prices, routing, engines, notifications, the
mirror's policy, declared models, offline, federation, facility budgets) is
a versioned document in the store, the same on every control plane.
`llmgw ctl config` shows, sets, edits, lists and rolls it back, and the
console's Settings screen edits it. A change is checked, kept with who made
it and why, and applied: guardrails live, anything else by every control
plane starting again on it, standbys first and the leader last. A change to
a machine's own keys in its server.yaml is noticed and applied the same way.
No "restart now?" any more, and no second copy to keep in step. The first
start against a store brings in the file's cluster keys, with the baseline
guard policy when the file had none, so an upgraded cluster gets its
guardrails; after that, those keys in a server.yaml are ignored, with a
warning.
### Fixed
- **A standby that becomes the leader reads the store again**: models added
or removed, engine builds chosen and domains held through the leader before
it, rather than placing on what it read when it started.
- **A node on another network joins**: `node add` set up the node's firewall
but left the control plane's to the networks it was installed with, so a node
on another subnet was installed and then never heard from (lost). It now lets
@@ -32,22 +53,6 @@ APIs change between minors without a deprecation period.
when a router is between them — and says so; a network already let in
changes nothing.
- **An upgraded control plane gets the guardrails**: `ctl upgrade` replaces the
binary but leaves server.yaml as it is, and one written before guardrails has no
`guard:` section, so nothing was checked. `llmgw server init`, run again on such
a machine, now adds the baseline policy (and fetches the judge model), keeping
the old file beside it as `server.yaml.bak-<time>`. A file that already says
`guard:`, even empty, is left alone.
- `ctl upgrade` on a control plane whose binary was installed by hand said
"this machine already runs" it and left the old server running: it read
its own binary's version. It now asks the server what it runs, and
restarts it when only the binary is new.
- `ctl upgrade` gave a node 15 seconds to answer the upgrade, while the
control plane waits up to five minutes for the agent to fetch and check
the release; a node that was upgrading stopped the run. It now waits as
long as the control plane does.
## v0.11.4 — 2026-10-05
Two control planes, run in anger on the reference cluster.
docs/admin-api.openapi.json
+221 -1
View File
@@ -83,6 +83,9 @@
{
"name": "restart"
},
{
"name": "settings"
},
{
"name": "summary"
},
@@ -1156,7 +1159,7 @@
"delete": {
"operationId": "deleteModelsName",
"summary": "Remove a model added at runtime",
"description": "A model declared in configuration is refused: it would come back on the next restart.",
"description": "A model declared in the cluster's settings is refused: it is changed there.",
"tags": [
"models"
],
@@ -2963,6 +2966,165 @@
}
}
},
"/api/v1/settings": {
"get": {
"operationId": "getSettings",
"summary": "The cluster's settings in force, or one earlier version",
"tags": [
"settings"
],
"parameters": [
{
"name": "version",
"in": "query",
"description": "An earlier version; the newest when absent.",
"schema": {
"type": "string"
}
}
],
"responses": {
"200": {
"description": "OK",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/SettingsView"
}
}
}
},
"401": {
"$ref": "#/components/responses/Unauthorized"
},
"default": {
"$ref": "#/components/responses/Error"
}
}
},
"put": {
"operationId": "putSettings",
"summary": "Change the cluster's settings: in force on every control plane within seconds",
"tags": [
"settings"
],
"parameters": [
{
"$ref": "#/components/parameters/Actor"
}
],
"requestBody": {
"required": true,
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/SettingsRequest"
}
}
}
},
"responses": {
"200": {
"description": "OK",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/SettingsView"
}
}
}
},
"401": {
"$ref": "#/components/responses/Unauthorized"
},
"default": {
"$ref": "#/components/responses/Error"
}
}
}
},
"/api/v1/settings/history": {
"get": {
"operationId": "getSettingsHistory",
"summary": "Versions of the cluster's settings, newest first",
"tags": [
"settings"
],
"parameters": [
{
"name": "limit",
"in": "query",
"description": "How many; 20 when absent.",
"schema": {
"type": "string"
}
}
],
"responses": {
"200": {
"description": "OK",
"content": {
"application/json": {
"schema": {
"type": "array",
"items": {
"$ref": "#/components/schemas/SettingsView"
}
}
}
}
},
"401": {
"$ref": "#/components/responses/Unauthorized"
},
"default": {
"$ref": "#/components/responses/Error"
}
}
}
},
"/api/v1/settings/rollback": {
"post": {
"operationId": "postSettingsRollback",
"summary": "Put an earlier version of the cluster's settings back in force",
"tags": [
"settings"
],
"parameters": [
{
"$ref": "#/components/parameters/Actor"
}
],
"requestBody": {
"required": true,
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/SettingsRollbackRequest"
}
}
}
},
"responses": {
"200": {
"description": "OK",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/SettingsView"
}
}
}
},
"401": {
"$ref": "#/components/responses/Unauthorized"
},
"default": {
"$ref": "#/components/responses/Error"
}
}
}
},
"/api/v1/summary": {
"get": {
"operationId": "getSummary",
@@ -5350,6 +5512,64 @@
}
}
},
"SettingsRequest": {
"type": "object",
"description": "SettingsRequest is a new version of the cluster's settings, made from the version Base, which must still be the newest.",
"properties": {
"base": {
"type": "integer",
"format": "int64"
},
"body": {
"type": "string"
},
"note": {
"type": "string"
}
}
},
"SettingsRollbackRequest": {
"type": "object",
"description": "SettingsRollbackRequest puts an earlier version's settings back in force, as a new version.",
"properties": {
"version": {
"type": "integer",
"format": "int64"
},
"note": {
"type": "string"
}
}
},
"SettingsView": {
"type": "object",
"description": "SettingsView is one version of the cluster's settings.",
"properties": {
"version": {
"type": "integer",
"format": "int64"
},
"body": {
"type": "string"
},
"note": {
"type": "string"
},
"at": {
"type": "string",
"format": "date-time"
},
"by": {
"type": "string"
},
"local_keys": {
"type": "array",
"items": {
"type": "string"
}
}
}
},
"ShardView": {
"type": "object",
"description": "ShardView is one shard of a replica.",
docs/console.md
+14 -8
View File
@@ -270,14 +270,20 @@ page's text.
## Settings
**server.yaml** (on a node, **agent.yaml**) is every key of the file, as a
tree of menus built from the configuration reference: the value set, or the
one that applies when nothing is, its type, and what it is for. Enter changes
a value or opens a block; on a list or map of blocks, Enter opens an entry and
**D** removes one. **U** puts a key back to its default. **S** saves: the file
is checked the way the service reads it before anything is written, and its
comments are kept. The service reads it when it starts, so the console offers
to restart it. Another control plane keeps its own copy.
**server.yaml** (on a node, **agent.yaml**) is every key, as a tree of menus
built from the configuration reference: the value set, or the one that
applies when nothing is, its type, and what it is for. Enter changes a value
or opens a block; on a list or map of blocks, Enter opens an entry and **D**
removes one. **U** puts a key back to its default. **S** saves.
On a control plane, the cluster's keys are read from and saved to the
cluster's settings in the store — the same on every control plane, in force
on all of them within seconds (`llmgw ctl config` does the same from a
shell) — and this machine's own keys (listeners, certificates, the store's
address) to its server.yaml, which the server notices and applies by itself.
Either is checked before anything is kept, and the file's comments survive.
On a node, agent.yaml is read when the agent starts, so the console offers
to restart it.
## Do
docs/guardrails.md
+7 -3
View File
@@ -5,9 +5,13 @@ policy can check, what it can do, and how one is attached to a team.
`llmgw server init` writes a `baseline` policy that applies to every
request, with Llama Guard 3 1B as its judge on a CPU; see
[install.md](install.md#2-the-control-plane) for what it checks. Without a
`guard:` section in `server.yaml` nothing is checked and nothing costs
anything. With one, every request is judged once it is authorised and
[install.md](install.md#2-the-control-plane) for what it checks. The
policies are part of the cluster's settings, kept in the store: `llmgw ctl
config edit` (or the console's Settings) changes them, and the change is in
force on every control plane within seconds, without a restart. A cluster
whose settings were brought in from a server.yaml without a `guard:` section
gets the `baseline` policy, without a judge. With no policies, nothing is
checked and nothing costs anything. With one, every request is judged once it is authorised and
within its limits, with its stored conversation spliced in, and before it is
sent to a model; and the answer is judged before the client sees it.
docs/install.md
+32 -2
View File
@@ -241,8 +241,38 @@ password, change it, and issue yourself a key from the profile page.
### The configuration
`/etc/llmgw/server.yaml` is yours from here on. Three things in it are worth
knowing before you edit it:
There are two halves, and each change to either is in force within seconds,
on every machine it concerns, without anyone restarting anything.
**The cluster's settings** — guardrails, limits and prices, routing, the
reconciler, engines, notifications, the mirror's policy, models declared by
hand, offline, federation, facility budgets, backups' recipients — are kept in
the store, so every control plane runs on the same ones. Change them with
```
llmgw ctl config # what is in force, and its version
llmgw ctl config set router.retries 3 --note "flaky upstream"
llmgw ctl config edit # the whole document in $EDITOR
llmgw ctl config history # who changed what, when, and why
llmgw ctl config rollback 4 # back to version 4, as a new version
```
or on the console's **Settings** screen. A change is checked before it is
kept — one that would not load is refused — and recorded in the audit log.
Guardrails change live; anything else has every control plane start again on
the new settings, the standbys first and the leader half a minute later, so
one of them is always serving. Engines keep running through it.
The first control plane to start against a store brings in its server.yaml's
cluster keys — with a `baseline` guard policy, if the file had none — and
from then on those keys in any server.yaml are not read (the server says so
when it starts).
**This machine's own keys** stay in `/etc/llmgw/server.yaml`: `listen`,
`admin_listen`, `metrics_listen`, `agent_listen`, `agent_tls`, `auth.store`,
`secrets`, `tracing`, `monitoring`, `mirror.api` and `mirror.tls`, the backup
directory and key. The server notices the file change and starts again on it
by itself. Three things are worth knowing before you edit it:
- `admin_listen` is on loopback on purpose, and the server refuses any other
address unless `admin_remote: true`. Its API under `/api` adds models and
internal/cli/backup.go
+1 -1
View File
@@ -54,7 +54,7 @@ func newBackupCmd() *cobra.Command {
"the newest archive as llmgw_backup_last_success_timestamp_seconds, and\n" +
"BackupTooOld fires when it is more than a day old.",
RunE: func(cmd *cobra.Command, _ []string) error {
cfg, err := config.Load(configPath)
cfg, err := loadEffectiveConfig(configPath)
if err != nil {
return err
}
internal/cli/bundle.go
+1 -1
View File
@@ -323,7 +323,7 @@ func loadServerConfig(path string) (config.Server, bool) {
if _, err := os.Stat(path); err != nil {
return config.Server{}, false
}
cfg, err := config.Load(path)
cfg, err := loadEffectiveConfig(path)
if err != nil {
return config.Server{}, false
}
internal/cli/clustersettings.go
+361
View File
@@ -0,0 +1,361 @@
package cli
import (
"context"
"crypto/sha256"
"errors"
"fmt"
"log/slog"
"os"
"slices"
"strings"
"time"
"gitea.mynodes.xyz/ilker/llmgw/internal/config"
"gitea.mynodes.xyz/ilker/llmgw/internal/settings"
storepkg "gitea.mynodes.xyz/ilker/llmgw/internal/store"
)
// settingsPoll is how often a control plane asks whether the cluster's
// settings, or its own server.yaml, changed.
const settingsPoll = 5 * time.Second
// clusterSettings is what a control plane runs on, and where it came from:
// its server.yaml for its own keys, the store's settings for the rest.
type clusterSettings struct {
cfg config.Server
eff map[string]any // the settings, as written, cfg was made from
cluster map[string]any // the store's document, nil without a store
version int64
fileHash [32]byte
}
// loadClusterSettings reads the configuration a control plane starts on.
// Without a store it is the file. With one, the store's settings take the
// place of the file's cluster keys; a store with none yet is given the
// file's, so the first control plane to start against it brings them in.
func loadClusterSettings(ctx context.Context, path string, log *slog.Logger) (clusterSettings, error) {
var s clusterSettings
b, file, err := readSettingsFile(path)
if err != nil {
return s, err
}
s.fileHash = sha256.Sum256(b)
fileCfg, err := config.FromMap(file)
if err != nil {
return s, fileError(path, err)
}
if fileCfg.Auth.Store == "" {
s.cfg, s.eff = fileCfg, file
return s, nil
}
st, err := storepkg.Open(ctx, fileCfg.Auth.Store)
if err != nil {
return s, fmt.Errorf("the store, for the cluster's settings: %w", err)
}
defer st.Close()
doc, err := st.ClusterSettings(ctx)
if err != nil {
return s, fmt.Errorf("reading the cluster's settings: %w", err)
}
if doc.Version == 0 {
if doc, err = importSettings(ctx, st, path, file, log); err != nil {
return s, err
}
} else if ignored := config.Ignored(file); len(ignored) > 0 {
log.Warn("server.yaml sets keys the cluster's settings now hold; they are not read "+
"(llmgw ctl config shows what is in force)", "keys", strings.Join(ignored, ", "))
}
// The newest version that this machine can run on. One that cannot — a
// path only another control plane has — must not keep this one down,
// where nothing could be used to put it right.
versions := []storepkg.SettingsVersion{doc}
if older, err := st.ClusterSettingsHistory(ctx, 20); err == nil && len(older) > 1 {
versions = append(versions, older[1:]...)
}
var firstErr error
for _, v := range versions {
cluster, err := config.Parse([]byte(v.Body))
if err == nil {
eff := config.Layer(file, cluster)
var cfg config.Server
if cfg, err = config.FromMap(eff); err == nil {
if v.Version != doc.Version {
log.Error("the cluster's newest settings do not load on this machine; running on an older version",
"newest", doc.Version, "running", v.Version, "err", firstErr)
}
s.cfg, s.eff, s.cluster, s.version = cfg, eff, cluster, doc.Version
return s, nil
}
}
if firstErr == nil {
firstErr = err
}
}
return s, fmt.Errorf("the cluster's settings (version %d) do not load on this machine: %w", doc.Version, firstErr)
}
// importSettings records a server.yaml's cluster keys as the cluster's first
// settings, with the baseline guard policy when the file has none: a cluster
// installed before guardrails gets them on its first start after an upgrade.
func importSettings(ctx context.Context, st *storepkg.DB, path string, file map[string]any,
log *slog.Logger,
) (storepkg.SettingsVersion, error) {
_, cluster := config.Split(file)
note := "imported from " + path
if _, ok := cluster["guard"]; !ok {
baseline, err := config.Parse([]byte(guardSection(false)))
if err == nil {
cluster["guard"] = baseline["guard"]
note += ", with the baseline guard policy it did not have"
}
}
body, err := config.Marshal(cluster)
if err != nil {
return storepkg.SettingsVersion{}, err
}
host, _ := os.Hostname()
_, err = st.PutClusterSettings(ctx, 0, string(body), host, note)
switch {
case errors.Is(err, storepkg.ErrSettingsChanged):
// Another control plane brought its own in first: those are the
// cluster's.
case err != nil:
return storepkg.SettingsVersion{}, fmt.Errorf("recording the cluster's settings: %w", err)
default:
log.Info("the cluster's settings are now kept in the store; server.yaml keeps this machine's own",
"from", path)
}
return st.ClusterSettings(ctx)
}
func readSettingsFile(path string) ([]byte, map[string]any, error) {
if path == "" {
return nil, map[string]any{}, nil
}
b, err := os.ReadFile(path) //nolint:gosec // the configuration
if err != nil {
return nil, nil, fmt.Errorf("config file: %w", err)
}
m, err := config.Parse(b)
if err != nil {
return nil, nil, fmt.Errorf("parsing %s: %w", path, err)
}
return b, m, nil
}
func fileError(path string, err error) error {
if path == "" {
return err
}
return fmt.Errorf("%s: %w", path, err)
}
// settingsWatch follows the cluster's settings and this machine's
// server.yaml, and puts every change in force.
type settingsWatch struct {
path string
st *storepkg.DB // nil: the file only
cur clusterSettings
// live applies a change without a restart, reporting whether it could.
live func(cfg config.Server, changed []string) bool
// restart has the process start again on the new settings.
restart func(reason string)
log *slog.Logger
}
func (w *settingsWatch) run(ctx context.Context) {
t := time.NewTicker(settingsPoll)
defer t.Stop()
for {
select {
case <-ctx.Done():
return
case <-t.C:
w.check(ctx)
}
}
}
// check reads what changed, if anything, and applies it.
func (w *settingsWatch) check(ctx context.Context) {
version := w.cur.version
if w.st != nil {
v, err := w.st.ClusterSettingsVersion(ctx)
if err != nil {
return // asked again in a few seconds
}
version = v
}
b, file, err := readSettingsFile(w.path)
if err != nil {
w.log.Error("server.yaml cannot be read; running on what it said before", "err", err)
return
}
hash := sha256.Sum256(b)
if version == w.cur.version && hash == w.cur.fileHash {
return
}
cluster := w.cur.cluster
eff := file
if w.st != nil {
if version != w.cur.version {
doc, err := w.st.ClusterSettings(ctx)
if err != nil {
return
}
if cluster, err = config.Parse([]byte(doc.Body)); err != nil {
w.log.Error("the cluster's settings do not parse; running on what was in force",
"version", doc.Version, "err", err)
w.cur.version, w.cur.fileHash = version, hash
return
}
}
eff = config.Layer(file, cluster)
}
cfg, err := config.FromMap(eff)
if err != nil {
w.log.Error("the changed settings do not load on this machine; running on what was in force",
"version", version, "err", err)
w.cur.version, w.cur.fileHash = version, hash
return
}
changed := config.Changed(w.cur.eff, eff)
w.cur = clusterSettings{cfg: cfg, eff: eff, cluster: cluster, version: version, fileHash: hash}
if len(changed) == 0 {
return
}
if w.live != nil && w.live(cfg, changed) {
w.log.Info("settings changed and in force", "version", version, "changed", strings.Join(changed, ", "))
return
}
w.restart(fmt.Sprintf("settings version %d changed %s", version, strings.Join(changed, ", ")))
}
// liveSettings are the keys a control plane puts in force without starting
// again: what it reads at every use, or not at all.
var liveSettings = []string{
"guard",
// Read by server init, node add and backups, not by the server.
"firewall_networks", "backup_recipients", "backup_copy_to",
}
// onlyLive is whether every changed key is one of liveSettings.
func onlyLive(changed []string) bool {
for _, c := range changed {
top, _, _ := strings.Cut(c, ".")
if !slices.Contains(liveSettings, top) {
return false
}
}
return true
}
// checkClusterSettings refuses a settings document that would not load,
// laid over the server.yaml at path, or that sets one machine's keys.
func checkClusterSettings(path string, body []byte) error {
cluster, err := config.Parse(body)
if err != nil {
return err
}
if err := config.CheckCluster(cluster); err != nil {
return err
}
_, file, err := readSettingsFile(path)
if err != nil {
return err
}
_, err = config.FromMap(config.Layer(file, cluster))
return err
}
// loadEffectiveConfig is what the control plane on this machine runs on:
// its server.yaml with the cluster's settings from the store laid over it,
// for the commands beside it that read cluster keys — init, node add,
// backups. The file alone when the store cannot be reached or holds none.
func loadEffectiveConfig(path string) (config.Server, error) {
_, file, err := readSettingsFile(path)
if err != nil {
return config.Default(), err
}
fileCfg, err := config.FromMap(file)
if err != nil {
return fileCfg, fileError(path, err)
}
cluster, ok := storedSettings(fileCfg)
if !ok {
return fileCfg, nil
}
cfg, err := config.FromMap(config.Layer(file, cluster))
if err != nil {
return fileCfg, nil //nolint:nilerr // the file still says what this machine is
}
return cfg, nil
}
// storedSettings are the cluster's settings, when the store a configuration
// names can be reached and holds some.
func storedSettings(cfg config.Server) (map[string]any, bool) {
if cfg.Auth.Store == "" {
return nil, false
}
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
st, err := storepkg.Open(ctx, cfg.Auth.Store)
if err != nil {
return nil, false
}
defer st.Close()
doc, err := st.ClusterSettings(ctx)
if err != nil || doc.Version == 0 {
return nil, false
}
cluster, err := config.Parse([]byte(doc.Body))
return cluster, err == nil
}
// changeStoredSettings changes the cluster's settings from a command run on
// a control plane (init), when the store holds them: changed is false when
// it does not, and the file is then where the change belongs.
func changeStoredSettings(path, note string, change func(doc *settings.Doc) error) (changed bool, err error) {
_, file, err := readSettingsFile(path)
if err != nil {
return false, err
}
fileCfg, err := config.FromMap(file)
if err != nil || fileCfg.Auth.Store == "" {
return false, nil //nolint:nilerr // no store to keep them in
}
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
st, err := storepkg.Open(ctx, fileCfg.Auth.Store)
if err != nil {
return false, nil //nolint:nilerr // not set up yet: the file it is
}
defer st.Close()
cur, err := st.ClusterSettings(ctx)
if err != nil || cur.Version == 0 {
return false, err
}
doc, err := settings.FromBytes("the cluster's settings", []byte(cur.Body))
if err != nil {
return false, err
}
if err := change(doc); err != nil {
return false, err
}
body, err := doc.Bytes()
if err != nil {
return false, err
}
if err := checkClusterSettings(path, body); err != nil {
return false, err
}
who := "llmgw"
if u := os.Getenv("SUDO_USER"); u != "" {
who = u
}
_, err = st.PutClusterSettings(ctx, cur.Version, string(body), who, note)
return err == nil, err
}
internal/cli/clustersettings_test.go
+139
View File
@@ -0,0 +1,139 @@
package cli
import (
"context"
"io"
"log/slog"
"os"
"path/filepath"
"strings"
"testing"
"gitea.mynodes.xyz/ilker/llmgw/internal/config"
storepkg "gitea.mynodes.xyz/ilker/llmgw/internal/store"
)
func settingsFixture(t *testing.T, extra string) (path, store string) {
t.Helper()
dir := t.TempDir()
store = filepath.Join(dir, "llmgw.db")
path = filepath.Join(dir, "server.yaml")
body := "listen: \":8080\"\nagent_insecure: true\nauth:\n enabled: true\n store: " + store +
"\nrouter:\n retries: 2\n" + extra
if err := os.WriteFile(path, []byte(body), 0o600); err != nil {
t.Fatal(err)
}
return path, store
}
var quiet = slog.New(slog.NewTextHandler(io.Discard, nil))
func TestTheFirstStartBringsTheFilesSettingsInWithGuardrails(t *testing.T) {
path, store := settingsFixture(t, "")
s, err := loadClusterSettings(t.Context(), path, quiet)
if err != nil {
t.Fatal(err)
}
if s.version != 1 || s.cfg.Router.Retries != 2 {
t.Errorf("version %d retries %d", s.version, s.cfg.Router.Retries)
}
if len(s.cfg.Guard.Policies) == 0 || s.cfg.Guard.Policies["baseline"].Judge != nil {
t.Errorf("a file without guard got %+v, want the baseline without a judge", s.cfg.Guard)
}
st, _ := storepkg.Open(t.Context(), store)
defer st.Close()
doc, _ := st.ClusterSettings(t.Context())
if strings.Contains(doc.Body, "listen") || strings.Contains(doc.Body, store) || !strings.Contains(doc.Body, "retries: 2") {
t.Errorf("the store holds a machine's keys, or lost the cluster's:\n%s", doc.Body)
}
// From then on the store's settings are the cluster's, whatever a
// file says.
_ = os.WriteFile(path, []byte(strings.Replace(mustRead(t, path), "retries: 2", "retries: 9", 1)), 0o600)
again, err := loadClusterSettings(t.Context(), path, quiet)
if err != nil || again.version != 1 || again.cfg.Router.Retries != 2 {
t.Errorf("second start: version %d retries %d err %v", again.version, again.cfg.Router.Retries, err)
}
}
func mustRead(t *testing.T, p string) string {
b, err := os.ReadFile(p)
if err != nil {
t.Fatal(err)
}
return string(b)
}
func TestEveryChangeIsAppliedLiveOrByStartingAgain(t *testing.T) {
path, store := settingsFixture(t, "")
s, err := loadClusterSettings(t.Context(), path, quiet)
if err != nil {
t.Fatal(err)
}
st, _ := storepkg.Open(t.Context(), store)
defer st.Close()
var lived [][]string
var restarts []string
w := &settingsWatch{path: path, st: st, cur: s, log: quiet,
live: func(_ config.Server, changed []string) bool {
lived = append(lived, changed)
return onlyLive(changed)
},
restart: func(why string) { restarts = append(restarts, why) },
}
ctx := context.Background()
put := func(body string) {
t.Helper()
v, _ := st.ClusterSettingsVersion(ctx)
if _, err := st.PutClusterSettings(ctx, v, body, "test", ""); err != nil {
t.Fatal(err)
}
}
cur, _ := st.ClusterSettings(ctx)
w.check(ctx) // nothing changed
if len(lived)+len(restarts) != 0 {
t.Fatalf("acted on no change: %v %v", lived, restarts)
}
put(strings.Replace(cur.Body, "injection: flag", "injection: block", 1))
w.check(ctx)
if len(restarts) != 0 || len(lived) != 1 || lived[0][0] != "guard" {
t.Fatalf("a guard change: lived %v restarts %v", lived, restarts)
}
cur, _ = st.ClusterSettings(ctx)
put(strings.Replace(cur.Body, "retries: 2", "retries: 4", 1))
w.check(ctx)
if len(restarts) != 1 || !strings.Contains(restarts[0], "router") || w.cur.cfg.Router.Retries != 4 {
t.Fatalf("a router change: restarts %v retries %d", restarts, w.cur.cfg.Router.Retries)
}
// A machine's own key, in its file.
_ = os.WriteFile(path, []byte(strings.Replace(mustRead(t, path), `":8080"`, `":8090"`, 1)), 0o600)
w.check(ctx)
if len(restarts) != 2 || !strings.Contains(restarts[1], "listen") {
t.Fatalf("a listener change: %v", restarts)
}
// Settings that do not load are not put in force.
cur, _ = st.ClusterSettings(ctx)
put(cur.Body + "control:\n timezone: Mars/Olympus\n")
w.check(ctx)
if len(restarts) != 2 || w.cur.cfg.Control.Timezone != "" {
t.Errorf("bad settings were applied: %v %q", restarts, w.cur.cfg.Control.Timezone)
}
}
func TestAClusterDocumentCannotSetAMachinesKeys(t *testing.T) {
path, _ := settingsFixture(t, "")
if err := checkClusterSettings(path, []byte("listen: \":1\"\n")); err == nil {
t.Error("listen was accepted")
}
if err := checkClusterSettings(path, []byte("control:\n timezone: Mars/Olympus\n")); err == nil {
t.Error("a timezone that is not one was accepted")
}
if err := checkClusterSettings(path, []byte("router:\n retries: 3\n")); err != nil {
t.Error(err)
}
}
internal/cli/ctl.go
+1
View File
@@ -173,6 +173,7 @@ func newCtlCmd() *cobra.Command {
newCapacityCmd(client, &asJSON),
newQuotaCmd(client, &asJSON),
newGuardCmd(client, &asJSON),
newConfigCmd(client, &asJSON),
newAuditCmd(client, &asJSON),
newNotifyCmd(client, &asJSON),
newUsageCmd(client, &asJSON),
internal/cli/ctl_config.go
+377
View File
@@ -0,0 +1,377 @@
package cli
import (
"bytes"
"encoding/json"
"errors"
"fmt"
"io"
"os"
"os/exec"
"path/filepath"
"strconv"
"strings"
"time"
"github.com/spf13/cobra"
"gitea.mynodes.xyz/ilker/llmgw/internal/config"
"gitea.mynodes.xyz/ilker/llmgw/internal/control/adminapi"
"gitea.mynodes.xyz/ilker/llmgw/internal/settings"
)
func newConfigCmd(client func() *ctlClient, asJSON *bool) *cobra.Command {
var version int64
c := &cobra.Command{
Use: "config",
Short: "The cluster's settings: show, change, and go back",
Long: "Everything in server.yaml that is not one machine's own — guardrails,\n" +
"limits, pricing, routing, engines, notifications, the mirror's policy —\n" +
"is the cluster's, kept in the store and the same on every control plane.\n" +
"A change is checked, kept as a new version, and in force on every control\n" +
"plane within seconds: live where it can be (guard), otherwise each control\n" +
"plane starts again on it, standbys first and the leader last.\n\n" +
"With no subcommand, prints the settings in force.\n\n" +
"A machine's own keys (" + strings.Join(config.LocalKeys(), ", ") + ")\n" +
"stay in its /etc/llmgw/server.yaml; a change there is noticed and applied\n" +
"on that machine the same way.",
Example: " llmgw ctl config\n" +
" llmgw ctl config get router.retries\n" +
" llmgw ctl config set router.retries 3 --note 'flaky upstream'\n" +
" llmgw ctl config set firewall_networks '[10.20.0.0/16]'\n" +
" llmgw ctl config edit\n" +
" llmgw ctl config history\n" +
" llmgw ctl config rollback 4",
Args: cobra.NoArgs,
RunE: func(cmd *cobra.Command, _ []string) error {
path := "/api/v1/settings"
if version > 0 {
path += "?version=" + strconv.FormatInt(version, 10)
}
var v adminapi.SettingsView
if err := client().get(cmd.Context(), path, &v); err != nil {
return err
}
if *asJSON {
return emitJSON(v)
}
fmt.Fprintf(cmd.ErrOrStderr(), "# version %d, %s by %s%s\n", v.Version, v.At.Local().Format(time.DateTime),
orUnknown(v.By), noteSuffix(v.Note))
fmt.Fprint(cmd.OutOrStdout(), v.Body)
return nil
},
}
c.Flags().Int64Var(&version, "version", 0, "an earlier version")
c.AddCommand(newConfigGetCmd(client), newConfigSetCmd(client), newConfigUnsetCmd(client),
newConfigEditCmd(client), newConfigHistoryCmd(client, asJSON), newConfigRollbackCmd(client))
return c
}
func orUnknown(s string) string {
if s == "" {
return "unknown"
}
return s
}
func noteSuffix(n string) string {
if n == "" {
return ""
}
return ": " + n
}
// keyPath reads a dotted key, refusing one machine's own.
func keyPath(key string) ([]string, error) {
path := strings.Split(key, ".")
if config.IsLocal(path) {
return nil, fmt.Errorf("%s is each machine's own: change it in that machine's /etc/llmgw/server.yaml, "+
"where a change is applied by itself too", key)
}
return path, nil
}
func newConfigGetCmd(client func() *ctlClient) *cobra.Command {
return &cobra.Command{
Use: "get KEY",
Short: "One key of the cluster's settings, as YAML",
Args: cobra.ExactArgs(1),
RunE: func(cmd *cobra.Command, args []string) error {
path, err := keyPath(args[0])
if err != nil {
return err
}
var v adminapi.SettingsView
if err := client().get(cmd.Context(), "/api/v1/settings", &v); err != nil {
return err
}
doc, err := settings.FromBytes("settings", []byte(v.Body))
if err != nil {
return err
}
n := doc.Node(path)
if n == nil {
return fmt.Errorf("%s is not set in the cluster's settings; its default applies (docs/reference.md)", args[0])
}
sub, err := settings.FromBytes("value", nil)
if err != nil {
return err
}
if err := sub.SetNode([]string{"v"}, n); err != nil {
return err
}
b, err := sub.Bytes()
if err != nil {
return err
}
out := strings.TrimPrefix(string(b), "v:")
fmt.Fprintln(cmd.OutOrStdout(), strings.TrimSpace(dedent(out)))
return nil
},
}
}
// dedent takes off the two spaces a block nested under one key carries.
func dedent(s string) string {
lines := strings.Split(s, "\n")
for i, l := range lines {
lines[i] = strings.TrimPrefix(l, " ")
}
return strings.Join(lines, "\n")
}
// changeSettings reads the newest settings, changes them, and puts the
// result in force, saying what that does.
func changeSettings(cmd *cobra.Command, client func() *ctlClient, note string,
change func(doc *settings.Doc) error,
) error {
ctx := cmd.Context()
var cur adminapi.SettingsView
if err := client().get(ctx, "/api/v1/settings", &cur); err != nil {
return err
}
doc, err := settings.FromBytes("settings", []byte(cur.Body))
if err != nil {
return err
}
if err := change(doc); err != nil {
return err
}
b, err := doc.Bytes()
if err != nil {
return err
}
return putSettings(cmd, client, cur, b, note)
}
func putSettings(cmd *cobra.Command, client func() *ctlClient, cur adminapi.SettingsView, body []byte, note string) error {
before, _ := config.Parse([]byte(cur.Body))
after, err := config.Parse(body)
if err != nil {
return err
}
changed := config.Changed(before, after)
if len(changed) == 0 {
fmt.Fprintln(cmd.OutOrStdout(), "nothing changed")
return nil
}
req, _ := json.Marshal(adminapi.SettingsRequest{Base: cur.Version, Body: string(body), Note: note})
var v adminapi.SettingsView
if err := client().put(cmd.Context(), "/api/v1/settings", req, &v); err != nil {
return err
}
fmt.Fprintf(cmd.OutOrStdout(), "version %d: %s\n", v.Version, strings.Join(changed, ", "))
if onlyLive(changed) {
fmt.Fprintln(cmd.OutOrStdout(), "in force on every control plane within seconds")
} else {
fmt.Fprintln(cmd.OutOrStdout(), "every control plane starts again on it within seconds, standbys first, "+
"the leader half a minute later; engines keep serving")
}
return nil
}
func newConfigSetCmd(client func() *ctlClient) *cobra.Command {
var note string
c := &cobra.Command{
Use: "set KEY VALUE",
Short: "Change one key of the cluster's settings; VALUE is YAML",
Example: " llmgw ctl config set router.retries 3\n" +
" llmgw ctl config set guard.default '[baseline, finance]'\n" +
" llmgw ctl config set notify.webhooks '[{url: https://hooks.example/llmgw}]'",
Args: cobra.ExactArgs(2),
RunE: func(cmd *cobra.Command, args []string) error {
path, err := keyPath(args[0])
if err != nil {
return err
}
return changeSettings(cmd, client, note, func(doc *settings.Doc) error {
return doc.SetYAML(path, args[1])
})
},
}
c.Flags().StringVar(&note, "note", "", "why, kept with the version")
return c
}
func newConfigUnsetCmd(client func() *ctlClient) *cobra.Command {
var note string
c := &cobra.Command{
Use: "unset KEY",
Short: "Remove one key from the cluster's settings, so its default applies",
Args: cobra.ExactArgs(1),
RunE: func(cmd *cobra.Command, args []string) error {
path, err := keyPath(args[0])
if err != nil {
return err
}
return changeSettings(cmd, client, note, func(doc *settings.Doc) error {
if doc.Node(path) == nil {
return fmt.Errorf("%s is not set", args[0])
}
doc.Unset(path)
return nil
})
},
}
c.Flags().StringVar(&note, "note", "", "why, kept with the version")
return c
}
func newConfigEditCmd(client func() *ctlClient) *cobra.Command {
var note string
c := &cobra.Command{
Use: "edit",
Short: "Edit the cluster's settings in $EDITOR",
Long: "Opens the settings in force in $EDITOR (vi when unset). Saved, they are\n" +
"checked and kept as a new version; a document that does not load is\n" +
"opened again with the reason, until it does or is left unchanged.",
Args: cobra.NoArgs,
RunE: func(cmd *cobra.Command, _ []string) error {
var cur adminapi.SettingsView
if err := client().get(cmd.Context(), "/api/v1/settings", &cur); err != nil {
return err
}
dir, err := os.MkdirTemp("", "llmgw-settings")
if err != nil {
return err
}
defer os.RemoveAll(dir)
file := filepath.Join(dir, "settings.yaml")
body := []byte(cur.Body)
problem := ""
for {
text := body
if problem != "" {
text = append([]byte("# NOT SAVED: "+strings.ReplaceAll(problem, "\n", " ")+"\n"), stripProblem(body)...)
}
if err := os.WriteFile(file, text, 0o600); err != nil {
return err
}
if err := runEditor(file, cmd.InOrStdin(), cmd.OutOrStdout(), cmd.ErrOrStderr()); err != nil {
return err
}
edited, err := os.ReadFile(file) //nolint:gosec // written above
if err != nil {
return err
}
edited = stripProblem(edited)
if bytes.Equal(bytes.TrimSpace(edited), bytes.TrimSpace([]byte(cur.Body))) {
fmt.Fprintln(cmd.OutOrStdout(), "nothing changed")
return nil
}
err = putSettings(cmd, client, cur, edited, note)
if err == nil {
return nil
}
if strings.Contains(err.Error(), "changed since they were read") {
// Somebody else's change came first: opening the editor
// again on the old base would only be refused again.
return fmt.Errorf("%w; your edit is not saved", err)
}
body, problem = edited, err.Error()
}
},
}
c.Flags().StringVar(&note, "note", "", "why, kept with the version")
return c
}
func stripProblem(b []byte) []byte {
if bytes.HasPrefix(b, []byte("# NOT SAVED: ")) {
if i := bytes.IndexByte(b, '\n'); i >= 0 {
return b[i+1:]
}
}
return b
}
func runEditor(file string, in io.Reader, out, errOut io.Writer) error {
editor := os.Getenv("VISUAL")
if editor == "" {
editor = os.Getenv("EDITOR")
}
if editor == "" {
editor = "vi"
}
parts := strings.Fields(editor)
c := exec.Command(parts[0], append(parts[1:], file)...) //nolint:gosec // the operator's own editor
c.Stdin, c.Stdout, c.Stderr = in, out, errOut
if err := c.Run(); err != nil {
return fmt.Errorf("%s: %w", editor, err)
}
return nil
}
func newConfigHistoryCmd(client func() *ctlClient, asJSON *bool) *cobra.Command {
var limit int
c := &cobra.Command{
Use: "history",
Short: "Versions of the cluster's settings, newest first",
Args: cobra.NoArgs,
RunE: func(cmd *cobra.Command, _ []string) error {
var list []adminapi.SettingsView
if err := client().get(cmd.Context(), "/api/v1/settings/history?limit="+strconv.Itoa(limit), &list); err != nil {
return err
}
if *asJSON {
return emitJSON(list)
}
w := tabTo(cmd.OutOrStdout())
fmt.Fprintln(w, "VERSION\tWHEN\tBY\tNOTE")
for _, v := range list {
fmt.Fprintf(w, "%d\t%s\t%s\t%s\n", v.Version, v.At.Local().Format(time.DateTime), orUnknown(v.By), v.Note)
}
return w.Flush()
},
}
c.Flags().IntVar(&limit, "limit", 20, "how many")
return c
}
func newConfigRollbackCmd(client func() *ctlClient) *cobra.Command {
var note string
c := &cobra.Command{
Use: "rollback VERSION",
Short: "Put an earlier version of the cluster's settings back in force, as a new version",
Args: cobra.ExactArgs(1),
RunE: func(cmd *cobra.Command, args []string) error {
n, err := strconv.ParseInt(args[0], 10, 64)
if err != nil || n < 1 {
return errors.New("VERSION is a number from 'llmgw ctl config history'")
}
var cur, old adminapi.SettingsView
if err := client().get(cmd.Context(), "/api/v1/settings", &cur); err != nil {
return err
}
if err := client().get(cmd.Context(), "/api/v1/settings?version="+args[0], &old); err != nil {
return err
}
if note == "" {
note = "back to version " + args[0]
}
return putSettings(cmd, client, cur, []byte(old.Body), note)
},
}
c.Flags().StringVar(&note, "note", "", "why, kept with the version")
return c
}
internal/cli/node.go
+3 -3
View File
@@ -167,7 +167,7 @@ func (na *nodeAdd) prepare() error {
// are not known until the node says what it has, and are not applied
// here; doctor's verdict on vLLM is a warning either way.
if o.VLLMVersion == "" {
if cfg, err := config.Load(na.serverCfg); err == nil {
if cfg, err := loadEffectiveConfig(na.serverCfg); err == nil {
o.VLLMVersion = cfg.WantedEngines()["vllm"]
if o.VLLMVersion == "" {
o.VLLMVersion = "none"
@@ -175,7 +175,7 @@ func (na *nodeAdd) prepare() error {
}
}
// The node's ports — its engines', its metrics — for the cluster only.
if cfg, err := config.Load(na.serverCfg); err == nil && !o.NoFirewall {
if cfg, err := loadEffectiveConfig(na.serverCfg); err == nil && !o.NoFirewall {
from, to := cfg.EnginePorts()
o.FirewallPorts = []firewall.Port{firewall.Engines(from, to), firewall.NodeExporter}
o.FirewallAllow, _ = cfg.FirewallAllow()
@@ -183,7 +183,7 @@ func (na *nodeAdd) prepare() error {
// Where the control plane upgrades nodes from, so a node joins on the
// same release from the same place: an offline cluster's mirror.
if o.Releases == "" {
if cfg, err := config.Load(na.serverCfg); err == nil {
if cfg, err := loadEffectiveConfig(na.serverCfg); err == nil {
o.Releases = cfg.ReleasesBase()
}
if o.Releases == "" {
internal/cli/server.go
+80 -4
View File
@@ -16,6 +16,7 @@ import (
"slices"
"strings"
"sync"
"sync/atomic"
"syscall"
"time"
@@ -80,10 +81,13 @@ func newServerCmd() *cobra.Command {
RunE: func(cmd *cobra.Command, _ []string) error {
log := logger()
cfg, err := config.Load(configPath)
// server.yaml for this machine's own keys, the store's settings
// for the cluster's: the same on every control plane.
boot, err := loadClusterSettings(cmd.Context(), configPath, log)
if err != nil {
return err
}
cfg := boot.cfg
if listen != "" {
cfg.Listen = listen
}
@@ -166,7 +170,7 @@ func newServerCmd() *cobra.Command {
}
gate.SetJudge(rt.Judge, rt.GuardFormat)
rt.SetGuard(gate)
rt.SetGuardLibrary(cfg.Guard)
rt.SetGuardLibrary(gate.Library)
if err := rt.RegisterMetrics(gate); err != nil {
return fmt.Errorf("guard metrics: %w", err)
}
@@ -175,6 +179,10 @@ func newServerCmd() *cobra.Command {
ctx, stop := signal.NotifyContext(cmd.Context(), syscall.SIGINT, syscall.SIGTERM)
defer stop()
// Cancelled to start again on changed settings.
ctx, cancelRun := context.WithCancel(ctx)
defer cancelRun()
var restartReason atomic.Pointer[string]
rt.Start(ctx)
defer rt.Stop()
@@ -341,12 +349,16 @@ func newServerCmd() *cobra.Command {
evals = newEvalService(cfg, rt, admin, log)
api.SetEval(evals)
if st := rt.Store(); st != nil && gate != nil {
api.SetGuard(&adminapi.GuardAdmin{Library: cfg.Guard, Store: st, Invalidate: gate.Invalidate})
api.SetGuard(&adminapi.GuardAdmin{Library: gate.Library, Store: st, Invalidate: gate.Invalidate})
}
if st := rt.Store(); st != nil {
// Removing a node revokes its certificate.
api.SetRevoker(st.RevokeNode)
api.SetAudit(st)
// The cluster's settings, each change checked against this
// machine's own keys before it is kept.
api.SetSettings(&adminapi.Settings{Store: st, LocalKeys: config.LocalKeys(),
Check: func(body []byte) error { return checkClusterSettings(configPath, body) }})
}
// Catalog changes need somewhere to persist, so this is wired
// only when there is a store. Without one, the endpoints say so
@@ -378,6 +390,17 @@ func newServerCmd() *cobra.Command {
} else {
admin.ReplaceCatalog(ctx, built)
}
// And again on becoming the leader: models added or
// removed through the leader before are in the store.
if l := admin.Leadership(); l != nil {
l.OnLead(func() {
if built, err := svc.Load(ctx); err != nil {
log.Error("could not read the stored models on becoming the leader", "err", err)
} else {
admin.ReplaceCatalog(ctx, built)
}
})
}
api.SetModels(svc)
catalogSvc = svc
// And the mirror page, so a model that has just been
@@ -627,6 +650,46 @@ func newServerCmd() *cobra.Command {
}
}
// Every change to the settings in force: what can change live
// does, the rest starts this process again, which systemd brings
// back on them.
watch := &settingsWatch{path: configPath, st: rt.Store(), cur: boot, log: log,
live: func(next config.Server, changed []string) bool {
if !onlyLive(changed) {
return false
}
if slices.ContainsFunc(changed, func(c string) bool { return strings.HasPrefix(c, "guard") }) {
if gate == nil {
// Built at start, and only when there are policies.
return len(next.Guard.Policies) == 0
}
if err := gate.Replace(next.Guard); err != nil {
log.Error("the changed guard policies are not valid; the old ones stay in force", "err", err)
}
}
return true
},
restart: func(reason string) {
if os.Getenv("INVOCATION_ID") == "" {
log.Warn("the settings changed in a way this process applies only when it starts; "+
"it is not run by systemd, so start it again by hand", "why", reason)
return
}
// The leader last, so its standbys are already on the new
// settings when it hands over to one of them.
delay := time.Duration(0)
if admin != nil && admin.Leadership() != nil && admin.Leadership().IsLeader() && rt.Store() != nil {
if cps, err := rt.Store().ControlPlanes(ctx); err == nil && len(cps) > 1 {
delay = 30 * time.Second
}
}
log.Warn("starting again to apply the settings", "why", reason, "in", delay)
restartReason.Store(&reason)
time.AfterFunc(delay, cancelRun)
},
}
go watch.run(ctx)
g, gctx := errgroup.WithContext(ctx)
g.Go(func() error { return serve(apiSrv, apiLn) })
g.Go(func() error { return serve(adminSrv, adminLn) })
@@ -658,7 +721,12 @@ func newServerCmd() *cobra.Command {
_ = adminSrv.Shutdown(sctx)
return apiSrv.Shutdown(sctx)
})
return g.Wait()
err = g.Wait()
if r := restartReason.Load(); r != nil {
// Not a failure, but said as one, so systemd starts it again.
return fmt.Errorf("stopped to start again on changed settings (%s)", *r)
}
return err
},
}
@@ -756,6 +824,14 @@ func startControlPlane(ctx context.Context, cfg config.Server, cat *catalog.Live
ctl.SetEngineChoices(control.NewEngineChoices(ctx, st, log))
// Domains held for maintenance and labels set on nodes.
ctl.Facility().UseStore(ctx, st)
// A standby read these when it started; the leader before it may
// have changed them since.
lead.OnLead(func() {
if ch := ctl.EngineChoices(); ch != nil {
ch.Reload(ctx)
}
ctl.Facility().Reload(ctx)
})
log.Info("competing to be the control plane that places work",
"owner", lead.Owner())
}
internal/cli/serveradd.go
+1 -1
View File
@@ -155,7 +155,7 @@ func (sa *serverAdd) run(ctx context.Context) error {
// firewall keeps the new control plane's internal ports to the networks it
// reaches this one on. Not fatal: it serves without one.
func (sa *serverAdd) firewall(ctx context.Context) error {
cfg, err := config.Load(filepath.Join(sa.etcDir, "server.yaml"))
cfg, err := loadEffectiveConfig(filepath.Join(sa.etcDir, "server.yaml"))
if err != nil {
return err
}
internal/cli/serverinit.go
+2 -14
View File
@@ -144,9 +144,7 @@ func newServerInitCmd() *cobra.Command {
"SQLite or PostgreSQL, is moved in. Run again, it changes nothing.\n\n" +
"Each step is skipped when its result already exists, so it is safe to run\n" +
"again: to finish an install that stopped, or with another --host to\n" +
"reissue the certificate. It never overwrites server.yaml or the authority;\n" +
"to a server.yaml from before guardrails it adds the guard section, the old\n" +
"file kept beside it.\n\n" +
"reissue the certificate. It never overwrites server.yaml or the authority.\n\n" +
"The authority's key stays in " + defaultPKIDir + " on this machine. Move it\n" +
"somewhere safer once the nodes are onboarded if this machine is exposed.",
Args: cobra.NoArgs,
@@ -290,7 +288,7 @@ func (in *serverInit) judgeWanted() bool {
// configuration written before init installed one is never changed, so the
// model is not fetched for nothing.
func (in *serverInit) configJudgesWith(model string) bool {
cfg, err := config.Load(in.configPath())
cfg, err := loadEffectiveConfig(in.configPath())
if err != nil {
return false
}
@@ -550,16 +548,6 @@ func (in *serverInit) ensureConfig(context.Context) error {
path := in.configPath()
if _, err := os.Stat(path); err == nil {
in.kept("%s (never overwritten)", path)
// A file from before guardrails has no guard: at all, and an
// upgrade alone leaves it unguarded. One that says guard:, even
// empty, was decided on and stays as it is.
added, err := appendGuardIfMissing(path, in.judgeWanted())
if err != nil {
return fmt.Errorf("%s: %w", path, err)
}
if added {
in.done("server.yaml: guardrails added (baseline policy), the old file kept beside it")
}
} else {
contact := in.admin
if contact == "" {
internal/cli/serverinit_bundle.go
+21 -2
View File
@@ -12,6 +12,7 @@ import (
"gitea.mynodes.xyz/ilker/llmgw/internal/bundle"
"gitea.mynodes.xyz/ilker/llmgw/internal/config"
"gitea.mynodes.xyz/ilker/llmgw/internal/mirror"
"gitea.mynodes.xyz/ilker/llmgw/internal/settings"
"gitea.mynodes.xyz/ilker/llmgw/internal/tlsconf"
)
@@ -60,7 +61,7 @@ func (in *serverInit) offline() bool {
if in.bundle != "" {
return true
}
cfg, err := config.Load(in.configPath())
cfg, err := loadEffectiveConfig(in.configPath())
return err == nil && cfg.Offline.Enabled()
}
@@ -76,7 +77,7 @@ func (in *serverInit) mirrorRoot() string {
// server.yaml that the cluster is offline. With a mirror on another
// machine, the files are imported there; here only the checksums are kept.
func (in *serverInit) ensureBundle(ctx context.Context) error {
cfg, err := config.Load(in.configPath())
cfg, err := loadEffectiveConfig(in.configPath())
if err != nil {
return err
}
@@ -129,6 +130,24 @@ func (in *serverInit) ensureBundle(ctx context.Context) error {
if sums == config.DefaultOfflineChecksums {
sums = ""
}
// In the cluster's settings once the store holds them; in server.yaml
// before, where the first start brings it in.
stored, err := changeStoredSettings(in.configPath(), "offline, from server init --bundle", func(doc *settings.Doc) error {
if err := doc.SetYAML([]string{"offline", "mirror"}, files); err != nil {
return err
}
if sums != "" {
return doc.SetYAML([]string{"offline", "checksums"}, sums)
}
return nil
})
if err != nil {
return err
}
if stored {
in.done("the cluster's settings: offline, nodes install from the mirror at %s", files)
return nil
}
if err := appendOfflineBlock(in.configPath(), files, sums); err != nil {
return err
}
internal/cli/serverinit_cluster.go
-38
View File
@@ -13,8 +13,6 @@ import (
"strings"
"time"
"go.yaml.in/yaml/v3"
"gitea.mynodes.xyz/ilker/llmgw/internal/agent/discovery"
"gitea.mynodes.xyz/ilker/llmgw/internal/config"
"gitea.mynodes.xyz/ilker/llmgw/internal/control/adminapi"
@@ -667,39 +665,3 @@ func waitFor(ctx context.Context, limit time.Duration, what string, done func()
}
}
}
// appendGuardIfMissing adds init's guard section to a server.yaml that has
// no guard key, keeping the file as it was beside it.
func appendGuardIfMissing(path string, judge bool) (bool, error) {
b, err := os.ReadFile(path) //nolint:gosec // server.yaml
if err != nil {
return false, err
}
var top map[string]any
if err := yaml.Unmarshal(b, &top); err != nil {
return false, err
}
if _, ok := top["guard"]; ok {
return false, nil
}
if err := appendBlock(path, b, guardSection(judge)); err != nil {
return false, err
}
return true, nil
}
// appendBlock writes b plus block to path, with b kept in a dated backup.
func appendBlock(path string, b []byte, block string) error {
info, err := os.Stat(path)
if err != nil {
return err
}
backup := fmt.Sprintf("%s.bak-%s", path, time.Now().Format("20060102-150405"))
if err := os.WriteFile(backup, b, 0o640); err != nil {
return err
}
if len(b) > 0 && b[len(b)-1] != '\n' {
b = append(b, '\n')
}
return os.WriteFile(path, append(append(b, '\n'), block...), info.Mode().Perm())
}
internal/cli/serverinit_cluster_test.go
-37
View File
@@ -207,40 +207,3 @@ func TestInitSaysAtOnceWhenAModelFitsNowhere(t *testing.T) {
t.Errorf("took %s to say so", time.Since(start))
}
}
func TestAnOldConfigurationGetsGuardrailsOnce(t *testing.T) {
body := fmt.Sprintf(serverConfigTemplate, "/etc/llmgw", "/etc/llmgw", "/etc/llmgw",
"/var/lib/llmgw/llmgw.db", "ops@example.com", "")
path := filepath.Join(t.TempDir(), "server.yaml")
if err := os.WriteFile(path, []byte(body), 0o640); err != nil {
t.Fatal(err)
}
added, err := appendGuardIfMissing(path, true)
if err != nil || !added {
t.Fatalf("added=%v err=%v, want the guard section added", added, err)
}
cfg, err := config.Load(path)
if err != nil {
t.Fatal(err)
}
if err := cfg.Validate(); err != nil {
t.Fatal(err)
}
if len(cfg.Guard.Default) != 1 || cfg.Guard.Policies["baseline"].Judge == nil {
t.Errorf("guard = %+v, want the baseline with its judge", cfg.Guard)
}
if backups, _ := filepath.Glob(path + ".bak-*"); len(backups) != 1 {
t.Errorf("backups = %v, want the old file kept", backups)
}
if added, err := appendGuardIfMissing(path, true); err != nil || added {
t.Errorf("second run: added=%v err=%v, want nothing changed", added, err)
}
// guard: written by hand, even empty, is a decision and stays.
if err := os.WriteFile(path, []byte(body+"guard: {}\n"), 0o640); err != nil {
t.Fatal(err)
}
if added, err := appendGuardIfMissing(path, true); err != nil || added {
t.Errorf("empty guard: added=%v err=%v, want it left alone", added, err)
}
}
internal/cli/serverinit_firewall.go
+1 -1
View File
@@ -19,7 +19,7 @@ import (
// the engines and etcd when they are here. The client API is not among
// them; it is meant to be reached.
func (in *serverInit) ensureFirewall(ctx context.Context) error {
cfg, err := config.Load(in.configPath())
cfg, err := loadEffectiveConfig(in.configPath())
if err != nil {
return err
}
internal/cli/serverinit_test.go
+2 -4
View File
@@ -126,10 +126,8 @@ func TestInitNeverOverwritesTheConfiguration(t *testing.T) {
if err := in.run(context.Background()); err != nil {
t.Fatalf("%v\n%s", err, out)
}
// Kept as written; only the guardrails it predates are added after it.
got, _ := os.ReadFile(in.configPath())
if !strings.HasPrefix(string(got), mine) || !strings.Contains(string(got[len(mine):]), "\nguard:\n") {
t.Errorf("server.yaml is not the old file with guard: after it:\n%s", got)
if got, _ := os.ReadFile(in.configPath()); string(got) != mine {
t.Error("an existing server.yaml was rewritten")
}
// And the store it names is the one used.
if _, err := os.Stat(filepath.Join(in.stateDir, "mine.db")); err != nil {
internal/cli/tui_console.go
+81 -25
View File
@@ -4,6 +4,7 @@ import (
"bufio"
"bytes"
"context"
"encoding/json"
"fmt"
"io"
"net"
@@ -138,12 +139,15 @@ type consoleModel struct {
node bool
// The settings.
file string
doc *settings.Doc
cfg any // the file as loaded, for the values that apply
levels []settingsLevel
docErr error
saveNote string
file string
doc *settings.Doc
cfg any // the file as loaded, for the values that apply
// settingsBase is the version of the cluster's settings the screen
// read, when the cluster keeps them (doc is then layered over them).
settingsBase int64
levels []settingsLevel
docErr error
saveNote string
// A question open as a dialog.
dialog string // "", "target", "edit", "addname", "saved", "confirm-leave", "error"
@@ -219,6 +223,26 @@ func (m *consoleModel) loadSettings() {
} else if !m.node {
m.cfg = config.Default()
}
// The cluster's keys from the cluster's settings, when the control
// plane answers with them; its own keys from its file.
if !m.node && m.client != nil {
ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
defer cancel()
var v adminapi.SettingsView
if err := m.client.get(ctx, "/api/v1/settings", &v); err == nil && v.Version > 0 {
if cdoc, err := settings.FromBytes("the cluster's settings", []byte(v.Body)); err == nil {
m.doc.Layer(cdoc, config.IsLocal)
m.settingsBase = v.Version
if file, err := config.ReadFile(m.file); err == nil {
if cluster, err := config.Parse([]byte(v.Body)); err == nil {
if cfg, err := config.FromMap(config.Layer(file, cluster)); err == nil {
m.cfg = cfg
}
}
}
}
}
}
if root, err := settings.Root(m.file); err == nil && len(m.levels) == 0 {
m.levels = []settingsLevel{{title: filepathBase(m.file), block: root, kind: settings.Block}}
}
@@ -643,10 +667,10 @@ func (m consoleModel) peopleView() bool {
}
func (m consoleModel) leave() (tea.Model, tea.Cmd) {
if m.doc != nil && m.doc.Changed > 0 {
if m.doc != nil && m.doc.Pending() > 0 {
m.dialog, m.dialogTitle = "confirm-leave", "Unsaved changes"
m.dialogText = fmt.Sprintf("%d change(s) to %s are not saved.\nLeave without saving them?\n\n%s",
m.doc.Changed, m.file, sDim.Render("Y leave · N stay"))
m.dialogText = fmt.Sprintf("%d change(s) are not saved.\nLeave without saving them?\n\n%s",
m.doc.Pending(), sDim.Render("Y leave · N stay"))
return m, nil
}
return m, tea.Quit
@@ -663,7 +687,7 @@ func (m consoleModel) ask(title, question string, n int) (tea.Model, tea.Cmd) {
func (m consoleModel) dialogKey(k tea.KeyMsg) (tea.Model, tea.Cmd) {
switch m.dialog {
case "error":
case "error", "note":
m.dialog = ""
return m, nil
case "confirm-leave":
@@ -917,28 +941,60 @@ func (m consoleModel) settingsKey(k tea.KeyMsg) (tea.Model, tea.Cmd) {
}
func (m consoleModel) save() (tea.Model, tea.Cmd) {
if m.doc.Changed == 0 {
if m.doc.Pending() == 0 {
m.dialog, m.dialogTitle, m.dialogText = "error", "Nothing to save", "No change has been made."
return m, nil
}
backup, err := m.doc.Save(func(p string) error { _, err := m.load(p); return err })
if err != nil {
notSaved := func(err error) (tea.Model, tea.Cmd) {
m.dialog, m.dialogTitle = "error", "Not saved"
m.dialogText = "The configuration would not load:\n\n" + sBad.Render(err.Error()) + "\n\n" + sDim.Render("Nothing was written. Any key to go back.")
m.dialogText = "The configuration would not load:\n\n" + sBad.Render(err.Error()) + "\n\n" +
sDim.Render("Nothing was written. Any key to go back.")
return m, nil
}
var said []string
// The cluster's settings first: checked by the control plane against
// its own file, and in force everywhere once kept.
if c := m.doc.Cluster(); c != nil && c.Changed > 0 {
body, err := c.Bytes()
if err != nil {
return notSaved(err)
}
req, _ := json.Marshal(adminapi.SettingsRequest{Base: m.settingsBase, Body: string(body), Note: "console"})
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
var v adminapi.SettingsView
if err := m.client.put(ctx, "/api/v1/settings", req, &v); err != nil {
return notSaved(err)
}
c.Changed = 0
said = append(said, fmt.Sprintf("The cluster's settings are version %d: in force on every control plane "+
"within seconds.\nllmgw ctl config history lists the versions; rollback goes back to one.", v.Version))
}
fileSaved := m.doc.Changed > 0
if fileSaved {
backup, err := m.doc.Save(func(p string) error { _, err := m.load(p); return err })
if err != nil {
return notSaved(err)
}
line := "Saved " + m.file + "."
if backup != "" {
line += " The old file is " + backup + "."
}
said = append(said, line)
}
m.loadSettings()
note := ""
if backup != "" {
note = "The old file is " + backup + ".\n"
if !m.node {
if fileSaved {
said = append(said, "llmgw-server notices the change and applies it within seconds.")
}
m.dialog, m.dialogTitle = "note", "Saved"
m.dialogText = strings.Join(said, "\n\n") + "\n\n" + sDim.Render("Any key to go back.")
return m, nil
}
m.dialog, m.dialogTitle = "saved", "Saved"
m.dialogText = "Saved " + m.file + ".\n" + note +
fmt.Sprintf("\n%s reads it when it starts. Restart it now?\n", m.service())
if !m.node {
m.dialogText += sDim.Render("Another control plane keeps its own copy: make the change there too.") + "\n"
}
m.dialogText += "\n" + sDim.Render("Y restart · N later")
m.dialogText = strings.Join(said, "\n") + "\n" +
fmt.Sprintf("\n%s reads it when it starts. Restart it now?\n", m.service()) +
"\n" + sDim.Render("Y restart · N later")
return m, nil
}
@@ -1011,8 +1067,8 @@ func (m consoleModel) View() string {
if n := runningJobs(m.data.jobs); i == cJobs && n > 0 {
label += fmt.Sprintf(" (%d)", n)
}
if i == cSettings && m.doc != nil && m.doc.Changed > 0 {
label += fmt.Sprintf(" (%d)", m.doc.Changed)
if i == cSettings && m.doc != nil && m.doc.Pending() > 0 {
label += fmt.Sprintf(" (%d)", m.doc.Pending())
}
nav = append(nav, navItem{Label: label, Current: i == m.menu, Dim: !m.entryOK(i)})
}
internal/cli/tui_hf.go
+1 -2
View File
@@ -13,7 +13,6 @@ import (
tea "github.com/charmbracelet/bubbletea"
"gitea.mynodes.xyz/ilker/llmgw/internal/config"
"gitea.mynodes.xyz/ilker/llmgw/internal/control/modelserve"
"gitea.mynodes.xyz/ilker/llmgw/internal/hf"
"gitea.mynodes.xyz/ilker/llmgw/internal/modelplan"
@@ -57,7 +56,7 @@ var hfClients sync.Map
func (m consoleModel) registry() *hf.Client {
endpoint := ""
if cfg, err := config.Load(m.file); err == nil {
if cfg, err := loadEffectiveConfig(m.file); err == nil {
endpoint = cfg.Mirror.Registry
}
c, _ := hfClients.LoadOrStore(endpoint, hf.New(endpoint, 20*time.Second))
internal/cli/tui_people.go
+1 -2
View File
@@ -12,7 +12,6 @@ import (
tea "github.com/charmbracelet/bubbletea"
"gitea.mynodes.xyz/ilker/llmgw/internal/config"
"gitea.mynodes.xyz/ilker/llmgw/internal/control/adminapi"
"gitea.mynodes.xyz/ilker/llmgw/internal/guard"
"gitea.mynodes.xyz/ilker/llmgw/internal/store"
@@ -118,7 +117,7 @@ func (m consoleModel) fetchPeople() tea.Cmd {
}
}
}
if cfg, err := config.Load(file); err == nil {
if cfg, err := loadEffectiveConfig(file); err == nil {
for name := range cfg.Guard.Policies {
d.guardPolicies = append(d.guardPolicies, name)
}
internal/cli/tui_setup_test.go
+57 -1
View File
@@ -1,6 +1,9 @@
package cli
import (
"encoding/json"
"net/http"
"net/http/httptest"
"os"
"path/filepath"
"slices"
@@ -164,7 +167,7 @@ func TestTheConsoleWatchesTheClusterAndEditsItsSettings(t *testing.T) {
cm = mm.(consoleModel)
cm.prompt.SetValue("5")
mm = press(cm, "enter", "s")
if d := mm.(consoleModel).dialog; d != "saved" {
if d := mm.(consoleModel).dialog; d != "note" {
t.Fatalf("after saving the dialog is %q: %s", d, mm.(consoleModel).dialogText)
}
b, _ := os.ReadFile(file)
@@ -244,3 +247,56 @@ func TestTheConsoleShowsTheGuardrails(t *testing.T) {
t.Errorf("window %q", guardWindows[c.guard.window])
}
}
func TestTheConsoleEditsTheClustersSettingsInTheStore(t *testing.T) {
dir := t.TempDir()
file := filepath.Join(dir, "server.yaml")
if err := os.WriteFile(file, []byte("listen: \":8080\"\n"), 0o644); err != nil {
t.Fatal(err)
}
var put adminapi.SettingsRequest
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
switch {
case r.Method == "GET" && r.URL.Path == "/api/v1/settings":
_ = json.NewEncoder(w).Encode(adminapi.SettingsView{Version: 7, Body: "router:\n retries: 2\n"})
case r.Method == "PUT" && r.URL.Path == "/api/v1/settings":
_ = json.NewDecoder(r.Body).Decode(&put)
_ = json.NewEncoder(w).Encode(adminapi.SettingsView{Version: 8, Body: put.Body})
default:
http.NotFound(w, r)
}
}))
defer srv.Close()
m := newConsoleModel(strings.TrimPrefix(srv.URL, "http://"), file)
m.width, m.height, m.loaded = 120, 40, true
m.menu, m.inPanel = cSettings, true
m.loadSettings()
for i, r := range m.settingsRows() {
if r.label == "router" {
m.row = i
}
}
var mm tea.Model = press(m, "enter")
cm := mm.(consoleModel)
for i, r := range cm.settingsRows() {
if r.label == "retries" {
cm.row = i
}
}
if v := cm.View(); !strings.Contains(v, "2") {
t.Errorf("the cluster's value is not shown:\n%s", v)
}
mm = press(cm, "enter")
cm = mm.(consoleModel)
cm.prompt.SetValue("5")
mm = press(cm, "enter", "s")
if d := mm.(consoleModel).dialog; d != "note" || !strings.Contains(mm.(consoleModel).dialogText, "version 8") {
t.Fatalf("after saving: %q %s", d, mm.(consoleModel).dialogText)
}
if put.Base != 7 || !strings.Contains(put.Body, "retries: 5") {
t.Errorf("sent %+v", put)
}
if b, _ := os.ReadFile(file); string(b) != "listen: \":8080\"\n" {
t.Errorf("the cluster's key went into the machine's file:\n%s", b)
}
}
internal/config/config.go
+21 -7
View File
@@ -14,7 +14,6 @@ import (
"github.com/knadh/koanf/parsers/yaml"
env "github.com/knadh/koanf/providers/env/v2"
"github.com/knadh/koanf/providers/file"
"github.com/knadh/koanf/v2"
"gitea.mynodes.xyz/ilker/llmgw/internal/control/catalog"
@@ -823,15 +822,30 @@ func Default() Server {
// An empty path loads the defaults, which is what makes `llmgw server` with a
// couple of --endpoint flags usable for a smoke test.
func Load(path string) (Server, error) {
cfg := Default()
k := koanf.New(".")
var b []byte
if path != "" {
if _, err := os.Stat(path); err != nil {
return cfg, fmt.Errorf("config file: %w", err)
return Default(), fmt.Errorf("config file: %w", err)
}
if err := k.Load(file.Provider(path), yaml.Parser()); err != nil {
return cfg, fmt.Errorf("parsing %s: %w", path, err)
var err error
if b, err = os.ReadFile(path); err != nil { //nolint:gosec // the configuration
return Default(), fmt.Errorf("config file: %w", err)
}
}
cfg, err := load(b)
if err != nil && path != "" && strings.HasPrefix(err.Error(), "parsing settings") {
return cfg, fmt.Errorf("parsing %s:%s", path, strings.TrimPrefix(err.Error(), "parsing settings:"))
}
return cfg, err
}
// load reads a YAML document of settings over the defaults.
func load(b []byte) (Server, error) {
cfg := Default()
k := koanf.New(".")
if len(b) > 0 {
if err := k.Load(bytesProvider(b), yaml.Parser()); err != nil {
return cfg, fmt.Errorf("parsing settings: %w", err)
}
}
internal/config/layer.go
+256
View File
@@ -0,0 +1,256 @@
package config
import (
"bytes"
"errors"
"fmt"
"maps"
"os"
"reflect"
"slices"
"strings"
"go.yaml.in/yaml/v3"
)
// localKeys are the keys of server.yaml that belong to one machine: where it
// listens, its certificates, how it reaches the store, its own secrets and
// files. Everything else is the cluster's, kept in the store and the same on
// every control plane (see Layer).
var localKeys = [][]string{
{"listen"}, {"admin_listen"}, {"metrics_listen"}, {"agent_listen"},
{"max_connections"}, {"admin_token_file"}, {"admin_remote"}, {"debug_pprof"},
{"agent_insecure"}, {"agent_tls"},
{"auth", "store"}, {"threads", "store"},
{"secrets"}, {"backup_dir"}, {"backup_copy_key"},
{"tracing"}, {"monitoring"},
{"mirror", "api"}, {"mirror", "tls"},
{"endpoints"},
}
// IsLocal is whether a key path is one machine's own, kept in its
// server.yaml, rather than the cluster's.
func IsLocal(path []string) bool {
for _, l := range localKeys {
if len(path) >= len(l) && slices.Equal(path[:len(l)], l) {
return true
}
}
return false
}
// LocalKeys are the per-machine keys, dotted, for saying which they are.
func LocalKeys() []string {
out := make([]string, len(localKeys))
for i, l := range localKeys {
out[i] = strings.Join(l, ".")
}
return out
}
// ReadFile is a configuration file as YAML maps, empty for no path.
func ReadFile(path string) (map[string]any, error) {
if path == "" {
return map[string]any{}, nil
}
b, err := os.ReadFile(path) //nolint:gosec // the configuration
if err != nil {
return nil, fmt.Errorf("config file: %w", err)
}
m, err := Parse(b)
if err != nil {
return nil, fmt.Errorf("parsing %s: %w", path, err)
}
return m, nil
}
// Parse reads a YAML document of settings.
func Parse(b []byte) (map[string]any, error) {
m := map[string]any{}
if len(bytes.TrimSpace(b)) == 0 {
return m, nil
}
if err := yaml.Unmarshal(b, &m); err != nil {
return nil, err
}
if m == nil {
m = map[string]any{}
}
return m, nil
}
// Split separates settings into one machine's and the cluster's.
func Split(m map[string]any) (local, cluster map[string]any) {
local, cluster = map[string]any{}, map[string]any{}
split(m, nil, local, cluster)
return local, cluster
}
func split(m map[string]any, path []string, local, cluster map[string]any) {
for k, v := range m {
p := append(slices.Clone(path), k)
switch {
case IsLocal(p):
local[k] = v
case hasLocalBelow(p):
sub, ok := v.(map[string]any)
if !ok {
cluster[k] = v
continue
}
l, c := map[string]any{}, map[string]any{}
split(sub, p, l, c)
if len(l) > 0 {
local[k] = l
}
if len(c) > 0 {
cluster[k] = c
}
default:
cluster[k] = v
}
}
}
// hasLocalBelow is whether some local key lies under a path: auth holds
// store, which is local, beside its policies, which are not.
func hasLocalBelow(path []string) bool {
for _, l := range localKeys {
if len(l) > len(path) && slices.Equal(l[:len(path)], path) {
return true
}
}
return false
}
// Layer is what a control plane runs on: its own keys from its server.yaml
// and every other one from the cluster's settings. The file's cluster keys
// are not read once the cluster has settings of its own; Ignored names them.
func Layer(file, cluster map[string]any) map[string]any {
local, _ := Split(file)
out := clone(local)
merge(out, cluster)
return out
}
// Ignored are the cluster keys a file has that the cluster's settings, once
// there, take the place of.
func Ignored(file map[string]any) []string {
_, c := Split(file)
return flatKeys(c, "")
}
func flatKeys(m map[string]any, prefix string) []string {
var out []string
for _, k := range slices.Sorted(maps.Keys(m)) {
p := prefix + k
if sub, ok := m[k].(map[string]any); ok && hasLocalBelow(strings.Split(p, ".")) {
out = append(out, flatKeys(sub, p+".")...)
continue
}
out = append(out, p)
}
return out
}
// CheckCluster refuses a cluster document that sets one machine's keys: a
// listener or a certificate path written there would be every control
// plane's, and none of them could start on it.
func CheckCluster(m map[string]any) error {
local, _ := Split(m)
if keys := flatKeys(local, ""); len(keys) > 0 {
return fmt.Errorf("%s belong to each machine's server.yaml, not to the cluster's settings",
strings.Join(keys, ", "))
}
return nil
}
// Changed are the keys whose values differ between two sets of settings,
// at the top level, and below it where a key mixes local and cluster ones.
func Changed(a, b map[string]any) []string {
var out []string
changed(a, b, nil, &out)
slices.Sort(out)
return out
}
func changed(a, b map[string]any, path []string, out *[]string) {
keys := map[string]bool{}
for k := range a {
keys[k] = true
}
for k := range b {
keys[k] = true
}
for k := range keys {
p := append(slices.Clone(path), k)
va, vb := a[k], b[k]
if reflect.DeepEqual(va, vb) {
continue
}
ma, okA := va.(map[string]any)
mb, okB := vb.(map[string]any)
if hasLocalBelow(p) && (okA || va == nil) && (okB || vb == nil) {
changed(ma, mb, p, out)
continue
}
*out = append(*out, strings.Join(p, "."))
}
}
// Marshal writes settings as YAML.
func Marshal(m map[string]any) ([]byte, error) {
if len(m) == 0 {
return nil, nil
}
var b bytes.Buffer
enc := yaml.NewEncoder(&b)
enc.SetIndent(2)
if err := enc.Encode(m); err != nil {
return nil, err
}
return b.Bytes(), enc.Close()
}
// FromMap is the configuration settings make: laid over the defaults, with
// LLMGW_ environment overrides, and validated.
func FromMap(m map[string]any) (Server, error) {
b, err := Marshal(m)
if err != nil {
return Default(), err
}
return load(b)
}
// bytesProvider hands koanf a document already read.
type bytesProvider []byte
func (b bytesProvider) ReadBytes() ([]byte, error) { return b, nil }
func (b bytesProvider) Read() (map[string]any, error) {
return nil, errors.New("bytesProvider reads bytes only")
}
func clone(m map[string]any) map[string]any {
out := make(map[string]any, len(m))
for k, v := range m {
if sub, ok := v.(map[string]any); ok {
v = clone(sub)
}
out[k] = v
}
return out
}
// merge lays src over dst: maps key by key, everything else replaced.
func merge(dst, src map[string]any) {
for k, v := range src {
if sv, ok := v.(map[string]any); ok {
if dv, ok := dst[k].(map[string]any); ok {
merge(dv, sv)
continue
}
v = clone(sv)
}
dst[k] = v
}
}
internal/config/layer_test.go
+99
View File
@@ -0,0 +1,99 @@
package config
import (
"slices"
"strings"
"testing"
)
const machineFile = `
listen: ":8080"
agent_listen: ":7443"
agent_insecure: true
auth:
enabled: true
store: /var/lib/llmgw/llmgw.db
default_rpm: 60
mirror:
api: http://127.0.0.1:8082
update_check: 6h
guard:
default: [baseline]
policies:
baseline:
input:
injection: flag
router:
retries: 2
`
func TestAFileSplitsIntoTheMachinesAndTheClusters(t *testing.T) {
file, err := Parse([]byte(machineFile))
if err != nil {
t.Fatal(err)
}
local, cluster := Split(file)
for _, k := range []string{"listen", "agent_listen", "agent_insecure"} {
if _, ok := local[k]; !ok {
t.Errorf("%s is not the machine's", k)
}
}
if local["auth"].(map[string]any)["store"] == nil || cluster["auth"].(map[string]any)["store"] != nil {
t.Errorf("auth.store went to the cluster: local %v cluster %v", local["auth"], cluster["auth"])
}
if cluster["auth"].(map[string]any)["default_rpm"] != 60 {
t.Errorf("auth's policy stayed with the machine: %v", cluster["auth"])
}
if local["mirror"].(map[string]any)["api"] == nil || cluster["mirror"].(map[string]any)["update_check"] == nil {
t.Errorf("mirror split wrong: local %v cluster %v", local["mirror"], cluster["mirror"])
}
for _, k := range []string{"guard", "router"} {
if _, ok := cluster[k]; !ok {
t.Errorf("%s is not the cluster's", k)
}
}
if err := CheckCluster(cluster); err != nil {
t.Errorf("the cluster half is refused: %v", err)
}
if err := CheckCluster(file); err == nil || !strings.Contains(err.Error(), "auth.store") {
t.Errorf("a cluster document with a machine's keys: %v", err)
}
}
func TestTheClustersSettingsTakeThePlaceOfTheFiles(t *testing.T) {
file, _ := Parse([]byte(machineFile))
cluster, _ := Parse([]byte("router:\n retries: 5\nauth:\n enabled: true\n default_rpm: 10\n"))
eff := Layer(file, cluster)
cfg, err := FromMap(eff)
if err != nil {
t.Fatal(err)
}
if cfg.Router.Retries != 5 || cfg.Auth.DefaultRPM != 10 {
t.Errorf("cluster values not in force: retries %d rpm %d", cfg.Router.Retries, cfg.Auth.DefaultRPM)
}
if cfg.Auth.Store != "/var/lib/llmgw/llmgw.db" || cfg.Mirror.API != "http://127.0.0.1:8082" || cfg.Listen != ":8080" {
t.Errorf("the machine's own keys lost: %+v", cfg.Auth)
}
if len(cfg.Guard.Policies) != 0 {
t.Error("the file's guard applies though the cluster has settings without one")
}
if got := Ignored(file); !slices.Contains(got, "guard") || !slices.Contains(got, "auth.default_rpm") ||
slices.Contains(got, "auth.store") {
t.Errorf("ignored: %v", got)
}
}
func TestChangedNamesWhatMoved(t *testing.T) {
a, _ := Parse([]byte(machineFile))
b, _ := Parse([]byte(strings.Replace(machineFile, "retries: 2", "retries: 3", 1)))
if got := Changed(a, b); !slices.Equal(got, []string{"router"}) {
t.Errorf("changed %v", got)
}
c, _ := Parse([]byte(strings.Replace(machineFile, "store: /var/lib/llmgw/llmgw.db", "store: /x.db", 1)))
if got := Changed(a, c); !slices.Equal(got, []string{"auth.store"}) {
t.Errorf("changed %v", got)
}
if got := Changed(a, a); len(got) != 0 {
t.Errorf("unchanged: %v", got)
}
}
internal/control/adminapi/api.go
+3 -1
View File
@@ -67,7 +67,8 @@ type API struct {
// nil without a store.
federation *federation.Service
// guard is the guardrail policies' library and attachments.
guard *GuardAdmin
guard *GuardAdmin
settings *Settings
// revoke records a removed node, so its certificate is refused.
revoke func(ctx context.Context, nodeID string, at time.Time) error
@@ -111,6 +112,7 @@ func (a *API) Register(mux Mux) {
mux.HandleFunc("POST /api/v1/models/{name}/resources", a.setResources)
a.registerQuotas(mux)
a.registerGuard(mux)
a.registerSettings(mux)
mux.HandleFunc("GET /api/v1/audit", a.auditEvents)
mux.HandleFunc("GET /api/v1/usage/export", a.usageExport)
mux.HandleFunc("POST /api/v1/notify/test", a.notifyTest)
internal/control/adminapi/guard.go
+11 -10
View File
@@ -21,10 +21,11 @@ type GuardStore interface {
DetachGuard(ctx context.Context, a guard.Attachment) (bool, error)
}
// GuardAdmin is what the guard endpoints need: the library from server.yaml,
// the store, and a way to have the gate read attachments again at once.
// GuardAdmin is what the guard endpoints need: the library in force (the
// cluster's settings), the store, and a way to have the gate read
// attachments again at once.
type GuardAdmin struct {
Library guard.Config
Library func() guard.Config
Store GuardStore
Invalidate func()
}
@@ -199,7 +200,7 @@ type GuardAttachmentRequest struct {
func (a *API) registerGuard(mux Mux) {
mux.HandleFunc("GET /api/v1/guard", func(w http.ResponseWriter, r *http.Request) {
if a.guard == nil {
writeError(w, http.StatusNotImplemented, "no guard policies: define them under guard: in server.yaml")
writeError(w, http.StatusNotImplemented, "no guard policies: define them under guard in the cluster's settings (llmgw ctl config edit)")
return
}
att, err := a.guard.Store.GuardAttachments(r.Context())
@@ -207,7 +208,7 @@ func (a *API) registerGuard(mux Mux) {
writeError(w, http.StatusInternalServerError, err.Error())
return
}
v := GuardView{Default: a.guard.Library.Default, Attachments: []GuardAttachmentView{}}
v := GuardView{Default: a.guard.Library().Default, Attachments: []GuardAttachmentView{}}
var names map[string]string
if a.quotas != nil {
names, _ = a.scopeNames(r.Context())
@@ -215,12 +216,12 @@ func (a *API) registerGuard(mux Mux) {
for _, at := range att {
v.Attachments = append(v.Attachments, GuardAttachmentView{Attachment: at, Name: names[at.Scope+":"+at.ScopeID]})
}
for name := range a.guard.Library.Policies {
for name := range a.guard.Library().Policies {
v.Policies = append(v.Policies, name)
}
slices.Sort(v.Policies)
for _, name := range v.Policies {
v.Details = append(v.Details, policyView(name, a.guard.Library.Policies[name]))
v.Details = append(v.Details, policyView(name, a.guard.Library().Policies[name]))
}
writeJSON(w, http.StatusOK, v)
})
@@ -275,7 +276,7 @@ func (a *API) registerGuard(mux Mux) {
change := func(attach bool) http.HandlerFunc {
return func(w http.ResponseWriter, r *http.Request) {
if a.guard == nil {
writeError(w, http.StatusNotImplemented, "no guard policies: define them under guard: in server.yaml")
writeError(w, http.StatusNotImplemented, "no guard policies: define them under guard in the cluster's settings (llmgw ctl config edit)")
return
}
var at GuardAttachmentRequest
@@ -289,8 +290,8 @@ func (a *API) registerGuard(mux Mux) {
return
}
att.Policy = at.Policy
if _, ok := a.guard.Library.Policies[att.Policy]; !ok && attach {
writeError(w, http.StatusBadRequest, fmt.Sprintf("no policy %q in server.yaml's guard.policies", att.Policy))
if _, ok := a.guard.Library().Policies[att.Policy]; !ok && attach {
writeError(w, http.StatusBadRequest, fmt.Sprintf("no policy %q in the cluster's guard.policies", att.Policy))
return
}
changed := true
internal/control/adminapi/guard_test.go
+7 -4
View File
@@ -36,8 +36,10 @@ func TestGuardPoliciesAreAttachedByTarget(t *testing.T) {
st := &memGuard{}
invalidated := 0
a.SetGuard(&GuardAdmin{
Library: guard.Config{Default: []string{"base"}, Policies: map[string]guard.Policy{"base": {}, "finance": {}}},
Store: st, Invalidate: func() { invalidated++ },
Library: func() guard.Config {
return guard.Config{Default: []string{"base"}, Policies: map[string]guard.Policy{"base": {}, "finance": {}}}
},
Store: st, Invalidate: func() { invalidated++ },
})
if code := call(t, a, "POST", "/api/v1/guard/attachments", `{"target":"key:k1","policy":"finance"}`, nil); code != 200 {
t.Fatalf("attach: %d", code)
@@ -138,10 +140,11 @@ func (m *memGuardHistory) GuardCounts(context.Context, time.Time) ([]store.Guard
func TestTheGuardrailsSayWhatEachPolicyDoesAndWhatTheyDid(t *testing.T) {
a, _ := setup(t, testModels())
h := &memGuardHistory{}
a.SetGuard(&GuardAdmin{Store: h, Library: guard.Config{Policies: map[string]guard.Policy{
lib := guard.Config{Policies: map[string]guard.Policy{
"base": {Fail: "closed", Input: guard.Rules{PII: map[string]string{"email": "redact"}, Injection: "block"},
Judge: &guard.JudgePolicy{Model: "llama-guard-3-1b", Input: "block", Categories: []string{"S1", "S9"}}},
}}})
}}
a.SetGuard(&GuardAdmin{Store: h, Library: func() guard.Config { return lib }})
var v GuardView
if code := call(t, a, "GET", "/api/v1/guard", "", &v); code != 200 || len(v.Details) != 1 {
internal/control/adminapi/notify.go
+1 -1
View File
@@ -12,7 +12,7 @@ func (a *API) SetNotifier(n *notify.Notifier) { a.notifier = n }
// notifyTest sends a test event to every webhook and says how each answered.
func (a *API) notifyTest(w http.ResponseWriter, r *http.Request) {
if a.notifier == nil {
writeError(w, http.StatusNotFound, "no webhooks: server.yaml has no notify.webhooks")
writeError(w, http.StatusNotFound, "no webhooks: the cluster's settings have no notify.webhooks (llmgw ctl config set notify.webhooks ...)")
return
}
writeJSON(w, http.StatusOK, a.notifier.Test(r.Context()))
internal/control/adminapi/routes.go
+14 -1
View File
@@ -131,7 +131,7 @@ var Routes = []Route{
Description: "With the defaults the catalog filled in. By its name only, not an alias.",
Response: catalog.Model{}},
{Method: "DELETE", Path: "/api/v1/models/{name}", Summary: "Remove a model added at runtime",
Description: "A model declared in configuration is refused: it would come back on the next restart.",
Description: "A model declared in the cluster's settings is refused: it is changed there.",
Response: Removed{}},
{Method: "POST", Path: "/api/v1/models/{name}/scale", Summary: "Set how many replicas a model runs",
Request: ScaleRequest{}, Response: ScaleResponse{}},
@@ -259,6 +259,19 @@ var Routes = []Route{
{Method: "DELETE", Path: "/api/v1/guard/attachments", Summary: "Detach a guardrail policy",
Request: GuardAttachmentRequest{}, Response: guard.Attachment{}},
// The cluster's settings: everything in server.yaml that is not one
// machine's own, the same on every control plane.
{Method: "GET", Path: "/api/v1/settings", Summary: "The cluster's settings in force, or one earlier version",
Query: []Param{{Name: "version", Description: "An earlier version; the newest when absent."}},
Response: SettingsView{}},
{Method: "GET", Path: "/api/v1/settings/history", Summary: "Versions of the cluster's settings, newest first",
Query: []Param{{Name: "limit", Description: "How many; 20 when absent."}},
Response: []SettingsView{}},
{Method: "PUT", Path: "/api/v1/settings", Summary: "Change the cluster's settings: in force on every control plane within seconds",
Request: SettingsRequest{}, Response: SettingsView{}},
{Method: "POST", Path: "/api/v1/settings/rollback", Summary: "Put an earlier version of the cluster's settings back in force",
Request: SettingsRollbackRequest{}, Response: SettingsView{}},
// Usage, requests and the record of changes.
{Method: "GET", Path: "/api/v1/usage", Summary: "What each group of the tenancy tree spent",
Query: []Param{
internal/control/adminapi/settings.go
+179
View File
@@ -0,0 +1,179 @@
package adminapi
import (
"context"
"encoding/json"
"errors"
"net/http"
"strconv"
"time"
"gitea.mynodes.xyz/ilker/llmgw/internal/store"
)
// SettingsStore keeps the cluster's settings.
type SettingsStore interface {
ClusterSettings(ctx context.Context) (store.SettingsVersion, error)
ClusterSettingsAt(ctx context.Context, version int64) (store.SettingsVersion, error)
ClusterSettingsHistory(ctx context.Context, limit int) ([]store.SettingsVersion, error)
PutClusterSettings(ctx context.Context, base int64, body, by, note string) (int64, error)
}
// Settings is what the settings endpoints need: where they are kept, and a
// check that a document would load — laid over this control plane's own
// server.yaml, validated as it would be at start.
type Settings struct {
Store SettingsStore
Check func(body []byte) error
// LocalKeys are those each machine's server.yaml keeps, which the
// document cannot set.
LocalKeys []string
}
// SetSettings enables the settings endpoints.
func (a *API) SetSettings(s *Settings) { a.settings = s }
// SettingsView is one version of the cluster's settings.
type SettingsView struct {
Version int64 `json:"version"`
Body string `json:"body"`
Note string `json:"note,omitempty"`
At time.Time `json:"at"`
By string `json:"by,omitempty"`
LocalKeys []string `json:"local_keys,omitempty"`
}
// SettingsRequest is a new version of the cluster's settings, made from the
// version Base, which must still be the newest.
type SettingsRequest struct {
Base int64 `json:"base"`
Body string `json:"body"`
Note string `json:"note,omitempty"`
}
// SettingsRollbackRequest puts an earlier version's settings back in force,
// as a new version.
type SettingsRollbackRequest struct {
Version int64 `json:"version"`
Note string `json:"note,omitempty"`
}
func settingsView(v store.SettingsVersion) SettingsView {
return SettingsView{Version: v.Version, Body: v.Body, Note: v.Note, At: v.At, By: v.By}
}
func (a *API) registerSettings(mux Mux) {
need := func(w http.ResponseWriter) bool {
if a.settings == nil {
writeError(w, http.StatusNotImplemented, "this control plane has no store to keep the cluster's settings in")
return false
}
return true
}
mux.HandleFunc("GET /api/v1/settings", func(w http.ResponseWriter, r *http.Request) {
if !need(w) {
return
}
var v store.SettingsVersion
var err error
if q := r.URL.Query().Get("version"); q != "" {
n, perr := strconv.ParseInt(q, 10, 64)
if perr != nil {
writeError(w, http.StatusBadRequest, "version is a number")
return
}
v, err = a.settings.Store.ClusterSettingsAt(r.Context(), n)
if err == nil && v.Version == 0 {
writeError(w, http.StatusNotFound, "there is no such version")
return
}
} else {
v, err = a.settings.Store.ClusterSettings(r.Context())
}
if err != nil {
writeError(w, http.StatusInternalServerError, err.Error())
return
}
out := settingsView(v)
out.LocalKeys = a.settings.LocalKeys
writeJSON(w, http.StatusOK, out)
})
mux.HandleFunc("GET /api/v1/settings/history", func(w http.ResponseWriter, r *http.Request) {
if !need(w) {
return
}
limit := 20
if q := r.URL.Query().Get("limit"); q != "" {
if n, err := strconv.Atoi(q); err == nil && n > 0 && n <= 500 {
limit = n
}
}
list, err := a.settings.Store.ClusterSettingsHistory(r.Context(), limit)
if err != nil {
writeError(w, http.StatusInternalServerError, err.Error())
return
}
out := make([]SettingsView, len(list))
for i, v := range list {
out[i] = settingsView(v)
}
writeJSON(w, http.StatusOK, out)
})
put := func(w http.ResponseWriter, r *http.Request, base int64, body, note string) {
if err := a.settings.Check([]byte(body)); err != nil {
writeError(w, http.StatusBadRequest, err.Error())
return
}
n, err := a.settings.Store.PutClusterSettings(r.Context(), base, body, actorOf(r), note)
switch {
case errors.Is(err, store.ErrSettingsChanged):
writeError(w, http.StatusConflict, err.Error())
return
case err != nil:
writeError(w, http.StatusInternalServerError, err.Error())
return
}
v, err := a.settings.Store.ClusterSettingsAt(r.Context(), n)
if err != nil {
writeError(w, http.StatusInternalServerError, err.Error())
return
}
writeJSON(w, http.StatusOK, settingsView(v))
}
mux.HandleFunc("PUT /api/v1/settings", func(w http.ResponseWriter, r *http.Request) {
if !need(w) {
return
}
var req SettingsRequest
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
writeError(w, http.StatusBadRequest, err.Error())
return
}
put(w, r, req.Base, req.Body, req.Note)
})
mux.HandleFunc("POST /api/v1/settings/rollback", func(w http.ResponseWriter, r *http.Request) {
if !need(w) {
return
}
var req SettingsRollbackRequest
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
writeError(w, http.StatusBadRequest, err.Error())
return
}
old, err := a.settings.Store.ClusterSettingsAt(r.Context(), req.Version)
if err == nil && old.Version == 0 {
writeError(w, http.StatusNotFound, "there is no such version")
return
}
cur, err2 := a.settings.Store.ClusterSettings(r.Context())
if err = errors.Join(err, err2); err != nil {
writeError(w, http.StatusInternalServerError, err.Error())
return
}
note := req.Note
if note == "" {
note = "back to version " + strconv.FormatInt(req.Version, 10)
}
put(w, r, cur.Version, old.Body, note)
})
}
internal/control/adminapi/settings_test.go
+78
View File
@@ -0,0 +1,78 @@
package adminapi
import (
"context"
"errors"
"strings"
"testing"
"time"
"gitea.mynodes.xyz/ilker/llmgw/internal/store"
)
type memSettings struct{ versions []store.SettingsVersion }
func (m *memSettings) ClusterSettings(context.Context) (store.SettingsVersion, error) {
if len(m.versions) == 0 {
return store.SettingsVersion{}, nil
}
return m.versions[len(m.versions)-1], nil
}
func (m *memSettings) ClusterSettingsAt(_ context.Context, v int64) (store.SettingsVersion, error) {
if v < 1 || int(v) > len(m.versions) {
return store.SettingsVersion{}, nil
}
return m.versions[v-1], nil
}
func (m *memSettings) ClusterSettingsHistory(context.Context, int) ([]store.SettingsVersion, error) {
var out []store.SettingsVersion
for i := len(m.versions) - 1; i >= 0; i-- {
out = append(out, m.versions[i])
}
return out, nil
}
func (m *memSettings) PutClusterSettings(_ context.Context, base int64, body, by, note string) (int64, error) {
if base != int64(len(m.versions)) {
return 0, store.ErrSettingsChanged
}
m.versions = append(m.versions, store.SettingsVersion{Version: base + 1, Body: body, By: by, Note: note, At: time.Now()})
return base + 1, nil
}
func TestSettingsAreCheckedKeptAndRolledBack(t *testing.T) {
a, _ := setup(t, testModels())
st := &memSettings{}
a.SetSettings(&Settings{Store: st, Check: func(b []byte) error {
if strings.Contains(string(b), "listen") {
return errors.New("listen is each machine's")
}
return nil
}})
var v SettingsView
if code := call(t, a, "PUT", "/api/v1/settings", `{"base":0,"body":"router:\n retries: 2\n"}`, &v); code != 200 || v.Version != 1 {
t.Fatalf("first: %d %+v", code, v)
}
if code := call(t, a, "PUT", "/api/v1/settings", `{"base":1,"body":"listen: :1\n"}`, nil); code != 400 {
t.Errorf("a document the check refuses: %d", code)
}
if code := call(t, a, "PUT", "/api/v1/settings", `{"base":0,"body":"router: {}\n"}`, nil); code != 409 {
t.Errorf("a change from an old version: %d", code)
}
if code := call(t, a, "PUT", "/api/v1/settings", `{"base":1,"body":"router:\n retries: 5\n"}`, nil); code != 200 {
t.Fatalf("second: %d", code)
}
if code := call(t, a, "POST", "/api/v1/settings/rollback", `{"version":1}`, &v); code != 200 ||
v.Version != 3 || v.Body != "router:\n retries: 2\n" || v.Note != "back to version 1" {
t.Errorf("rollback: %d %+v", code, v)
}
var h []SettingsView
if code := call(t, a, "GET", "/api/v1/settings/history", "", &h); code != 200 || len(h) != 3 || h[0].Version != 3 {
t.Errorf("history: %d %+v", code, h)
}
if code := call(t, a, "GET", "/api/v1/settings?version=9", "", nil); code != 404 {
t.Errorf("a version that is not there: %d", code)
}
}
internal/control/enginechoice.go
+33
View File
@@ -43,6 +43,39 @@ func NewEngineChoices(ctx context.Context, st EngineChoiceStore, log *slog.Logge
return c
}
// Reload reads the choices again, as a standby becoming the leader does, and
// tells each node whose choice changed.
func (c *EngineChoices) Reload(ctx context.Context) {
got, err := c.st.NodeEngines(ctx)
if err != nil {
c.log.Error("could not read the engine builds chosen per node again; keeping the ones read before", "err", err)
return
}
c.mu.Lock()
var changed []string
for node := range got {
if !maps.Equal(got[node], c.byNode[node]) {
changed = append(changed, node)
}
}
for node := range c.byNode {
if _, ok := got[node]; !ok {
changed = append(changed, node)
}
}
c.byNode = got
notify := c.notify
c.mu.Unlock()
if notify == nil {
return
}
for _, node := range changed {
if err := notify(node); err != nil {
c.log.Warn("could not tell a node its engine builds", "node", node, "err", err)
}
}
}
// OnChange is called with a node whose choice changed, to tell it.
func (c *EngineChoices) OnChange(f func(node string) error) {
c.mu.Lock()
internal/control/facility.go
+17
View File
@@ -68,6 +68,23 @@ func (f *Facility) UseStore(ctx context.Context, st FacilityStore) {
f.mu.Lock()
defer f.mu.Unlock()
f.st = st
f.loadLocked(ctx)
}
// Reload reads the holds and labels in the store again, as a standby
// becoming the leader does.
func (f *Facility) Reload(ctx context.Context) {
f.mu.Lock()
defer f.mu.Unlock()
if f.st == nil {
return
}
f.holds = map[facility.Domain]facility.Hold{}
f.loadLocked(ctx)
}
func (f *Facility) loadLocked(ctx context.Context) {
st := f.st
if holds, err := st.FacilityHolds(ctx); err != nil {
f.log.Error("could not read the domains held for maintenance; none is held until they are set again", "err", err)
} else {
internal/control/leadership.go
+24
View File
@@ -7,7 +7,9 @@ import (
"errors"
"log/slog"
"os"
"slices"
"strings"
"sync"
"sync/atomic"
"time"
@@ -54,6 +56,27 @@ type Leadership struct {
leader atomic.Bool
// stopped is set by StepDown: nothing takes the lease again after it.
stopped atomic.Bool
hookMu sync.Mutex
onLead []func()
}
// OnLead is called each time this process becomes the leader, after
// standing by: what it read at start may have been changed since by the
// leader before it, and is read again.
func (l *Leadership) OnLead(f func()) {
l.hookMu.Lock()
l.onLead = append(l.onLead, f)
l.hookMu.Unlock()
}
func (l *Leadership) becameLeader() {
l.hookMu.Lock()
hooks := slices.Clone(l.onLead)
l.hookMu.Unlock()
for _, f := range hooks {
go f()
}
}
// NewLeadership prepares leadership over a store.
@@ -133,6 +156,7 @@ func (l *Leadership) attempt(ctx context.Context) {
case err == nil:
if l.leader.CompareAndSwap(false, true) {
l.log.Info("this control plane is now placing work", "owner", l.owner)
l.becameLeader()
}
case errors.Is(err, store.ErrNotLeader):
if l.leader.CompareAndSwap(true, false) {
internal/control/leadership_test.go
+26
View File
@@ -49,3 +49,29 @@ func TestAStoppingLeaderHandsOverAtOnceAndNeverTakesItBack(t *testing.T) {
t.Errorf("holder name %q from %q", got, b.Owner())
}
}
func TestAStandbyIsToldWhenItBecomesTheLeader(t *testing.T) {
st, err := store.Open(t.Context(), filepath.Join(t.TempDir(), "s.db"))
if err != nil {
t.Fatal(err)
}
defer st.Close()
log := slog.New(slog.DiscardHandler)
a := NewLeadership(st, time.Minute, log)
b := NewLeadership(st, time.Minute, log)
led := make(chan struct{}, 4)
b.OnLead(func() { led <- struct{}{} })
ctx := t.Context()
a.attempt(ctx)
b.attempt(ctx)
if !a.IsLeader() || b.IsLeader() {
t.Fatal("a should lead, b stand by")
}
a.StepDown()
b.attempt(ctx)
select {
case <-led:
case <-time.After(time.Second):
t.Fatal("b became the leader and was not told")
}
}
internal/control/modelsvc/modelsvc.go
+3 -3
View File
@@ -109,7 +109,7 @@ func (s *Service) Add(ctx context.Context, m catalog.Model, by string) error {
defer s.mu.Unlock()
if s.Declared(m.Name) {
return fmt.Errorf("model %q is declared in configuration; edit that rather than adding a second definition", m.Name)
return fmt.Errorf("model %q is declared in the cluster's settings; edit it there (llmgw ctl config edit) rather than adding a second definition", m.Name)
}
// A model with no GPU is all host memory, and one that declares none
// fits everywhere as far as placement can tell — however many of it
@@ -170,7 +170,7 @@ func (s *Service) Remove(ctx context.Context, name string) error {
defer s.mu.Unlock()
if s.Declared(name) {
return fmt.Errorf("model %q is declared in configuration; remove it there, or it comes back on the next restart", name)
return fmt.Errorf("model %q is declared in the cluster's settings, under models; remove it there: llmgw ctl config edit", name)
}
stored, err := s.storedModels(ctx)
@@ -368,7 +368,7 @@ func (s *Service) update(ctx context.Context, name, by, what string, change func
if s.Declared(name) {
return catalog.Model{}, fmt.Errorf(
"model %q is declared in configuration; %s there, or it reverts on the next restart", name, what)
"model %q is declared in the cluster's settings, under models; %s there: llmgw ctl config edit", name, what)
}
stored, err := s.storedModels(ctx)
if err != nil {
internal/control/modelsvc/modelsvc_test.go
+3 -3
View File
@@ -96,7 +96,7 @@ func TestConfigurationWinsOverTheDatabase(t *testing.T) {
if err == nil {
t.Fatal("a stored model was allowed to shadow a declared one")
}
if !strings.Contains(err.Error(), "declared in configuration") {
if !strings.Contains(err.Error(), "declared in the cluster's settings") {
t.Errorf("the error does not say why: %v", err)
}
}
@@ -109,7 +109,7 @@ func TestADeclaredModelCannotBeRemovedAtRuntime(t *testing.T) {
if err == nil {
t.Fatal("a declared model was removed at runtime")
}
if !strings.Contains(err.Error(), "comes back") {
if !strings.Contains(err.Error(), "llmgw ctl config edit") {
t.Errorf("the error does not explain the consequence: %v", err)
}
}
@@ -264,7 +264,7 @@ func TestADeclaredModelCannotBeScaledHere(t *testing.T) {
if err == nil {
t.Fatal("expected scaling a declared model to be refused")
}
if !strings.Contains(err.Error(), "configuration") {
if !strings.Contains(err.Error(), "cluster's settings") {
t.Errorf("the error should say where to change it; got %q", err)
}
}
internal/guard/gate.go
+44 -6
View File
@@ -9,6 +9,7 @@ import (
"sort"
"strings"
"sync"
"sync/atomic"
"time"
"gitea.mynodes.xyz/ilker/llmgw/internal/guard/detect"
@@ -36,7 +37,7 @@ type Gate struct {
log *slog.Logger
now func() time.Time
deny map[string]*regexp.Regexp // by pattern regex
deny atomic.Pointer[map[string]*regexp.Regexp] // by pattern regex
tools toolMemo
judger *judge
m *gateMetrics
@@ -59,16 +60,53 @@ func NewGate(cfg Config, store Attachments, log *slog.Logger) (*Gate, error) {
log = slog.Default()
}
g := &Gate{cfg: cfg, store: store, log: log, now: time.Now,
deny: map[string]*regexp.Regexp{}, merged: map[string]Policy{}, m: newGateMetrics()}
merged: map[string]Policy{}, m: newGateMetrics()}
g.deny.Store(compileDeny(cfg))
g.reload(context.Background())
return g, nil
}
func compileDeny(cfg Config) *map[string]*regexp.Regexp {
deny := map[string]*regexp.Regexp{}
for _, p := range cfg.Policies {
for _, r := range []Rules{p.Input, p.Output} {
for _, d := range r.Deny {
g.deny[d.Regex] = regexp.MustCompile(d.Regex)
deny[d.Regex] = regexp.MustCompile(d.Regex)
}
}
}
return &deny
}
// Replace puts a new library of policies in force, for requests from the
// next one on: the cluster's settings changed. Refused, and the old one
// kept, when it is not valid.
func (g *Gate) Replace(cfg Config) error {
if err := cfg.Validate(); err != nil {
return err
}
// The patterns first: a policy naming one must find it compiled.
old := *g.deny.Load()
next := *compileDeny(cfg)
for k, v := range old {
if _, ok := next[k]; !ok {
next[k] = v
}
}
g.deny.Store(&next)
g.mu.Lock()
g.cfg = cfg
g.merged = map[string]Policy{}
g.mu.Unlock()
g.reload(context.Background())
return g, nil
return nil
}
// Library is the policies in force.
func (g *Gate) Library() Config {
g.mu.Lock()
defer g.mu.Unlock()
return g.cfg
}
// attachmentsTTL is how stale an attachment may be: one made in the console
@@ -92,7 +130,7 @@ func (g *Gate) reload(ctx context.Context) {
g.attached = map[string][]string{}
for _, a := range list {
if _, ok := g.cfg.Policies[a.Policy]; !ok {
g.log.Warn("a guard policy is attached that server.yaml does not define; ignored",
g.log.Warn("a guard policy is attached that the cluster's settings do not define; ignored",
"policy", a.Policy, "scope", a.Scope, "id", a.ScopeID)
continue
}
@@ -188,7 +226,7 @@ func (g *Gate) judge(text string, r Rules) ([]finding, []detect.Span) {
if len(r.Deny) > 0 {
norm := detect.Normalize(text)
for _, d := range r.Deny {
re := g.deny[d.Regex]
re := (*g.deny.Load())[d.Regex]
if re == nil {
continue
}
internal/router/handlers.go
+1 -1
View File
@@ -239,7 +239,7 @@ func (r *Router) Handler() http.Handler {
return names, p.NoStore && p.Mode == "enforce"
})
}
if r.store != nil && len(r.guardLibrary.Policies) > 0 {
if r.store != nil && r.guardLibrary != nil {
pages.RegisterAdminGuard(mux, r.store, r.guardLibrary)
}
pages.Register(mux)
internal/router/router.go
+2 -2
View File
@@ -108,7 +108,7 @@ type Router struct {
// 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
guardLibrary guard.Config
guardLibrary func() guard.Config
// oidc is sign-in through an identity provider, and scim provisioning
// by it.
oidc config.OIDC
@@ -187,7 +187,7 @@ func (r *Router) SetMirror(m web.MirrorAdmin) { r.mirror = m }
func (r *Router) SetGuard(g guard.Guard) { r.guard = g }
// SetGuardLibrary is the policies the guard was built from, for the pages.
func (r *Router) SetGuardLibrary(c guard.Config) { r.guardLibrary = c }
func (r *Router) SetGuardLibrary(c func() guard.Config) { r.guardLibrary = c }
// SetRegistry gives the mirror page somewhere to look models up.
func (r *Router) SetRegistry(reg web.Registry) { r.registry = reg }
internal/settings/settings.go
+99 -1
View File
@@ -126,15 +126,61 @@ type Doc struct {
doc yaml.Node
// Changed counts edits not yet saved.
Changed int
// cluster holds the keys that are not the machine's own, when the
// cluster keeps them elsewhere (the store); isLocal says which are.
cluster *Doc
isLocal func([]string) bool
}
// Layer has the keys isLocal does not claim read and written in cluster
// instead of this document: one screen over a machine's file and the
// cluster's settings.
func (d *Doc) Layer(cluster *Doc, isLocal func([]string) bool) {
d.cluster, d.isLocal = cluster, isLocal
}
// Cluster is the layered cluster document, nil when there is none.
func (d *Doc) Cluster() *Doc { return d.cluster }
// Pending counts edits not yet saved, in this document and its layer.
func (d *Doc) Pending() int {
if d.cluster != nil {
return d.Changed + d.cluster.Changed
}
return d.Changed
}
// at is the document a path is kept in.
func (d *Doc) at(path []string) *Doc {
if d.cluster != nil && len(path) > 0 && !d.isLocal(path) {
return d.cluster
}
return d
}
// Open reads a file; a missing one is an empty document.
func Open(path string) (*Doc, error) {
d := &Doc{Path: path}
b, err := os.ReadFile(path)
if err != nil && !os.IsNotExist(err) {
return nil, err
}
return parseDoc(path, b)
}
// FromBytes reads a document held somewhere other than a file: the
// cluster's settings, kept in the store. It has no Path, so it is not Saved.
func FromBytes(name string, b []byte) (*Doc, error) {
d, err := parseDoc(name, b)
if err != nil {
return nil, err
}
d.Path = ""
return d, nil
}
func parseDoc(path string, b []byte) (*Doc, error) {
d := &Doc{Path: path}
if len(bytes.TrimSpace(b)) > 0 {
if err := yaml.Unmarshal(b, &d.doc); err != nil {
return nil, fmt.Errorf("%s: %w", path, err)
@@ -163,6 +209,9 @@ func (d *Doc) Bytes() ([]byte, error) {
// Node is the node at a path of keys and list indexes, or nil.
func (d *Doc) Node(path []string) *yaml.Node {
if t := d.at(path); t != d {
return t.Node(path)
}
n := d.root
for _, p := range path {
n = child(n, p)
@@ -209,6 +258,9 @@ func (d *Doc) ensure(path []string) *yaml.Node {
// Set writes one key's value: a scalar, a list of lines or a map of
// "key = value" lines, by its kind. It checks the value first.
func (d *Doc) Set(path []string, k Kind, value string) error {
if t := d.at(path); t != d {
return t.Set(path, k, value)
}
if len(path) == 0 {
return fmt.Errorf("no key")
}
@@ -216,6 +268,42 @@ func (d *Doc) Set(path []string, k Kind, value string) error {
if err != nil {
return err
}
return d.place(path, n)
}
// SetYAML writes one key's value given as YAML: a scalar, [a, b], {k: v},
// or a whole block.
func (d *Doc) SetYAML(path []string, value string) error {
if t := d.at(path); t != d {
return t.SetYAML(path, value)
}
if len(path) == 0 {
return fmt.Errorf("no key")
}
var doc yaml.Node
if err := yaml.Unmarshal([]byte(value), &doc); err != nil {
return fmt.Errorf("%q is not YAML: %w", value, err)
}
n := &yaml.Node{Kind: yaml.ScalarNode, Tag: "!!null", Value: ""}
if len(doc.Content) > 0 {
n = doc.Content[0]
}
return d.place(path, n)
}
// SetNode writes a node, as read from another document, at path.
func (d *Doc) SetNode(path []string, n *yaml.Node) error {
if t := d.at(path); t != d {
return t.SetNode(path, n)
}
if len(path) == 0 {
return fmt.Errorf("no key")
}
c := *n
return d.place(path, &c)
}
func (d *Doc) place(path []string, n *yaml.Node) error {
parent := d.ensure(path[:len(path)-1])
if parent == nil {
return fmt.Errorf("%s is not a block", strings.Join(path[:len(path)-1], "."))
@@ -237,6 +325,10 @@ func (d *Doc) Set(path []string, k Kind, value string) error {
// Unset removes a key, so its default applies.
func (d *Doc) Unset(path []string) {
if t := d.at(path); t != d {
t.Unset(path)
return
}
if len(path) == 0 {
return
}
@@ -264,6 +356,9 @@ func (d *Doc) Unset(path []string) {
// Append adds an empty block to the list at path, and returns its index.
func (d *Doc) Append(path []string) int {
if t := d.at(path); t != d {
return t.Append(path)
}
parent := d.ensure(path[:len(path)-1])
key := path[len(path)-1]
seq := child(parent, key)
@@ -283,6 +378,9 @@ func (d *Doc) Append(path []string) int {
// AddEntry adds an empty block under a name to the map at path.
func (d *Doc) AddEntry(path []string, name string) error {
if t := d.at(path); t != d {
return t.AddEntry(path, name)
}
m := d.ensure(path)
if m == nil || m.Kind != yaml.MappingNode {
return fmt.Errorf("%s is not a map", strings.Join(path, "."))
internal/store/copy.go
+2
View File
@@ -58,6 +58,8 @@ var copyOrder = []string{
// Or a removed node's certificate would work again in the new store.
"revoked_nodes",
"guard_attachments",
// Or the cluster would start the new store on server.yaml's settings again.
"cluster_settings",
// Who changed what is not lost by moving the store.
"audit_events",
"key_ttl",
internal/store/migrations/postgres/0032_cluster_settings.sql
+10
View File
@@ -0,0 +1,10 @@
-- Ported from the SQLite migration of the same number.
-- The cluster's settings: everything in server.yaml that is not one
-- machine's own, as a versioned YAML document every control plane follows.
CREATE TABLE cluster_settings (
version BIGINT PRIMARY KEY,
body TEXT NOT NULL,
note TEXT NOT NULL DEFAULT '',
created_at BIGINT NOT NULL,
created_by TEXT NOT NULL DEFAULT ''
);
internal/store/migrations/sqlite/0032_cluster_settings.sql
+13
View File
@@ -0,0 +1,13 @@
-- The cluster's settings: everything in server.yaml that is not one
-- machine's own (listeners, certificates, the store's address), as a YAML
-- document every control plane reads and follows. Each change is a new row,
-- so what was in force before is there to compare with and go back to, and
-- two edits made at once cannot both win: the second is refused for naming
-- a version that is no longer the newest.
CREATE TABLE cluster_settings (
version INTEGER PRIMARY KEY,
body TEXT NOT NULL,
note TEXT NOT NULL DEFAULT '',
created_at INTEGER NOT NULL,
created_by TEXT NOT NULL DEFAULT ''
);
internal/store/settings.go
+98
View File
@@ -0,0 +1,98 @@
package store
import (
"context"
"database/sql"
"errors"
"time"
)
// ErrSettingsChanged is a change made against a version of the cluster's
// settings that is no longer the newest: somebody else changed them first.
var ErrSettingsChanged = errors.New("the cluster's settings were changed since they were read; read them again")
// SettingsVersion is one version of the cluster's settings.
type SettingsVersion struct {
Version int64
Body string
Note string
At time.Time
By string
}
// ClusterSettings is the newest version, Version 0 when there is none yet.
func (s *DB) ClusterSettings(ctx context.Context) (SettingsVersion, error) {
return s.settingsWhere(ctx, `version = (SELECT MAX(version) FROM cluster_settings)`)
}
// ClusterSettingsAt is one version, Version 0 when there is no such version.
func (s *DB) ClusterSettingsAt(ctx context.Context, version int64) (SettingsVersion, error) {
return s.settingsWhere(ctx, `version = ?`, version)
}
func (s *DB) settingsWhere(ctx context.Context, where string, args ...any) (SettingsVersion, error) {
var v SettingsVersion
var at int64
err := s.db.QueryRowContext(ctx,
`SELECT version, body, note, created_at, created_by FROM cluster_settings WHERE `+where, args...).
Scan(&v.Version, &v.Body, &v.Note, &at, &v.By)
if errors.Is(err, sql.ErrNoRows) {
return SettingsVersion{}, nil
}
v.At = time.UnixMilli(at)
return v, err
}
// ClusterSettingsVersion is the newest version's number, 0 when there is
// none: what every control plane asks every few seconds.
func (s *DB) ClusterSettingsVersion(ctx context.Context) (int64, error) {
var v sql.NullInt64
err := s.db.QueryRowContext(ctx, `SELECT MAX(version) FROM cluster_settings`).Scan(&v)
return v.Int64, err
}
// PutClusterSettings records a new version made from base, the version the
// change was read from, and returns its number. Refused with
// ErrSettingsChanged when base is not the newest.
func (s *DB) PutClusterSettings(ctx context.Context, base int64, body, by, note string) (int64, error) {
cur, err := s.ClusterSettingsVersion(ctx)
if err != nil {
return 0, err
}
if cur != base {
return 0, ErrSettingsChanged
}
// The key settles a race the read above cannot: of two writers from the
// same base, one inserts base+1 and the other is refused.
_, err = s.db.ExecContext(ctx,
`INSERT INTO cluster_settings (version, body, note, created_at, created_by) VALUES (?, ?, ?, ?, ?)`,
base+1, body, note, time.Now().UnixMilli(), by)
if isUniqueViolation(err) {
return 0, ErrSettingsChanged
}
if err != nil {
return 0, err
}
return base + 1, nil
}
// ClusterSettingsHistory is the newest versions first, at most limit.
func (s *DB) ClusterSettingsHistory(ctx context.Context, limit int) ([]SettingsVersion, error) {
rows, err := s.db.QueryContext(ctx,
`SELECT version, body, note, created_at, created_by FROM cluster_settings ORDER BY version DESC LIMIT ?`, limit)
if err != nil {
return nil, err
}
defer rows.Close()
var out []SettingsVersion
for rows.Next() {
var v SettingsVersion
var at int64
if err := rows.Scan(&v.Version, &v.Body, &v.Note, &at, &v.By); err != nil {
return nil, err
}
v.At = time.UnixMilli(at)
out = append(out, v)
}
return out, rows.Err()
}
internal/store/settings_test.go
+39
View File
@@ -0,0 +1,39 @@
package store
import (
"errors"
"testing"
)
func TestSettingsAreVersionedAndTwoEditsCannotBothWin(t *testing.T) {
s := capacityStore(t)
ctx := t.Context()
if v, err := s.ClusterSettings(ctx); err != nil || v.Version != 0 {
t.Fatalf("empty store: %+v %v", v, err)
}
v1, err := s.PutClusterSettings(ctx, 0, "router: {}\n", "import", "from server.yaml")
if err != nil || v1 != 1 {
t.Fatalf("first: %d %v", v1, err)
}
if _, err := s.PutClusterSettings(ctx, 0, "x: 1\n", "late", ""); !errors.Is(err, ErrSettingsChanged) {
t.Fatalf("a change from an old version: %v", err)
}
if _, err := s.PutClusterSettings(ctx, 1, "guard: {}\n", "ilker", "guard"); err != nil {
t.Fatal(err)
}
cur, _ := s.ClusterSettings(ctx)
if cur.Version != 2 || cur.Body != "guard: {}\n" || cur.By != "ilker" {
t.Errorf("newest: %+v", cur)
}
if n, _ := s.ClusterSettingsVersion(ctx); n != 2 {
t.Errorf("version %d", n)
}
old, _ := s.ClusterSettingsAt(ctx, 1)
if old.Body != "router: {}\n" || old.Note != "from server.yaml" {
t.Errorf("version 1: %+v", old)
}
h, _ := s.ClusterSettingsHistory(ctx, 10)
if len(h) != 2 || h[0].Version != 2 {
t.Errorf("history: %+v", h)
}
}
internal/web/admin_guard.go
+4 -3
View File
@@ -49,17 +49,18 @@ var guardWindows = map[string]time.Duration{"24h": 24 * time.Hour, "7d": 7 * 24
// RegisterAdminGuard mounts /admin/guard: what the guardrails did, where, by
// which policy — names and counts, never text.
func (p *AccountPages) RegisterAdminGuard(mux *http.ServeMux, s GuardAdminSource, library guard.Config) {
func (p *AccountPages) RegisterAdminGuard(mux *http.ServeMux, s GuardAdminSource, library func() guard.Config) {
p.hasGuard = true
mux.HandleFunc("GET /admin/guard", p.adminOnly(func(w http.ResponseWriter, r *http.Request, admin store.User) {
ctx, l := r.Context(), langOf(r)
lib := library()
d := guardPageData{Nav: p.navFor(r, admin, "guard"), Version: p.version, Window: r.URL.Query().Get("window"),
Windows: []string{"24h", "7d", "30d"}, Default: library.Default}
Windows: []string{"24h", "7d", "30d"}, Default: lib.Default}
since, ok := guardWindows[d.Window]
if !ok {
d.Window, since = "7d", guardWindows["7d"]
}
for name := range library.Policies {
for name := range lib.Policies {
d.Policies = append(d.Policies, name)
}
slices.Sort(d.Policies)
internal/web/admin_mirror_test.go
+3 -2
View File
@@ -591,8 +591,9 @@ func TestHowAModelRunsIsChangedWhereItsFileIsShown(t *testing.T) {
if !strings.Contains(body, `<option value="always" selected>always on</option>`) {
t.Error("the running model's mode is not shown as chosen")
}
// Declared in server.yaml: shown, not offered, since it would revert.
if !strings.Contains(body, `title="set in server.yaml, which is read again on every start">on demand · yaml</span>`) {
// Declared in the cluster's settings: shown, not offered, since it is
// changed there.
if !strings.Contains(body, `title="set in the cluster&#39;s settings (llmgw ctl config)">on demand · yaml</span>`) {
t.Error("a declared model should say where its mode is set")
}
if strings.Count(body, `name="mode" class="mode"`) != 1 {
internal/web/i18n_people_test.go
+4 -3
View File
@@ -138,14 +138,15 @@ func TestTheGuardrailsPageSpeaksTurkishWhenAsked(t *testing.T) {
p := NewAccountPages(a, nil, nil, "v0", slog.New(slog.NewTextHandler(io.Discard, nil)))
mux := http.NewServeMux()
at := time.Date(2026, 10, 4, 9, 5, 7, 0, time.UTC)
p.RegisterAdminGuard(mux, guardSource{at: at}, guard.Config{
Default: []string{"base"}, Policies: map[string]guard.Policy{"base": {}, "finance": {}}})
p.RegisterAdminGuard(mux, guardSource{at: at}, func() guard.Config {
return guard.Config{Default: []string{"base"}, Policies: map[string]guard.Policy{"base": {}, "finance": {}}}
})
body := peopleInTurkish(t, mux, http.MethodGet, "/admin/guard", nil)
for _, want := range []string{
"<title>Koruma kuralları · llmgw</title>",
"<h2>Politikalar</h2>",
"<code>server.yaml</code> içinde tanımlı:",
"Cluster ayarlarında tanımlı (<code>llmgw ctl config</code>):",
"<td>proje</td><td>Acme/fin/payroll</td>", // the scope's kind, in Turkish
"<td>kişisel</td><td>maskelendi</td>",
`<td class="num">1.500</td>`, // a count written the Turkish way
internal/web/locales/en/guard.json
+2 -2
View File
@@ -2,7 +2,7 @@
"guard.title": "Guardrails",
"guard.sub": "What the policies did, where, and by which rule — names and counts, never what was written.",
"guard.policies": "Policies",
"guard.defined_in": "Defined in <code>server.yaml</code>:",
"guard.defined_in": "Defined in the cluster's settings (<code>llmgw ctl config</code>):",
"guard.applied": "Applied to every request:",
"guard.col.scope": "Scope",
"guard.col.who": "Who",
@@ -13,7 +13,7 @@
"guard.scope.user": "user",
"guard.scope.key": "key",
"guard.attached_nowhere": "Attached nowhere else. <code>llmgw ctl guard attach project:ORG/TEAM/PROJECT POLICY</code> attaches one.",
"guard.no_policies": "No policies: add a <code>guard:</code> section to <code>server.yaml</code> (docs/guardrails.md).",
"guard.no_policies": "No policies: add them under <code>guard</code> in the cluster's settings, <code>llmgw ctl config edit</code> (docs/guardrails.md).",
"guard.what_they_did": "What they did",
"guard.over_last": "Over the last",
internal/web/locales/en/mirror.json
+3 -3
View File
@@ -28,7 +28,7 @@
"mirror.percent_of": "%d%% of %s",
"mirror.so_far": "%s so far",
"mirror.use.as": "as %s",
"mirror.use.yaml_title": "set in server.yaml, which is read again on every start",
"mirror.use.yaml_title": "set in the cluster's settings (llmgw ctl config)",
"mirror.use.yaml": "%s · yaml",
"mirror.mode.always": "always on",
"mirror.mode.ondemand": "on demand",
@@ -183,7 +183,7 @@
"mirror.adapters.adapter_file": "Adapter file",
"mirror.adapters.attach": "Attach",
"mirror.adapters.no_choice": "No held adapter file for this model's architecture. Fetch one from Hugging Face first.",
"mirror.detail.declared": "Set in server.yaml, which is read again on every start: change its replicas there.",
"mirror.detail.declared": "Set in the cluster's settings: change its replicas there, with llmgw ctl config edit.",
"mirror.form.min": "Always running",
"mirror.form.max": "Up to, under load",
"mirror.form.class": "Class",
@@ -214,7 +214,7 @@
"mirror.err.replicas_whole": "replicas are whole numbers",
"mirror.err.min_over_max": "always running (%d) cannot be more than the most under load (%d)",
"mirror.err.at_most": "at most %d replicas",
"mirror.err.declared": "%s is set in server.yaml; change it there",
"mirror.err.declared": "%s is set in the cluster's settings; change it there (llmgw ctl config edit)",
"mirror.err.no_model": "no model %q",
"mirror.err.nothing_fits": "nothing more fits",
"mirror.err.no_room": "%s can always run at most %d now (%d running, room for %d more): %s",
internal/web/locales/tr/guard.json
+2 -2
View File
@@ -2,7 +2,7 @@
"guard.title": "Koruma kuralları",
"guard.sub": "Politikaların ne yaptığı, nerede ve hangi kuralla — yalnızca adlar ve sayılar, yazılanlar asla.",
"guard.policies": "Politikalar",
"guard.defined_in": "<code>server.yaml</code> içinde tanımlı:",
"guard.defined_in": "Cluster ayarlarında tanımlı (<code>llmgw ctl config</code>):",
"guard.applied": "Her isteğe uygulanan:",
"guard.col.scope": "Kapsam",
"guard.col.who": "Kim",
@@ -13,7 +13,7 @@
"guard.scope.user": "kişi",
"guard.scope.key": "anahtar",
"guard.attached_nowhere": "Başka hiçbir yere bağlı değil. Bağlamak için: <code>llmgw ctl guard attach project:ORG/TEAM/PROJECT POLICY</code>",
"guard.no_policies": "Politika yok: <code>server.yaml</code> dosyasına bir <code>guard:</code> bölümü ekleyin (docs/guardrails.md).",
"guard.no_policies": "Politika yok: cluster ayarlarında <code>guard</code> altına ekleyin, <code>llmgw ctl config edit</code> (docs/guardrails.md).",
"guard.what_they_did": "Ne yaptılar",
"guard.over_last": "Son",
internal/web/locales/tr/mirror.json
+3 -3
View File
@@ -26,7 +26,7 @@
"mirror.percent_of": "%[2]s içinden %%%[1]d",
"mirror.so_far": "şu ana dek %s",
"mirror.use.as": "%s adıyla",
"mirror.use.yaml_title": "server.yaml içinde tanımlı; bu dosya her başlangıçta yeniden okunur",
"mirror.use.yaml_title": "cluster ayarlarında tanımlı (llmgw ctl config)",
"mirror.use.yaml": "%s · yaml",
"mirror.mode.always": "sürekli açık",
"mirror.mode.ondemand": "istek üzerine",
@@ -178,7 +178,7 @@
"mirror.adapters.adapter_file": "Adaptör dosyası",
"mirror.adapters.attach": "Ekle",
"mirror.adapters.no_choice": "Aynada bu modelin mimarisi için adaptör dosyası yok. Önce Hugging Face'ten bir tane indirin.",
"mirror.detail.declared": "server.yaml içinde tanımlı; bu dosya her başlangıçta yeniden okunur: replikalarını orada değiştirin.",
"mirror.detail.declared": "cluster ayarlarında tanımlı: replikalarını orada değiştirin, llmgw ctl config edit ile.",
"mirror.form.min": "Sürekli çalışan",
"mirror.form.max": "Yük altında en çok",
"mirror.form.class": "Sınıf",
@@ -209,7 +209,7 @@
"mirror.err.replicas_whole": "replika sayıları tam sayı olmalı",
"mirror.err.min_over_max": "sürekli çalışan (%d), yük altındaki en yüksek sayıdan (%d) fazla olamaz",
"mirror.err.at_most": "en çok %d replika",
"mirror.err.declared": "%s, server.yaml içinde tanımlı; orada değiştirin",
"mirror.err.declared": "%s, cluster ayarlarında tanımlı; orada değiştirin (llmgw ctl config edit)",
"mirror.err.no_model": "%q adında model yok",
"mirror.err.nothing_fits": "daha fazlası sığmıyor",
"mirror.err.no_room": "%s şu anda en çok %d replikayla sürekli çalışabilir (%d çalışıyor, %d tane daha için yer var): %s",