From 3680e5b5bc29d342759037299e1efb1c0bece020 Mon Sep 17 00:00:00 2001 From: Josh Siviour Date: Fri, 28 Aug 2026 15:19:35 +1000 Subject: [PATCH] feat: render predastore admin_port and gate deploys on /readyz (#936) * feat: render predastore admin_port and consult /readyz in health probe Spinifex now allocates predastore's admin listener port (8660, after the existing gate/blob/meta bases) and renders admin_port into both the single-node and multi-node predastore.toml templates. The daemon's health probe consults /readyz over that port on the two weak paths (single-host gate fallback and hosts running no meta node), falling back to the prior behavior when no admin port is configured or the listener refuses. * build: pin predastore to the release carrying Host.AdminPort The admin port field this branch reads is new in predastore, so CI could not compile against the pinned version: go.work hid the gap locally by resolving predastore to the checkout instead of the module. --- cmd/spinifex/cmd/admin.go | 3 + .../cmd/templates/predastore-multinode.toml | 1 + cmd/spinifex/cmd/templates/predastore.toml | 1 + go.mod | 2 +- go.sum | 4 +- spinifex/admin/admin.go | 13 ++ spinifex/daemon/health.go | 87 +++++++- spinifex/daemon/health_test.go | 206 ++++++++++++++++++ 8 files changed, 312 insertions(+), 5 deletions(-) diff --git a/cmd/spinifex/cmd/admin.go b/cmd/spinifex/cmd/admin.go index 4c7d9f430..bd34dbe2a 100644 --- a/cmd/spinifex/cmd/admin.go +++ b/cmd/spinifex/cmd/admin.go @@ -1601,6 +1601,7 @@ func runAdminInit(cmd *cobra.Command, args []string) { ClusterName: clusterName, PredastoreHostID: predastoreHostID, + PredastoreAdminPort: admin.PredastoreAdminPort, CompactionIntervalSeconds: compactionInterval, Services: services, @@ -1902,6 +1903,7 @@ func runAdminInitMultiNode(cmd *cobra.Command, accessKey, secretKey, accountID, ClusterName: clusterName, PredastoreHostID: predastoreHostID, + PredastoreAdminPort: admin.PredastoreAdminPort, CompactionIntervalSeconds: compactionInterval, Services: services, RemoteNodes: buildRemoteNodes(allNodes, node, northstarConfigPath), @@ -2461,6 +2463,7 @@ func runAdminJoin(cmd *cobra.Command, args []string) { ClusterName: creds.ClusterName, PredastoreHostID: predastoreHostID, + PredastoreAdminPort: admin.PredastoreAdminPort, CompactionIntervalSeconds: compactionInterval, Services: services, RemoteNodes: buildRemoteNodes(statusResp.Nodes, node, northstarConfigPath), diff --git a/cmd/spinifex/cmd/templates/predastore-multinode.toml b/cmd/spinifex/cmd/templates/predastore-multinode.toml index 6cb9fb7ff..622c3413e 100644 --- a/cmd/spinifex/cmd/templates/predastore-multinode.toml +++ b/cmd/spinifex/cmd/templates/predastore-multinode.toml @@ -36,6 +36,7 @@ data_dir = "{{$.PredastoreDataDir}}" tls_cert = "{{$.ConfigDir}}/server.pem" tls_key = "{{$.ConfigDir}}/server.key" encryption_key = "{{$.ConfigDir}}/predastore/encryption.key" +admin_port = {{$.AdminPort}} {{- range $.ClusterNodes}} {{- if eq .HostID $host.ID}} diff --git a/cmd/spinifex/cmd/templates/predastore.toml b/cmd/spinifex/cmd/templates/predastore.toml index 8379c34b4..dcb165b93 100644 --- a/cmd/spinifex/cmd/templates/predastore.toml +++ b/cmd/spinifex/cmd/templates/predastore.toml @@ -32,6 +32,7 @@ data_dir = "{{.PredastoreDataDir}}" tls_cert = "{{.ConfigDir}}/server.pem" tls_key = "{{.ConfigDir}}/server.key" encryption_key = "{{.ConfigDir}}/predastore/encryption.key" +admin_port = {{.PredastoreAdminPort}} # The gate serves the S3 API. Without one the host answers no S3 request. # S3 is the public plane, so it binds every interface while the host's diff --git a/go.mod b/go.mod index 9171db20e..c7b6462b1 100644 --- a/go.mod +++ b/go.mod @@ -23,7 +23,7 @@ require ( github.com/miekg/dns v1.1.73 github.com/mulgadc/bluebottle v1.18.1-0.20260827045410-e7d5cf022748 github.com/mulgadc/northstar v1.18.0 - github.com/mulgadc/predastore v1.18.1-0.20260826045109-26ce1b091112 + github.com/mulgadc/predastore v1.18.1-0.20260828045402-1dba12d04866 github.com/mulgadc/viperblock v1.18.0 github.com/nats-io/nats-server/v2 v2.14.5 github.com/nats-io/nats.go v1.53.1 diff --git a/go.sum b/go.sum index f71aaeec6..7fff65eee 100644 --- a/go.sum +++ b/go.sum @@ -1331,8 +1331,8 @@ github.com/mulgadc/bluebottle v1.18.1-0.20260827045410-e7d5cf022748 h1:kvoYwPtGX github.com/mulgadc/bluebottle v1.18.1-0.20260827045410-e7d5cf022748/go.mod h1:mvvV4d2w+kGLXxO1ymptNlxqUlNJKB2oAhk6PkpVLj8= github.com/mulgadc/northstar v1.18.0 h1:ksRjIKE6qtrw/ZoYaJh8JRBh+xdnuDw8FsmFLFCiXbg= github.com/mulgadc/northstar v1.18.0/go.mod h1:z6iol77R1vsIQyIBc1JQboUPhSJ6eG4bipzfbZZbLEM= -github.com/mulgadc/predastore v1.18.1-0.20260826045109-26ce1b091112 h1:eCzKJDL1lmw6DNwjbDCQBoT3ZHIsYE1lMj5qitjySrY= -github.com/mulgadc/predastore v1.18.1-0.20260826045109-26ce1b091112/go.mod h1:xDJMcTuHTG4aKfO2PrSlkpID/UJrvYj1yfLZ0Pi8CSU= +github.com/mulgadc/predastore v1.18.1-0.20260828045402-1dba12d04866 h1:xR8X8FR2vHxjT9y47zT4/ka40d3uLD+qCzwi4huiAzQ= +github.com/mulgadc/predastore v1.18.1-0.20260828045402-1dba12d04866/go.mod h1:Az85JvrDqSk5pIF0A3Y06GwCmx53tktRSXEUHZIlsZ0= github.com/mulgadc/viperblock v1.18.0 h1:u6yu5lPfZ+RSmfzlg1OXI3whlRBnGp8t685xUzzxlns= github.com/mulgadc/viperblock v1.18.0/go.mod h1:OjqMFZba0TJl9zcW2oKmz2/MrApIxRIFNcafiv5vTkM= github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 h1:C3w9PqII01/Oq1c1nUAm88MOHcQC9l5mIlSMApZMrHA= diff --git a/spinifex/admin/admin.go b/spinifex/admin/admin.go index 241e40fdf..514eeb2ce 100644 --- a/spinifex/admin/admin.go +++ b/spinifex/admin/admin.go @@ -63,6 +63,11 @@ type ConfigSettings struct { // Empty → callers fall back to BindIP. AdvertiseIP string + // PredastoreAdminPort is rendered as predastore.toml's host-level + // admin_port, serving /healthz and /readyz on the host's cluster + // bind_addr. Zero runs no admin listener. + PredastoreAdminPort int + // Cluster settings ClusterBindIP string ClusterRoutes []string @@ -978,6 +983,12 @@ const ( predastoreMetaBasePort = 7660 ) +// PredastoreAdminPort is the port predastore's admin listener serves +// /healthz and /readyz on. It continues the base ports above's progression by +// 1000, so it can never collide with a gate, blob or meta node on the same +// host; exported so single-node config generation can render it too. +const PredastoreAdminPort = 8660 + // PredastoreTopology derives the cluster nodes for a set of machines: each // machine hosts a gate, one blob node and one meta node. Node IDs are unique // across the whole file, so gates take 1..n, blobs n+1..2n and metas 2n+1..3n. @@ -1027,6 +1038,7 @@ func GenerateMultiNodePredastoreConfig(templateStr string, nodes []PredastoreNod NorthstarAccessKey string NorthstarSecretKey string NorthstarBucket string + AdminPort int }{ Nodes: nodes, ClusterNodes: PredastoreTopology(nodes), AccessKey: accessKey, SecretKey: secretKey, Region: region, @@ -1037,6 +1049,7 @@ func GenerateMultiNodePredastoreConfig(templateStr string, nodes []PredastoreNod NorthstarAccessKey: northstar.AccessKey, NorthstarSecretKey: northstar.SecretKey, NorthstarBucket: northstar.Bucket, + AdminPort: PredastoreAdminPort, } tmpl, err := template.New("predastore-multinode").Parse(templateStr) diff --git a/spinifex/daemon/health.go b/spinifex/daemon/health.go index 30de76507..82540a9b5 100644 --- a/spinifex/daemon/health.go +++ b/spinifex/daemon/health.go @@ -6,6 +6,7 @@ import ( "fmt" "log/slog" "net" + "net/http" "path/filepath" "strconv" "sync" @@ -50,6 +51,22 @@ var gateDialFn = func(ctx context.Context, network, addr string) (net.Conn, erro return d.DialContext(ctx, network, addr) } +// readyzFn performs the /readyz HTTP GET against a predastore admin listener +// and reports its status code, indirected so tests can stub the transport and +// exercise every verdict without a live listener. +var readyzFn = func(ctx context.Context, addr string) (int, error) { + req, err := http.NewRequestWithContext(ctx, http.MethodGet, "http://"+addr+"/readyz", nil) + if err != nil { + return 0, err + } + resp, err := http.DefaultClient.Do(req) + if err != nil { + return 0, err + } + defer func() { _ = resp.Body.Close() }() + return resp.StatusCode, nil +} + // probeFunc reports one service's real health, in place of the "ok" every // service otherwise reports simply for being named in spinifex.toml. type probeFunc func(ctx context.Context, d *Daemon) string @@ -128,6 +145,14 @@ func computePredastoreHealth(ctx context.Context, d *Daemon) string { return predastoreHealthUnreachable } if len(nodes) == 0 { + // This host runs predastore but no meta node — nothing raft-local to + // probe. Its admin listener, if named, still answers for the roles it + // does run, so ask that rather than reporting ok unconditionally. + if cfg != nil { + if verdict, ok := probeReadyz(ctx, cfg, hostID); ok { + return verdict + } + } return predastoreHealthOK } @@ -176,11 +201,17 @@ func predastoreIsSingleHost(cfg *pds.Config, hostID pds.HostID) bool { return true } -// probeLocalGate reports predastore's health on a single-host install by -// connecting to the local S3 gate — its one cross-process listener. A host +// probeLocalGate reports predastore's health on a single-host install. Its +// admin listener, when named, is asked first since it observes the meta and +// blob roles the pipe-only meta node cannot expose; otherwise this falls back +// to connecting to the local S3 gate, its one cross-process listener. A host // running no gate exposes no socket to probe and reports ok rather than // failing a service it cannot answer for on a channel that cannot exist. func probeLocalGate(ctx context.Context, d *Daemon, cfg *pds.Config, hostID pds.HostID) string { + if verdict, ok := probeReadyz(ctx, cfg, hostID); ok { + return verdict + } + addr, ok := localGateAddr(cfg, hostID) if !ok { return predastoreHealthOK @@ -198,6 +229,58 @@ func probeLocalGate(ctx context.Context, d *Daemon, cfg *pds.Config, hostID pds. return predastoreHealthOK } +// probeReadyz consults predastore's /readyz on this host's admin listener, +// when the host names one. 200 maps to ok; 503 maps to unreachable, the +// closest existing verdict for "the process is running but cannot serve". +// ok is false whenever no admin port is configured, the listener could not be +// reached at all, or it answered with neither status: in every case the +// caller must fall back to its own probe, so that an older predastore build +// with no admin listener is never read as unhealthy. +func probeReadyz(ctx context.Context, cfg *pds.Config, hostID pds.HostID) (string, bool) { + addr, ok := localAdminAddr(cfg, hostID) + if !ok { + return "", false + } + + probeCtx, cancel := context.WithTimeout(ctx, predastoreProbeTimeout) + defer cancel() + + status, err := readyzFn(probeCtx, addr) + if err != nil { + slog.Debug("predastore health probe: admin listener unreachable, falling back", "addr", addr, "err", err) + return "", false + } + switch status { + case http.StatusOK: + return predastoreHealthOK, true + case http.StatusServiceUnavailable: + return predastoreHealthUnreachable, true + default: + slog.Warn("predastore health probe: unexpected /readyz status, falling back", "addr", addr, "status", status) + return "", false + } +} + +// localAdminAddr is the dial address of this host's predastore admin +// listener, and whether the host names one. A wildcard or empty bind_addr is +// dialled on the loopback, since the daemon and predastore share the host. +func localAdminAddr(cfg *pds.Config, hostID pds.HostID) (string, bool) { + for _, h := range cfg.Hosts { + if h.ID != hostID { + continue + } + if h.AdminPort == 0 { + return "", false + } + host := pds.HostBindAddr(h) + if host == "" || host == "0.0.0.0" || host == "::" { + host = "127.0.0.1" + } + return net.JoinHostPort(host, strconv.Itoa(h.AdminPort)), true + } + return "", false +} + // localGateAddr is the dial address of the S3 gate on this host, and whether // the host runs one. A wildcard or empty bind is dialled on the loopback, // since the daemon and predastore share the host. diff --git a/spinifex/daemon/health_test.go b/spinifex/daemon/health_test.go index e7a0bfcff..e7aef2691 100644 --- a/spinifex/daemon/health_test.go +++ b/spinifex/daemon/health_test.go @@ -8,7 +8,9 @@ package daemon import ( "context" "crypto/x509" + "fmt" "net" + "net/http" "os" "path/filepath" "testing" @@ -97,6 +99,81 @@ role = "meta" port = 7660 ` +// singleHostWithAdminTOML is singleHostTOML plus an admin_port, so the probe +// can reach /readyz instead of falling back to the bare gate dial. +const singleHostWithAdminTOML = ` +version = 1 +region = "ap-southeast-2" + +[rs] +data = 1 +parity = 0 + +[[host]] +id = 1 +bind_addr = "127.0.0.1" +addr = "127.0.0.1" +data_dir = "/var/lib/spinifex/predastore/cluster" +admin_port = 8660 + +[[host.node]] +id = 1 +role = "gate" +port = 8443 +bind_addr = "0.0.0.0" + +[[host.node]] +id = 2 +role = "blob" +port = 6660 + +[[host.node]] +id = 3 +role = "meta" +port = 7660 +` + +// gateOnlyWithAdminTOML pins nothing but a gate to host 1 — the other weak +// path, a host with no meta node at all — and names an admin_port so the +// probe has somewhere to ask instead of reporting ok unconditionally. +const gateOnlyWithAdminTOML = ` +version = 1 +region = "ap-southeast-2" + +[rs] +data = 1 +parity = 0 + +[[host]] +id = 1 +bind_addr = "0.0.0.0" +addr = "10.0.0.1" +data_dir = "/var/lib/spinifex/predastore/cluster" +admin_port = 8660 + +[[host.node]] +id = 1 +role = "gate" +port = 8443 +bind_addr = "0.0.0.0" + +[[host]] +id = 2 +bind_addr = "0.0.0.0" +addr = "10.0.0.2" +data_dir = "/var/lib/spinifex/predastore/cluster" + +[[host.node]] +id = 2 +role = "blob" +port = 6660 + +[[host.node]] +id = 3 +role = "meta" +port = 7660 +` + // newHealthTestDaemon writes healthTestTOML and a CA cert under a temp // config dir and returns a Daemon whose predastore host id is hostID. func newHealthTestDaemon(t *testing.T, hostID int) *Daemon { @@ -139,6 +216,20 @@ func setGateDialFn(t *testing.T, fn func(context.Context, string, string) (net.C return &calls } +// setReadyzFn stubs readyzFn for the test's lifetime and returns a pointer to +// a call counter, mirroring setGateDialFn. +func setReadyzFn(t *testing.T, fn func(context.Context, string) (int, error)) *int { + t.Helper() + calls := 0 + orig := readyzFn + readyzFn = func(ctx context.Context, addr string) (int, error) { + calls++ + return fn(ctx, addr) + } + t.Cleanup(func() { readyzFn = orig }) + return &calls +} + // setNodeStatusFn stubs nodeStatusFn for the test's lifetime, restoring the // real pds.NodeStatus on cleanup, and returns a pointer to a call counter. func setNodeStatusFn(t *testing.T, fn func(context.Context, *pds.Config, pds.NodeID, *x509.CertPool) (pds.Status, error)) *int { @@ -269,6 +360,121 @@ func TestComputePredastoreHealth_SingleHostGateUnreachable(t *testing.T) { assert.Equal(t, predastoreHealthUnreachable, got) } +func TestComputePredastoreHealth_ReadyzReadyMapsToOK(t *testing.T) { + d := newHealthTestDaemonTOML(t, 1, singleHostWithAdminTOML) + + readyzCalls := setReadyzFn(t, func(context.Context, string) (int, error) { + return http.StatusOK, nil + }) + gateCalls := setGateDialFn(t, func(context.Context, string, string) (net.Conn, error) { + return nil, assert.AnError + }) + + got := computePredastoreHealth(t.Context(), d) + assert.Equal(t, predastoreHealthOK, got) + assert.Equal(t, 1, *readyzCalls) + assert.Equal(t, 0, *gateCalls, "a ready /readyz must short-circuit the gate dial") +} + +func TestComputePredastoreHealth_ReadyzUnavailableMapsToUnreachable(t *testing.T) { + d := newHealthTestDaemonTOML(t, 1, singleHostWithAdminTOML) + + setReadyzFn(t, func(context.Context, string) (int, error) { + return http.StatusServiceUnavailable, nil + }) + + got := computePredastoreHealth(t.Context(), d) + assert.Equal(t, predastoreHealthUnreachable, got) +} + +func TestComputePredastoreHealth_NoAdminPortFallsBackToGateDial(t *testing.T) { + // singleHostTOML names no admin_port, so the probe must never reach for + // /readyz and must fall back to the pre-existing gate dial exactly as + // before this feature existed. + d := newHealthTestDaemonTOML(t, 1, singleHostTOML) + + readyzCalls := setReadyzFn(t, func(context.Context, string) (int, error) { + return http.StatusOK, nil + }) + gateCalls := setGateDialFn(t, func(context.Context, string, string) (net.Conn, error) { + server, client := net.Pipe() + t.Cleanup(func() { _ = server.Close() }) + return client, nil + }) + + got := computePredastoreHealth(t.Context(), d) + assert.Equal(t, predastoreHealthOK, got) + assert.Equal(t, 0, *readyzCalls, "no admin_port configured must never dial /readyz") + assert.Equal(t, 1, *gateCalls) +} + +func TestComputePredastoreHealth_ReadyzConnectionRefusedFallsBackToGateDial(t *testing.T) { + // A refused admin port must read as "not running this build's listener", + // not as unhealthy — a mixed fleet with an older predastore must still + // deploy off the gate dial's verdict. + d := newHealthTestDaemonTOML(t, 1, singleHostWithAdminTOML) + + setReadyzFn(t, func(context.Context, string) (int, error) { + return 0, &net.OpError{Op: "dial", Err: fmt.Errorf("connection refused")} + }) + gateCalls := setGateDialFn(t, func(context.Context, string, string) (net.Conn, error) { + server, client := net.Pipe() + t.Cleanup(func() { _ = server.Close() }) + return client, nil + }) + + got := computePredastoreHealth(t.Context(), d) + assert.Equal(t, predastoreHealthOK, got) + assert.Equal(t, 1, *gateCalls, "a refused admin port must fall back to the gate dial") +} + +func TestComputePredastoreHealth_NoMetaNodeOnHostAsksAdminListener(t *testing.T) { + // The other weak path: a host running predastore but no meta node at all. + // With an admin_port named, it must ask /readyz rather than reporting ok + // unconditionally. + d := newHealthTestDaemonTOML(t, 1, gateOnlyWithAdminTOML) + + readyzCalls := setReadyzFn(t, func(context.Context, string) (int, error) { + return http.StatusServiceUnavailable, nil + }) + + got := computePredastoreHealth(t.Context(), d) + assert.Equal(t, predastoreHealthUnreachable, got) + assert.Equal(t, 1, *readyzCalls) +} + +func TestProbePredastore_CachesReadyzWithinTTL(t *testing.T) { + d := newHealthTestDaemonTOML(t, 1, singleHostWithAdminTOML) + + origTTL := predastoreHealthCacheTTL + predastoreHealthCacheTTL = time.Minute + t.Cleanup(func() { predastoreHealthCacheTTL = origTTL }) + + readyzCalls := setReadyzFn(t, func(context.Context, string) (int, error) { + return http.StatusOK, nil + }) + + first := probePredastore(t.Context(), d) + second := probePredastore(t.Context(), d) + + assert.Equal(t, predastoreHealthOK, first) + assert.Equal(t, predastoreHealthOK, second) + assert.Equal(t, 1, *readyzCalls, "a second poll within the TTL must not re-probe /readyz") +} + +func TestLocalAdminAddr(t *testing.T) { + // A wildcard host bind is dialled on the loopback the daemon shares. + cfg := &pds.Config{Hosts: []pds.HostConfig{{ID: 1, BindAddr: "0.0.0.0", AdminPort: 8660}}} + addr, ok := localAdminAddr(cfg, 1) + require.True(t, ok) + assert.Equal(t, "127.0.0.1:8660", addr) + + // A host naming no admin_port has nothing to dial. + noAdmin := &pds.Config{Hosts: []pds.HostConfig{{ID: 1, BindAddr: "0.0.0.0"}}} + _, ok = localAdminAddr(noAdmin, 1) + assert.False(t, ok) +} + func TestLocalGateAddr(t *testing.T) { // A wildcard gate bind is dialled on the loopback the daemon shares. cfg := &pds.Config{Hosts: []pds.HostConfig{{