diff --git a/CHANGELOG.md b/CHANGELOG.md index 56fc636f..2201aa28 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -84,6 +84,26 @@ planned switchover cost one failed lease renewal; switching off the primary's machine promoted the standby in 24 seconds with the gateway serving throughout. Example configurations are beside it. +**`llmgw server init --patroni user@host --patroni user@host`** sets it up: +PostgreSQL, Patroni and etcd on those machines over ssh — this one as etcd's +third vote — the role and database, and the store already in use moved in. +Every step looks first; run against a cluster that is up, it changes nothing. +Tried from scratch on three new machines (two minutes, then a failover by +switching the primary off) and against the reference cluster (nothing +touched). + +**Watching it**: for a store of several hosts the control plane publishes +`llmgw_store_member`, `llmgw_store_replica_lag_bytes` and +`llmgw_store_patroni_up` from each member's Patroni, a member that stopped +showing as `missing` rather than vanishing; `llmgw ctl status` has a STORE +line; and four alerts — `StoreHasNoLeader`, `StorePatroniUnreachable`, +`StoreReplicaNotStreaming`, `StoreReplicaBehind`. + +**Making an administrator on PostgreSQL** (`init --admin`, `users promote`) +failed: the flag was sent as a bool to an integer column, which SQLite takes +and PostgreSQL refuses. Every existing PostgreSQL store got its administrator +by being copied from SQLite, which is why nobody had met it. + `llmgw server backup` reads a store address of several hosts. It took the list for one host name and reached the database only by falling through to the second; it now hands libpq the list, and dumps from a standby when there diff --git a/deploy/ha/README.md b/deploy/ha/README.md index c3df86ee..fa679a66 100644 --- a/deploy/ha/README.md +++ b/deploy/ha/README.md @@ -104,13 +104,14 @@ On the reference cluster: | llmgw-pg01 | member | Patroni | | llmgw-ctl02 (second control plane) | member | Patroni | -`etcd.default.example` is one member's `/etc/default/etcd` (Ubuntu's -`etcd-server`), and `patroni.yml.example` one node's `/etc/patroni/config.yml` -(Ubuntu's `patroni`); the other node differs in `name` and its addresses. -Start Patroni on the first node, which initialises the cluster, then on the -second, which clones it. Create the `llmgw` role and database on the leader, -and move an existing store into it with `pg_dump --no-owner` and -`pg_restore --no-owner`. +`llmgw server init --patroni user@host --patroni user@host` sets all of it up +over ssh and moves the store in — see [install.md](../../docs/install.md). +It looks before it acts at every step, so it is also how an existing cluster +is checked: run against one that is up, it changes nothing. What it writes +is in `etcd.default.example` (one member's `/etc/default/etcd`) and +`patroni.yml.example` (one node's `/etc/patroni/config.yml`), for anyone +doing it by hand; passwords are generated into `/root/llmgw-pg` on each +database machine. **Give llmgw every member, and ask for the writable one:** @@ -133,6 +134,14 @@ Measured on the reference cluster, with the gateway serving throughout: out without losing leadership or a replica. Started again, the old primary rejoined as a standby by itself. +**Watching it.** For a store of several hosts the control plane reads the +cluster from each member's Patroni (port 8008) and publishes +`llmgw_store_member{member,host,role,state}`, `llmgw_store_replica_lag_bytes` +and `llmgw_store_patroni_up`; a host of the address that the cluster no longer +lists is a member in state `missing`. `llmgw ctl status` prints it as one +line, and `deploy/monitoring/llmgw-rules.yml` alerts on no leader, a member +not streaming or missing, and a standby too far behind to be promoted. + Not done here, and worth doing in production: a watchdog (`softdog`) so a primary that has lost the lock but not died cannot keep accepting writes, TLS on etcd and between PostgreSQL members, and more than one standby. diff --git a/deploy/monitoring/llmgw-rules.yml b/deploy/monitoring/llmgw-rules.yml index e8770193..c43e0786 100644 --- a/deploy/monitoring/llmgw-rules.yml +++ b/deploy/monitoring/llmgw-rules.yml @@ -350,3 +350,51 @@ groups: Working as designed when the cluster is full: the lowest band is held first. Worth knowing about because the callers being refused do not know why. + + # The store, when it is a replicated cluster under Patroni (llmgw server + # init --patroni). Published only for a store of several hosts. + - name: llmgw-store + rules: + - alert: StoreHasNoLeader + expr: llmgw_store_patroni_up == 1 and on() count(llmgw_store_member{role="leader"}) == 0 + for: 1m + labels: { severity: critical } + annotations: + summary: "The store's cluster has no primary" + description: >- + Every write fails — keys, usage, budgets, leadership. Patroni + promotes a standby within its 30-second TTL; a minute without one + means it cannot: etcd has lost its majority, or no standby is fit. + `patronictl -c /etc/patroni/config.yml list` on a member says which. + + - alert: StorePatroniUnreachable + expr: llmgw_store_patroni_up == 0 + for: 2m + labels: { severity: warning } + annotations: + summary: "No member of the store's cluster answers on its Patroni port" + description: >- + The control plane reads the cluster's state from port 8008 on each + member. The database may still be serving; nothing here can tell. + + - alert: StoreReplicaNotStreaming + expr: llmgw_store_member{role!="leader", state!="streaming"} == 1 + for: 5m + labels: { severity: warning } + annotations: + summary: "{{ $labels.member }} is {{ $labels.state }}, not streaming from the primary" + description: >- + A standby that is not streaming is not a standby: if the primary + is lost now, there is nothing to promote. "missing" is a host of + the store address the cluster no longer lists — usually a machine + that is down. + + - alert: StoreReplicaBehind + expr: llmgw_store_replica_lag_bytes > 64 * 1024 * 1024 or llmgw_store_replica_lag_bytes == -1 + for: 10m + labels: { severity: warning } + annotations: + summary: "{{ $labels.member }} is {{ $value | humanize1024 }}B behind the primary" + description: >- + Patroni will not promote a standby more than 1 MB behind, so a + failover now would find none fit. -1 is a lag Patroni cannot read. diff --git a/docs/install.md b/docs/install.md index 97e03754..91216d45 100644 --- a/docs/install.md +++ b/docs/install.md @@ -121,6 +121,20 @@ Add `--postgres` to keep accounts and keys in PostgreSQL rather than SQLite: init installs it and sets it up, and on a machine already running on SQLite it moves everything across. See [PostgreSQL](#postgresql-when-sqlite-is-no-longer-the-right-shape). +Or `--patroni`, given two machines (or more), to keep it in PostgreSQL +replicated across them, so losing one loses nothing: + +``` +sudo --preserve-env=SSH_AUTH_SOCK llmgw server init --patroni ubuntu@10.0.0.149 --patroni ubuntu@10.0.0.150 +``` + +Over ssh with passwordless sudo, as `node add` works, it installs PostgreSQL, +Patroni and etcd there — this machine is etcd's third vote — and moves the +store already in use, SQLite or PostgreSQL, into the cluster. A standby is +promoted within about thirty seconds of the primary being lost, and the +control plane follows it by itself. `llmgw ctl status` says who leads; see +[deploy/ha](../deploy/ha/README.md) for what it sets up. + Every step looks before it acts. Running it again finishes an install that stopped halfway and changes nothing on one that did not; it never overwrites `server.yaml`, the authority, or an existing account's password. diff --git a/go.mod b/go.mod index 5c59dba0..4a3c0161 100644 --- a/go.mod +++ b/go.mod @@ -54,6 +54,7 @@ require ( github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect github.com/jackc/puddle/v2 v2.2.2 // indirect github.com/knadh/koanf/maps v0.1.2 // indirect + github.com/kylelemons/godebug v1.1.0 // indirect github.com/lucasb-eyer/go-colorful v1.3.0 // indirect github.com/mattn/go-localereader v0.0.1 // indirect github.com/mattn/go-runewidth v0.0.19 // indirect diff --git a/internal/cli/backup.go b/internal/cli/backup.go index ff214ab8..d409e5ca 100644 --- a/internal/cli/backup.go +++ b/internal/cli/backup.go @@ -8,7 +8,6 @@ import ( "fmt" "io" "io/fs" - "net/url" "os" "os/exec" "path/filepath" @@ -190,42 +189,14 @@ func pgDump(ctx context.Context, dsn, out string) error { return nil } -// pgEnv turns a store address into libpq's environment. -// -// Several hosts are allowed, as pgx allows them — "h1:5432,h2:5432", the -// members of a replicated cluster — and become libpq's own lists in PGHOST and -// PGPORT. url.Parse does not read such a host: it took everything before the -// last colon as one name, which libpq could not resolve, and a backup worked -// only because it fell through to the second host. +// pgEnv turns a store address into libpq's environment. Several hosts — the +// members of a replicated cluster — become libpq's own lists in PGHOST and +// PGPORT; read as one URL host, the list was one name libpq could not +// resolve, and a backup worked only by falling through to the second. func pgEnv(dsn string) ([]string, error) { - scheme, rest, ok := strings.Cut(dsn, "://") - if !ok { - return nil, fmt.Errorf("the store address is not a URL") - } - authority, tail := rest, "" - if i := strings.IndexAny(rest, "/?"); i >= 0 { - authority, tail = rest[:i], rest[i:] - } - userinfo, hostlist := "", authority - if i := strings.LastIndexByte(authority, '@'); i >= 0 { - userinfo, hostlist = authority[:i+1], authority[i+1:] - } - var hosts, ports []string - for _, hp := range strings.Split(hostlist, ",") { - u, err := url.Parse("x://" + hp) - if err != nil || u.Hostname() == "" { - return nil, fmt.Errorf("the store address has a host that is not one: %q", hp) - } - port := u.Port() - if port == "" { - port = "5432" - } - hosts, ports = append(hosts, u.Hostname()), append(ports, port) - } - // The rest of the address with one placeholder host, which url.Parse reads. - u, err := url.Parse(scheme + "://" + userinfo + "host" + tail) + hosts, ports, u, err := store.SplitPostgres(dsn) if err != nil { - return nil, fmt.Errorf("the store address is not a URL") + return nil, err } var env []string set := func(k, v string) { diff --git a/internal/cli/ctl_status_test.go b/internal/cli/ctl_status_test.go index e579ee47..0c80d1d1 100644 --- a/internal/cli/ctl_status_test.go +++ b/internal/cli/ctl_status_test.go @@ -1,9 +1,14 @@ package cli import ( + "bytes" "net/http" "net/http/httptest" + "strings" "testing" + + "gitea.mynodes.xyz/ilker/llmgw/internal/control/adminapi" + "github.com/spf13/cobra" ) func TestAQueuedChangeIsNotReportedAsAnError(t *testing.T) { @@ -22,3 +27,27 @@ func TestAQueuedChangeIsNotReportedAsAnError(t *testing.T) { t.Errorf("the body of a 202 was not read: %+v", out) } } + +func TestStatusSaysWhoLeadsTheStoreAndWhetherTheRestKeepUp(t *testing.T) { + render := func(s adminapi.Summary) string { + cmd := &cobra.Command{} + var b bytes.Buffer + cmd.SetOut(&b) + printSummary(cmd, s) + return b.String() + } + out := render(adminapi.Summary{Store: []adminapi.StoreMember{ + {Name: "llmgw-ctl02", Role: "replica", State: "streaming", Lag: 2048}, + {Name: "llmgw-pg01", Role: "leader", State: "running"}, + }}) + if !strings.Contains(out, "STORE llmgw-ctl02 streaming (2 KB behind), llmgw-pg01 leads") { + t.Errorf("status:\n%s", out) + } + out = render(adminapi.Summary{Store: []adminapi.StoreMember{{Name: "llmgw-pg01", Role: "replica", State: "stopped"}}}) + if !strings.Contains(out, "NO LEADER") { + t.Errorf("a store with no leader was not called out:\n%s", out) + } + if out := render(adminapi.Summary{}); strings.Contains(out, "STORE") { + t.Error("a single database got a store line") + } +} diff --git a/internal/cli/server.go b/internal/cli/server.go index 70daa17c..e8586fee 100644 --- a/internal/cli/server.go +++ b/internal/cli/server.go @@ -19,6 +19,7 @@ import ( "gitea.mynodes.xyz/ilker/llmgw/internal/control/inventory" "gitea.mynodes.xyz/ilker/llmgw/internal/control/modelserve" "gitea.mynodes.xyz/ilker/llmgw/internal/gguf" + "gitea.mynodes.xyz/ilker/llmgw/internal/pgha" "gitea.mynodes.xyz/ilker/llmgw/internal/sourcecheck" "github.com/prometheus/client_golang/prometheus" @@ -127,6 +128,9 @@ func newServerCmd() *cobra.Command { var agentHandler http.Handler var agentPath string var admin *control.Controller + // The store's cluster, when it is one; read by the metrics and + // by `ctl status`. + var storeHA *pgha.Collector if controlMode { cat, err := cfg.Catalog() if err != nil { @@ -179,6 +183,17 @@ func newServerCmd() *cobra.Command { if err := rt.RegisterMetrics(fleetmetrics.New(ctl)); err != nil { return err } + // A store of several hosts is a replicated cluster: what state + // it is in is published beside everything else, read from its + // Patroni. One host is a single PostgreSQL, which has none. + if addr, err := storepkg.ResolveAddress(cfg.Auth.Store); err == nil && storepkg.IsPostgres(addr) { + if hosts, _, _, err := storepkg.SplitPostgres(strings.TrimSpace(addr)); err == nil && len(hosts) > 1 { + storeHA = pgha.NewCollector(hosts) + if err := rt.RegisterMetrics(storeHA); err != nil { + return err + } + } + } // So a request for a model that is configured but not running // starts it, rather than being told nothing serves it. rt.SetActivator(ctl.Reconciler()) @@ -201,6 +216,16 @@ func newServerCmd() *cobra.Command { ) if admin != nil { api := adminapi.New(admin) + if storeHA != nil { + api.SetStoreStatus(func() ([]adminapi.StoreMember, error) { + c, err := storeHA.Read() + var out []adminapi.StoreMember + for _, m := range c.Members { + out = append(out, adminapi.StoreMember{Name: m.Name, Host: m.Host, Role: m.Role, State: m.State, Lag: m.Lag}) + } + return out, err + }) + } adminAPI = api api.SetReleases(cfg.Releases) // Catalog changes need somewhere to persist, so this is wired diff --git a/internal/cli/serverinit.go b/internal/cli/serverinit.go index 7d3a9109..23d0111d 100644 --- a/internal/cli/serverinit.go +++ b/internal/cli/serverinit.go @@ -69,6 +69,9 @@ type serverInit struct { // postgres keeps the store in a PostgreSQL on this machine, installed if // need be, moving whatever SQLite already holds. postgres bool + // patroni are user@host of the machines to keep the store on under + // Patroni, replicated; this machine is the third etcd vote for two. + patroni []string // noMirror, noNode and noModel turn off the parts of a first install that // make it serve something: a mirror, this machine as a node, and a small // model on it. @@ -111,6 +114,12 @@ func newServerInitCmd() *cobra.Command { "installed if there is none, with its own role and a generated password\n" + "kept in /etc/llmgw/secrets. An existing SQLite store is copied into it,\n" + "so accounts and keys survive, and server.yaml is pointed at it.\n\n" + + "With --patroni, given two or more machines, the store is PostgreSQL\n" + + "replicated across them under Patroni, which promotes a standby when the\n" + + "primary is lost. etcd holds its leader lock on those machines — and on\n" + + "this one, when they are two, as the third vote. Installed over ssh with\n" + + "passwordless sudo, as 'llmgw node add' does; the store already in use,\n" + + "SQLite or PostgreSQL, is moved in. Run again, it changes nothing.\n\n" + "Each step is skipped when its result already exists, so it is safe to run\n" + "again: to finish an install that stopped, or with another --host to\n" + "reissue the certificate. It never overwrites server.yaml or the authority.\n\n" + @@ -144,6 +153,9 @@ func newServerInitCmd() *cobra.Command { f.BoolVar(&in.noModel, "no-model", false, "do not fetch and serve the starter model ("+starterName+")") f.BoolVar(&in.postgres, "postgres", false, "keep the store in PostgreSQL on this machine, installing it and moving an existing SQLite store into it") + f.StringArrayVar(&in.patroni, "patroni", nil, + "user@host of a machine to keep the store on, replicated under Patroni; give two or more. "+ + "Installs PostgreSQL, Patroni and etcd there over ssh, and moves the existing store in") return cmd } diff --git a/internal/cli/serverinit_patroni.go b/internal/cli/serverinit_patroni.go new file mode 100644 index 00000000..3342dea2 --- /dev/null +++ b/internal/cli/serverinit_patroni.go @@ -0,0 +1,165 @@ +package cli + +import ( + "context" + "fmt" + "net" + "os" + "os/exec" + "strings" + + "gitea.mynodes.xyz/ilker/llmgw/internal/config" + "gitea.mynodes.xyz/ilker/llmgw/internal/onboard" + "gitea.mynodes.xyz/ilker/llmgw/internal/pgha" + "gitea.mynodes.xyz/ilker/llmgw/internal/store" +) + +// ensurePatroni puts the store in PostgreSQL under Patroni on the machines +// given with --patroni, and moves whatever store server.yaml names into it: +// SQLite, or a PostgreSQL on one machine. This machine is the third etcd vote +// when they are two. +func (in *serverInit) ensurePatroni(ctx context.Context) (err error) { + setup := pgha.Setup{Emit: func(step, detail string, changed bool) { + if changed { + in.done("%s: %s", step, detail) + } else { + in.kept("%s: %s", step, detail) + } + }} + for _, t := range in.patroni { + user, host, ok := strings.Cut(t, "@") + if !ok { + user, host = "", t + } + remote := &onboard.SSH{Addr: host, User: user} + out, err := remote.Run(ctx, "hostname && sudo -n true", nil) + if err != nil { + return fmt.Errorf("%s: cannot log in and use sudo without a password: %v: %s", + t, err, strings.TrimSpace(string(out))) + } + ip, err := ipv4Of(host) + if err != nil { + return err + } + setup.DB = append(setup.DB, pgha.Node{ + Member: pgha.Member{Name: strings.TrimSpace(strings.SplitN(string(out), "\n", 2)[0]), IP: ip}, + Remote: remote, + }) + } + self, err := ipv4Of(in.reachableAddr()) + if err != nil { + return err + } + name, _ := os.Hostname() + setup.Witness = &pgha.Node{Member: pgha.Member{Name: name, IP: self}, Remote: pgha.Local{}} + setup.Clients = []string{self} + + dsn, err := setup.Run(ctx) + if err != nil { + return err + } + + cfg, err := config.Load(in.configPath()) + if err != nil { + return err + } + current := cfg.Auth.Store + if current == "" { + current = in.storeAddress() + } + resolved, _ := store.ResolveAddress(current) + if strings.TrimSpace(resolved) == dsn { + in.kept("store in the Patroni cluster (%s)", in.dsnPath()) + return nil + } + + // Still while it moves, and started again on the old store if the move + // fails: a gateway down over a database that did not work out is worse + // than one left where it was. + stopped, err := in.stopService(ctx) + if err != nil { + return err + } + defer func() { + if err != nil && stopped { + _ = exec.Command("systemctl", "start", "llmgw-server").Run() + fmt.Fprintf(in.out, " .. llmgw-server started again on its old store\n") + } + }() + if err := in.moveInto(ctx, current, dsn); err != nil { + return err + } + + // The old address is kept beside the new one: the way back, if the + // cluster is not what was wanted. + if b, err := os.ReadFile(in.dsnPath()); err == nil && strings.TrimSpace(string(b)) != dsn { + if err := os.WriteFile(in.dsnPath()+".pre-patroni", b, 0o600); err != nil { + return err + } + _ = in.chown(in.dsnPath()+".pre-patroni", true) + } + if err := in.writeDSN(dsn); err != nil { + return err + } + if current != "@"+in.dsnPath() { + if err := setAuthStore(in.configPath(), "@"+in.dsnPath()); err != nil { + return err + } + } + // Started again by the service step that follows, on the new address. + in.done("server.yaml now keeps the store in the Patroni cluster") + return nil +} + +// moveInto copies a store into the cluster, unless the cluster already holds +// one — from a run that copied and stopped before changing the address. +func (in *serverInit) moveInto(ctx context.Context, from, to string) error { + src, err := store.Open(ctx, from) + if err != nil { + // Nothing there yet: a first install, with nothing to move. + in.kept("no store at %s to move", from) + return nil + } + defer src.Close() + dst, err := store.Open(ctx, to) + if err != nil { + return fmt.Errorf("opening the cluster: %w", err) + } + defer dst.Close() + empty, err := dst.Empty(ctx) + if err != nil { + return err + } + if !empty { + in.kept("the cluster already holds a store; %s is left as it is", from) + return nil + } + res, err := src.CopyTo(ctx, dst) + if err != nil { + return fmt.Errorf("copying the store into the cluster: %w", err) + } + var rows int64 + for _, r := range res { + rows += r.Rows + } + in.done("copied %d rows into the cluster (the old store is kept as it was)", rows) + return nil +} + +// ipv4Of is the address a host name or address stands for, as the other +// machines will be told to reach it. +func ipv4Of(host string) (string, error) { + if ip := net.ParseIP(host); ip != nil { + return ip.String(), nil + } + ips, err := net.LookupIP(host) + if err != nil { + return "", fmt.Errorf("%s does not resolve: %w", host, err) + } + for _, ip := range ips { + if v4 := ip.To4(); v4 != nil { + return v4.String(), nil + } + } + return "", fmt.Errorf("%s has no IPv4 address", host) +} diff --git a/internal/cli/serverinit_postgres.go b/internal/cli/serverinit_postgres.go index 138cd272..9f37f7a8 100644 --- a/internal/cli/serverinit_postgres.go +++ b/internal/cli/serverinit_postgres.go @@ -44,6 +44,12 @@ func (in *serverInit) storeAddress() string { // means reissuing every key and being unable to say which was which; and the // administrator init created a minute ago would be gone too. func (in *serverInit) ensureDatabase(ctx context.Context) (err error) { + if len(in.patroni) > 0 { + if in.postgres { + return fmt.Errorf("--postgres keeps the store on this machine and --patroni on others: give one") + } + return in.ensurePatroni(ctx) + } if !in.postgres { return nil } diff --git a/internal/cli/status.go b/internal/cli/status.go index acb35ff7..c9c5ae37 100644 --- a/internal/cli/status.go +++ b/internal/cli/status.go @@ -2,6 +2,7 @@ package cli import ( "fmt" + "strings" "time" "github.com/spf13/cobra" @@ -94,6 +95,31 @@ 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) + // The store, when it is a replicated cluster: one line, saying who leads + // and whether every other member keeps up. + switch { + case s.StoreError != "": + fmt.Fprintf(out, "STORE cannot read its cluster: %s\n", s.StoreError) + case len(s.Store) > 0: + var parts []string + leader := false + for _, m := range s.Store { + switch { + case m.Role == "leader": + leader = true + parts = append(parts, m.Name+" leads") + case m.State == "streaming" && m.Lag >= 0: + parts = append(parts, fmt.Sprintf("%s streaming (%s behind)", m.Name, bytesShort(m.Lag))) + default: + parts = append(parts, fmt.Sprintf("%s %s", m.Name, m.State)) + } + } + if !leader { + parts = append([]string{"NO LEADER — writes fail"}, parts...) + } + fmt.Fprintf(out, "STORE %s\n", strings.Join(parts, ", ")) + } + if len(s.Parked) > 0 { fmt.Fprintln(out, "\nPARKED — failed every time they started; not retried until fixed") for _, p := range s.Parked { @@ -169,3 +195,16 @@ func unhealthyCell(n int) string { } return fmt.Sprintf("%d", n) } + +// bytesShort is a byte count as a person reads it: 0 B, 12 KB, 3.4 MB. +func bytesShort(n int64) string { + switch { + case n < 1024: + return fmt.Sprintf("%d B", n) + case n < 1<<20: + return fmt.Sprintf("%d KB", n>>10) + case n < 1<<30: + return fmt.Sprintf("%.1f MB", float64(n)/(1<<20)) + } + return fmt.Sprintf("%.1f GB", float64(n)/(1<<30)) +} diff --git a/internal/control/adminapi/api.go b/internal/control/adminapi/api.go index b1d5146b..3a18611a 100644 --- a/internal/control/adminapi/api.go +++ b/internal/control/adminapi/api.go @@ -40,6 +40,8 @@ import ( type API struct { // releases is where release binaries are fetched from, for upgrades. releases string + // storeStatus reads a replicated store's members; nil for one database. + storeStatus func() ([]StoreMember, error) // serve runs registry-to-replica jobs; nil without a mirror and a store. serve *modelserve.Server @@ -698,6 +700,9 @@ type Usage interface { // SetUsage supplies the usage reporting source. func (a *API) SetUsage(u Usage) { a.usage = u } +// SetStoreStatus supplies the state of a replicated store, for the summary. +func (a *API) SetStoreStatus(f func() ([]StoreMember, error)) { a.storeStatus = f } + // ScopeUsageView is one group's spend, with its quota when it has one. type ScopeUsageView struct { Scope string `json:"scope"` diff --git a/internal/control/adminapi/summary.go b/internal/control/adminapi/summary.go index 897568ca..27b88b48 100644 --- a/internal/control/adminapi/summary.go +++ b/internal/control/adminapi/summary.go @@ -31,6 +31,20 @@ type Summary struct { // 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"` + // Store is the members of the store's cluster when it is replicated, and + // StoreError why it could not be read; both empty for a single database. + Store []StoreMember `json:"store,omitempty"` + StoreError string `json:"store_error,omitempty"` +} + +// StoreMember is one member of a replicated store, as its Patroni reports it. +type StoreMember struct { + Name string `json:"name"` + Host string `json:"host"` + Role string `json:"role"` + State string `json:"state"` + // Lag is bytes behind the leader; -1 unknown, 0 for the leader. + Lag int64 `json:"lag"` } // NodeTotals counts nodes by the states that need different responses. @@ -107,6 +121,13 @@ const idleThreshold = 10 * time.Minute // Summary builds the fleet summary. func (a *API) Summary() Summary { var s Summary + if a.storeStatus != nil { + members, err := a.storeStatus() + s.Store = members + if err != nil { + s.StoreError = err.Error() + } + } nodes := a.Nodes() byClass := map[string]*ClassUse{} diff --git a/internal/pgha/config.go b/internal/pgha/config.go new file mode 100644 index 00000000..bd4d284e --- /dev/null +++ b/internal/pgha/config.go @@ -0,0 +1,171 @@ +// Package pgha sets up the store as PostgreSQL under Patroni: two or more +// machines, one primary streaming to the others, and a standby promoted when +// the primary is lost. Patroni keeps its leader lock in etcd, which decides +// only with a majority, so etcd runs on three machines — the database +// machines and, when they are two, the control plane as the third vote. +// +// Everything is done over the same ssh the rest of onboarding uses, and every +// step looks before it acts: run against a cluster that is already up, it +// changes nothing. +package pgha + +import ( + "fmt" + "net/url" + "strings" +) + +// Scope is the Patroni cluster's name, and the etcd cluster token. +const Scope = "llmgw" + +// PatroniPort is where Patroni's REST API listens on every member: what the +// control plane reads the cluster's state from. +const PatroniPort = 8008 + +// Role and Database are what the control plane connects as. +const ( + Role = "llmgw" + Database = "llmgw" +) + +// Member is one machine of the cluster, by the address the others reach it on. +type Member struct { + Name string + IP string +} + +// Secrets are the cluster's three passwords. +type Secrets struct { + Superuser, Replicator, App string +} + +// Etcd is a member's /etc/default/etcd, for Ubuntu's etcd-server. +func Etcd(m Member, all []Member) string { + var peers []string + for _, p := range all { + peers = append(peers, fmt.Sprintf("%s=http://%s:2380", p.Name, p.IP)) + } + return fmt.Sprintf(`# llmgw: one member of the etcd that holds Patroni's leader lock. +ETCD_NAME=%[1]s +ETCD_DATA_DIR=/var/lib/etcd/default +ETCD_LISTEN_PEER_URLS=http://%[2]s:2380 +ETCD_LISTEN_CLIENT_URLS=http://%[2]s:2379,http://127.0.0.1:2379 +ETCD_INITIAL_ADVERTISE_PEER_URLS=http://%[2]s:2380 +ETCD_ADVERTISE_CLIENT_URLS=http://%[2]s:2379 +ETCD_INITIAL_CLUSTER=%[3]s +ETCD_INITIAL_CLUSTER_STATE=new +ETCD_INITIAL_CLUSTER_TOKEN=%[4]s +`, m.Name, m.IP, strings.Join(peers, ","), Scope+"-patroni") +} + +// Patroni is a database member's /etc/patroni/config.yml. +// +// clients are the addresses the control planes connect from; the members +// themselves are always allowed, since a second control plane usually runs on +// one of them. +func Patroni(m Member, db, etcd []Member, clients []string, version string, s Secrets) string { + var hosts []string + for _, e := range etcd { + hosts = append(hosts, e.IP+":2379") + } + var hba []string + hba = append(hba, "local all all peer", "host all all 127.0.0.1/32 scram-sha-256") + for _, d := range db { + hba = append(hba, + "host replication replicator "+d.IP+"/32 scram-sha-256", + "host all postgres "+d.IP+"/32 scram-sha-256") + } + seen := map[string]bool{} + for _, c := range append(append([]string{}, clients...), ips(db)...) { + if c == "" || seen[c] { + continue + } + seen[c] = true + hba = append(hba, fmt.Sprintf("host %s %s %s/32 scram-sha-256", Database, Role, c)) + } + var b strings.Builder + fmt.Fprintf(&b, `# llmgw: the control planes' store, kept by Patroni, its leader lock in etcd. +scope: %[1]s +namespace: /service/ +name: %[2]s + +restapi: + listen: %[3]s:%[4]d + connect_address: %[3]s:%[4]d + +etcd3: + hosts: %[5]s + +bootstrap: + dcs: + ttl: 30 + loop_wait: 10 + retry_timeout: 10 + maximum_lag_on_failover: 1048576 + postgresql: + use_pg_rewind: true + use_slots: true + parameters: + wal_level: replica + hot_standby: "on" + max_wal_senders: 10 + max_replication_slots: 10 + wal_keep_size: 1GB + wal_log_hints: "on" + password_encryption: scram-sha-256 + initdb: + - encoding: UTF8 + - data-checksums + +postgresql: + listen: 127.0.0.1,%[3]s:5432 + connect_address: %[3]s:5432 + data_dir: /var/lib/postgresql/%[6]s/llmgw + bin_dir: /usr/lib/postgresql/%[6]s/bin + pgpass: /var/lib/postgresql/.pgpass_patroni + authentication: + superuser: + username: postgres + password: "%[7]s" + replication: + username: replicator + password: "%[8]s" + rewind: + username: postgres + password: "%[7]s" + parameters: + unix_socket_directories: /var/run/postgresql + pg_hba: +`, Scope, m.Name, m.IP, PatroniPort, strings.Join(hosts, ","), version, s.Superuser, s.Replicator) + for _, h := range hba { + fmt.Fprintf(&b, " - %s\n", h) + } + b.WriteString(` +tags: + nofailover: false + noloadbalance: false + clonefrom: false + nosync: false +`) + return b.String() +} + +// DSN is the store address the control planes are given: every member, and +// the driver told to use whichever accepts writes, so a failover is followed +// without anything in front of the database. +func DSN(db []Member, password string) string { + var hosts []string + for _, d := range db { + hosts = append(hosts, d.IP+":5432") + } + return fmt.Sprintf("postgres://%s:%s@%s/%s?sslmode=disable&target_session_attrs=read-write", + Role, url.QueryEscape(password), strings.Join(hosts, ","), Database) +} + +func ips(ms []Member) []string { + out := make([]string, len(ms)) + for i, m := range ms { + out[i] = m.IP + } + return out +} diff --git a/internal/pgha/metrics.go b/internal/pgha/metrics.go new file mode 100644 index 00000000..5ffa77b8 --- /dev/null +++ b/internal/pgha/metrics.go @@ -0,0 +1,100 @@ +package pgha + +import ( + "context" + "sync" + "time" + + "github.com/prometheus/client_golang/prometheus" +) + +// Collector publishes the store cluster's state as the control plane sees it +// through Patroni's API: who leads, who streams, how far behind. Read at most +// every fifteen seconds, however often it is scraped. +type Collector struct { + hosts []string + status func(ctx context.Context, hosts []string) (Cluster, error) + + mu sync.Mutex + at time.Time + last Cluster + err error + + up, member, lag *prometheus.Desc +} + +// NewCollector reads the cluster from the Patroni of each host in turn. +func NewCollector(hosts []string) *Collector { + return &Collector{ + hosts: hosts, + status: HTTPStatus, + up: prometheus.NewDesc("llmgw_store_patroni_up", + "1 when a member's Patroni answered for the store's cluster.", nil, nil), + member: prometheus.NewDesc("llmgw_store_member", + "Each member of the store's cluster, by role (leader, replica) and state (running, streaming, …); always 1.", + []string{"member", "host", "role", "state"}, nil), + lag: prometheus.NewDesc("llmgw_store_replica_lag_bytes", + "How far each replica is behind the leader; -1 when Patroni does not know.", + []string{"member"}, nil), + } +} + +// Describe implements prometheus.Collector. +func (c *Collector) Describe(ch chan<- *prometheus.Desc) { + ch <- c.up + ch <- c.member + ch <- c.lag +} + +// Collect implements prometheus.Collector. +func (c *Collector) Collect(ch chan<- prometheus.Metric) { + cl, err := c.read() + if err != nil { + ch <- prometheus.MustNewConstMetric(c.up, prometheus.GaugeValue, 0) + return + } + ch <- prometheus.MustNewConstMetric(c.up, prometheus.GaugeValue, 1) + for _, m := range cl.Members { + ch <- prometheus.MustNewConstMetric(c.member, prometheus.GaugeValue, 1, m.Name, m.Host, m.Role, m.State) + if m.Role != "leader" { + ch <- prometheus.MustNewConstMetric(c.lag, prometheus.GaugeValue, float64(m.Lag), m.Name) + } + } +} + +// Read is the cluster's state, at most fifteen seconds old. +func (c *Collector) Read() (Cluster, error) { return c.read() } + +func (c *Collector) read() (Cluster, error) { + c.mu.Lock() + defer c.mu.Unlock() + if time.Since(c.at) < 15*time.Second { + return c.last, c.err + } + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + c.last, c.err = c.status(ctx, c.hosts) + if c.err == nil { + c.last = withMissing(c.last, c.hosts) + } + c.at = time.Now() + return c.last, c.err +} + +// withMissing adds every host of the store address that the cluster does not +// list, as a member in state "missing". A member that stops is dropped from +// Patroni's list once its key expires — so without this a cluster down to +// one machine reads as a healthy cluster of one, and nothing says a standby +// is gone. +func withMissing(c Cluster, hosts []string) Cluster { + seen := map[string]bool{} + for _, m := range c.Members { + seen[m.Host] = true + } + for _, h := range hosts { + if !seen[h] { + c.Members = append(c.Members, MemberState{Name: h, Host: h, Role: "replica", State: "missing", Lag: -1}) + } + } + return c +} diff --git a/internal/pgha/pgha_test.go b/internal/pgha/pgha_test.go new file mode 100644 index 00000000..a2bc2913 --- /dev/null +++ b/internal/pgha/pgha_test.go @@ -0,0 +1,210 @@ +package pgha + +import ( + "context" + "encoding/json" + "strings" + "sync" + "testing" + "time" + + "github.com/prometheus/client_golang/prometheus/testutil" + "github.com/prometheus/common/expfmt" +) + +var ( + pg01 = Member{"llmgw-pg01", "10.0.0.149"} + ctl02 = Member{"llmgw-ctl02", "10.0.0.150"} + ctl = Member{"llmgw-ctl", "10.0.0.140"} +) + +func TestTheConfigurationsNameEveryMemberAndOnlyTheClientsGiven(t *testing.T) { + sec := Secrets{"su", "rep", "app"} + cfg := Patroni(pg01, []Member{pg01, ctl02}, []Member{pg01, ctl02, ctl}, []string{"10.0.0.140"}, "16", sec) + for _, want := range []string{ + "name: llmgw-pg01", "listen: 10.0.0.149:8008", + "hosts: 10.0.0.149:2379,10.0.0.150:2379,10.0.0.140:2379", + "data_dir: /var/lib/postgresql/16/llmgw", `password: "rep"`, + "- host replication replicator 10.0.0.150/32 scram-sha-256", + "- host llmgw llmgw 10.0.0.140/32 scram-sha-256", + // A member can run a control plane too. + "- host llmgw llmgw 10.0.0.150/32 scram-sha-256", + } { + if !strings.Contains(cfg, want) { + t.Errorf("patroni config lacks %q", want) + } + } + if strings.Count(cfg, "host llmgw llmgw 10.0.0.149/32") != 1 { + t.Error("a member's address is allowed twice") + } + e := Etcd(ctl, []Member{pg01, ctl02, ctl}) + if !strings.Contains(e, "ETCD_NAME=llmgw-ctl\n") || + !strings.Contains(e, "llmgw-pg01=http://10.0.0.149:2380,llmgw-ctl02=http://10.0.0.150:2380,llmgw-ctl=http://10.0.0.140:2380") { + t.Errorf("etcd defaults:\n%s", e) + } + if got := DSN([]Member{pg01, ctl02}, "pw"); got != + "postgres://llmgw:pw@10.0.0.149:5432,10.0.0.150:5432/llmgw?sslmode=disable&target_session_attrs=read-write" { + t.Errorf("dsn %s", got) + } +} + +func TestAClusterIsCompleteWithALeaderAndEveryOtherMemberStreaming(t *testing.T) { + var c Cluster + if err := json.Unmarshal([]byte(`{"members":[ + {"name":"llmgw-ctl02","role":"replica","state":"streaming","host":"10.0.0.150","lag":0}, + {"name":"llmgw-pg01","role":"leader","state":"running","host":"10.0.0.149"}]}`), &c); err != nil { + t.Fatal(err) + } + if c.Leader() != "llmgw-pg01" || !c.Complete(2) || c.Complete(3) { + t.Errorf("%s: leader %q", c, c.Leader()) + } + c.Members[0].State = "starting" + if c.Complete(2) { + t.Error("a replica still starting counted as streaming") + } + if lagOf(json.RawMessage(`"unknown"`)) != -1 || lagOf(json.RawMessage(`1024`)) != 1024 { + t.Error("lag not read") + } +} + +// machine is a fake host: it records what it was asked to run and answers +// from a table of script fragments. +type machine struct { + name string + mu sync.Mutex + ran []string + answers map[string]string +} + +func (m *machine) Run(_ context.Context, script string, _ []byte) ([]byte, error) { + m.mu.Lock() + defer m.mu.Unlock() + m.ran = append(m.ran, script) + for frag, out := range m.answers { + if strings.Contains(script, frag) { + return []byte(out), nil + } + } + return nil, nil +} + +func (m *machine) ranAny(frag string) bool { + for _, s := range m.ran { + if strings.Contains(s, frag) { + return true + } + } + return false +} + +const pws = "superuser=a\nreplicator=b\nllmgw=c\n" + +func healthy(n int) string { + return strings.Repeat("http://x:2379 is healthy: successfully committed proposal\n", n) +} + +func TestARunningClusterIsLeftAsItIs(t *testing.T) { + // What init does when run again, or against the reference cluster: find + // it up, change nothing, and give the address. + a := &machine{name: "pg01", answers: map[string]string{"openssl rand": pws, "endpoint health": healthy(3)}} + b := &machine{name: "ctl02", answers: map[string]string{"sudo cat /root/llmgw-pg": pws}} + w := &machine{name: "ctl"} + up := Cluster{Members: []MemberState{{Name: "llmgw-pg01", Role: "leader", State: "running"}, + {Name: "llmgw-ctl02", Role: "replica", State: "streaming"}}} + s := Setup{ + DB: []Node{{pg01, a}, {ctl02, b}}, + Witness: &Node{ctl, w}, + Status: func(context.Context, []string) (Cluster, error) { return up, nil }, + } + dsn, err := s.Run(context.Background()) + if err != nil { + t.Fatal(err) + } + if !strings.Contains(dsn, "llmgw:c@10.0.0.149:5432,10.0.0.150:5432") { + t.Errorf("dsn %s", dsn) + } + for _, m := range []*machine{a, b, w} { + for _, frag := range []string{"apt-get", "systemctl restart", "/etc/default/etcd", "/etc/patroni/config.yml"} { + if m.ranAny(frag) { + t.Errorf("%s was changed (%q) though the cluster is up", m.name, frag) + } + } + } + if !a.ranAny("CREATE ROLE") { + t.Error("the role was not made sure of on the leader") + } +} + +func TestTwoDatabaseMachinesWithoutAThirdVoteAreRefused(t *testing.T) { + s := Setup{DB: []Node{{pg01, &machine{}}, {ctl02, &machine{}}}} + if _, err := s.Run(context.Background()); err == nil || !strings.Contains(err.Error(), "third etcd vote") { + t.Errorf("got %v", err) + } +} + +func TestAMachineHoldingOtherPasswordsIsNotOverwritten(t *testing.T) { + a := &machine{answers: map[string]string{"openssl rand": pws}} + b := &machine{answers: map[string]string{"sudo cat /root/llmgw-pg": "superuser=x\nreplicator=y\nllmgw=z\n"}} + s := Setup{DB: []Node{{pg01, a}, {ctl02, b}}, Witness: &Node{ctl, &machine{}}} + _, err := s.Run(context.Background()) + if err == nil || !strings.Contains(err.Error(), "differ") { + t.Fatalf("got %v", err) + } + if b.ranAny("cat > /root/llmgw-pg") { + t.Error("the other cluster's passwords were replaced") + } +} + +func TestAPostgreSQLWithDatabasesInItIsNotReplaced(t *testing.T) { + a := &machine{answers: map[string]string{"openssl rand": pws, "endpoint health": healthy(3), + "pg_dropcluster": "holds-data\n"}} + b := &machine{answers: map[string]string{"sudo cat /root/llmgw-pg": pws}} + s := Setup{DB: []Node{{pg01, a}, {ctl02, b}}, Witness: &Node{ctl, &machine{}}, + Status: func(context.Context, []string) (Cluster, error) { return Cluster{}, nil }, + Wait: time.Second} + _, err := s.Run(context.Background()) + if err == nil || !strings.Contains(err.Error(), "databases in it") { + t.Fatalf("got %v", err) + } + if a.ranAny("systemctl restart patroni") { + t.Error("patroni was started over somebody's data") + } +} + +func TestTheCollectorSaysWhoLeadsAndHowFarBehindTheRestAre(t *testing.T) { + c := NewCollector([]string{"10.0.0.149"}) + c.status = func(context.Context, []string) (Cluster, error) { + return Cluster{Members: []MemberState{{Name: "llmgw-pg01", Host: "10.0.0.149", Role: "leader", State: "running"}, + {Name: "llmgw-ctl02", Host: "10.0.0.150", Role: "replica", State: "streaming", Lag: 2048}}}, nil + } + got, err := testutil.CollectAndFormat(c, expfmt.TypeTextPlain, "llmgw_store_patroni_up", "llmgw_store_member", "llmgw_store_replica_lag_bytes") + if err != nil { + t.Fatal(err) + } + for _, want := range []string{ + "llmgw_store_patroni_up 1", + `llmgw_store_member{host="10.0.0.149",member="llmgw-pg01",role="leader",state="running"} 1`, + `llmgw_store_replica_lag_bytes{member="llmgw-ctl02"} 2048`, + } { + if !strings.Contains(string(got), want) { + t.Errorf("missing %s in\n%s", want, got) + } + } + if strings.Contains(string(got), `lag_bytes{member="llmgw-pg01"}`) { + t.Error("the leader was given a lag") + } +} + +func TestAMemberThatStoppedIsReportedMissingNotForgotten(t *testing.T) { + c := NewCollector([]string{"10.0.0.153", "10.0.0.154"}) + c.status = func(context.Context, []string) (Cluster, error) { + return Cluster{Members: []MemberState{{Name: "hatest-pg2", Host: "10.0.0.154", Role: "leader", State: "running"}}}, nil + } + cl, err := c.Read() + if err != nil { + t.Fatal(err) + } + if len(cl.Members) != 2 || cl.Members[1].Host != "10.0.0.153" || cl.Members[1].State != "missing" { + t.Errorf("%s", cl) + } +} diff --git a/internal/pgha/setup.go b/internal/pgha/setup.go new file mode 100644 index 00000000..2dca4daa --- /dev/null +++ b/internal/pgha/setup.go @@ -0,0 +1,417 @@ +package pgha + +import ( + "bytes" + "context" + "fmt" + "os/exec" + "strings" + "sync" + "time" + + "gitea.mynodes.xyz/ilker/llmgw/internal/onboard" +) + +// Node is a machine the setup reaches, and how. +type Node struct { + Member + Remote onboard.Remote +} + +// Setup builds, or finds, a Patroni cluster. +type Setup struct { + // DB are the database machines; the first is where the cluster starts. + DB []Node + // Witness is the third etcd vote when there are only two database + // machines: the control plane, run locally. Ignored with three or more. + Witness *Node + // Clients are the addresses control planes connect from. + Clients []string + // Status reads the cluster from Patroni's API; nil asks over HTTP. + Status func(ctx context.Context, hosts []string) (Cluster, error) + // Emit reports progress: a step, what happened, and whether anything + // was changed or found as it should be. + Emit func(step, detail string, changed bool) + // Wait bounds each wait for the cluster to come up. + Wait time.Duration +} + +// Local is a Remote for the machine this runs on, which init runs as root. +type Local struct{} + +// Run runs script with sh here. +func (Local) Run(ctx context.Context, script string, stdin []byte) ([]byte, error) { + cmd := exec.CommandContext(ctx, "sh", "-c", script) + if stdin != nil { + cmd.Stdin = bytes.NewReader(stdin) + } + var out, errb bytes.Buffer + cmd.Stdout, cmd.Stderr = &out, &errb + if err := cmd.Run(); err != nil { + return append(out.Bytes(), errb.Bytes()...), err + } + return out.Bytes(), nil +} + +func (s *Setup) emit(step string, changed bool, format string, a ...any) { + if s.Emit != nil { + s.Emit(step, fmt.Sprintf(format, a...), changed) + } +} + +// etcdMembers are the machines that vote: the database machines, and the +// witness when they are only two. +func (s *Setup) etcdMembers() []Node { + out := append([]Node{}, s.DB...) + if len(s.DB) == 2 && s.Witness != nil { + out = append(out, *s.Witness) + } + return out +} + +// Run sets the cluster up, or confirms it is, and returns the store address. +func (s *Setup) Run(ctx context.Context) (string, error) { + if len(s.DB) < 2 { + return "", fmt.Errorf("a replicated store needs at least two database machines, not %d", len(s.DB)) + } + if len(s.DB) == 2 && s.Witness == nil { + return "", fmt.Errorf("two database machines need a third etcd vote, or losing either stops failover") + } + if s.Wait == 0 { + s.Wait = 3 * time.Minute + } + if s.Status == nil { + s.Status = HTTPStatus + } + sec, err := s.secrets(ctx) + if err != nil { + return "", fmt.Errorf("passwords: %w", err) + } + if err := s.etcd(ctx); err != nil { + return "", fmt.Errorf("etcd: %w", err) + } + if err := s.patroni(ctx, sec); err != nil { + return "", fmt.Errorf("patroni: %w", err) + } + if err := s.database(ctx); err != nil { + return "", fmt.Errorf("database: %w", err) + } + members := make([]Member, len(s.DB)) + for i, d := range s.DB { + members[i] = d.Member + } + return DSN(members, sec.App), nil +} + +// secretDir holds the passwords on every database machine, readable by root +// alone. Generated on the first, carried to the others. +const secretDir = "/root/llmgw-pg" + +var secretNames = []string{"superuser", "replicator", "llmgw"} + +func (s *Setup) secrets(ctx context.Context) (Secrets, error) { + script := `set -e +sudo install -d -m 700 ` + secretDir + ` +for f in superuser replicator llmgw; do + sudo test -s ` + secretDir + `/$f.pw || sudo sh -c "umask 077; openssl rand -hex 24 > ` + secretDir + `/$f.pw" +done +for f in superuser replicator llmgw; do printf '%s=' $f; sudo cat ` + secretDir + `/$f.pw; done` + out, err := s.DB[0].Remote.Run(ctx, script, nil) + if err != nil { + return Secrets{}, fmt.Errorf("on %s: %v: %s", s.DB[0].Name, err, strings.TrimSpace(string(out))) + } + got := parseSecrets(out) + if got.Superuser == "" || got.Replicator == "" || got.App == "" { + return Secrets{}, fmt.Errorf("could not read them back from %s", s.DB[0].Name) + } + for _, d := range s.DB[1:] { + have, err := d.Remote.Run(ctx, `for f in superuser replicator llmgw; do + if sudo test -s `+secretDir+`/$f.pw; then printf '%s=' $f; sudo cat `+secretDir+`/$f.pw; fi +done`, nil) + if err != nil { + return Secrets{}, fmt.Errorf("on %s: %v", d.Name, err) + } + if h := parseSecrets(have); h != (Secrets{}) { + if h != got { + // Not overwritten: a member holding other passwords belongs + // to another cluster, or to an earlier one somebody meant to + // keep, and replacing them would lock it out of itself. + return Secrets{}, fmt.Errorf("%s holds passwords in %s that differ from %s's; "+ + "remove them there if that machine is meant to join this cluster", d.Name, secretDir, s.DB[0].Name) + } + continue + } + body := fmt.Sprintf("%s\n%s\n%s\n", got.Superuser, got.Replicator, got.App) + if out, err := d.Remote.Run(ctx, `set -e +sudo install -d -m 700 `+secretDir+` +read su; read rep; read app +for kv in "superuser $su" "replicator $rep" "llmgw $app"; do + set -- $kv + printf '%s\n' "$2" | sudo sh -c "umask 077; cat > `+secretDir+`/$1.pw" +done`, []byte(body)); err != nil { + return Secrets{}, fmt.Errorf("on %s: %v: %s", d.Name, err, strings.TrimSpace(string(out))) + } + s.emit("passwords", true, "carried to %s", d.Name) + } + s.emit("passwords", false, "kept in %s on every database machine", secretDir) + return got, nil +} + +func parseSecrets(out []byte) Secrets { + var s Secrets + for _, line := range strings.Split(string(out), "\n") { + k, v, ok := strings.Cut(strings.TrimSpace(line), "=") + if !ok { + continue + } + switch k { + case "superuser": + s.Superuser = v + case "replicator": + s.Replicator = v + case "llmgw": + s.App = v + } + } + return s +} + +// etcd brings up the voting members, unless every one of them is already +// healthy in one cluster — in which case nothing is touched, since rewriting +// a member's configuration means emptying its data. +func (s *Setup) etcd(ctx context.Context) error { + nodes := s.etcdMembers() + var members []Member + var endpoints []string + for _, n := range nodes { + members = append(members, n.Member) + endpoints = append(endpoints, "http://"+n.IP+":2379") + } + if s.etcdHealthy(ctx, nodes[0], endpoints, len(nodes)) { + s.emit("etcd", false, "%d members healthy; left as they are", len(nodes)) + return nil + } + + changed := make([]bool, len(nodes)) + for i, n := range nodes { + out, err := n.Remote.Run(ctx, `set -e +if ! command -v etcd >/dev/null 2>&1; then + sudo DEBIAN_FRONTEND=noninteractive apt-get -qq update >/dev/null + sudo DEBIAN_FRONTEND=noninteractive apt-get -qq install -y etcd-server etcd-client >/dev/null +fi +new=$(cat) +if [ "$(sudo cat /etc/default/etcd 2>/dev/null)" != "$new" ]; then + sudo systemctl stop etcd >/dev/null 2>&1 || true + sudo rm -rf /var/lib/etcd/default + printf '%s\n' "$new" | sudo tee /etc/default/etcd >/dev/null + echo changed +fi +sudo systemctl enable etcd >/dev/null 2>&1`, []byte(Etcd(n.Member, members))) + if err != nil { + return fmt.Errorf("on %s: %v: %s", n.Name, err, strings.TrimSpace(string(out))) + } + changed[i] = strings.Contains(string(out), "changed") + } + // Started together: a new member waits for the others before it says it + // is up, so starting them one after another would wait for ever on the + // first. + var wg sync.WaitGroup + errs := make([]error, len(nodes)) + for i, n := range nodes { + wg.Add(1) + go func(i int, n Node) { + defer wg.Done() + verb := "start" + if changed[i] { + verb = "restart" + } + if out, err := n.Remote.Run(ctx, "sudo systemctl "+verb+" --no-block etcd", nil); err != nil { + errs[i] = fmt.Errorf("on %s: %v: %s", n.Name, err, strings.TrimSpace(string(out))) + } + }(i, n) + } + wg.Wait() + for _, err := range errs { + if err != nil { + return err + } + } + deadline := time.Now().Add(s.Wait) + for !s.etcdHealthy(ctx, nodes[0], endpoints, len(nodes)) { + if time.Now().After(deadline) { + return fmt.Errorf("the %d members did not become healthy within %s; `journalctl -u etcd` on each says why", len(nodes), s.Wait) + } + if err := sleep(ctx, 2*time.Second); err != nil { + return err + } + } + s.emit("etcd", true, "%d members up", len(nodes)) + return nil +} + +func (s *Setup) etcdHealthy(ctx context.Context, n Node, endpoints []string, want int) bool { + out, err := n.Remote.Run(ctx, "ETCDCTL_API=3 etcdctl --dial-timeout=3s --endpoints="+ + strings.Join(endpoints, ",")+" endpoint health 2>&1", nil) + return err == nil && strings.Count(string(out), "is healthy") == want +} + +// patroni installs PostgreSQL and Patroni on every database machine, unless +// the cluster already has a leader and every member streaming. +func (s *Setup) patroni(ctx context.Context, sec Secrets) error { + hosts := make([]string, len(s.DB)) + for i, d := range s.DB { + hosts[i] = d.IP + } + if c, err := s.Status(ctx, hosts); err == nil && c.Complete(len(s.DB)) { + s.emit("patroni", false, "%s leads, %d streaming; left as it is", c.Leader(), len(s.DB)-1) + return nil + } + + var version string + for _, d := range s.DB { + v, err := s.preparePostgres(ctx, d) + if err != nil { + return err + } + if version != "" && v != version { + // Streaming replication is between one major version. + return fmt.Errorf("%s has PostgreSQL %s and %s has %s; they must match", s.DB[0].Name, version, d.Name, v) + } + version = v + } + + var etcd, db []Member + for _, n := range s.etcdMembers() { + etcd = append(etcd, n.Member) + } + for _, d := range s.DB { + db = append(db, d.Member) + } + for i, d := range s.DB { + cfg := Patroni(d.Member, db, etcd, s.Clients, version, sec) + if out, err := d.Remote.Run(ctx, `set -e +sudo install -d -m 755 /etc/patroni +sudo install -d -o postgres -g postgres -m 700 /var/lib/postgresql/`+version+`/llmgw +sudo sh -c 'umask 027; cat > /etc/patroni/config.yml.new' +sudo chown root:postgres /etc/patroni/config.yml.new +sudo chmod 640 /etc/patroni/config.yml.new +sudo mv -f /etc/patroni/config.yml.new /etc/patroni/config.yml +sudo systemctl enable patroni >/dev/null 2>&1 +sudo systemctl restart patroni`, []byte(cfg)); err != nil { + return fmt.Errorf("on %s: %v: %s", d.Name, err, strings.TrimSpace(string(out))) + } + // The first starts the cluster; the rest clone it, which they can + // only do once it has a leader. + want := func(c Cluster) bool { return c.Leader() != "" } + if i > 0 { + want = func(c Cluster) bool { return c.Complete(i + 1) } + } + if err := s.waitFor(ctx, hosts, want); err != nil { + return fmt.Errorf("after starting %s: %w", d.Name, err) + } + s.emit("patroni", true, "%s up", d.Name) + } + return nil +} + +// preparePostgres installs the packages and clears the distribution's own +// cluster, which would hold the port Patroni's needs — unless it holds a +// database somebody made, which is not this setup's to delete. +func (s *Setup) preparePostgres(ctx context.Context, d Node) (string, error) { + out, err := d.Remote.Run(ctx, `set -e +need="" +command -v patroni >/dev/null 2>&1 || need="$need patroni" +ls /usr/lib/postgresql/*/bin/postgres >/dev/null 2>&1 || need="$need postgresql" +if [ -n "$need" ]; then + sudo DEBIAN_FRONTEND=noninteractive apt-get -qq update >/dev/null + sudo DEBIAN_FRONTEND=noninteractive apt-get -qq install -y $need >/dev/null +fi +v=$(ls /usr/lib/postgresql | sort -n | tail -1) +if [ -d /var/lib/postgresql/$v/main ]; then + if sudo systemctl is-active --quiet postgresql@$v-main; then + n=$(sudo -u postgres psql -Atc "select count(*) from pg_database where datname not in ('postgres','template0','template1')") + if [ "$n" != "0" ]; then echo "holds-data"; exit 0; fi + fi + sudo systemctl disable --now postgresql@$v-main postgresql >/dev/null 2>&1 || true + sudo pg_dropcluster --stop $v main >/dev/null 2>&1 || sudo rm -rf /var/lib/postgresql/$v/main +fi +sudo systemctl disable --now postgresql >/dev/null 2>&1 || true +echo "version=$v"`, nil) + if err != nil { + return "", fmt.Errorf("on %s: %v: %s", d.Name, err, strings.TrimSpace(string(out))) + } + if strings.Contains(string(out), "holds-data") { + return "", fmt.Errorf("%s runs a PostgreSQL with databases in it; move or drop them first — "+ + "this setup replaces that cluster", d.Name) + } + for _, line := range strings.Split(string(out), "\n") { + if v, ok := strings.CutPrefix(strings.TrimSpace(line), "version="); ok && v != "" { + return v, nil + } + } + return "", fmt.Errorf("no PostgreSQL found on %s after installing it", d.Name) +} + +// database makes the role and database on whichever member leads. +func (s *Setup) database(ctx context.Context) error { + hosts := make([]string, len(s.DB)) + for i, d := range s.DB { + hosts[i] = d.IP + } + c, err := s.Status(ctx, hosts) + if err != nil { + return err + } + var leader *Node + for i := range s.DB { + if s.DB[i].Name == c.Leader() { + leader = &s.DB[i] + } + } + if leader == nil { + return fmt.Errorf("no member leads") + } + out, err := leader.Remote.Run(ctx, `set -e +pw=$(sudo cat `+secretDir+`/llmgw.pw) +sudo -u postgres psql -q -v ON_ERROR_STOP=1 -v pw="$pw" <<'SQL' +SELECT format('CREATE ROLE %I LOGIN', 'llmgw') WHERE NOT EXISTS (SELECT 1 FROM pg_roles WHERE rolname = 'llmgw') \gexec +ALTER ROLE llmgw PASSWORD :'pw'; +SELECT format('CREATE DATABASE %I OWNER %I', 'llmgw', 'llmgw') WHERE NOT EXISTS (SELECT 1 FROM pg_database WHERE datname = 'llmgw') \gexec +SQL`, nil) + if err != nil { + return fmt.Errorf("on %s: %v: %s", leader.Name, err, strings.TrimSpace(string(out))) + } + s.emit("database", true, "role and database %s on %s", Database, leader.Name) + return nil +} + +func (s *Setup) waitFor(ctx context.Context, hosts []string, ok func(Cluster) bool) error { + deadline := time.Now().Add(s.Wait) + var last error + for { + c, err := s.Status(ctx, hosts) + if err == nil && ok(c) { + return nil + } + if err != nil { + last = err + } else { + last = fmt.Errorf("the cluster is %s", c) + } + if time.Now().After(deadline) { + return fmt.Errorf("not up within %s: %v; `journalctl -u patroni` says why", s.Wait, last) + } + if err := sleep(ctx, 3*time.Second); err != nil { + return err + } + } +} + +func sleep(ctx context.Context, d time.Duration) error { + select { + case <-ctx.Done(): + return ctx.Err() + case <-time.After(d): + return nil + } +} diff --git a/internal/pgha/status.go b/internal/pgha/status.go new file mode 100644 index 00000000..74c1b163 --- /dev/null +++ b/internal/pgha/status.go @@ -0,0 +1,114 @@ +package pgha + +import ( + "context" + "encoding/json" + "fmt" + "net" + "net/http" + "strconv" + "strings" + "time" +) + +// MemberState is one member as Patroni reports it. +type MemberState struct { + Name string `json:"name"` + // Role is "leader", "replica" or "sync_standby". + Role string `json:"role"` + // State is "running" for a leader, "streaming" for a replica keeping up, + // and anything else for one that is not. + State string `json:"state"` + Host string `json:"host"` + Timeline int `json:"timeline"` + // Lag is how far behind the leader a replica is, in bytes; -1 when + // Patroni does not know, which is itself worth an alert. + Lag int64 `json:"-"` + RawLag json.RawMessage `json:"lag"` +} + +// Cluster is the cluster's state, as any member's API reports it. +type Cluster struct { + Members []MemberState `json:"members"` +} + +// Leader is the leading member's name, or "". +func (c Cluster) Leader() string { + for _, m := range c.Members { + if m.Role == "leader" || m.Role == "standby_leader" { + return m.Name + } + } + return "" +} + +// Complete reports whether n members are up: one leading, the rest streaming. +func (c Cluster) Complete(n int) bool { + if c.Leader() == "" || len(c.Members) < n { + return false + } + streaming := 0 + for _, m := range c.Members { + if m.Role != "leader" && m.State == "streaming" { + streaming++ + } + } + return streaming >= n-1 +} + +func (c Cluster) String() string { + if len(c.Members) == 0 { + return "empty" + } + var parts []string + for _, m := range c.Members { + parts = append(parts, m.Name+" "+m.Role+"/"+m.State) + } + return strings.Join(parts, ", ") +} + +// HTTPStatus asks each host's Patroni in turn and returns the first answer: +// any member describes the whole cluster. +func HTTPStatus(ctx context.Context, hosts []string) (Cluster, error) { + client := &http.Client{Timeout: 3 * time.Second} + var last error + for _, h := range hosts { + u := "http://" + net.JoinHostPort(h, strconv.Itoa(PatroniPort)) + "/cluster" + req, err := http.NewRequestWithContext(ctx, http.MethodGet, u, nil) + if err != nil { + return Cluster{}, err + } + resp, err := client.Do(req) + if err != nil { + last = err + continue + } + var c Cluster + err = json.NewDecoder(resp.Body).Decode(&c) + resp.Body.Close() + if err != nil { + last = fmt.Errorf("%s: %w", u, err) + continue + } + for i := range c.Members { + c.Members[i].Lag = lagOf(c.Members[i].RawLag) + } + return c, nil + } + if last == nil { + last = fmt.Errorf("no hosts to ask") + } + return Cluster{}, fmt.Errorf("no member's Patroni answered: %w", last) +} + +// lagOf reads Patroni's lag: a number of bytes, or "unknown". +func lagOf(raw json.RawMessage) int64 { + if len(raw) == 0 { + return 0 + } + var n int64 + if json.Unmarshal(raw, &n) == nil { + return n + } + return -1 +} diff --git a/internal/store/pghosts.go b/internal/store/pghosts.go new file mode 100644 index 00000000..55d002a1 --- /dev/null +++ b/internal/store/pghosts.go @@ -0,0 +1,44 @@ +package store + +import ( + "fmt" + "net/url" + "strings" +) + +// SplitPostgres reads a PostgreSQL address that may name several hosts — +// "postgres://u:p@h1:5432,h2:5432/db?…", the members of a replicated cluster, +// as pgx and libpq both accept — into its hosts, their ports, and the rest of +// the address with one placeholder host, which url.Parse can read. url.Parse +// cannot read the list itself: it took everything before the last colon for +// one host name. +func SplitPostgres(dsn string) (hosts, ports []string, rest *url.URL, err error) { + scheme, after, ok := strings.Cut(dsn, "://") + if !ok { + return nil, nil, nil, fmt.Errorf("the store address is not a URL") + } + authority, tail := after, "" + if i := strings.IndexAny(after, "/?"); i >= 0 { + authority, tail = after[:i], after[i:] + } + userinfo, hostlist := "", authority + if i := strings.LastIndexByte(authority, '@'); i >= 0 { + userinfo, hostlist = authority[:i+1], authority[i+1:] + } + for _, hp := range strings.Split(hostlist, ",") { + u, err := url.Parse("x://" + hp) + if err != nil || u.Hostname() == "" { + return nil, nil, nil, fmt.Errorf("the store address has a host that is not one: %q", hp) + } + port := u.Port() + if port == "" { + port = "5432" + } + hosts, ports = append(hosts, u.Hostname()), append(ports, port) + } + rest, err = url.Parse(scheme + "://" + userinfo + "host" + tail) + if err != nil { + return nil, nil, nil, fmt.Errorf("the store address is not a URL") + } + return hosts, ports, rest, nil +} diff --git a/internal/store/postgres_test.go b/internal/store/postgres_test.go index 32465de3..273caf98 100644 --- a/internal/store/postgres_test.go +++ b/internal/store/postgres_test.go @@ -252,3 +252,27 @@ func TestPostgresBucketsUsageByEveryGrouping(t *testing.T) { } } } + +func TestPostgresMakesAnAdministrator(t *testing.T) { + // init --postgres and --patroni create the first administrator straight + // in PostgreSQL, which refused the bool SetAdmin sent. + ctx := context.Background() + s := openPG(t) + if err := s.CreateUser(ctx, User{ID: "u1", Email: "a@x", Name: "a", PasswordHash: "h", + State: UserActive, CreatedAt: time.Now()}); err != nil { + t.Fatal(err) + } + if err := s.SetAdmin(ctx, "u1", true); err != nil { + t.Fatalf("making an administrator: %v", err) + } + u, err := s.UserByEmail(ctx, "a@x") + if err != nil || !u.IsAdmin { + t.Fatalf("%+v, %v", u, err) + } + if err := s.SetAdmin(ctx, "u1", false); err != nil { + t.Fatal(err) + } + if u, _ := s.UserByEmail(ctx, "a@x"); u.IsAdmin { + t.Error("still an administrator") + } +} diff --git a/internal/store/users.go b/internal/store/users.go index 2a87f39a..72bd5a1c 100644 --- a/internal/store/users.go +++ b/internal/store/users.go @@ -175,7 +175,14 @@ func (s *DB) SetUserState(ctx context.Context, id string, state UserState) error // account is what closes sessions, and that is the lever for someone who should // not be here at all. func (s *DB) SetAdmin(ctx context.Context, id string, admin bool) error { - res, err := s.db.ExecContext(ctx, `UPDATE users SET is_admin = ? WHERE id = ?`, admin, id) + // An integer, not a bool: the column is an integer on both dialects, + // and PostgreSQL refuses a bool for it where SQLite took one — so making + // an administrator worked in every test and failed on a fresh install. + flag := 0 + if admin { + flag = 1 + } + res, err := s.db.ExecContext(ctx, `UPDATE users SET is_admin = ? WHERE id = ?`, flag, id) if err != nil { return err }