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.
This commit is contained in:
Josh Siviour
2026-08-28 15:19:35 +10:00
committed by GitHub
parent 1d0a53a4b8
commit 3680e5b5bc
8 changed files with 312 additions and 5 deletions
+3
View File
@@ -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),
@@ -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}}
@@ -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
+1 -1
View File
@@ -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
+2 -2
View File
@@ -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=
+13
View File
@@ -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)
+85 -2
View File
@@ -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.
+206
View File
@@ -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{{