Compare commits

...
4 Commits
Author SHA1 Message Date
Ilker ManapandClaude Opus 5.5 37a80a9b44 Park a model that fails every time it starts, and say so everywhere
Release / build (push) Successful in 1m3s
Release / notes (push) Successful in 3s
CI / test (push) Successful in 3m34s
On the live cluster kev-0.5b — an architecture the nodes' llama.cpp did not
know — was started 155 times in eight hours, failing the same way each time,
and phi-4, which fitted on no node, logged "no placement" 250 times in two
days. Neither showed up anywhere an operator looked: the backoff never gave
up, no metric said a model was failing or unplaceable, and the one alert for
a model with nothing serving could not fire, because its series was published
only while a replica was ready.

After five failures in a row a model is parked: not placed again until its
definition changes — a new source or profile — or `llmgw ctl retry` is run.
Status, the model list and the admin API show it with its last error. The
placement warning is logged when its reason changes or every ten minutes.

A request for a parked or unplaceable model is answered at once with why,
rather than "loading; retry shortly", which a client retries forever. Each
request still clears the previous placement result and triggers a fresh
attempt, so a model that fits once room is made is not refused on a stale
answer; the router waits for that attempt, not for its full activation
window.

Metrics now cover model health (desired, ready at 0 for what should run,
consecutive start failures, parked, unplaceable), engine installs per node,
leadership, and certificate expiry read from disk at each scrape, with alerts
for each. `ctl status` counts a model against what it should have now, so an
on-demand model that was asked for and fitted nowhere no longer reads as
satisfied.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
2026-09-30 01:21:57 +02:00
Ilker ManapandClaude Opus 5.5 5e1b6ecea0 Read a queued change as a success in ctl
ctl accepted only 200, so the first real `ctl restart --all` printed
"Error: 202 Accepted" over a request that had queued nine restarts, all of
which then happened. Any 2xx is now a success, and its body is read.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
2026-09-30 01:14:12 +02:00
Ilker ManapandClaude Opus 5.5 5dac1854c2 Publish small fixes as release candidates of the next version
Release / build (push) Successful in 1m1s
Release / notes (push) Successful in 3s
CI / test (push) Successful in 3m30s
Every fix so far became its own patch release, and the natural alternative —
v0.10.3-a — sorts before v0.10.3 by SemVer's rules: a hyphen marks a
pre-release, so every tool that orders versions would take the fix for the
older one. A release candidate of the next version, v0.10.4-rc.1, sorts
after v0.10.3 and before v0.10.4, which is what it is.

CI already builds any v* tag. Its release notes now come from a
"## Unreleased" section in CHANGELOG.md, where changes gather until the
version they belong to is released, and the release is marked as a
pre-release. Onboarding, in Go and in the shell script, counts a candidate
as a release and downloads it rather than copying the operator's binary.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
2026-09-30 01:00:11 +02:00
Ilker ManapandClaude Opus 5.5 ad6d2d348d Restart replicas without taking them out of service first
There was no way to restart a running replica. A running engine picks up an
engine key, an edited profile or a new engine version only when it starts,
so after v0.10.3 the nine replicas already running stayed without a key —
reachable from any node — with nothing to do about it short of draining
nodes, which on a small CPU fleet has nowhere to move a model to.

`llmgw ctl restart REPLICA... | --model M | --all` queues replicas to be
replaced. A replica being replaced keeps serving but stops counting toward
its model, so the reconciler places a replacement beside it; once that is
ready the old one is drained and stopped as a node drain would. Replicas of
one model go one at a time, so a model never needs the memory of every
replica twice. Where no second copy fits anywhere, the old replica is
stopped after the drain grace period and the new one starts in its place,
and the log says so, since that is the one case with an outage.

`ctl replicas` shows "(restart queued)" and "(restart replacing)" while it
happens.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
2026-09-30 00:59:26 +02:00
23 changed files with 1157 additions and 36 deletions

No files matched your search

