llmgw stop, start and restart act on the whole system

Issue #3 asked for commands to stop and start the system, and 05aa21c gave
only one machine's services. Now:

- llmgw stop sets serving.stopped in the cluster's settings, live: the
  reconciler wants every model at no replicas, freeing every GPU, and the
  gateway answers every POST under /v1/ with 503 system_stopped. The control
  planes, agents and mirror keep running, and no model's definition changes.
- llmgw start clears it; every model is placed again as it asks.
- llmgw restart has every agent restart over its stream (RestartAgent), then
  the other control planes over ssh, standbys before a leader, then this
  one; engines keep serving.
- llmgw status is the cluster's summary, saying when it is stopped.

One machine's services moved to llmgw service status|start|stop|restart.

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-07 18:01:45 +02:00
1 parent 05aa21c4d4
commit 7245dbe1dd
29 files changed
+891 -254

No files matched your search

+9 -4
View File
@@ -28,10 +28,15 @@ What was left of the field report from v0.10.6.
### Added
- **`llmgw status`, `start`, `stop` and `restart`** for this machine's llmgw
services — server, agent, mirror, or those named — in the order they
depend on each other. Stopping the agent leaves the engines serving unless
`--engines` is given; stopping a leader hands over to a standby.
- **The whole system stops, starts and restarts from one command**: `llmgw
stop` takes every model down and answers every request for one with 503
(`system_stopped`), keeping the control planes, agents and mirror running
and no model's definition changed; `llmgw start` brings every model back.
It is the cluster's setting `serving.stopped`, live on every control plane.
`llmgw restart` restarts every node's agent and every control plane, the
leader last, with the engines serving throughout. `llmgw status` says
whether the system serves. One machine's services are `llmgw service
status|start|stop|restart`.
- **A replica says what it computes on**: `ctl replicas` (ON), the console
and the admin API show `CUDA` with the GPUs its processes hold memory on
and how much, as nvidia-smi sees them, or `CPU` — for every engine, kept by
File diff suppressed because it is too large. Load diff
+7
View File
@@ -72,6 +72,7 @@ message ControlMessage {
UpgradeAgent upgrade = 10;
ControlPlanes control_planes = 11;
Firewall firewall = 13;
RestartAgent restart_agent = 14;
}
// trace is the W3C trace context (traceparent, and tracestate when there
// is one) of the control plane's span that sent this command, so what the
@@ -81,6 +82,12 @@ message ControlMessage {
map<string, string> trace = 12;
}
// RestartAgent has the agent start again, as llmgw restart asks of every
// node: its engines keep serving and are taken over again.
message RestartAgent {
string reason = 1;
}
// Firewall is whether llmgw's firewall is kept on the node — its nftables
// table, which lets the cluster's networks alone reach the engines' ports and
// the node exporter — sent when an agent registers and whenever the
+51
View File
@@ -17,6 +17,9 @@
}
],
"tags": [
{
"name": "agents"
},
{
"name": "audit"
},
@@ -100,6 +103,38 @@
}
],
"paths": {
"/api/v1/agents/restart": {
"post": {
"operationId": "postAgentsRestart",
"summary": "Have every node's agent start again; their engines keep serving",
"tags": [
"agents"
],
"parameters": [
{
"$ref": "#/components/parameters/Actor"
}
],
"responses": {
"200": {
"description": "OK",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/AgentRestartView"
}
}
}
},
"401": {
"$ref": "#/components/responses/Unauthorized"
},
"default": {
"$ref": "#/components/responses/Error"
}
}
}
},
"/api/v1/audit": {
"get": {
"operationId": "getAudit",
@@ -3554,6 +3589,18 @@
}
}
},
"AgentRestartView": {
"type": "object",
"description": "AgentRestartView names the nodes whose agents were told to restart.",
"properties": {
"nodes": {
"type": "array",
"items": {
"type": "string"
}
}
}
},
"CanaryPercentRequest": {
"type": "object",
"description": "CanaryPercentRequest changes the share of a model's requests its canary takes.",
@@ -5928,6 +5975,10 @@
},
"description": "Judges are the guardrails' judge models with something wrong."
},
"stopped": {
"type": "boolean",
"description": "Stopped says the operators have stopped the service (llmgw stop): no model runs, and requests for one are answered 503."
},
"guard_off": {
"type": "boolean",
"description": "GuardOff says the guardrails are turned off (guard.enabled: false): no request is checked, whatever the policies say."
+31 -9
View File
@@ -2294,20 +2294,42 @@ A cluster set up online becomes offline the same way: `bundle import` on the
mirror and control plane, then the `offline:` block in `server.yaml` and a
restart.
## Stopping and starting a machine
## Stopping and starting the system
From a control plane:
```
sudo llmgw status # each llmgw service here, and whether it runs
sudo llmgw stop # agent, control plane, mirror — whichever this machine has
sudo llmgw start
sudo llmgw restart agent # or name one: server, agent, mirror
llmgw status # does it serve; its control planes, nodes, models, store
llmgw stop --note "maintenance" # every model taken down, every request answered 503
llmgw start # every model back as its catalog entry asks
sudo llmgw restart # every agent and control plane started again, engines serving throughout
```
`llmgw stop` frees every GPU and answers requests for a model with 503
(`system_stopped`, with Retry-After) until `llmgw start`. The control planes,
agents and mirror keep running, so the console, the admin pages and `llmgw
start` keep working, and no model's definition changes. It is the cluster's
setting `serving.stopped`, so a restart does not undo it, and it is in the
audit with its note. `--wait` waits until every replica is down, or every
model has its replicas again.
`llmgw restart` restarts every node's agent, then the other control planes
over ssh (`--ssh-user`, the one sudo was run by), then this one — the leader
last, so it hands over to a standby already restarted. Engines keep serving
and are taken over again; nothing is placed anew.
One machine's own services are `llmgw service`:
```
sudo llmgw service status
sudo llmgw service stop # agent, control plane, mirror — whichever this machine has
sudo llmgw service start
sudo llmgw service restart agent # or name one: server, agent, mirror
```
Stopping a node's agent leaves its engines serving — the agent takes them
over again when it starts — unless `llmgw stop --engines` stops them too, and
the control plane then places their replicas elsewhere. Stopping the leading
control plane hands over to a standby within seconds. The whole cluster's
state is `llmgw ctl status`.
over again when it starts — unless `llmgw service stop --engines` stops them
too, and the control plane then places their replicas elsewhere.
## Removing llmgw from a machine
+23
View File
@@ -290,6 +290,7 @@ rather than repeated.
| `secrets` | `Secrets` | [Server](#server) |
| `notify` | `notify.Config` | [Server](#server) |
| `mail` | `mail.Config` | [Server](#server) |
| `serving` | `ServingSettings` | [Server](#server) |
| `firewall` | `FirewallSettings` | [Server](#server) |
| `firewall_networks` | `[]string` | [Server](#server) |
| `backup_recipients` | `[]string` | [Server](#server) |
@@ -302,6 +303,7 @@ rather than repeated.
| `facility` | `Facility` | [Server](#server) |
| `federation` | `Federation` | [Server](#server) |
| `endpoints` | `[]Endpoint` | [Server](#server) |
| `stopped` | `bool` | [ServingSettings](#servingsettings) |
| `bin` | `string` | [SimConfig](#simconfig) |
| `python` | `string` | [TRTLLMConfig](#trtllmconfig) |
| `store` | `string` | [Threads](#threads) |
@@ -2481,6 +2483,15 @@ forgotten password. Unset, none is sent and people ask an operator.
Takes a [mail.Config](#mailconfig) block.
### `serving`
`ServingSettings`
Serving stops and starts the whole cluster's service (llmgw stop, llmgw
start).
Takes a [ServingSettings](#servingsettings) block.
### `firewall`
`FirewallSettings`
@@ -2579,6 +2590,18 @@ Endpoints is the phase-1 static replica list. It is only consulted when
no model declares a profile, which is what makes a router-only smoke test
against sim-engines possible without a control plane.
## ServingSettings
### `stopped`
`bool`
Stopped takes every model down and answers every request for one with
503 (system_stopped), leaving the control planes, agents and mirror
running and every model's definition as it is: llmgw stop sets it,
llmgw start clears it, and the models come back as their catalog
entries ask. In force on every control plane within seconds.
## SimConfig
### `bin`
+8
View File
@@ -869,6 +869,14 @@ func (a *Agent) handle(ctx context.Context, msg *llmgwv1.ControlMessage, out cha
a.setControlPlanes(m.ControlPlanes)
case *llmgwv1.ControlMessage_Firewall:
go a.applyFirewall(m.Firewall)
case *llmgwv1.ControlMessage_RestartAgent:
go func() {
a.log.Info("restarting, as the control plane asks; the engines keep serving",
"reason", m.RestartAgent.GetReason())
// A moment for the log line to leave before the process does.
time.Sleep(300 * time.Millisecond)
exitForUpgrade()
}()
case *llmgwv1.ControlMessage_Logs:
reply := a.logsReply(m.Logs)
// Off the receive loop: a full send queue must not stop commands being
+2
View File
@@ -237,6 +237,8 @@ func (w *settingsWatch) check(ctx context.Context) {
// again: what it reads at every use, or not at all.
var liveSettings = []string{
"guard",
// Stopping and starting the service (llmgw stop, llmgw start).
"serving",
// Sent to the nodes and this machine's firewall unit as they change.
"firewall", "firewall_networks",
// Read by server init, node add and backups, not by the server.
+4 -2
View File
@@ -63,8 +63,10 @@ func Root() *cobra.Command {
newWipeCmd(),
newDocsCmd(),
)
// status, start, stop and restart: this machine's llmgw services.
root.AddCommand(newServiceCmds()...)
// status, start, stop and restart: the whole system; service: this
// machine's services.
root.AddCommand(newSystemCmds()...)
root.AddCommand(newServiceCmd())
return root
}
+14
View File
@@ -186,6 +186,7 @@ func newServerCmd() *cobra.Command {
rt.Start(ctx)
defer rt.Stop()
rt.SetStopped(cfg.Serving.Stopped)
// Which cluster this is, to federated peers: the id in the hop
// header that stops a request going round two clusters that
@@ -252,6 +253,8 @@ func newServerCmd() *cobra.Command {
}
defer ctl.Close()
agentSvc = svc
// Stopped by the operators (llmgw stop): nothing placed.
ctl.Reconciler().SetStopped(cfg.Serving.Stopped)
// Whether nodes keep llmgw's firewall, as the settings say now.
svc.SetFirewall(func() *llmgwv1.Firewall {
c := current.Load()
@@ -367,6 +370,9 @@ func newServerCmd() *cobra.Command {
if st := rt.Store(); st != nil {
// Removing a node revokes its certificate.
api.SetRevoker(st.RevokeNode)
if agentSvc != nil {
api.SetAgentRestarter(agentSvc.RestartAgents)
}
api.SetAudit(st)
// The cluster's settings, each change checked against this
// machine's own keys before it is kept.
@@ -676,6 +682,14 @@ func newServerCmd() *cobra.Command {
return false
}
current.Store(&next)
if slices.ContainsFunc(changed, func(c string) bool { return strings.HasPrefix(c, "serving") }) {
rt.SetStopped(next.Serving.Stopped)
if admin != nil {
admin.Reconciler().SetStopped(next.Serving.Stopped)
}
log.Warn("the service is " + map[bool]string{true: "stopped", false: "started"}[next.Serving.Stopped] +
" by the cluster's settings")
}
if slices.ContainsFunc(changed, func(c string) bool { return strings.HasPrefix(c, "firewall") }) {
if err := writeFirewallState(next); err != nil {
log.Warn("could not record the firewall's settings for this machine", "err", err)
+16 -3
View File
@@ -78,13 +78,26 @@ func hasUnit(list []string, s string) bool {
return false
}
func newServiceCmds() []*cobra.Command {
// newServiceCmd is llmgw service: this machine's llmgw services, one by one.
// The whole system is llmgw status, start, stop and restart.
func newServiceCmd() *cobra.Command {
c := &cobra.Command{
Use: "service",
Short: "This machine's llmgw services: status, start, stop, restart",
Long: "The control plane (server), node agent (agent) and model mirror (mirror) on\n" +
"this machine alone, by systemd. The whole system is 'llmgw status', 'llmgw\n" +
"stop', 'llmgw start' and 'llmgw restart', run on a control plane.",
}
c.AddCommand(serviceVerbs()...)
return c
}
func serviceVerbs() []*cobra.Command {
status := &cobra.Command{
Use: "status [server|agent|mirror]...",
Short: "Whether llmgw's services on this machine run",
Long: "Each llmgw service this machine has — the control plane (server), the\n" +
"node agent (agent), the model mirror (mirror) — and whether it runs.\n" +
"The cluster as a whole is 'llmgw ctl status'.",
"node agent (agent), the model mirror (mirror) — and whether it runs.",
RunE: func(cmd *cobra.Command, args []string) error {
units, err := localUnits(cmd, args)
if err != nil {
+51 -1
View File
@@ -3,11 +3,18 @@ package cli
import (
"bytes"
"context"
"encoding/json"
"net/http"
"net/http/httptest"
"os"
"path/filepath"
"runtime"
"strings"
"testing"
"github.com/spf13/cobra"
"gitea.mynodes.xyz/ilker/llmgw/internal/control/adminapi"
)
func TestTheServicesHereAreFoundAndAskedAbout(t *testing.T) {
@@ -42,7 +49,7 @@ func TestTheServicesHereAreFoundAndAskedAbout(t *testing.T) {
return
}
var out bytes.Buffer
c := newServiceCmds()[0]
c := serviceVerbs()[0]
c.SetOut(&out)
c.SetContext(context.Background())
if err := c.RunE(c, nil); err != nil {
@@ -52,3 +59,46 @@ func TestTheServicesHereAreFoundAndAskedAbout(t *testing.T) {
t.Errorf("%s", out.String())
}
}
func TestStopSetsTheClustersServingStopped(t *testing.T) {
var put adminapi.SettingsRequest
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
switch {
case r.URL.Path == "/api/v1/summary":
_ = json.NewEncoder(w).Encode(adminapi.Summary{})
case r.Method == "GET" && r.URL.Path == "/api/v1/settings":
_ = json.NewEncoder(w).Encode(adminapi.SettingsView{Version: 3, 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: 4, Body: put.Body})
default:
http.NotFound(w, r)
}
}))
defer srv.Close()
dir := t.TempDir()
cfg := filepath.Join(dir, "server.yaml")
_ = os.WriteFile(filepath.Join(dir, "token"), []byte("t"), 0o600)
_ = os.WriteFile(cfg, []byte("listen: \":8080\"\nagent_insecure: true\nadmin_listen: "+
strings.TrimPrefix(srv.URL, "http://")+"\nadmin_remote: true\nadmin_token_file: "+filepath.Join(dir, "token")+"\n"), 0o600)
var stop *cobra.Command
for _, c := range newSystemCmds() {
if c.Name() == "stop" {
stop = c
}
}
var out bytes.Buffer
stop.SetOut(&out)
stop.SetArgs([]string{"--config", cfg, "--note", "maintenance"})
if err := stop.ExecuteContext(context.Background()); err != nil {
t.Fatalf("%v: %s", err, out.String())
}
if put.Base != 3 || !strings.Contains(put.Body, "stopped: true") || !strings.Contains(put.Body, "retries: 2") ||
put.Note != "maintenance" {
t.Errorf("sent %+v", put)
}
if !strings.Contains(out.String(), "stopping") {
t.Errorf("%s", out.String())
}
}
+3
View File
@@ -153,6 +153,9 @@ func printSummary(cmd *cobra.Command, s adminapi.Summary) {
fmt.Fprintf(out, "STORE %s\n", strings.Join(parts, ", "))
}
if s.Stopped {
fmt.Fprintln(out, "SERVICE STOPPED: no model runs and requests are answered 503 (llmgw start)")
}
if s.GuardOff {
fmt.Fprintln(out, "GUARD off: no request is checked, whatever the policies say (llmgw ctl guard on)")
}
+216
View File
@@ -0,0 +1,216 @@
package cli
import (
"errors"
"fmt"
"os"
"os/exec"
"strings"
"time"
"github.com/spf13/cobra"
"gitea.mynodes.xyz/ilker/llmgw/internal/control/adminapi"
"gitea.mynodes.xyz/ilker/llmgw/internal/onboard"
"gitea.mynodes.xyz/ilker/llmgw/internal/settings"
)
// systemClient is the admin API of the control plane on this machine, as its
// server.yaml says, with its token.
func systemClient(configPath string) func() *ctlClient {
return func() *ctlClient {
addr := "127.0.0.1:9090"
var tokens []string
if cfg, ok := loadServerConfig(configPath); ok {
if cfg.AdminListen != "" {
addr = cfg.AdminListen
if strings.HasPrefix(addr, ":") {
addr = "127.0.0.1" + addr
}
}
tokens = append(tokens, adminTokenPath(configPath, cfg))
}
return newCtlClient(addr, tokens...)
}
}
// newSystemCmds are llmgw status, stop, start and restart: the whole system,
// from a control plane.
func newSystemCmds() []*cobra.Command {
var configPath string
client := systemClient("/etc/llmgw/server.yaml")
bind := func(c *cobra.Command) *cobra.Command {
c.Flags().StringVar(&configPath, "config", "/etc/llmgw/server.yaml", "this control plane's configuration")
c.PreRun = func(*cobra.Command, []string) { client = systemClient(configPath) }
return c
}
call := func() *ctlClient { return client() }
status := bind(&cobra.Command{
Use: "status",
Short: "The whole system: whether it serves, its control planes, nodes, models and store",
Long: "Run on a control plane. Says whether the service is stopped (llmgw stop),\n" +
"and everything 'llmgw ctl status' does. One machine's services are\n" +
"'llmgw service status'.",
Args: cobra.NoArgs,
RunE: func(cmd *cobra.Command, _ []string) error {
var s adminapi.Summary
if err := call().get(cmd.Context(), "/api/v1/summary", &s); err != nil {
return notAControlPlane(err)
}
printSummary(cmd, s)
return nil
},
})
switchService := func(stop bool) *cobra.Command {
use, short, long := "start", "Start the whole system serving again, after llmgw stop",
"Every model comes back as its catalog entry asks: placed, loaded and\n"+
"serving again, on every control plane within seconds. Run on a control\n"+
"plane. The services themselves are never stopped by llmgw stop; a\n"+
"machine's own are started with 'llmgw service start'."
if stop {
use, short, long = "stop", "Stop the whole system serving: every model taken down, requests answered 503",
"Every model's replicas are taken down, freeing every GPU, and every\n"+
"request for a model is answered 503 (system_stopped) until 'llmgw start'.\n"+
"The control planes, agents and mirror keep running, so the console, the\n"+
"admin pages and llmgw start still work, and no model's definition is\n"+
"changed. In force on every control plane within seconds; kept in the\n"+
"cluster's settings (serving.stopped), so a restart does not undo it."
}
var note string
var wait bool
c := bind(&cobra.Command{
Use: use,
Short: short,
Long: long,
Args: cobra.NoArgs,
RunE: func(cmd *cobra.Command, _ []string) error {
if note == "" {
note = "llmgw " + use
}
var s adminapi.Summary
if err := call().get(cmd.Context(), "/api/v1/summary", &s); err != nil {
return notAControlPlane(err)
}
if s.Stopped == stop {
fmt.Fprintf(cmd.OutOrStdout(), "the system is already %s\n", map[bool]string{true: "stopped", false: "serving"}[stop])
return nil
}
if err := changeSettings(cmd, call, note, func(doc *settings.Doc) error {
return doc.SetYAML([]string{"serving", "stopped"}, fmt.Sprint(stop))
}); err != nil {
return err
}
out := cmd.OutOrStdout()
if !wait {
if stop {
fmt.Fprintln(out, "stopping: every model's replicas are being taken down; llmgw status follows it")
} else {
fmt.Fprintln(out, "starting: every model is placed again as it asks; llmgw status follows it")
}
return nil
}
return waitForService(cmd, call, stop)
},
})
c.Flags().StringVar(&note, "note", "", "why, kept with the settings' version and in the audit")
c.Flags().BoolVar(&wait, "wait", false, "wait until every replica is down (stop) or the models have theirs (start)")
return c
}
var sshUser string
restart := bind(&cobra.Command{
Use: "restart",
Short: "Restart every control plane and node agent; the engines keep serving",
Long: "Every node's agent starts again, then the other control planes over ssh\n" +
"as --ssh-user, then this one — the leader last, so it hands over to a\n" +
"standby already restarted. The engines keep serving throughout and are\n" +
"taken over again; nothing is placed anew. Run on a control plane, with\n" +
"sudo. One machine's services are 'llmgw service restart'.",
Args: cobra.NoArgs,
RunE: func(cmd *cobra.Command, _ []string) error {
if os.Geteuid() != 0 {
return errors.New("restarting this control plane needs root: run it with sudo")
}
ctx, out := cmd.Context(), cmd.OutOrStdout()
var s adminapi.Summary
if err := call().get(ctx, "/api/v1/summary", &s); err != nil {
return notAControlPlane(err)
}
var agents adminapi.AgentRestartView
if err := call().post(ctx, "/api/v1/agents/restart", []byte("{}"), &agents); err != nil {
return fmt.Errorf("restarting the agents: %w", err)
}
fmt.Fprintf(out, "agents %d told to restart: %s\n", len(agents.Nodes), strings.Join(agents.Nodes, ", "))
// The other control planes, standbys before a leader.
var others []adminapi.ControlPlane
for _, cp := range s.ControlPlanes.Members {
if !cp.This {
others = append(others, cp)
}
}
for pass := 0; pass < 2; pass++ {
for _, cp := range others {
leading := cp.State == "leading"
if leading != (pass == 1) {
continue
}
r := &onboard.SSH{Addr: cp.Address, User: sshUser}
if res, err := r.Run(ctx, "sudo systemctl restart llmgw-server", nil); err != nil {
return fmt.Errorf("%s (%s) could not be restarted over ssh as %q: %v %s; "+
"run `sudo llmgw service restart server` there", cp.Name, cp.Address, sshUser, err, strings.TrimSpace(string(res)))
}
fmt.Fprintf(out, "%-10s restarted (%s)\n", cp.Name, cp.State)
time.Sleep(5 * time.Second)
}
}
if _, err := os.Stat(unitDir + "/llmgw-server.service"); err == nil {
fmt.Fprintln(out, "this restarting")
if b, err := exec.CommandContext(ctx, "systemctl", "restart", "llmgw-server").CombinedOutput(); err != nil { //nolint:gosec // fixed command
return fmt.Errorf("restarting this control plane: %v: %s", err, strings.TrimSpace(string(b)))
}
fmt.Fprintln(out, "this restarted")
}
fmt.Fprintln(out, "\nllmgw status shows the nodes coming back; their replicas are taken over as they are.")
return nil
},
})
restart.Flags().StringVar(&sshUser, "ssh-user", os.Getenv("SUDO_USER"), "who to log in to the other control planes as")
return []*cobra.Command{status, switchService(true), switchService(false), restart}
}
// waitForService waits until no replica runs (stopped) or every model has
// the replicas it asks for (started).
func waitForService(cmd *cobra.Command, call func() *ctlClient, stop bool) error {
deadline := time.Now().Add(15 * time.Minute)
for time.Now().Before(deadline) {
var s adminapi.Summary
if err := call().get(cmd.Context(), "/api/v1/summary", &s); err == nil {
if stop && s.Replicas.Ready+s.Replicas.Starting+s.Replicas.Draining == 0 {
fmt.Fprintln(cmd.OutOrStdout(), "stopped: no replica runs")
return nil
}
if !stop && s.Models.Satisfied >= s.Models.Total {
fmt.Fprintf(cmd.OutOrStdout(), "serving: %d of %d models have the replicas they ask for\n",
s.Models.Satisfied, s.Models.Total)
return nil
}
}
select {
case <-cmd.Context().Done():
return cmd.Context().Err()
case <-time.After(5 * time.Second):
}
}
return errors.New("not there within fifteen minutes; llmgw status says where it stands")
}
// notAControlPlane is a failure to reach the admin API, said as what it
// most often means.
func notAControlPlane(err error) error {
return fmt.Errorf("%w\n(the whole system is stopped and started from a control plane; this machine's own "+
"services are 'llmgw service status|start|stop|restart')", err)
}
+4
View File
@@ -151,6 +151,10 @@ type Server struct {
// forgotten password. Unset, none is sent and people ask an operator.
Mail mail.Config `koanf:"mail"`
// Serving stops and starts the whole cluster's service (llmgw stop, llmgw
// start).
Serving ServingSettings `koanf:"serving"`
// Firewall turns llmgw's firewall on every machine on or off.
Firewall FirewallSettings `koanf:"firewall"`
+11
View File
@@ -0,0 +1,11 @@
package config
// ServingSettings is whether the cluster serves at all.
type ServingSettings struct {
// Stopped takes every model down and answers every request for one with
// 503 (system_stopped), leaving the control planes, agents and mirror
// running and every model's definition as it is: llmgw stop sets it,
// llmgw start clears it, and the models come back as their catalog
// entries ask. In force on every control plane within seconds.
Stopped bool `koanf:"stopped"`
}
+3
View File
@@ -70,6 +70,8 @@ type API struct {
// guard is the guardrail policies' library and attachments.
guard *GuardAdmin
settings *Settings
// restartAgents tells every connected agent to start again.
restartAgents func(reason string) []string
// revoke records a removed node, so its certificate is refused.
revoke func(ctx context.Context, nodeID string, at time.Time) error
@@ -114,6 +116,7 @@ func (a *API) Register(mux Mux) {
a.registerQuotas(mux)
a.registerGuard(mux)
a.registerSettings(mux)
mux.HandleFunc("POST /api/v1/agents/restart", a.restartAgentsHandler)
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)
+3
View File
@@ -263,6 +263,9 @@ var Routes = []Route{
{Method: "DELETE", Path: "/api/v1/guard/attachments", Summary: "Detach a guardrail policy",
Request: GuardAttachmentRequest{}, Response: guard.Attachment{}},
{Method: "POST", Path: "/api/v1/agents/restart", Summary: "Have every node's agent start again; their engines keep serving",
Response: AgentRestartView{}},
// 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",
+4 -1
View File
@@ -44,6 +44,9 @@ type Summary struct {
ControlPlanes ControlPlanes `json:"control_planes,omitzero"`
// Judges are the guardrails' judge models with something wrong.
Judges []JudgeView `json:"judges,omitempty"`
// Stopped says the operators have stopped the service (llmgw stop): no
// model runs, and requests for one are answered 503.
Stopped bool `json:"stopped,omitempty"`
// GuardOff says the guardrails are turned off (guard.enabled: false):
// no request is checked, whatever the policies say.
GuardOff bool `json:"guard_off,omitempty"`
@@ -153,7 +156,7 @@ const idleThreshold = 10 * time.Minute
// Summary builds the fleet summary.
func (a *API) Summary() Summary {
s := Summary{Version: version.Version}
s := Summary{Version: version.Version, Stopped: a.ctl.Reconciler().Stopped()}
if a.guard != nil && a.guard.Library != nil && !a.guard.Library().On() {
s.GuardOff = true
}
+22
View File
@@ -0,0 +1,22 @@
package adminapi
import (
"net/http"
)
// AgentRestartView names the nodes whose agents were told to restart.
type AgentRestartView struct {
Nodes []string `json:"nodes"`
}
// SetAgentRestarter enables POST /api/v1/agents/restart.
func (a *API) SetAgentRestarter(f func(reason string) []string) { a.restartAgents = f }
func (a *API) restartAgentsHandler(w http.ResponseWriter, r *http.Request) {
if a.restartAgents == nil {
writeError(w, http.StatusNotImplemented, "this control plane has no agents to restart")
return
}
nodes := a.restartAgents("llmgw restart, by " + actorOf(r))
writeJSON(w, http.StatusAccepted, AgentRestartView{Nodes: nodes})
}
+19
View File
@@ -1,6 +1,8 @@
package agentsvc
import (
"time"
llmgwv1 "gitea.mynodes.xyz/ilker/llmgw/api/gen/llmgw/v1"
)
@@ -40,3 +42,20 @@ func (s *Service) BroadcastFirewall() {
}
}
}
// RestartAgents has every connected agent start again, one at a time a
// moment apart, and names those it told. Their engines keep serving.
func (s *Service) RestartAgents(reason string) []string {
msg := &llmgwv1.ControlMessage{Msg: &llmgwv1.ControlMessage_RestartAgent{
RestartAgent: &llmgwv1.RestartAgent{Reason: reason}}}
var told []string
for _, node := range s.ConnectedNodes() {
if err := s.send(node, msg); err != nil {
s.log.Warn("could not tell a node to restart its agent", "node", node, "err", err)
continue
}
told = append(told, node)
time.Sleep(time.Second)
}
return told
}
+4
View File
@@ -81,6 +81,10 @@ func (r *Reconciler) Want(model string) bool {
// feature: a cluster with more models than GPUs can hold every model in its
// catalog and only the wanted ones in its cards.
func (r *Reconciler) desiredFor(m catalog.Model) int {
// Stopped by the operators: nothing runs, whatever the catalog asks.
if r.stopped.Load() {
return 0
}
base := r.baseFor(m)
if base == 0 {
return 0
+14
View File
@@ -307,6 +307,9 @@ type Reconciler struct {
startedAt time.Time
lastNodeCount int
steadySince time.Time
// stopped is the operators having stopped the service: every model is
// wanted at no replicas until they start it again.
stopped atomic.Bool
// stoodBy is whether the last pass found this process a standby: one
// that then takes over settles again, as at start.
stoodBy bool
@@ -484,6 +487,17 @@ func (r *Reconciler) stopReplicasOfRemovedModels(ctx context.Context) {
}
}
// SetStopped stops every model, or lets each run as its catalog entry
// asks again, from the next pass.
func (r *Reconciler) SetStopped(stopped bool) {
if r.stopped.Swap(stopped) != stopped {
r.Notify()
}
}
// Stopped is whether the operators have stopped the service.
func (r *Reconciler) Stopped() bool { return r.stopped.Load() }
// resettle opens the settling window again from now, for a control plane
// that has just become the leader.
func (r *Reconciler) resettle() {
@@ -2133,3 +2133,27 @@ func TestAnAdoptedShardKeepsWhatItComputesOn(t *testing.T) {
t.Fatalf("%+v", reps)
}
}
func TestStoppingTheServiceTakesEveryModelDownAndStartingBringsItBack(t *testing.T) {
h := newHarness(t, testModel(1), gpuNode("1", 8))
h.r.Reconcile(t.Context())
h.markReady()
if len(h.r.ReadyReplicas()) != 1 {
t.Fatal("nothing ready to stop")
}
h.r.SetStopped(true)
h.r.Reconcile(t.Context())
for _, rep := range h.r.Replicas() {
if rep.State == Ready {
t.Fatalf("stopped, a replica still serves: %+v", rep)
}
}
h.r.SetStopped(false)
for range 3 {
h.r.Reconcile(t.Context())
h.markReady()
}
if len(h.r.ReadyReplicas()) != 1 {
t.Errorf("started again, %d ready", len(h.r.ReadyReplicas()))
}
}
+4
View File
@@ -368,6 +368,7 @@ var Reference = []Block{
{Key: "secrets", Type: "Secrets", Doc: "Secrets is where secrets kept outside files are read from, and the key\nthe store's own secrets are sealed under.", Nested: "config.Secrets"},
{Key: "notify", Type: "notify.Config", Doc: "Notify sends what needs a person — a model parked, a node lost, a\nbudget spent, new weights published — to webhooks, signed, without a\nPrometheus and Alertmanager in between. Empty sends nothing.", Nested: "notify.Config"},
{Key: "mail", Type: "mail.Config", Doc: "Mail is where the gateway's few emails go out from: a link to set a\nforgotten password. Unset, none is sent and people ask an operator.", Nested: "mail.Config"},
{Key: "serving", Type: "ServingSettings", Doc: "Serving stops and starts the whole cluster's service (llmgw stop, llmgw\nstart).", Nested: "config.ServingSettings"},
{Key: "firewall", Type: "FirewallSettings", Doc: "Firewall turns llmgw's firewall on every machine on or off.", Nested: "config.FirewallSettings"},
{Key: "firewall_networks", Type: "[]string", Doc: "FirewallNetworks are networks, besides those each machine reaches the\ncluster on, that llmgw's internal ports answer: a monitoring host on\nanother subnet, nodes behind a router. Written into every machine's\nfirewall by `server init`, `node add` and `server add`.", Nested: ""},
{Key: "backup_recipients", Type: "[]string", Doc: "BackupRecipients encrypt every archive `llmgw server backup` writes, to\nwhoever holds one of their keys: age recipients (age1…) or SSH public\nkeys (ssh-ed25519 …, ssh-rsa …), as `age` takes them. An archive holds\nthe certificate authority's key and every secret in the store, and is\ncopied off the machine; encrypted, the copy is safe wherever it lands.\nEmpty writes archives unencrypted, readable by root only.", Nested: ""},
@@ -381,6 +382,9 @@ var Reference = []Block{
{Key: "federation", Type: "Federation", Doc: "Federation is how this cluster takes part in federation with other\nllmgw clusters. The peers themselves are added at runtime, with\n`llmgw ctl federation add`, and kept in the store.", Nested: "config.Federation"},
{Key: "endpoints", Type: "[]Endpoint", Doc: "Endpoints is the phase-1 static replica list. It is only consulted when\nno model declares a profile, which is what makes a router-only smoke test\nagainst sim-engines possible without a control plane.", Nested: ""},
}},
{Package: "config", Name: "ServingSettings", Title: "ServingSettings", Doc: "", Fields: []Field{
{Key: "stopped", Type: "bool", Doc: "Stopped takes every model down and answers every request for one with\n503 (system_stopped), leaving the control planes, agents and mirror\nrunning and every model's definition as it is: llmgw stop sets it,\nllmgw start clears it, and the models come back as their catalog\nentries ask. In force on every control plane within seconds.", Nested: ""},
}},
{Package: "engines", Name: "SimConfig", Title: "SimConfig", Doc: "", Fields: []Field{
{Key: "bin", Type: "string", Doc: "Bin is the llmgw binary itself; the engine is one of its subcommands.\nEmpty means whatever started this agent, which is right almost always.", Nested: ""},
}},
+1
View File
@@ -31,6 +31,7 @@ var Sources = []string{
"internal/guard/judge.go",
"internal/config/oidc.go",
"internal/config/firewall.go",
"internal/config/serving.go",
"internal/mail/mail.go",
"internal/notify/notify.go",
"internal/secrets/secrets.go",
+17 -2
View File
@@ -310,7 +310,7 @@ func (r *Router) Handler() http.Handler {
}
if r.auth == nil {
return r.refuseLoops(mux)
return r.refuseWhileStopped(r.refuseLoops(mux))
}
exempt := []string{"/health", "/ready", "/leader", "/", "/about", "/_static/", "/_lang/", "/claim/",
"/signup", "/login", "/login/", "/logout", "/profile", "/profile/", "/scim/",
@@ -324,7 +324,22 @@ func (r *Router) Handler() http.Handler {
// Wrapping the whole mux rather than checking inside each handler: adding a
// route must not be able to add an unauthenticated one by omission. Health
// stays open because a load balancer has no key.
return r.refuseLoops(auth.Middleware(r.auth, r.log, exempt...)(mux))
return r.refuseWhileStopped(r.refuseLoops(auth.Middleware(r.auth, r.log, exempt...)(mux)))
}
// refuseWhileStopped answers every request for a model with 503 while the
// operators have stopped the service (llmgw stop): the models are being taken
// down, and a client is better told so plainly than left to time out.
func (r *Router) refuseWhileStopped(next http.Handler) http.Handler {
return http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
if r.stopped.Load() && req.Method == http.MethodPost && strings.HasPrefix(req.URL.Path, "/v1/") {
w.Header().Set("Retry-After", "300")
writeErrorCode(w, http.StatusServiceUnavailable, "server_error", "system_stopped",
"this gateway's service has been stopped by its operators; it answers again once they start it")
return
}
next.ServeHTTP(w, req)
})
}
// parsedRequest is what the router needs from a request. Everything else in the
+6 -1
View File
@@ -115,7 +115,9 @@ type Router struct {
oidc config.OIDC
// mail sends what the portal mails: a password reset link.
mail mail.Config
scim config.SCIM
// stopped is the operators having stopped the service (llmgw stop).
stopped atomic.Bool
scim config.SCIM
// threads holds conversations and memory notes. It is store when auth is on,
// its own file when only the threads block is configured, and nil when
@@ -192,6 +194,9 @@ 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 func() guard.Config) { r.guardLibrary = c }
// SetStopped refuses, or again serves, every request for a model.
func (r *Router) SetStopped(stopped bool) { r.stopped.Store(stopped) }
// SetRegistry gives the mirror page somewhere to look models up.
func (r *Router) SetRegistry(reg web.Registry) { r.registry = reg }
+21
View File
@@ -748,3 +748,24 @@ func TestModelsEndpointListsOnlyWhatTheKeyMayUse(t *testing.T) {
t.Errorf("listed %v; the key may use real and its alias only", ids)
}
}
func TestAStoppedServiceAnswers503AndStillListsModels(t *testing.T) {
e := newFakeEngine(t, "m")
rt := newTestRouter(t, baseConfig(config.Endpoint{ID: "a", URL: e.srv.URL, Models: []string{"m"}}))
h := rt.Handler()
rt.SetStopped(true)
rec := post(t, h, map[string]any{"model": "m", "messages": []map[string]string{{"role": "user", "content": "hi"}}}, nil)
if rec.Code != http.StatusServiceUnavailable || !strings.Contains(rec.Body.String(), "system_stopped") ||
rec.Header().Get("Retry-After") == "" {
t.Fatalf("stopped: %d %s", rec.Code, rec.Body.String())
}
list := httptest.NewRecorder()
h.ServeHTTP(list, httptest.NewRequest(http.MethodGet, "/v1/models", nil))
if list.Code == http.StatusServiceUnavailable {
t.Error("listing models was refused too")
}
rt.SetStopped(false)
if rec := post(t, h, map[string]any{"model": "m", "messages": []map[string]string{{"role": "user", "content": "hi"}}}, nil); rec.Code == http.StatusServiceUnavailable {
t.Errorf("started again, still refused: %s", rec.Body.String())
}
}