A second mirror: llmgw mirror add-peer, and on a store machine at setup
CI / test (push) Successful in 9m29s

mirror add-peer, run on the mirror's machine, sets up a mirror on
another over ssh, kept in step with this one: llmgw there if it has
none, mirror init --peer, this mirror told of it and allowed to be read
by it, and mirror.peers in the cluster's settings. The setup screens
offer one on either database machine, as server init --second-mirror.

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-29 13:43:27 +01:00
1 parent 8a76a53afb
commit 1c16f2337d
11 files changed
+367 -11

No files matched your search

+4 -1
View File
@@ -30,7 +30,10 @@ What is left, most needed first. The detail is in the item each names.
**Soon: the customer's experience, and HA.**
4. ~~**C24**~~: done 2026-10-21. The store's, catalog's and mirror's errors, percent signs, footer times and stale
forms are in the page's language. Left: the /docs documents, and a language kept per person (cookie only today).
5. **J4**: the model mirror on more than one machine, the last single point of an HA install. (M)
5. **J4**: the model mirror on more than one machine. Code done 2026-10-29: mirrors kept in step (`mirror serve
--peer`, removals carried), `mirror.peers` and failover in the control plane, nodes given the file on a mirror that
answers, `mirror add-peer` and the setup's second mirror. Left: the "done when" on machines, the first mirror off
while a model is added and served.
6. **F5**: large imports limited by the embedding engine's slots. (M)
**From Kong** (section L): L1–L5 and L9 done (v0.15.0); left of L4 is routing by cost between outside
+12
View File
@@ -27,6 +27,18 @@ APIs change between minors without a deprecation period.
## Unreleased
### Added
- **A second model mirror, kept in step.** In a three-machine install the setup screens put one on a database machine
(`--second-mirror 1|2`). An install with one mirror gets one with `sudo llmgw mirror add-peer user@host`.
- **In step:** each mirror reads the other every thirty seconds. It copies what it lacks, checksum checked, and
removes what the other removed, even if it was down at the time (`llmgw mirror serve --peer`).
- **Control planes:** they name both (`mirror.peers`) and use whichever answers.
- **Nodes:** while one mirror is down, a node is given a model's file on the other.
This takes away the last single point of an HA install: with the first machine off, models are still added and
fetched.
**Upgrading.**
- A model whose weights are in a pickle format (`.bin`, `.pt`, `.pth`,
`.ckpt`, `.pkl`) stops being placed: its node refuses the source. Give it
+25 -2
View File
@@ -2522,8 +2522,31 @@ lost: a standby leads within seconds, Patroni promotes the standby database
if it was the primary, and the floating address follows whichever leads. The
mirror's management API answers the three control planes' addresses, not
only its own machine's, so a standby that comes to lead can still fetch and
add models. The mirror itself stays on the first machine: while it is down,
models the nodes already have keep serving, and new ones wait for it.
add models.
**A second mirror.** The mirror is on the first machine, and without a
second one, models the nodes already have keep serving while that machine
is down, and new ones wait for it. With a second, the cluster carries on.
- **Setting it up:** the screens offer one on either database machine
("A second mirror"); by command, `--second-mirror 1` (or `2`). For an
install that has one mirror, run this on the mirror's machine:
```
sudo llmgw mirror add-peer ubuntu@10.0.0.7 --control-plane 10.0.0.5 --control-plane 10.0.0.6 --control-plane 10.0.0.7
```
- **Kept in step:** each mirror reads the other every thirty seconds and
copies what it lacks, keeping a copy only if its checksum matches.
- A model removed from one is removed from the other, even if that one
was down at the time and comes back later.
- A model added again after it was removed stands on both.
- Neither mirror leads.
- **Control planes:** they name both (`mirror.peers` in the cluster's
settings) and send the mirror's management calls to the first that
answers.
- **Nodes:** while the mirror a model's address names is down, a node is
given the same file on the other.
- **Disk:** the second machine needs room for every model, as the first
does.
## Two control planes
+1
View File
@@ -51,6 +51,7 @@ func newMirrorCmd() *cobra.Command {
newMirrorRemoveCmd(&dir),
newMirrorServeCmd(&dir),
newMirrorInitCmd(&dir),
newMirrorAddPeerCmd(&dir),
)
return c
}
+209
View File
@@ -0,0 +1,209 @@
package cli
import (
"context"
"errors"
"fmt"
"io"
"net"
"os"
"regexp"
"slices"
"strings"
"github.com/spf13/cobra"
"gitea.mynodes.xyz/ilker/llmgw/internal/onboard"
"gitea.mynodes.xyz/ilker/llmgw/internal/settings"
)
// A second mirror, so models can still be added and fetched while the first
// machine is down. Set up from the machine of the first, which it is kept in
// step with: the first is told of it, it of the first, and the control
// planes of both.
func newMirrorAddPeerCmd(dir *string) *cobra.Command {
var controlPlanes []string
var base string
cmd := &cobra.Command{
Use: "add-peer [user@]host",
Short: "Set up a second mirror on another machine, kept in step with this one",
Long: `Run on the machine of the cluster's mirror, as root. It logs in to HOST with
ssh, installs llmgw there if it is not, and sets up a mirror that keeps in
step with this one: each copies what the other holds, checksum checked, and
a model removed from one is removed from the other. This mirror is told of
it, and the cluster's settings name it (mirror.peers), so while either
machine is down the control planes use the other and nodes fetch from it.
HOST needs room for every model this mirror holds: it copies them all.`,
Example: " sudo llmgw mirror add-peer ubuntu@10.0.0.12 --control-plane 10.0.0.10 --control-plane 10.0.0.12",
Args: cobra.ExactArgs(1),
RunE: func(cmd *cobra.Command, args []string) error {
if os.Geteuid() != 0 {
return errors.New("add-peer changes this mirror's service and the cluster's settings: run it as root")
}
user, host, ok := strings.Cut(args[0], "@")
if !ok {
user, host = os.Getenv("SUDO_USER"), args[0]
}
return addMirrorPeer(cmd.Context(), cmd.OutOrStdout(), &onboard.SSH{Addr: host, User: user}, host,
*dir, base, controlPlanes, "/etc/llmgw/server.yaml")
},
}
cmd.Flags().StringArrayVar(&controlPlanes, "control-plane", nil,
"address of a control plane that manages the mirrors; repeatable (default: this machine)")
cmd.Flags().StringVar(&base, "base", "", "the address nodes fetch from on HOST (default: http://HOST:8081)")
return cmd
}
// addMirrorPeer sets up a mirror on a machine reached by remote, kept in
// step with the one on this machine.
func addMirrorPeer(ctx context.Context, out io.Writer, remote onboard.Remote, host, dir, base string,
controlPlanes []string, configPath string,
) error {
ip, err := ipv4Of(host)
if err != nil {
return err
}
here := primaryAddr()
if len(controlPlanes) == 0 {
controlPlanes = []string{here}
}
if base == "" {
base = "http://" + net.JoinHostPort(ip, "8081")
}
thisAPI := "http://" + net.JoinHostPort(here, "8082")
peerAPI := "http://" + net.JoinHostPort(ip, "8082")
// llmgw there, the same as here, when it has none.
if _, err := remote.Run(ctx, "command -v llmgw >/dev/null", nil); err != nil {
exe, err := os.Executable()
if err != nil {
return err
}
bin, err := os.ReadFile(exe) //nolint:gosec // this program
if err != nil {
return err
}
fmt.Fprintf(out, " .. llmgw to %s\n", host)
if _, err := remote.Run(ctx, "sudo install -m 755 /dev/stdin /usr/local/bin/llmgw", bin); err != nil {
return fmt.Errorf("installing llmgw on %s: %w", host, err)
}
}
// The mirror there: its management API for the control planes and this
// mirror, which reads it; kept in step with this one.
allow := append([]string{}, controlPlanes...)
if !slices.Contains(allow, here) {
allow = append(allow, here)
}
script := "sudo llmgw mirror init --dir " + shellQuote(dir) + " --base " + shellQuote(base) + " --peer " + shellQuote(thisAPI)
for _, a := range allow {
script += " --control-plane " + shellQuote(a)
}
fmt.Fprintf(out, " .. the mirror on %s\n", host)
if b, err := remote.Run(ctx, script, nil); err != nil {
return fmt.Errorf("the mirror on %s: %w\n%s", host, err, b)
}
// This mirror told of it, and to let it read this one's API.
changed, err := addToMirrorUnit(mirrorUnitPath, peerAPI, ip)
if err != nil {
return err
}
if changed {
in := &serverInit{out: out}
if err := in.systemctl(ctx, "daemon-reload"); err != nil {
return err
}
if err := in.systemctl(ctx, "restart", "llmgw-mirror"); err != nil {
return err
}
}
fmt.Fprintf(out, " ok this mirror keeps in step with %s\n", peerAPI)
// The control planes told of both.
if err := addMirrorPeerSetting(configPath, peerAPI); err != nil {
return err
}
fmt.Fprintf(out, " ok mirror.peers names %s; the control planes use it while this one is down\n", peerAPI)
return nil
}
// addToMirrorUnit adds a peer and the address allowed to read this mirror's
// API to its systemd unit, and says whether it changed.
func addToMirrorUnit(path, peer, allow string) (bool, error) {
b, err := os.ReadFile(path) //nolint:gosec // the mirror's unit
if err != nil {
return false, fmt.Errorf("this machine has no mirror to keep in step (%s): run add-peer on the mirror's machine", path)
}
unit := string(b)
re := regexp.MustCompile(`(?m)^ExecStart=.*$`)
line := re.FindString(unit)
if line == "" {
return false, fmt.Errorf("%s has no ExecStart", path)
}
next := line
if !strings.Contains(next, "--peer "+peer) {
next += " --peer " + peer
}
if !strings.Contains(next, "--api-allow "+allow+" ") && !strings.HasSuffix(next, "--api-allow "+allow) &&
strings.Contains(next, "--api-listen :") {
next += " --api-allow " + allow
}
if next == line {
return false, nil
}
return true, os.WriteFile(path, []byte(strings.Replace(unit, line, next, 1)), 0o644)
}
// addMirrorPeerSetting names a peer in mirror.peers: in the cluster's
// settings when the store holds them, else in server.yaml.
func addMirrorPeerSetting(path, peer string) error {
add := func(doc *settings.Doc) error {
var peers []string
if n := doc.Node([]string{"mirror", "peers"}); n != nil {
_ = n.Decode(&peers)
}
if slices.Contains(peers, peer) {
return nil
}
list := "[" + strings.Join(append(peers, peer), ", ") + "]"
return doc.SetYAML([]string{"mirror", "peers"}, list)
}
changed, err := changeStoredSettings(path, "a second mirror (llmgw mirror add-peer)", add)
if err != nil || changed {
return err
}
doc, err := settings.Open(path)
if err != nil {
return err
}
if err := add(doc); err != nil {
return err
}
_, err = doc.Save(nil)
return err
}
// ensureSecondMirror sets up the mirror kept in step with this machine's on
// the store machine chosen, reached as the store's setup reached it.
func (in *serverInit) ensureSecondMirror(ctx context.Context) error {
if !in.standbys {
return errors.New("it needs --standbys: without them this mirror's API answers this machine alone, which the second cannot read")
}
i := in.secondMirror - 1
if i < 0 || i >= len(in.patroniNodes) {
return fmt.Errorf("there is no database machine %d to put it on", in.secondMirror)
}
if in.noMirror || in.mirrorURL != "" {
return errors.New("it keeps in step with a mirror on this machine, and there is none")
}
n := in.patroniNodes[i]
fmt.Fprintf(in.out, " .. a second mirror on %s\n", n.Name)
if err := addMirrorPeer(ctx, in.out, n.Remote, n.IP, in.mirrorRoot(), "", in.mirrorAllow(), in.configPath()); err != nil {
return err
}
in.done("a second mirror on %s (%s), kept in step with this one", n.Name, n.IP)
return nil
}
+74
View File
@@ -0,0 +1,74 @@
package cli
import (
"os"
"path/filepath"
"strings"
"testing"
)
func TestAPeerIsAddedToTheMirrorsUnitOnce(t *testing.T) {
path := filepath.Join(t.TempDir(), "llmgw-mirror.service")
unit := mirrorUnitFor("/srv/models", ":8082", "http://10.0.0.10:8081", []string{"10.0.0.10"}, nil, "test")
if err := os.WriteFile(path, []byte(unit), 0o644); err != nil {
t.Fatal(err)
}
changed, err := addToMirrorUnit(path, "http://10.0.0.12:8082", "10.0.0.12")
if err != nil || !changed {
t.Fatalf("changed %v %v", changed, err)
}
b, _ := os.ReadFile(path)
if !strings.Contains(string(b), "--peer http://10.0.0.12:8082") || !strings.Contains(string(b), "--api-allow 10.0.0.12") {
t.Errorf("unit:\n%s", b)
}
if changed, _ := addToMirrorUnit(path, "http://10.0.0.12:8082", "10.0.0.12"); changed {
t.Error("added twice")
}
// A mirror whose API is on loopback alone is not opened by the back door.
lo := mirrorUnitFor("/srv/models", "127.0.0.1:8082", "http://10.0.0.10:8081", nil, nil, "test")
_ = os.WriteFile(path, []byte(lo), 0o644)
_, _ = addToMirrorUnit(path, "http://10.0.0.12:8082", "10.0.0.12")
if b, _ := os.ReadFile(path); strings.Contains(string(b), "--api-allow") {
t.Errorf("a loopback mirror was opened:\n%s", b)
}
}
func TestASecondMirrorIsNamedInTheSettings(t *testing.T) {
path := filepath.Join(t.TempDir(), "server.yaml")
if err := os.WriteFile(path, []byte("listen: \":8080\"\nmirror:\n api: http://10.0.0.10:8082\n"), 0o600); err != nil {
t.Fatal(err)
}
for range 2 {
if err := addMirrorPeerSetting(path, "http://10.0.0.12:8082"); err != nil {
t.Fatal(err)
}
}
b, _ := os.ReadFile(path)
if strings.Count(string(b), "10.0.0.12:8082") != 1 || !strings.Contains(string(b), "peers") {
t.Errorf("server.yaml:\n%s", b)
}
}
func TestTheSetupOffersASecondMirrorOnAStoreMachine(t *testing.T) {
in := newDefaultServerInit()
f := newInstallForm(&in, "ctl", []string{"10.0.0.5"}, true)
f.secondMirror.Choice = 1
if !f.secondMirror.Show() {
t.Fatal("not offered with the mirror on this machine")
}
f.mirrorWhere.Choice = 1
if f.secondMirror.Show() {
t.Error("offered with the mirror elsewhere")
}
f.mirrorWhere.Choice = 0
in.secondMirror = 1
in.patroni = []string{"ubuntu@10.0.0.11", "ubuntu@10.0.0.12"}
if got := strings.Join(in.initFlags(), " "); !strings.Contains(got, "--second-mirror 1") {
t.Errorf("flags %s", got)
}
// Without standbys this mirror answers itself alone: the second could not read it.
in.standbys = false
if err := in.ensureSecondMirror(t.Context()); err == nil || !strings.Contains(err.Error(), "standbys") {
t.Errorf("without standbys: %v", err)
}
}
+7 -3
View File
@@ -29,6 +29,7 @@ import (
func newMirrorInitCmd(dir *string) *cobra.Command {
var (
controlPlanes []string
peers []string
base string
noFirewall bool
)
@@ -61,7 +62,7 @@ Safe to run again: it changes only what differs.`,
if _, err := parseAllow(controlPlanes); err != nil {
return err
}
if err := mirrorInit(cmd.Context(), cmd.OutOrStdout(), *dir, base, controlPlanes); err != nil {
if err := mirrorInit(cmd.Context(), cmd.OutOrStdout(), *dir, base, controlPlanes, peers); err != nil {
return err
}
if !noFirewall {
@@ -80,6 +81,9 @@ Safe to run again: it changes only what differs.`,
}
cmd.Flags().StringArrayVar(&controlPlanes, "control-plane", nil,
"address of the control plane that manages this mirror; repeatable")
cmd.Flags().StringArrayVar(&peers, "peer", nil,
"another mirror's management API to keep in step with, http://host:8082; repeatable. "+
"Its machine must be among --control-plane, to read this one")
cmd.Flags().BoolVar(&noFirewall, "no-firewall", false,
"leave the firewall alone, instead of keeping the mirror's ports to the cluster's networks")
cmd.Flags().StringVar(&base, "base", "",
@@ -87,7 +91,7 @@ Safe to run again: it changes only what differs.`,
return cmd
}
func mirrorInit(ctx context.Context, out io.Writer, dir, base string, controlPlanes []string) error {
func mirrorInit(ctx context.Context, out io.Writer, dir, base string, controlPlanes, peers []string) error {
in := &serverInit{user: "llmgw", stateDir: "/var/lib/llmgw", out: out}
if err := in.ensureUser(ctx); err != nil {
return err
@@ -105,7 +109,7 @@ func mirrorInit(ctx context.Context, out io.Writer, dir, base string, controlPla
}
base = "http://" + net.JoinHostPort(addr, "8081")
}
unit := mirrorUnitFor(dir, ":8082", base, controlPlanes,
unit := mirrorUnitFor(dir, ":8082", base, controlPlanes, peers,
"'llmgw mirror init'. Bytes on :8081 for every node; the\n# management API on :8082, for the control plane only.")
changed, err := in.writeUnit(mirrorUnitPath, unit)
if err != nil {
+11
View File
@@ -86,6 +86,9 @@ type serverInit struct {
// nodes labelled for them; judgeCPUs their cores, once added.
judgeMachines []judgeMachine
judgeCPUs []int
// secondMirror is which machine of the store, 1 or 2, keeps a second
// mirror in step with this machine's; 0 is none.
secondMirror int
// judgeNote is what timing the judge found, said before the next
// steps when it is not fine.
judgeNote string
@@ -243,6 +246,8 @@ func newServerInitCmd() *cobra.Command {
f.StringVar(&judgeCats, "judge-categories", strings.Join(defaultJudgeCategories, ","),
"what the judge refuses, comma-separated, of: "+strings.Join(judgeCategories, ", "))
f.IntVar(&in.judge.Replicas, "judge-replicas", 2, "how many copies of the judge run, each on CPUs")
f.IntVar(&in.secondMirror, "second-mirror", 0,
"with --patroni, a second mirror on the store's machine 1 or 2, kept in step with this one's (0: none)")
f.Var(judgeMachineFlag{&in.judgeMachines}, "judge-node",
"user@host of a machine for the guardrails' judges alone: added as a node, and the judges run only there; repeatable")
f.DurationVar(&in.judge.Budget, "judge-budget", 5*time.Second, "how long a request waits for the judge")
@@ -357,6 +362,12 @@ func (in *serverInit) run(ctx context.Context) error {
return fmt.Errorf("standby control planes: %w", err)
}
}
if hasSystemd && in.secondMirror > 0 {
if err := in.ensureSecondMirror(ctx); err != nil {
// Not fatal: the first mirror serves; said, and how to try again.
fmt.Fprintf(in.out, " !! second mirror: %v; later: sudo llmgw mirror add-peer <user>@<machine>\n", err)
}
}
if hasSystemd && in.judgeWanted() && in.configJudgesWith(guardModelName) {
in.checkJudge(ctx)
}
+3
View File
@@ -396,6 +396,9 @@ func (in *serverInit) initFlags() []string {
for _, m := range in.judgeMachines {
add("--judge-node", m.target)
}
if in.secondMirror > 0 {
add("--second-mirror", fmt.Sprint(in.secondMirror))
}
switch {
case in.mirrorURL != "":
add("--mirror", in.mirrorURL)
+6 -3
View File
@@ -123,10 +123,10 @@ func (in *serverInit) writeUnit(path, body string) (bool, error) {
func (in *serverInit) mirrorUnit() string {
base := "http://" + net.JoinHostPort(in.reachableAddr(), "8081")
if in.standbys {
return mirrorUnitFor(in.mirrorRoot(), ":8082", base, in.mirrorAllow(),
return mirrorUnitFor(in.mirrorRoot(), ":8082", base, in.mirrorAllow(), nil,
"'llmgw server init'. Bytes on :8081 for every node; the\n# management API for the control planes only, whichever leads.")
}
return mirrorUnitFor(in.mirrorRoot(), mirrorAPI, base, nil,
return mirrorUnitFor(in.mirrorRoot(), mirrorAPI, base, nil, nil,
"'llmgw server init'. Bytes on :8081 for every node; the\n# management API on loopback only, for the control plane beside it.")
}
@@ -154,11 +154,14 @@ func (in *serverInit) mirrorAllow() []string {
// mirrorUnitFor is a mirror's systemd unit. allow is who, besides the
// machine itself, may use the management API.
func mirrorUnitFor(dir, apiListen, base string, allow []string, about string) string {
func mirrorUnitFor(dir, apiListen, base string, allow, peers []string, about string) string {
exec := "/usr/local/bin/llmgw mirror serve --dir " + dir + " --listen :8081 --api-listen " + apiListen + " --base " + base
for _, a := range allow {
exec += " --api-allow " + a
}
for _, p := range peers {
exec += " --peer " + p
}
return `[Unit]
Description=llmgw model mirror
After=network-online.target
+15 -2
View File
@@ -62,6 +62,7 @@ type installForm struct {
admin, hosts, language, signup *setupField
storeKind *setupField
mirrorWhere, mirrorDirF, mirrorURL *setupField
secondMirror *setupField
starterPick, starterSource, starterCtx *setupField
judgeFail, judgeCopies, judgeBudget *setupField
judgeCats []*setupField
@@ -181,18 +182,26 @@ func newInstallForm(in *serverInit, host string, found []string, ha bool) *insta
f.mirrorDirF.Show = func() bool { return f.mirrorWhere.Choice == 0 }
f.mirrorURL = textSetupField("Its management API", in.mirrorURL, "http://10.0.0.20:8082", "")
f.mirrorURL.Show = func() bool { return f.mirrorWhere.Choice == 1 }
f.secondMirror = choiceField("A second mirror", []string{"none", "on database machine 1", "on database machine 2"},
in.secondMirror, "Kept in step with this one, each copying what the other has: while either machine is down, "+
"models are still added and fetched. It needs room for every model, as this one does.")
f.secondMirror.Show = func() bool { return ha && f.mirrorWhere.Choice == 0 }
f.mirror = &installPart{
title: "Model mirror", on: !in.noMirror || in.mirrorURL != "",
about: "Downloads each model once, from Hugging Face or wherever it comes from, and serves the files to " +
"every node on the network, so ten nodes do not download the same 40 GB ten times. " +
"The starter model, the judge and an offline bundle go into it.",
fields: []*setupField{f.mirrorWhere, f.mirrorDirF, f.mirrorURL},
fields: []*setupField{f.mirrorWhere, f.mirrorDirF, f.mirrorURL, f.secondMirror},
off: "each model's files must already be on a mirror somewhere; no starter model, no judge",
summary: func() string {
if f.mirrorWhere.Choice == 1 {
return "at " + orText(f.mirrorURL.Value(), "(an address)")
}
return "on this machine, in " + orText(f.mirrorDirF.Value(), mirrorDir)
s := "on this machine, in " + orText(f.mirrorDirF.Value(), mirrorDir)
if ha && f.secondMirror.Choice > 0 {
s += "; a second " + f.secondMirror.Options[f.secondMirror.Choice]
}
return s
},
}
@@ -636,6 +645,10 @@ func (f *installForm) apply(in *serverInit) error {
in.postgres = !f.ha && f.storeKind.Choice == 1
in.noMirror = !f.mirror.on || mirrorURL != ""
in.mirrorURL, in.mirrorPath = mirrorURL, mirrorPath
in.secondMirror = 0
if f.ha && f.mirror.on && f.mirrorWhere.Choice == 0 {
in.secondMirror = f.secondMirror.Choice
}
in.noNode = !f.node.on
in.noModel = !f.starter.on
in.starter, in.starterCtx = starter, ctx