+58
View File
@@ -20,6 +20,64 @@ is only released once its **exit criterion** is met, not when its code lands.
Until v1.0.0 the minor version is the breaking one: configuration formats and Go
APIs change between minors without a deprecation period.
## Unreleased
Changes since v0.10.3, published as release candidates (v0.10.4-rc.N) until
there are enough for v0.10.4.
### Restarting replicas
`llmgw ctl restart REPLICA... | --model M | --all` replaces replicas with new
ones. Each replacement is started beside the replica it replaces, which keeps
serving until the new one is ready; replicas of one model go one at a time.
Where no second copy fits anywhere, the old replica is stopped after the drain
grace period and the new one starts in its place. This is how running engines
pick up the keys introduced in v0.10.3, an edited profile, or a new engine
version.
### A model that keeps failing is parked, and everybody is told
A model whose replicas failed at every start was retried forever: kev-0.5b,
whose architecture the nodes' llama.cpp did not know, was started 155 times in
eight hours. After five failures in a row a model is now parked — not placed
again until its definition changes or `llmgw ctl retry MODEL` is run — and
`ctl status`, `ctl models` and the admin API say so with the last error.
A request for a model that will not come up — parked, or fitting on no node —
is answered at once with the reason, instead of "the model is loading; retry
shortly", which a client would retry forever. Each request still makes the
control plane try placing it afresh, so a model that fits once room is made is
not refused on an old answer. "no placement for model" is logged when its
reason changes or every ten minutes, not on every pass: it was 250 identical
lines in two days for one model.
### Metrics and alerts for what goes wrong
New series: `llmgw_model_replicas_desired`, `llmgw_model_start_failures`,
`llmgw_model_parked`, `llmgw_model_unplaceable`, `llmgw_node_engine_install`,
`llmgw_control_plane_leader` and `llmgw_certificate_expiry_timestamp_seconds`.
`llmgw_model_replicas_ready` is now published at 0 for every model that should
be running; it was published only while a replica was ready, so
ModelHasNothingServing could never fire. New alerts: ModelParked,
ModelFailingToStart, ModelCannotBePlaced, EngineInstallFailed,
ControlPlaneLeadership, CertificateExpiresSoon and CertificateExpiresVerySoon.
`ctl status` counts a model against the replicas it should have now rather
than its minimum, so an on-demand model that was asked for and fitted nowhere
no longer reads as satisfied.
### Fixes
- `ctl` reported a change the API had queued (202 Accepted) as an error, as
`ctl restart` did on its first real use.
### Releases
Small fixes are published as release candidates of the next version —
v0.10.4-rc.1, rc.2 — which sort after v0.10.3 and before v0.10.4. Their release
notes come from this section, and they are marked as pre-releases. Onboarding
treats a release candidate as a release and downloads it.
## v0.10.3 — 2026-09-30
Security fixes from a review of v0.10.2. Three of them were confirmed on the
+1 -1
View File
@@ -143,7 +143,7 @@ $SSH "set -e
# that is not a release (a developer's, with commits past the tag) has no such
# file, and is copied as before.
TAG="$("$BINARY" --version 2>/dev/null | awk '{print $NF}')"
if printf '%s' "$TAG" | grep -Eq '^v[0-9]+\.[0-9]+\.[0-9]+$'; then
if printf '%s' "$TAG" | grep -Eq '^v[0-9]+\.[0-9]+\.[0-9]+(-rc\.[0-9]+)?$'; then
say "installing llmgw $TAG from the release"
$SSH "set -e
arch=\$(dpkg --print-architecture)
+80 -4
View File
@@ -42,10 +42,86 @@ groups:
annotations:
summary: "{{ $labels.model }} has no replica ready"
description: >-
The model is placed and nothing is serving it. Ten minutes, because
a large model on CPU legitimately takes minutes to load. A model
that is deliberately stopped publishes no series at all, so this
cannot fire for one nobody asked for.
The model should be running and nothing is serving it. Ten minutes,
because a large model on CPU legitimately takes minutes to load. The
series is published at 0 for every model that should be running —
before v0.10.4 it was published only while a replica was ready, so
this could not fire at all. A model deliberately stopped, or an
on-demand one nobody is asking for, publishes nothing.
- alert: ModelParked
expr: llmgw_model_parked == 1
labels: { severity: critical }
annotations:
summary: "{{ $labels.model }} failed every time it started and is no longer retried"
description: >-
Five replicas in a row failed at start. Before this existed a model
whose weights its engine could not load was started 155 times over
eight hours. `llmgw ctl status` has the last error; once it is dealt
with, `llmgw ctl retry <model>`. Changing the model's definition
also retries it.
- alert: ModelFailingToStart
expr: llmgw_model_start_failures >= 2
for: 5m
labels: { severity: warning }
annotations:
summary: "{{ $labels.model }} has failed to start {{ $value }} times in a row"
description: >-
On its way to being parked. The engine's own words are in
`llmgw ctl logs <node> --source <shard>` for the next attempt, or in
the control plane's log for the last one.
- alert: ModelCannotBePlaced
expr: llmgw_model_unplaceable == 1
for: 10m
labels: { severity: warning }
annotations:
summary: "{{ $labels.model }} is wanted and fits on no node"
description: >-
Short of replicas, and the last attempt found nowhere with the
memory, cards or engine it needs. `llmgw ctl explain-placement
<model>` says what each node lacks. For an on-demand model, requests
for it are told this rather than asked to retry.
- alert: EngineInstallFailed
expr: llmgw_node_engine_install{state="failed"} == 1
for: 15m
labels: { severity: warning }
annotations:
summary: "{{ $labels.engine }} {{ $labels.version }} will not install on {{ $labels.node }}"
description: >-
The node takes no model needing it until it does. The agent retries
on its own; `llmgw ctl nodes` and the node's logs on the cluster
page say why it failed.
- alert: ControlPlaneLeadership
expr: sum(llmgw_control_plane_leader) != 1
for: 2m
labels: { severity: critical }
annotations:
summary: "{{ $value }} control planes believe they are placing work"
description: >-
None means nothing is placed or replaced; two means both are, each
into GPUs the other believes free. The lease is in the store: check
it is reachable from every control plane.
- alert: CertificateExpiresSoon
expr: (llmgw_certificate_expiry_timestamp_seconds - time()) < 30 * 86400
labels: { severity: warning }
annotations:
summary: "{{ $labels.cert }} ({{ $labels.subject }}) expires in {{ $value | humanizeDuration }}"
description: >-
Certificates last a year and nothing renews them. When the agent
server's runs out every agent refuses to connect at once. Reissue
with `llmgw ctl cert` and restart the control plane.
- alert: CertificateExpiresVerySoon
expr: (llmgw_certificate_expiry_timestamp_seconds - time()) < 7 * 86400
labels: { severity: critical }
annotations:
summary: "{{ $labels.cert }} ({{ $labels.subject }}) expires in {{ $value | humanizeDuration }}"
description: See CertificateExpiresSoon; a week is not long.
- alert: PoolSaturated
expr: llmgw_pool_saturation > 1
+13 -1
View File
@@ -83,7 +83,10 @@ func (c *ctlClient) do(ctx context.Context, method, path string, body, out any)
}
defer resp.Body.Close()
if resp.StatusCode != http.StatusOK {
// Any 2xx. The API answers a change it has queued rather than made — a
// restart — with 202, and treating that as a failure printed
// "Error: 202 Accepted" over a request that had worked.
if resp.StatusCode < 200 || resp.StatusCode > 299 {
var e struct {
Error string `json:"error"`
}
@@ -135,6 +138,8 @@ func newCtlCmd() *cobra.Command {
newStatusCmd(client, &asJSON),
newModelCmd(client),
newScaleCmd(client),
newRestartCmd(client),
newRetryCmd(client),
newEngineCheckCmd(),
newToolCheckCmd(),
newWeightsCmd(client, &asJSON),
@@ -244,6 +249,10 @@ func newNodesCmd(client func() *ctlClient, asJSON *bool) *cobra.Command {
// actually served, never what is merely catalogued.
func modelStatus(m adminapi.ModelView) string {
switch {
case m.Parked:
return "parked"
case m.Unplaceable:
return "cannot place"
case m.Min == 0 && m.Total == 0:
return "stopped"
case m.Ready >= m.Min && m.Min > 0:
@@ -370,6 +379,9 @@ func newReplicasCmd(client func() *ctlClient, asJSON *bool) *cobra.Command {
if r.Stale {
state += " (stale profile)"
}
if r.Restart != "" {
state += " (restart " + r.Restart + ")"
}
fmt.Fprintf(w, "%s\t%s\t%s\t%s\t%s\t%s\t%s\t%s\n",
r.ID, r.Model, state, r.Pool, join(nodes),
(time.Duration(r.AgeSec) * time.Second).String(),
+75
View File
@@ -0,0 +1,75 @@
package cli
import (
"encoding/json"
"errors"
"fmt"
"github.com/spf13/cobra"
"gitea.mynodes.xyz/ilker/llmgw/internal/control/adminapi"
)
func newRestartCmd(client func() *ctlClient) *cobra.Command {
var (
models []string
all bool
)
c := &cobra.Command{
Use: "restart [REPLICA...]",
Short: "Replace replicas with new ones, starting each replacement before stopping the old",
Long: "How a running engine picks up what only applies at start: an engine key,\n" +
"an edited profile, a new engine version.\n\n" +
"Each replacement is placed beside the replica it replaces, and the old one\n" +
"keeps serving until the new one is ready. Replicas of one model are\n" +
"replaced one at a time. Where nothing else has room for a second copy —\n" +
"the usual case on a small CPU fleet — the old one is stopped after the\n" +
"drain grace period (5 minutes by default) and the new one starts in its\n" +
"place; a model with only that replica is unserved while it loads.\n\n" +
" llmgw ctl restart qwen2.5-7b-3f2a91c0\n" +
" llmgw ctl restart --model qwen2.5-7b\n" +
" llmgw ctl restart --all",
RunE: func(cmd *cobra.Command, args []string) error {
if len(args) == 0 && len(models) == 0 && !all {
return errors.New("name replicas, --model, or --all")
}
body, _ := json.Marshal(adminapi.RestartRequest{Replicas: args, Models: models, All: all})
var res adminapi.RestartResult
if err := client().post(cmd.Context(), "/api/v1/restart", body, &res); err != nil {
return err
}
fmt.Printf("%d replica(s) queued for restart; watch with: llmgw ctl replicas\n", len(res.Queued))
return nil
},
}
c.Flags().StringArrayVar(&models, "model", nil, "every replica of this model; repeatable")
c.Flags().BoolVar(&all, "all", false, "every ready replica in the cluster")
return c
}
func newRetryCmd(client func() *ctlClient) *cobra.Command {
return &cobra.Command{
Use: "retry MODEL",
Short: "Try a parked model again, and clear its backoff",
Long: "A model whose replicas fail every time they start is parked after five\n" +
"failures in a row: it is not placed again until its definition changes\n" +
"or this is run. Run it once the cause — the weights, the engine, the\n" +
"profile — has been dealt with; `llmgw ctl status` lists parked models with\n" +
"their last error.",
Args: cobra.ExactArgs(1),
RunE: func(cmd *cobra.Command, args []string) error {
var res struct {
WasParked bool `json:"was_parked"`
}
if err := client().post(cmd.Context(), "/api/v1/models/"+args[0]+"/retry", []byte(`{}`), &res); err != nil {
return err
}
if res.WasParked {
fmt.Printf("%s unparked; it is placed on the next pass\n", args[0])
} else {
fmt.Printf("%s was not parked; its backoff is cleared\n", args[0])
}
return nil
},
}
}
+24
View File
@@ -0,0 +1,24 @@
package cli
import (
"net/http"
"net/http/httptest"
"testing"
)
func TestAQueuedChangeIsNotReportedAsAnError(t *testing.T) {
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
w.WriteHeader(http.StatusAccepted)
_, _ = w.Write([]byte(`{"queued":["r1"]}`))
}))
defer srv.Close()
var out struct {
Queued []string `json:"queued"`
}
if err := newCtlClient(srv.URL).post(t.Context(), "/api/v1/restart", []byte(`{}`), &out); err != nil {
t.Fatalf("202 read as a failure: %v", err)
}
if len(out.Queued) != 1 {
t.Errorf("the body of a 202 was not read: %+v", out)
}
}
+6
View File
@@ -36,6 +36,7 @@ import (
"gitea.mynodes.xyz/ilker/llmgw/internal/promq"
"gitea.mynodes.xyz/ilker/llmgw/internal/router"
storepkg "gitea.mynodes.xyz/ilker/llmgw/internal/store"
"gitea.mynodes.xyz/ilker/llmgw/internal/tlsconf"
"gitea.mynodes.xyz/ilker/llmgw/internal/version"
"gitea.mynodes.xyz/ilker/llmgw/internal/web"
)
@@ -131,6 +132,11 @@ func newServerCmd() *cobra.Command {
// On the same /metrics as the traffic figures: correlating
// "the cluster was full" with "requests were queueing" should
// not need two systems.
if err := rt.RegisterMetrics(tlsconf.ExpiryCollector(map[string]string{
"agent-server": cfg.AgentTLS.Cert, "agent-ca": cfg.AgentTLS.CA,
})); err != nil {
return err
}
if err := rt.RegisterMetrics(fleetmetrics.New(ctl)); err != nil {
return err
}
+8
View File
@@ -94,6 +94,14 @@ func printSummary(cmd *cobra.Command, s adminapi.Summary) {
fmt.Fprintf(out, "\nMODELS %d of %d have the replicas they asked for\n",
s.Models.Satisfied, s.Models.Total)
if len(s.Parked) > 0 {
fmt.Fprintln(out, "\nPARKED — failed every time they started; not retried until fixed")
for _, p := range s.Parked {
fmt.Fprintf(out, " %s (%d failures): %s\n", p.Model, p.Failures, p.Reason)
}
fmt.Fprintln(out, "\n Once fixed: llmgw ctl retry <model>")
}
if len(s.Unplaceable) > 0 {
fmt.Fprintln(out, "\nNOT PLACED")
for _, m := range s.Unplaceable {
+73 -2
View File
@@ -77,6 +77,8 @@ func (a *API) Register(mux *http.ServeMux) {
mux.HandleFunc("DELETE /api/v1/nodes/{id}", a.removeNode)
mux.HandleFunc("POST /api/v1/fabric-test", a.fabricTest)
mux.HandleFunc("GET /api/v1/nodes/{id}/logs", a.nodeLogs)
mux.HandleFunc("POST /api/v1/restart", a.restart)
mux.HandleFunc("POST /api/v1/models/{name}/retry", a.retryModel)
}
// --- responses ---
@@ -232,6 +234,18 @@ type ModelView struct {
Max int `json:"max_replicas"`
Ready int `json:"ready_replicas"`
Total int `json:"total_replicas"`
// Desired is how many it should have now: Min, or for an on-demand
// model one while it is being asked for and none otherwise.
Desired int `json:"desired_replicas"`
// Failures counts replicas that failed to start in a row.
Failures int `json:"failures,omitempty"`
// Parked is set once Failures reached the limit; the model is not retried
// until its definition changes or `ctl retry` is run. Reason is the last
// error.
Parked bool `json:"parked,omitempty"`
Reason string `json:"reason,omitempty"`
// Unplaceable is set when it is short and the last attempt found no node.
Unplaceable bool `json:"unplaceable,omitempty"`
Fallbacks []string `json:"fallbacks,omitempty"`
Profiles []string `json:"profiles"`
}
@@ -243,8 +257,11 @@ type ReplicaView struct {
// and anyone else holding the admin listener who needs to reach an engine
// directly. The client-facing pages never render it.
EngineKey string `json:"engine_key,omitempty"`
Model string `json:"model"`
Profile string `json:"profile"`
// Restart is "queued" or "replacing" while an operator's restart of this
// replica is under way.
Restart string `json:"restart,omitempty"`
Model string `json:"model"`
Profile string `json:"profile"`
// Stale is true when the profile has been edited since this replica was
// placed, so it is running a definition the catalog no longer holds.
//
@@ -438,13 +455,20 @@ func (a *API) Models() []ModelView {
}
}
health := map[string]reconciler.ModelHealth{}
for _, h := range a.ctl.Reconciler().ModelHealth() {
health[h.Model] = h
}
out := make([]ModelView, 0)
for _, m := range a.ctl.Catalog().Models() {
h := health[m.Name]
v := ModelView{
Name: m.Name, Description: m.Description, Aliases: m.Aliases, Source: m.Source,
Min: m.Scaling.Min, Max: m.Scaling.Max,
Ready: ready[m.Name], Total: total[m.Name],
Fallbacks: m.Fallbacks,
Desired: h.Desired, Failures: h.Failures,
Parked: h.Parked, Reason: h.Reason, Unplaceable: h.Unplaceable,
}
for _, p := range m.Profiles {
v.Profiles = append(v.Profiles, p.Name)
@@ -476,6 +500,7 @@ func (a *API) Replicas(model string) []ReplicaView {
State: rep.State.String(), Pool: rep.Pool, Fabric: rep.Fabric,
Endpoint: rep.Endpoint, Err: rep.Err,
EngineKey: a.ctl.EngineKey(rep.ID),
Restart: a.ctl.Reconciler().RestartState(rep.ID),
Stale: a.profileChanged(rep.Model, rep.Profile, rep.ProfileFP),
AgeSec: int64(time.Since(rep.CreatedAt).Seconds()),
IdleSec: -1,
@@ -1044,3 +1069,49 @@ func (a *API) Topology(id string) (TopologyView, bool) {
func (a *API) NVLink() nvlink.Report {
return nvlink.Analyze(a.ctl.Inventory().Nodes())
}
// RestartRequest names what to restart: replicas by id, every replica of some
// models, or — All — every ready replica in the cluster.
type RestartRequest struct {
Replicas []string `json:"replicas,omitempty"`
Models []string `json:"models,omitempty"`
All bool `json:"all,omitempty"`
}
// RestartResult lists the replicas queued.
type RestartResult struct {
Queued []string `json:"queued"`
}
func (a *API) restart(w http.ResponseWriter, r *http.Request) {
var req RestartRequest
if err := json.NewDecoder(io.LimitReader(r.Body, 1<<16)).Decode(&req); err != nil {
writeError(w, http.StatusBadRequest, "the restart request is not valid JSON: "+err.Error())
return
}
ids := slices.Clone(req.Replicas)
for _, rep := range a.ctl.Reconciler().ReadyReplicas() {
if req.All || slices.Contains(req.Models, rep.Model) {
ids = append(ids, rep.ID)
}
}
if len(ids) == 0 {
writeError(w, http.StatusBadRequest, "nothing ready matches; name replicas, models, or all")
return
}
if err := a.ctl.Reconciler().Restart(ids...); err != nil {
writeError(w, http.StatusNotFound, err.Error())
return
}
writeJSON(w, http.StatusAccepted, RestartResult{Queued: ids})
}
func (a *API) retryModel(w http.ResponseWriter, r *http.Request) {
name := r.PathValue("name")
if _, ok := a.ctl.Catalog().Resolve(name); !ok {
writeError(w, http.StatusNotFound, "no model called "+name)
return
}
was := a.ctl.Reconciler().Retry(name)
writeJSON(w, http.StatusOK, map[string]bool{"was_parked": was})
}
+18 -2
View File
@@ -25,6 +25,9 @@ type Summary struct {
// Unplaceable names models the scheduler currently cannot satisfy. This is
// the line that turns "something is wrong" into a place to look.
Unplaceable []string `json:"unplaceable,omitempty"`
// Parked are models that failed every time they started and are no
// longer retried, with the last error.
Parked []ParkedModel `json:"parked,omitempty"`
// IdleReplicas are serving replicas nothing has asked for in a while, which
// on a fleet with more models than GPUs is the first place to find room.
IdleReplicas []IdleReplica `json:"idle_replicas,omitempty"`
@@ -201,17 +204,30 @@ func (a *API) Summary() Summary {
for _, m := range a.Models() {
s.Models.Total++
if m.Ready >= m.Min {
if m.Parked {
s.Parked = append(s.Parked, ParkedModel{Model: m.Name, Failures: m.Failures, Reason: m.Reason})
}
// Against what it should have now, not its minimum: an on-demand
// model's minimum is 0, so one that was asked for and fitted nowhere
// used to count as satisfied.
if m.Ready >= m.Desired && !m.Parked {
s.Models.Satisfied++
continue
}
s.Models.Short++
// Short of replicas and unable to place more is the actionable case;
// short and still starting is just a cluster doing its job.
if m.Total <= m.Ready {
if m.Unplaceable || (m.Total <= m.Ready && !m.Parked) {
s.Unplaceable = append(s.Unplaceable, m.Name)
}
}
sort.Strings(s.Unplaceable)
return s
}
// ParkedModel is a model no longer retried, and why.
type ParkedModel struct {
Model string `json:"model"`
Failures int `json:"failures"`
Reason string `json:"reason"`
}
+3
View File
@@ -171,6 +171,9 @@ func (c *Controller) Nodes() []inventory.Node { return c.inv.Nodes() }
func (c *Controller) FreeGPUs(nodeID string) []int { return c.inv.FreeGPUs(nodeID) }
func (c *Controller) Replicas() []reconciler.Replica { return c.rec.Replicas() }
// ModelHealth is how each model in the catalog stands.
func (c *Controller) ModelHealth() []reconciler.ModelHealth { return c.rec.ModelHealth() }
// Reconciler exposes the reconciliation loop, for status reports and ctl.
func (c *Controller) Reconciler() *reconciler.Reconciler { return c.rec }
+76 -4
View File
@@ -39,9 +39,25 @@ type Source interface {
// derived from state that already exists, so computing them when asked is both
// simpler and fresher than maintaining a parallel copy that can drift. Nothing
// runs between scrapes.
// HealthSource adds what the reconciler knows about each model and where
// engine installs stand, and whether this control plane is the one placing
// work. Asked for by type assertion, so a source without it — a test's —
// publishes what it did before.
type HealthSource interface {
ModelHealth() []reconciler.ModelHealth
IsLeader() bool
}
type Collector struct {
src Source
modelDesired *prometheus.Desc
modelFailures *prometheus.Desc
modelParked *prometheus.Desc
modelUnplaceable *prometheus.Desc
engineInstall *prometheus.Desc
leader *prometheus.Desc
nodeUp *prometheus.Desc
nodeGPUs *prometheus.Desc
nodeGPUsFree *prometheus.Desc
@@ -137,7 +153,26 @@ func New(src Source) *Collector {
replicaState: prometheus.NewDesc("llmgw_replicas",
"Replicas by model and state.", []string{"model", "state"}, nil),
replicaReady: prometheus.NewDesc("llmgw_model_replicas_ready",
"Replicas of a model that are serving.", []string{"model"}, nil),
"Replicas of a model that are serving. Published, as 0 when none are, for every "+
"model that should be running; a stopped model, or an on-demand one nobody is "+
"asking for, publishes nothing.", []string{"model"}, nil),
modelDesired: prometheus.NewDesc("llmgw_model_replicas_desired",
"Replicas a model should have now: its minimum, or for an on-demand model 1 while "+
"it is being asked for.", []string{"model"}, nil),
modelFailures: prometheus.NewDesc("llmgw_model_start_failures",
"Replicas of a model that failed to start in a row. Resets when one comes up.",
[]string{"model"}, nil),
modelParked: prometheus.NewDesc("llmgw_model_parked",
"1 when a model failed so often in a row that it is no longer retried.",
[]string{"model"}, nil),
modelUnplaceable: prometheus.NewDesc("llmgw_model_unplaceable",
"1 when a model is short of replicas and the last attempt found no node to place one on.",
[]string{"model"}, nil),
engineInstall: prometheus.NewDesc("llmgw_node_engine_install",
"Where installing an engine stands on a node: 1 for the current state.",
[]string{"node", "engine", "version", "state"}, nil),
leader: prometheus.NewDesc("llmgw_control_plane_leader",
"1 when this control plane is the one placing work, 0 when it is a standby.", nil, nil),
replicaIdle: prometheus.NewDesc("llmgw_replica_idle_seconds",
"Seconds since a replica was last sent a request. Absent when nothing has been reported, "+
"which is not the same as idle.", []string{"model", "replica"}, nil),
@@ -201,6 +236,8 @@ func (c *Collector) Describe(ch chan<- *prometheus.Desc) {
c.gpuMemoryTotal, c.gpuMemoryUsed, c.gpuUtil, c.gpuTemp, c.gpuHealthy,
c.gpuNVLink, c.gpuECC,
c.replicaState, c.replicaReady, c.replicaIdle,
c.modelDesired, c.modelFailures, c.modelParked, c.modelUnplaceable,
c.engineInstall, c.leader,
} {
ch <- d
}
@@ -387,8 +424,43 @@ func (c *Collector) Collect(ch chan<- prometheus.Metric) {
ch <- prometheus.MustNewConstMetric(c.replicaState, prometheus.GaugeValue,
float64(v), k[0], k[1])
}
for model, v := range ready {
ch <- prometheus.MustNewConstMetric(c.replicaReady, prometheus.GaugeValue,
float64(v), model)
hs, ok := c.src.(HealthSource)
if !ok {
for model, v := range ready {
ch <- prometheus.MustNewConstMetric(c.replicaReady, prometheus.GaugeValue,
float64(v), model)
}
return
}
for _, n := range c.src.Nodes() {
for _, p := range n.Provision {
ch <- prometheus.MustNewConstMetric(c.engineInstall, prometheus.GaugeValue, 1,
n.ID, p.Engine, p.Version, p.State)
}
}
lead := 0.0
if hs.IsLeader() {
lead = 1
}
ch <- prometheus.MustNewConstMetric(c.leader, prometheus.GaugeValue, lead)
for _, h := range hs.ModelHealth() {
// Ready is published at zero for what should be running — the one
// value ModelHasNothingServing exists to catch, and which, emitted
// only for replicas that were Ready, it never saw.
if h.Desired > 0 || ready[h.Model] > 0 {
ch <- prometheus.MustNewConstMetric(c.replicaReady, prometheus.GaugeValue,
float64(ready[h.Model]), h.Model)
}
ch <- prometheus.MustNewConstMetric(c.modelDesired, prometheus.GaugeValue, float64(h.Desired), h.Model)
ch <- prometheus.MustNewConstMetric(c.modelFailures, prometheus.GaugeValue, float64(h.Failures), h.Model)
ch <- prometheus.MustNewConstMetric(c.modelParked, prometheus.GaugeValue, b2f(h.Parked), h.Model)
ch <- prometheus.MustNewConstMetric(c.modelUnplaceable, prometheus.GaugeValue, b2f(h.Unplaceable), h.Model)
}
}
func b2f(b bool) float64 {
if b {
return 1
}
return 0
}
@@ -196,3 +196,35 @@ func TestEveryIBPortIsReportedWithItsCounters(t *testing.T) {
}
}
}
type healthSource struct {
fakeSource
health []reconciler.ModelHealth
}
func (h healthSource) ModelHealth() []reconciler.ModelHealth { return h.health }
func (h healthSource) IsLeader() bool { return true }
func TestAModelThatShouldRunPublishesZeroReplicasReady(t *testing.T) {
// ModelHasNothingServing alerts on llmgw_model_replicas_ready == 0, and
// the series was only ever published while a replica was ready — so the
// alert could not fire for the case it was written for.
src := healthSource{health: []reconciler.ModelHealth{
{Model: "wanted", Desired: 1, Failures: 5, Parked: true, Reason: "boom"},
{Model: "stopped", Desired: 0},
}}
out := gather(t, src)
for _, want := range []string{
"llmgw_model_replicas_ready model=wanted -> 0",
"llmgw_model_parked model=wanted -> 1",
"llmgw_model_start_failures model=wanted -> 5",
"llmgw_control_plane_leader -> 1",
} {
if !strings.Contains(out, want) {
t.Errorf("missing %s", want)
}
}
if strings.Contains(out, "llmgw_model_replicas_ready model=stopped") {
t.Error("a stopped model publishes a ready count, so the alert would fire for it")
}
}
+336 -14
View File
@@ -3,10 +3,15 @@ package reconciler
import (
"context"
crypterand "crypto/rand"
"crypto/sha256"
"encoding/hex"
"encoding/json"
"errors"
"fmt"
"log/slog"
"math/rand/v2"
"slices"
"strings"
"sync"
"time"
@@ -65,6 +70,15 @@ type Config struct {
// Without a bound, draining a node whose model cannot be placed anywhere
// else would never finish.
DrainGrace time.Duration
// ParkAfter is how many failures in a row stop a model being retried.
//
// Backoff alone never gives up: it grows to MaxBackoff and keeps trying,
// and a model whose weights its engine cannot load — an architecture the
// installed llama.cpp does not know — was started 155 times over eight
// hours on the live cluster, each time failing the same way, with nothing
// to show for it but the log. A parked model is not placed again until its
// definition changes or an operator runs `llmgw ctl retry`.
ParkAfter int
// PortRange is the per-node port range for engine HTTP and dist-init.
PortRange [2]int
// MgmtInterface is the ethernet interface used for NCCL bootstrap.
@@ -125,6 +139,7 @@ func DefaultConfig() Config {
EvictionGrace: 2 * time.Minute,
MinLifetime: 5 * time.Minute,
DrainGrace: 5 * time.Minute,
ParkAfter: 5,
PortRange: [2]int{20000, 20999},
MgmtInterface: "eth0",
}
@@ -164,6 +179,9 @@ func (c *Config) normalize() {
if c.DrainGrace <= 0 {
c.DrainGrace = d.DrainGrace
}
if c.ParkAfter <= 0 {
c.ParkAfter = d.ParkAfter
}
if c.PortRange[1] <= c.PortRange[0] {
c.PortRange = d.PortRange
}
@@ -204,6 +222,15 @@ type Reconciler struct {
// because it is state about this moment and not about what the model is.
// A model nobody has wanted has no entry and is simply not running.
wanted map[string]time.Time
// restartQueue holds, per model, the replicas an operator asked to be
// restarted and not yet begun. One is replaced at a time: replacing them
// all at once would need the memory of every replica twice over.
restartQueue map[string][]string
// parked holds models that failed ParkAfter times in a row.
parked map[string]parkRecord
// placementLogged remembers the last "no placement" said per model, so it
// is said when it changes rather than on every five-second pass.
placementLogged map[string]placementNote
wake chan struct{}
stop chan struct{}
@@ -225,18 +252,21 @@ func New(cfg Config, cat *catalog.Live, inv *inventory.Store, disp Dispatcher, l
}
return &Reconciler{
cfg: cfg, cat: cat, inv: inv, disp: disp, log: log,
sched: placement.New(inv),
now: time.Now,
replicas: map[string]*Replica{},
nextAttempt: map[string]time.Time{},
attempts: map[string]int{},
reservedPorts: map[int]struct{}{},
wanted: map[string]time.Time{},
lastExplain: map[string]placement.Result{},
wake: make(chan struct{}, 1),
stop: make(chan struct{}),
done: make(chan struct{}),
startedAt: time.Now(),
sched: placement.New(inv),
now: time.Now,
replicas: map[string]*Replica{},
nextAttempt: map[string]time.Time{},
attempts: map[string]int{},
reservedPorts: map[int]struct{}{},
wanted: map[string]time.Time{},
restartQueue: map[string][]string{},
parked: map[string]parkRecord{},
placementLogged: map[string]placementNote{},
lastExplain: map[string]placement.Result{},
wake: make(chan struct{}, 1),
stop: make(chan struct{}),
done: make(chan struct{}),
startedAt: time.Now(),
}
}
@@ -458,6 +488,12 @@ func (r *Reconciler) Want(model string) bool {
}
r.mu.Lock()
r.wanted[m.Name] = r.now()
// A placement that failed before this request says nothing about now:
// forgotten, so the pass this triggers decides afresh and Unservable
// does not answer from a stale result.
if res, ok := r.lastExplain[m.Name]; ok && res.Placement == nil {
delete(r.lastExplain, m.Name)
}
r.mu.Unlock()
// Immediately, not on the next tick. Someone is waiting on this request,
// and five seconds of ticker is five seconds they spend staring at nothing.
@@ -663,6 +699,17 @@ func (r *Reconciler) reapFailed(ctx context.Context) {
// replica is about to disappear, and the backoff has to survive it.
r.attempts[rep.Model]++
r.nextAttempt[rep.Model] = r.now().Add(r.backoff(r.attempts[rep.Model]))
if n := r.attempts[rep.Model]; n >= r.cfg.ParkAfter {
if _, already := r.parked[rep.Model]; !already {
m, _ := r.cat.Load().Resolve(rep.Model)
r.parked[rep.Model] = parkRecord{
Since: r.now(), Reason: rep.Err, Failures: n, fp: modelFingerprint(m),
}
r.log.Error("model parked: it failed every time it started, and is not retried "+
"until its definition changes or `llmgw ctl retry` is run",
"model", rep.Model, "failures", n, "last_error", rep.Err)
}
}
}
}
r.mu.Unlock()
@@ -708,24 +755,54 @@ func (r *Reconciler) reconcileModel(ctx context.Context, m catalog.Model) {
// pointer stops being safe the moment it is in the map: an agent's status
// report can mutate it from another goroutine at any time.
r.mu.Lock()
var liveIDs, drainingIDs, avoid []string
var liveIDs, drainingIDs, retiringIDs, avoid []string
readyLive := 0
for _, rep := range r.replicas {
if rep.Model != m.Name {
continue
}
switch {
case rep.State.Counts() && !rep.RetiringAt.IsZero():
retiringIDs = append(retiringIDs, rep.ID)
avoid = append(avoid, rep.NodeIDs()...)
case rep.State.Counts():
liveIDs = append(liveIDs, rep.ID)
avoid = append(avoid, rep.NodeIDs()...)
if rep.State == Ready {
readyLive++
}
case rep.State == Draining:
drainingIDs = append(drainingIDs, rep.ID)
}
}
// The next restart an operator queued, once the last one is done.
if len(retiringIDs) == 0 {
if id, ok := r.nextRestartLocked(m.Name); ok {
r.replicas[id].RetiringAt = r.now()
retiringIDs = append(retiringIDs, id)
liveIDs = slices.DeleteFunc(liveIDs, func(x string) bool { return x == id })
readyLive--
r.log.Info("restarting a replica: placing its replacement first",
"replica", id, "model", m.Name)
}
}
blockedUntil := r.nextAttempt[m.Name]
park, isParked := r.parked[m.Name]
if isParked && park.fp != modelFingerprint(m) {
// Somebody changed the model — a new source, an edited profile — so
// the reason it kept failing may be gone. Worth one more streak.
delete(r.parked, m.Name)
r.attempts[m.Name] = 0
delete(r.nextAttempt, m.Name)
isParked, blockedUntil = false, time.Time{}
r.log.Info("model unparked: its definition changed", "model", m.Name)
}
r.mu.Unlock()
desired := r.desiredFor(m)
switch {
case len(liveIDs) < desired && isParked:
// Not placed. Its failure is on every status the operator reads.
case len(liveIDs) < desired:
if r.now().Before(blockedUntil) {
// Still in backoff. Rebuilding immediately would hot-loop on a
@@ -763,6 +840,112 @@ func (r *Reconciler) reconcileModel(ctx context.Context, m catalog.Model) {
r.stopReplica(ctx, id)
}
}
r.retire(ctx, retiringIDs, readyLive, len(liveIDs), desired)
}
// retire finishes restarting replicas.
//
// A retiring replica keeps serving until its replacement is ready, and is then
// drained like any other. When no replacement could be placed at all —
// nowhere else has the memory, which on a small CPU fleet is the usual case —
// it is stopped after DrainGrace so its replacement can take its place: a
// restart in place, with the model unserved while it loads if it had no other
// replica. Said in the log, since that is the one case with an outage.
func (r *Reconciler) retire(ctx context.Context, ids []string, readyLive, live, desired int) {
for _, id := range ids {
r.mu.Lock()
rep, ok := r.replicas[id]
var since time.Time
if ok {
since = rep.RetiringAt
}
r.mu.Unlock()
if !ok {
continue
}
switch {
case readyLive >= desired:
r.drainForRestart(ctx, id)
case live < desired && r.now().Sub(since) > r.cfg.DrainGrace:
r.log.Warn("restarting a replica in place: no room for its replacement beside it",
"replica", id, "after", r.cfg.DrainGrace)
r.drainForRestart(ctx, id)
r.stopReplica(ctx, id)
}
}
}
func (r *Reconciler) drainForRestart(ctx context.Context, id string) {
r.mu.Lock()
rep, ok := r.replicas[id]
var shards []Shard
if ok {
rep.State, rep.DrainingAt, rep.RetiringAt = Draining, r.now(), time.Time{}
shards = slices.Clone(rep.Shards)
}
r.mu.Unlock()
for _, s := range shards {
if err := r.disp.DrainShard(ctx, s.NodeID, s.ID); err != nil {
r.log.Warn("draining a shard failed", "shard", s.ID, "err", err)
}
}
}
// nextRestartLocked pops the next queued replica of a model that is still
// ready to be restarted. Called with mu held.
func (r *Reconciler) nextRestartLocked(model string) (string, bool) {
q := r.restartQueue[model]
for len(q) > 0 {
id := q[0]
q = q[1:]
if rep, ok := r.replicas[id]; ok && rep.State == Ready && rep.RetiringAt.IsZero() {
r.restartQueue[model] = q
return id, true
}
}
delete(r.restartQueue, model)
return "", false
}
// ErrNoSuchReplica means no ready replica has that id.
var ErrNoSuchReplica = errors.New("no ready replica with that id")
// Restart queues replicas to be replaced, each by a new one started beside it
// before it is stopped — to pick up an edited profile, an engine key, a new
// engine version. Replicas of one model are replaced one at a time.
func (r *Reconciler) Restart(ids ...string) error {
r.mu.Lock()
for _, id := range ids {
rep, ok := r.replicas[id]
if !ok || rep.State != Ready {
r.mu.Unlock()
return fmt.Errorf("%w: %s", ErrNoSuchReplica, id)
}
if !slices.Contains(r.restartQueue[rep.Model], id) && rep.RetiringAt.IsZero() {
r.restartQueue[rep.Model] = append(r.restartQueue[rep.Model], id)
}
}
r.mu.Unlock()
r.Notify()
return nil
}
// RestartState says where a replica's restart stands: "queued", "replacing",
// or "" when none was asked for.
func (r *Reconciler) RestartState(id string) string {
r.mu.Lock()
defer r.mu.Unlock()
rep, ok := r.replicas[id]
switch {
case !ok:
return ""
case !rep.RetiringAt.IsZero():
return "replacing"
case slices.Contains(r.restartQueue[rep.Model], id):
return "queued"
}
return ""
}
// scaleUp places and starts one replica per pass.
@@ -795,7 +978,7 @@ func (r *Reconciler) scaleUp(ctx context.Context, m catalog.Model, avoid []strin
if m.Scaling.OnDemand() && r.makeRoomFor(ctx, m) {
return
}
r.log.Warn("no placement for model", "model", m.Name, "reasons", res.Explain())
r.notePlacementFailure(m.Name, res.Explain())
return
}
if err := r.startReplica(ctx, m, res.Placement); err != nil {
@@ -1393,3 +1576,142 @@ func (r *Reconciler) engineKey(replicaID string) string {
}
return r.cfg.EngineKey(replicaID)
}
// parkRecord is why a model stopped being retried.
type parkRecord struct {
Since time.Time
Reason string
Failures int
// fp is the model's definition when it was parked; a different one
// unparks it.
fp string
}
type placementNote struct {
text string
at time.Time
}
// modelFingerprint identifies a model's definition, so a change to it can be
// told from a retry of the same thing.
func modelFingerprint(m catalog.Model) string {
b, _ := json.Marshal(m)
sum := sha256.Sum256(b)
return hex.EncodeToString(sum[:8])
}
// placementLogEvery is how often an unchanged placement failure is repeated.
const placementLogEvery = 10 * time.Minute
// notePlacementFailure logs why a model could not be placed, when the reason
// is new or has not been said for a while. Said on every pass it was 250
// identical lines in two days for one model, burying everything else.
func (r *Reconciler) notePlacementFailure(model, why string) {
r.mu.Lock()
last := r.placementLogged[model]
say := last.text != why || r.now().Sub(last.at) >= placementLogEvery
if say {
r.placementLogged[model] = placementNote{text: why, at: r.now()}
}
r.mu.Unlock()
if say {
r.log.Warn("no placement for model", "model", model, "reasons", why)
}
}
// Retry clears a parked model and its backoff, so it is placed on the next
// pass.
func (r *Reconciler) Retry(model string) bool {
r.mu.Lock()
_, was := r.parked[model]
delete(r.parked, model)
r.attempts[model] = 0
delete(r.nextAttempt, model)
r.mu.Unlock()
r.Notify()
return was
}
// ModelHealth is how one model stands, for status and metrics.
type ModelHealth struct {
Model string
// Desired is how many replicas it should have now; 0 for a stopped model
// or an on-demand one nobody is asking for.
Desired int
Ready int
// Failures counts replicas that failed in a row.
Failures int
// Parked is set when Failures reached ParkAfter; Reason is the last error.
Parked bool
ParkedSince time.Time
Reason string
// Unplaceable is set when it is short of replicas and the last attempt to
// place one found nowhere; Placement says why.
Unplaceable bool
Placement string
}
// ModelHealth reports every model in the catalog.
func (r *Reconciler) ModelHealth() []ModelHealth {
models := r.cat.Load().Models()
desired := map[string]int{}
for _, m := range models {
desired[m.Name] = r.desiredFor(m)
}
r.mu.Lock()
defer r.mu.Unlock()
ready, live := map[string]int{}, map[string]int{}
for _, rep := range r.replicas {
if rep.State == Ready {
ready[rep.Model]++
}
if rep.State.Counts() {
live[rep.Model]++
}
}
out := make([]ModelHealth, 0, len(models))
for _, m := range models {
h := ModelHealth{Model: m.Name, Desired: desired[m.Name], Ready: ready[m.Name],
Failures: r.attempts[m.Name]}
if p, ok := r.parked[m.Name]; ok {
h.Parked, h.ParkedSince, h.Reason = true, p.Since, p.Reason
}
if res, ok := r.lastExplain[m.Name]; ok && res.Placement == nil && live[m.Name] < h.Desired {
h.Unplaceable, h.Placement = true, res.Explain()
}
out = append(out, h)
}
return out
}
// Unservable says why a model cannot be served at all, so a request for it is
// told now rather than asked to retry something that will never come up.
func (r *Reconciler) Unservable(model string) (string, bool) {
m, ok := r.cat.Load().Resolve(model)
if !ok {
return "", false
}
r.mu.Lock()
defer r.mu.Unlock()
if p, ok := r.parked[m.Name]; ok {
return fmt.Sprintf("%s failed to start %d times in a row and is not being retried: %s",
m.Name, p.Failures, p.Reason), true
}
for _, rep := range r.replicas {
if rep.Model == m.Name && rep.State.Counts() {
return "", false // something is on its way
}
}
if res, ok := r.lastExplain[m.Name]; ok && res.Placement == nil {
return m.Name + " cannot be placed on any node: " + firstLine(res.Explain()), true
}
return "", false
}
func firstLine(s string) string {
s = strings.TrimSpace(s)
if i := strings.IndexByte(s, '\n'); i >= 0 {
return strings.TrimSpace(s[:i]) + " (and others; see llmgw ctl explain-placement)"
}
return s
}
@@ -1,6 +1,7 @@
package reconciler
import (
"bytes"
"context"
"fmt"
"log/slog"
@@ -1741,3 +1742,193 @@ func TestAnEngineIsStartedWithItsReplicasKey(t *testing.T) {
}
}
}
func replicaByID(h *harness, id string) (Replica, bool) {
for _, rep := range h.r.Replicas() {
if rep.ID == id {
return rep, true
}
}
return Replica{}, false
}
func TestARestartedReplicaServesUntilItsReplacementIsReady(t *testing.T) {
// Restarting is how a running engine picks up a key, an edited profile or
// a new engine version. Stopping first would leave a one-replica model
// unserved for as long as its replacement takes to load.
h := newHarness(t, testModel(1), gpuNode("1", 4), gpuNode("2", 4))
h.r.Reconcile(t.Context())
h.markReady()
old := h.r.ReadyReplicas()[0].ID
if err := h.r.Restart(old); err != nil {
t.Fatal(err)
}
h.r.Reconcile(t.Context())
if rep, _ := replicaByID(h, old); rep.State != Ready || h.r.RestartState(old) != "replacing" {
t.Fatalf("the old replica stopped serving before its replacement was up: %+v", rep)
}
if n := len(h.r.Replicas()); n != 2 {
t.Fatalf("no replacement was placed beside it: %d replicas", n)
}
h.markReady()
h.r.Reconcile(t.Context())
h.r.Reconcile(t.Context())
if _, still := replicaByID(h, old); still {
if rep, _ := replicaByID(h, old); rep.State != Draining && rep.State != Stopped {
t.Errorf("the old replica is %s after its replacement became ready", rep.State)
}
}
if got := len(h.r.ReadyReplicas()); got != 1 {
t.Errorf("%d ready replicas at the end; the model asks for 1", got)
}
}
func TestARestartWithNoRoomBesideItHappensInPlace(t *testing.T) {
// One node, room for one replica: the replacement can only start once the
// old one has gone. Waiting forever would mean never restarting at all.
h := newHarness(t, testModel(1), gpuNode("1", 4))
h.r.Reconcile(t.Context())
h.markReady()
old := h.r.ReadyReplicas()[0].ID
if err := h.r.Restart(old); err != nil {
t.Fatal(err)
}
h.r.Reconcile(t.Context())
if rep, _ := replicaByID(h, old); rep.State != Ready {
t.Fatalf("stopped before the grace period: %s", rep.State)
}
h.advance(h.r.cfg.DrainGrace + time.Second)
h.r.Reconcile(t.Context())
if rep, ok := replicaByID(h, old); ok && rep.State == Ready {
t.Fatal("still serving after the grace period with no replacement possible")
}
for i := 0; i < 3; i++ {
h.r.Reconcile(t.Context())
h.markReady()
}
ready := h.r.ReadyReplicas()
if len(ready) != 1 || ready[0].ID == old {
t.Errorf("expected one new replica in its place, got %+v", ready)
}
}
func TestReplicasOfAModelAreRestartedOneAtATime(t *testing.T) {
h := newHarness(t, testModel(2), gpuNode("1", 8), gpuNode("2", 8), gpuNode("3", 8))
for i := 0; i < 3; i++ {
h.r.Reconcile(t.Context())
h.markReady()
}
var ids []string
for _, rep := range h.r.ReadyReplicas() {
ids = append(ids, rep.ID)
}
if err := h.r.Restart(ids...); err != nil {
t.Fatal(err)
}
h.r.Reconcile(t.Context())
states := []string{h.r.RestartState(ids[0]), h.r.RestartState(ids[1])}
if !slices.Contains(states, "replacing") || !slices.Contains(states, "queued") {
t.Errorf("expected one replacing and one queued, got %v", states)
}
if err := h.r.Restart("no-such"); err == nil {
t.Error("an unknown replica was accepted")
}
}
// failRepeatedly starts and fails the model's replica n times over.
func failRepeatedly(t *testing.T, h *harness, n int, why string) {
t.Helper()
for i := 0; i < n; i++ {
h.disp.reset()
h.r.Reconcile(t.Context())
started := h.disp.startedShards()
if len(started) == 0 {
return // parked, or still in backoff
}
s := started[0]
h.r.OnShardStatus(s.ReplicaID, s.ShardID, supervisor.Failed, why)
h.r.Reconcile(t.Context())
h.advance(DefaultConfig().MaxBackoff + time.Second)
}
}
func TestAModelThatAlwaysFailsIsParkedNotRetriedForever(t *testing.T) {
// kev-0.5b: an architecture the node's llama.cpp did not know, started
// 155 times in eight hours, the same error every time.
h := newHarness(t, testModel(1), gpuNode("1", 8))
failRepeatedly(t, h, DefaultConfig().ParkAfter, "unknown model architecture: 'ggmlc'")
h.disp.reset()
h.advance(DefaultConfig().MaxBackoff + time.Second)
h.r.Reconcile(t.Context())
if n := len(h.disp.startedShards()); n != 0 {
t.Fatalf("a parked model was started again (%d shards)", n)
}
var mh ModelHealth
for _, x := range h.r.ModelHealth() {
if x.Model == "llama" {
mh = x
}
}
if !mh.Parked || !strings.Contains(mh.Reason, "ggmlc") || mh.Failures < DefaultConfig().ParkAfter {
t.Errorf("health %+v", mh)
}
if why, no := h.r.Unservable("llama"); !no || !strings.Contains(why, "ggmlc") {
t.Errorf("a request would not be told why: %q %v", why, no)
}
// Retried by the operator once fixed.
if !h.r.Retry("llama") {
t.Error("Retry did not report the model as parked")
}
h.r.Reconcile(t.Context())
if len(h.disp.startedShards()) == 0 {
t.Error("a retried model was not placed")
}
}
func TestChangingAParkedModelsDefinitionTriesItAgain(t *testing.T) {
h := newHarness(t, testModel(1), gpuNode("1", 8))
failRepeatedly(t, h, DefaultConfig().ParkAfter, "boom")
fixed := testModel(1)
fixed.Source = "/models/llama-fixed"
cat, err := catalog.New([]catalog.Model{fixed})
if err != nil {
t.Fatal(err)
}
h.r.cat.Store(cat)
h.disp.reset()
h.r.Reconcile(t.Context())
if len(h.disp.startedShards()) == 0 {
t.Error("a model whose definition changed stayed parked")
}
}
func TestAPlacementFailureIsLoggedWhenItChangesNotEveryPass(t *testing.T) {
var buf bytes.Buffer
h := newHarness(t, testModel(1), gpuNode("1", 1)) // tp4 needs 4 cards
h.r.log = slog.New(slog.NewTextHandler(&buf, nil))
for i := 0; i < 20; i++ {
h.r.Reconcile(t.Context())
h.advance(5 * time.Second)
}
if n := strings.Count(buf.String(), "no placement for model"); n != 1 {
t.Errorf("logged %d times over 20 passes", n)
}
h.advance(placementLogEvery)
h.r.Reconcile(t.Context())
if n := strings.Count(buf.String(), "no placement for model"); n != 2 {
t.Errorf("not repeated after %s: %d", placementLogEvery, n)
}
var mh ModelHealth
for _, x := range h.r.ModelHealth() {
mh = x
}
if !mh.Unplaceable || mh.Desired != 1 {
t.Errorf("health %+v", mh)
}
}
+4
View File
@@ -156,6 +156,10 @@ type Replica struct {
// DrainingAt is when the replica stopped taking new work, so the wait for a
// replacement can be bounded.
DrainingAt time.Time
// RetiringAt is set while the replica is being restarted: it keeps serving
// but no longer counts toward the model's replicas, so a replacement is
// placed beside it, and it is drained only once that replacement is ready.
RetiringAt time.Time
// Attempt counts how many times this model has failed to come up in a row,
// carried across rebuilds so the backoff does not reset on every attempt.
Attempt int
+4 -1
View File
@@ -332,7 +332,10 @@ func certNames(path string) ([]string, error) {
// --- binary ---
var releaseTag = regexp.MustCompile(`^v\d+\.\d+\.\d+$`)
// releaseTag is a version CI published: v0.10.4, or a release candidate of
// it, v0.10.4-rc.2. Anything else — a build with commits past its tag — has no
// release to download.
var releaseTag = regexp.MustCompile(`^v\d+\.\d+\.\d+(-rc\.\d+)?$`)
func (r *Runner) binary(ctx context.Context, res *Result) error {
v := r.Opts.Version
+14
View File
@@ -0,0 +1,14 @@
package onboard
import "testing"
func TestAReleaseCandidateIsARelease(t *testing.T) {
for v, want := range map[string]bool{
"v0.10.4": true, "v0.10.4-rc.1": true, "v0.10.4-rc.12": true,
"v0.10.3-5-gad6d2d3": false, "v0.10.4-rc": false, "v0.10.3-a": false, "dev": false,
} {
if got := releaseTag.MatchString(v); got != want {
t.Errorf("%s: %v", v, got)
}
}
}
+23
View File
@@ -17,6 +17,22 @@ type Activator interface {
Want(model string) bool
}
// unservable is an activator that can also say a model will not come up at
// all — it failed every time it started, or fits on no node — so a request is
// told so at once rather than asked to retry something that never arrives.
type unservable interface {
Unservable(model string) (string, bool)
}
// whyUnservable is the activator's reason a model cannot be served, if any.
func (r *Router) whyUnservable(model string) (string, bool) {
u, ok := r.activator.(unservable)
if !ok {
return "", false
}
return u.Unservable(model)
}
// SetActivator supplies the control plane that can load models on demand.
func (r *Router) SetActivator(a Activator) { r.activator = a }
@@ -32,6 +48,8 @@ func (r *Router) SetActivator(a Activator) { r.activator = a }
// a slow request. A fifteen-minute wait is a job pretending to be one, and the
// client should be told to come back rather than left holding a socket.
func (r *Router) activate(ctx context.Context, model string) bool {
// Asked first, always: asking is what makes the control plane try again,
// and a model that fitted nowhere an hour ago may fit now.
if r.activator == nil || !r.activator.Want(model) {
return false
}
@@ -56,6 +74,11 @@ func (r *Router) activate(ctx context.Context, model string) bool {
if len(r.reg.Snapshots(model)) > 0 {
return true
}
// Asking set a placement going; if it found nowhere, there is
// nothing to wait for.
if _, no := r.whyUnservable(model); no {
return false
}
}
}
}
+5
View File
@@ -629,6 +629,11 @@ func (r *Router) dispatchOne(ctx context.Context, w http.ResponseWriter, req *pa
}
}
if len(candidates) == 0 {
if why, no := r.whyUnservable(model); no && r.activator != nil {
// Not "loading; retry": it will not load, and a client told to
// retry keeps retrying something that never arrives.
return dispatchResult{err: fmt.Errorf("%w: %s", errNoCapacity, why)}
}
if r.activator != nil && r.activator.Want(model) {
// Asked for, still loading. Saying "no endpoint serves this" would
// be a lie, and a client told to retry will succeed where one told
+36
View File
@@ -619,3 +619,39 @@ func TestTheServedModelHeaderArrivesBeforeTheBody(t *testing.T) {
t.Errorf("served-model header = %q, want backup", got)
}
}
type unservableActivator struct {
fakeActivator
why map[string]string
}
func (u *unservableActivator) Unservable(model string) (string, bool) {
w, ok := u.why[model]
return w, ok
}
func TestAModelThatCannotComeUpIsSaidSoAtOnce(t *testing.T) {
// A model that fits on no node, or failed every time it started, was
// answered "loading; retry shortly" — and a client that retried did so
// forever.
e := newFakeEngine(t, "m")
cfg := baseConfig(config.Endpoint{ID: "a", URL: e.srv.URL, Models: []string{"m"}})
cfg.Router.ActivationWait = 10 * time.Second
rt := newTestRouter(t, cfg)
rt.SetActivator(&unservableActivator{
fakeActivator: fakeActivator{onDemand: map[string]bool{"phi-4": true}},
why: map[string]string{"phi-4": "phi-4 cannot be placed on any node: needs 15 GB"},
})
start := time.Now()
rec := post(t, rt.Handler(), map[string]any{"model": "phi-4", "messages": []any{}}, nil)
if took := time.Since(start); took > 2*time.Second {
t.Errorf("the request waited %s for a model that will not load", took)
}
if rec.Code != http.StatusServiceUnavailable || rec.Header().Get("Retry-After") != "" {
t.Errorf("status %d, Retry-After %q", rec.Code, rec.Header().Get("Retry-After"))
}
if !strings.Contains(rec.Body.String(), "needs 15 GB") {
t.Errorf("the reason was not given: %s", rec.Body.String())
}
}
+58
View File
@@ -0,0 +1,58 @@
package tlsconf
import (
"crypto/x509"
"encoding/pem"
"os"
"github.com/prometheus/client_golang/prometheus"
)
// ExpiryCollector publishes when each certificate in a set of files expires.
//
// Certificates here last a year and nothing renews them; the first sign of one
// running out was going to be every agent refusing to connect on the same day.
// Read at every scrape rather than once, so a certificate replaced on disk is
// seen without a restart. A file that cannot be read publishes nothing, which
// is its own thing to notice.
func ExpiryCollector(files map[string]string) prometheus.Collector {
return &expiry{files: files, desc: prometheus.NewDesc("llmgw_certificate_expiry_timestamp_seconds",
"When a certificate this control plane uses expires, as a Unix time.",
[]string{"cert", "subject"}, nil)}
}
type expiry struct {
files map[string]string
desc *prometheus.Desc
}
func (e *expiry) Describe(ch chan<- *prometheus.Desc) { ch <- e.desc }
func (e *expiry) Collect(ch chan<- prometheus.Metric) {
for name, path := range e.files {
if path == "" {
continue
}
b, err := os.ReadFile(path) //nolint:gosec // configured certificate paths
if err != nil {
continue
}
// Every certificate in the file: a CA bundle may hold several.
for {
var blk *pem.Block
blk, b = pem.Decode(b)
if blk == nil {
break
}
if blk.Type != "CERTIFICATE" {
continue
}
c, err := x509.ParseCertificate(blk.Bytes)
if err != nil {
continue
}
ch <- prometheus.MustNewConstMetric(e.desc, prometheus.GaugeValue,
float64(c.NotAfter.Unix()), name, c.Subject.CommonName)
}
}
}
+19 -7
View File
@@ -10,6 +10,11 @@ it, it does not replace it.
Idempotent: a release whose body already matches is left alone, a version with
no release is skipped, and nothing is created or deleted.
A release candidate — v0.10.4-rc.2 — takes the "## Unreleased" section, which
is where changes gather until the version they belong to is released, and is
marked as a pre-release. The final v0.10.4 takes its own section once
"Unreleased" has been renamed to it.
Usage:
GITEA_TOKEN=<token> scripts/gitea/release_notes.py [--tag v0.8.1] [--dry-run]
@@ -32,7 +37,8 @@ DEFAULT_REPO = "ilker/llmgw"
# "## v0.8.1 — 2026-09-28". The dash is whatever was typed; the version is what
# matters.
HEADING = re.compile(r"^## (v\d+\.\d+\.\d+)\b.*$", re.M)
HEADING = re.compile(r"^## (v\d+\.\d+\.\d+|Unreleased)\b.*$", re.M)
CANDIDATE = re.compile(r"^v\d+\.\d+\.\d+-rc\.\d+$")
def sections(changelog):
@@ -77,11 +83,16 @@ def main():
with open(args.changelog, encoding="utf-8") as f:
notes = sections(f.read())
if args.tag:
if args.tag not in notes:
section = "Unreleased" if CANDIDATE.match(args.tag) else args.tag
if section not in notes:
# A tag with no section is a release somebody forgot to describe.
# Failing says so where it is seen; an empty page says nothing.
sys.exit(f"{args.changelog} has no section for {args.tag}")
notes = {args.tag: notes[args.tag]}
sys.exit(f"{args.changelog} has no section for {args.tag} (looked for {section!r})")
notes = {args.tag: notes[section]}
else:
# Syncing everything: "Unreleased" belongs to whichever candidate is
# newest, which only a tag says.
notes.pop("Unreleased", None)
base = f"{args.instance.rstrip('/')}/api/v1/repos/{args.repo}/releases"
for tag, body in notes.items():
@@ -92,11 +103,12 @@ def main():
if rel is None:
print(f"skip {tag} (no release)")
continue
if (rel.get("body") or "").strip() == body.strip():
pre = bool(CANDIDATE.match(tag))
if (rel.get("body") or "").strip() == body.strip() and bool(rel.get("prerelease")) == pre:
print(f"same {tag}")
continue
request("PATCH", f"{base}/{rel['id']}", token, {"body": body})
print(f"set {tag}")
request("PATCH", f"{base}/{rel['id']}", token, {"body": body, "prerelease": pre})
print(f"set {tag}" + (" (pre-release)" if pre else ""))
if __name__ == "__main__":