From 560b437f3b472c207b0315be95e2487dcc332146 Mon Sep 17 00:00:00 2001 From: Josh Siviour Date: Thu, 13 Aug 2026 15:24:54 +1000 Subject: [PATCH] fix(health): probe predastore raft state instead of echoing the config (#786) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit /health reported service_health.predastore == "ok" for every node simply because predastore was named in spinifex.toml, without contacting it — on env19 all three nodes stayed green for the whole time raft had no leader and volume creates were failing. A probe registry now backs /health's predastore entry with a real raft status dial (ok/no_leader/unreachable, cached briefly and bounded by a short timeout), while every other service keeps its unprobed "ok" default unchanged. spx admin storage status had the same root problem from the other side: it dialed HTTPS /health and /status endpoints predastore's meta nodes never served (they speak an opcode-multiplexed RPC), so every meta node always reported unhealthy with blank raft fields. It now dials the same raft status RPC predastore#82 exposes, verified against the cluster CA, with the JSON output shape unchanged. --- go.mod | 2 +- go.sum | 8 +- spinifex/daemon/daemon.go | 8 +- spinifex/daemon/daemon_handlers_storage.go | 4 +- spinifex/daemon/health.go | 182 ++++++++++++++++++ spinifex/daemon/health_test.go | 203 +++++++++++++++++++++ spinifex/gateway/gateway.go | 12 +- spinifex/gateway/spinifex.go | 2 +- spinifex/gateway/spx/storage.go | 117 ++++++------ spinifex/gateway/spx/storage_test.go | 92 ++++------ spinifex/services/awsgw/awsgw.go | 12 ++ spinifex/utils/nats.go | 24 ++- 12 files changed, 532 insertions(+), 134 deletions(-) create mode 100644 spinifex/daemon/health.go create mode 100644 spinifex/daemon/health_test.go diff --git a/go.mod b/go.mod index 63b8f28c3..94149bc6c 100644 --- a/go.mod +++ b/go.mod @@ -23,7 +23,7 @@ require ( github.com/miekg/dns v1.1.72 github.com/mulgadc/bluebottle v1.16.1-0.20260812012137-683e88e465e3 github.com/mulgadc/northstar v1.16.1-0.20260812012604-5d7a622ad059 - github.com/mulgadc/predastore v1.16.1-0.20260812012531-12c90bd2e14a + github.com/mulgadc/predastore v1.16.1-0.20260813022738-0b4392b43cae github.com/mulgadc/viperblock v1.16.1-0.20260812124023-7e0df2323c75 github.com/nats-io/nats-server/v2 v2.14.4 github.com/nats-io/nats.go v1.52.0 diff --git a/go.sum b/go.sum index 3d2790804..f347e2111 100644 --- a/go.sum +++ b/go.sum @@ -1323,12 +1323,8 @@ github.com/mulgadc/bluebottle v1.16.1-0.20260812012137-683e88e465e3 h1:EJ8JMQB9Y github.com/mulgadc/bluebottle v1.16.1-0.20260812012137-683e88e465e3/go.mod h1:bhJitsX6VKkbmQ1dVh3vPUmoSeQsvtn5Jm4D8hUyq2s= github.com/mulgadc/northstar v1.16.1-0.20260812012604-5d7a622ad059 h1:CK3Ed0Rg5dGvpSeMFTGpE1nf+390Rbf5hePOcnm1+kM= github.com/mulgadc/northstar v1.16.1-0.20260812012604-5d7a622ad059/go.mod h1:vyfhxWlBlAnJfUeSEt+z/Z6KrfReC/uzZM3WTllorYM= -github.com/mulgadc/predastore v1.16.1-0.20260812012531-12c90bd2e14a h1:XSvL4FucGZaEDPUXCG/mUshfFskGO9tE69SYMJJ28L4= -github.com/mulgadc/predastore v1.16.1-0.20260812012531-12c90bd2e14a/go.mod h1:S5vcni7R3VYi3nccelMagKbWdX58WtbKf0I1R/7MmNI= -github.com/mulgadc/viperblock v1.16.1-0.20260812012547-bb1225320f0c h1:5PCU4696Gelc+4X4K+E+EzNVsyoLLWEiF/TcsIaAGi4= -github.com/mulgadc/viperblock v1.16.1-0.20260812012547-bb1225320f0c/go.mod h1:8NT5u+udY1CsoKz/JAPY2SSMcQ98Omi+42r8BkoA4FY= -github.com/mulgadc/viperblock v1.16.1-0.20260812033921-fc825c93eff6 h1:QfdcaqnrDDzsO4neJNX7BdL8vzyga/WPn4A/04QZcVw= -github.com/mulgadc/viperblock v1.16.1-0.20260812033921-fc825c93eff6/go.mod h1:8NT5u+udY1CsoKz/JAPY2SSMcQ98Omi+42r8BkoA4FY= +github.com/mulgadc/predastore v1.16.1-0.20260813022738-0b4392b43cae h1:itP5arN88Xv/xe4kI/5TWC/fkawnOeIR8MF1AL7aC48= +github.com/mulgadc/predastore v1.16.1-0.20260813022738-0b4392b43cae/go.mod h1:S5vcni7R3VYi3nccelMagKbWdX58WtbKf0I1R/7MmNI= github.com/mulgadc/viperblock v1.16.1-0.20260812124023-7e0df2323c75 h1:eqB0VOuFPldBHuQ7ELYZHSRxq+3MXbMMi9D5bfVUwe8= github.com/mulgadc/viperblock v1.16.1-0.20260812124023-7e0df2323c75/go.mod h1:8NT5u+udY1CsoKz/JAPY2SSMcQ98Omi+42r8BkoA4FY= github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 h1:C3w9PqII01/Oq1c1nUAm88MOHcQC9l5mIlSMApZMrHA= diff --git a/spinifex/daemon/daemon.go b/spinifex/daemon/daemon.go index ca36cea85..4b1099f2e 100644 --- a/spinifex/daemon/daemon.go +++ b/spinifex/daemon/daemon.go @@ -184,6 +184,9 @@ type Daemon struct { startTime time.Time configPath string + // predastoreHealth caches the /health predastore probe's verdict; see health.go. + predastoreHealth predastoreHealthCache + // System credentials for ALB agent SigV4 auth (loaded from system-credentials.json) systemAccessKey string systemSecretKey string @@ -2307,10 +2310,7 @@ func (d *Daemon) ClusterManager() error { return } - serviceHealth := make(map[string]string) - for _, svc := range d.config.GetServices() { - serviceHealth[svc] = "ok" - } + serviceHealth := d.probeServiceHealth(r.Context()) if !d.config.HasService("nats") { if d.natsConn != nil && d.natsConn.IsConnected() { serviceHealth["nats"] = "remote_ok" diff --git a/spinifex/daemon/daemon_handlers_storage.go b/spinifex/daemon/daemon_handlers_storage.go index 9f0b113ad..c9b5ce74f 100644 --- a/spinifex/daemon/daemon_handlers_storage.go +++ b/spinifex/daemon/daemon_handlers_storage.go @@ -3,7 +3,6 @@ package daemon import ( "log/slog" "os" - "path/filepath" "github.com/mulgadc/spinifex/spinifex/types" "github.com/nats-io/nats.go" @@ -55,8 +54,7 @@ type predastoreBucket struct { // handleStorageConfig responds with parsed predastore config (topology only, no creds). // Used by the gateway to build the GetStorageStatus response. func (d *Daemon) handleStorageConfig(msg *nats.Msg) { - configDir := filepath.Dir(d.configPath) - predastorePath := filepath.Join(configDir, "predastore", "predastore.toml") + predastorePath := predastoreConfigPath(d.configPath) data, err := os.ReadFile(predastorePath) if err != nil { diff --git a/spinifex/daemon/health.go b/spinifex/daemon/health.go new file mode 100644 index 000000000..7c4974829 --- /dev/null +++ b/spinifex/daemon/health.go @@ -0,0 +1,182 @@ +package daemon + +import ( + "context" + "crypto/x509" + "fmt" + "log/slog" + "path/filepath" + "sync" + "time" + + pds "github.com/mulgadc/predastore" + "github.com/mulgadc/spinifex/spinifex/utils" +) + +// Predastore health verdicts, as reported under service_health.predastore in +// the /health response. +const ( + predastoreHealthOK = "ok" + predastoreHealthNoLeader = "no_leader" + predastoreHealthUnreachable = "unreachable" +) + +// predastoreProbeTimeout bounds every dial a /health request makes to this +// host's own meta node(s), so a wedged or unreachable peer can never stall +// the handler. A var, not a const, so tests can shrink it to exercise the +// timeout path without a multi-second sleep. +var predastoreProbeTimeout = 2 * time.Second + +// predastoreHealthCacheTTL bounds how stale a cached predastore verdict may +// be. /health is a monitoring target polled far more often than raft state +// changes, and a fresh QUIC dial per poll would put that polling load on the +// raft plane; a few seconds of staleness is a good trade for that. A var for +// the same reason as predastoreProbeTimeout. +var predastoreHealthCacheTTL = 5 * time.Second + +// nodeStatusFn is pds.NodeStatus, indirected so tests can stub the raft dial +// and exercise every verdict — including the timeout path — without a live +// predastore process. +var nodeStatusFn = pds.NodeStatus + +// 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 + +// serviceProbes maps a service name to the probe that reports its real +// health. A service absent from this map keeps reporting "ok" as soon as it +// is named in spinifex.toml: service_health. == "ok" gates several +// ansible bootstrap playbooks, and changing that default in the same change +// that adds the first probe would stall them for a reason unrelated to any +// bug in those services. Probes for the other services are a separate piece +// of work. +var serviceProbes = map[string]probeFunc{ + "predastore": probePredastore, +} + +// probeServiceHealth reports every service named in spinifex.toml: a probed +// service returns its probe's verdict, and everything else returns "ok" — +// the same restatement of the config /health has always reported for +// services with no probe. +func (d *Daemon) probeServiceHealth(ctx context.Context) map[string]string { + services := d.config.GetServices() + health := make(map[string]string, len(services)) + for _, svc := range services { + if probe, ok := serviceProbes[svc]; ok { + health[svc] = probe(ctx, d) + continue + } + health[svc] = "ok" + } + return health +} + +// predastoreHealthCache holds the last predastore verdict this node computed, +// so concurrent or rapid /health polls do not each dial the local meta +// node(s) themselves. +type predastoreHealthCache struct { + mu sync.Mutex + at time.Time + result string +} + +// probePredastore reports the raft health of the meta node(s) MetaNodesOnHost +// pins to this host, serving a cached verdict when it is fresh enough. +func probePredastore(ctx context.Context, d *Daemon) string { + d.predastoreHealth.mu.Lock() + if !d.predastoreHealth.at.IsZero() && time.Since(d.predastoreHealth.at) < predastoreHealthCacheTTL { + result := d.predastoreHealth.result + d.predastoreHealth.mu.Unlock() + return result + } + d.predastoreHealth.mu.Unlock() + + result := computePredastoreHealth(ctx, d) + + d.predastoreHealth.mu.Lock() + d.predastoreHealth.result = result + d.predastoreHealth.at = time.Now() + d.predastoreHealth.mu.Unlock() + + return result +} + +// computePredastoreHealth dials every meta node MetaNodesOnHost pins to this +// host and reports the worst of them: unreachable beats no_leader beats ok, +// since any one of them failing is this host's predastore service failing. +// A host that runs no meta node at all — a pure gate/blob host, or a node +// that runs no predastore — has nothing local to probe and reports ok: that +// is not a failure of this host's own service, and the raft quorum as a +// whole is what `spx admin storage status` reports on, across every host +// that does run a meta node. +func computePredastoreHealth(ctx context.Context, d *Daemon) string { + nodes, cfg, err := localMetaNodes(d) + if err != nil { + slog.Warn("predastore health probe: could not resolve local meta nodes", "err", err) + return predastoreHealthUnreachable + } + if len(nodes) == 0 { + return predastoreHealthOK + } + + rootCAs, err := loadPredastoreTrustRoot(d) + if err != nil { + slog.Warn("predastore health probe: could not load cluster CA", "err", err) + return predastoreHealthUnreachable + } + + probeCtx, cancel := context.WithTimeout(ctx, predastoreProbeTimeout) + defer cancel() + + verdict := predastoreHealthOK + for _, node := range nodes { + status, err := nodeStatusFn(probeCtx, cfg, node, rootCAs) + if err != nil { + slog.Debug("predastore health probe: node unreachable", "node", node, "err", err) + return predastoreHealthUnreachable + } + if status.Leader == "" { + verdict = predastoreHealthNoLeader + } + } + return verdict +} + +// localMetaNodes returns the meta node ids pinned to this daemon's predastore +// host, and the config they were parsed from. A node with no predastore host +// id configured — this node runs no predastore at all — returns an empty +// slice and no error, which computePredastoreHealth treats as "ok" rather +// than "unreachable": it is not this host's job to answer for a service it +// was never asked to run. +func localMetaNodes(d *Daemon) ([]pds.NodeID, *pds.Config, error) { + if d.config.Predastore.HostID <= 0 { + return nil, nil, nil + } + + cfgPath := predastoreConfigPath(d.configPath) + cfg, err := pds.LoadConfig(cfgPath) + if err != nil { + return nil, nil, fmt.Errorf("load predastore config %s: %w", cfgPath, err) + } + + nodes := pds.MetaNodesOnHost(cfg, pds.HostID(d.config.Predastore.HostID)) + return nodes, cfg, nil +} + +// predastoreConfigPath is where the predastore process reads its own config: +// a fixed "predastore/predastore.toml" alongside the daemon's own +// spinifex.toml, matching how the predastore service and handleStorageConfig +// both resolve it. +func predastoreConfigPath(daemonConfigPath string) string { + return filepath.Join(filepath.Dir(daemonConfigPath), "predastore", "predastore.toml") +} + +// loadPredastoreTrustRoot returns the cluster CA pool predastore's meta +// replicas are issued from — the same CA (d.config.NATS.CACert, conventionally +// /etc/spinifex/ca.pem) the daemon already trusts NATS peers with. +func loadPredastoreTrustRoot(d *Daemon) (*x509.CertPool, error) { + if d.config.NATS.CACert == "" { + return nil, fmt.Errorf("cluster CA not configured (nats.cacert)") + } + return utils.LoadCertPool(d.config.NATS.CACert) +} diff --git a/spinifex/daemon/health_test.go b/spinifex/daemon/health_test.go new file mode 100644 index 000000000..9ba9329bd --- /dev/null +++ b/spinifex/daemon/health_test.go @@ -0,0 +1,203 @@ +// Exercises unexported probe internals (serviceProbes, computePredastoreHealth, +// nodeStatusFn, predastoreHealthCache) that have no exported surface to drive +// them through. +// +//test:in-package +package daemon + +import ( + "context" + "crypto/x509" + "os" + "path/filepath" + "testing" + "time" + + pds "github.com/mulgadc/predastore" + "github.com/mulgadc/spinifex/spinifex/config" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// healthTestTOML pins one meta node (id 3) to host 1, and gives host 2 no +// meta node at all — the two shapes localMetaNodes has to tell apart. +const healthTestTOML = ` +version = 1 +region = "ap-southeast-2" + +[rs] +data = 1 +parity = 1 + +[[host]] +id = 1 +bind_addr = "0.0.0.0" +addr = "10.0.0.1" +data_dir = "/var/lib/spinifex/predastore/cluster" + +[[host.node]] +id = 1 +role = "gate" +port = 8443 + +[[host.node]] +id = 2 +role = "blob" +port = 6660 + +[[host.node]] +id = 3 +role = "meta" +port = 7660 + +[[host]] +id = 2 +bind_addr = "0.0.0.0" +addr = "10.0.0.2" +data_dir = "/var/lib/spinifex/predastore/cluster" + +[[host.node]] +id = 4 +role = "blob" +port = 6660 +` + +// 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 { + t.Helper() + dir := t.TempDir() + require.NoError(t, os.MkdirAll(filepath.Join(dir, "predastore"), 0o750)) + require.NoError(t, os.WriteFile(filepath.Join(dir, "predastore", "predastore.toml"), []byte(healthTestTOML), 0o600)) + + certPEM, _ := generateTestCert(t) + caPath := filepath.Join(dir, "ca.pem") + require.NoError(t, os.WriteFile(caPath, certPEM, 0o600)) + + return &Daemon{ + configPath: filepath.Join(dir, "spinifex.toml"), + config: &config.Config{ + Predastore: config.PredastoreConfig{HostID: hostID}, + NATS: config.NATSConfig{CACert: caPath}, + }, + } +} + +// 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 { + t.Helper() + calls := 0 + orig := nodeStatusFn + nodeStatusFn = func(ctx context.Context, cfg *pds.Config, node pds.NodeID, rootCAs *x509.CertPool) (pds.Status, error) { + calls++ + return fn(ctx, cfg, node, rootCAs) + } + t.Cleanup(func() { nodeStatusFn = orig }) + return &calls +} + +func TestProbeServiceHealth_UnprobedServiceDefaultsToOK(t *testing.T) { + d := &Daemon{config: &config.Config{Services: []string{"viperblock"}}} + health := d.probeServiceHealth(t.Context()) + assert.Equal(t, "ok", health["viperblock"]) +} + +func TestComputePredastoreHealth_NoMetaNodeOnHost(t *testing.T) { + // Host 2 in healthTestTOML runs only a blob node — nothing local to + // probe, which is not this host's service failing. + d := newHealthTestDaemon(t, 2) + calls := setNodeStatusFn(t, func(context.Context, *pds.Config, pds.NodeID, *x509.CertPool) (pds.Status, error) { + return pds.Status{}, nil + }) + + got := computePredastoreHealth(t.Context(), d) + assert.Equal(t, predastoreHealthOK, got) + assert.Equal(t, 0, *calls, "a host with no meta node must never dial one") +} + +func TestComputePredastoreHealth_NoHostIDConfigured(t *testing.T) { + // HostID <= 0: this node runs no predastore at all. + d := newHealthTestDaemon(t, 0) + got := computePredastoreHealth(t.Context(), d) + assert.Equal(t, predastoreHealthOK, got) +} + +func TestComputePredastoreHealth_OK(t *testing.T) { + d := newHealthTestDaemon(t, 1) + setNodeStatusFn(t, func(context.Context, *pds.Config, pds.NodeID, *x509.CertPool) (pds.Status, error) { + return pds.Status{State: "Leader", Leader: "node-3", IsLeader: true}, nil + }) + + got := computePredastoreHealth(t.Context(), d) + assert.Equal(t, predastoreHealthOK, got) +} + +func TestComputePredastoreHealth_NoLeader(t *testing.T) { + d := newHealthTestDaemon(t, 1) + setNodeStatusFn(t, func(context.Context, *pds.Config, pds.NodeID, *x509.CertPool) (pds.Status, error) { + return pds.Status{State: "Follower", Leader: ""}, nil + }) + + got := computePredastoreHealth(t.Context(), d) + assert.Equal(t, predastoreHealthNoLeader, got) +} + +func TestComputePredastoreHealth_UnreachableOnError(t *testing.T) { + d := newHealthTestDaemon(t, 1) + setNodeStatusFn(t, func(context.Context, *pds.Config, pds.NodeID, *x509.CertPool) (pds.Status, error) { + return pds.Status{}, assert.AnError + }) + + got := computePredastoreHealth(t.Context(), d) + assert.Equal(t, predastoreHealthUnreachable, got) +} + +func TestComputePredastoreHealth_UnreachableOnTimeout(t *testing.T) { + d := newHealthTestDaemon(t, 1) + + // Shrink the probe timeout so a wedged peer is provably bounded rather + // than actually waiting the production default. + origTimeout := predastoreProbeTimeout + predastoreProbeTimeout = 50 * time.Millisecond + t.Cleanup(func() { predastoreProbeTimeout = origTimeout }) + + setNodeStatusFn(t, func(ctx context.Context, cfg *pds.Config, node pds.NodeID, rootCAs *x509.CertPool) (pds.Status, error) { + <-ctx.Done() + return pds.Status{}, ctx.Err() + }) + + start := time.Now() + got := computePredastoreHealth(t.Context(), d) + elapsed := time.Since(start) + + assert.Equal(t, predastoreHealthUnreachable, got) + assert.Less(t, elapsed, 2*time.Second, "a wedged peer must not stall the handler beyond the probe timeout") +} + +func TestComputePredastoreHealth_NoCAConfigured(t *testing.T) { + d := newHealthTestDaemon(t, 1) + d.config.NATS.CACert = "" + + got := computePredastoreHealth(t.Context(), d) + assert.Equal(t, predastoreHealthUnreachable, got) +} + +func TestProbePredastore_CachesWithinTTL(t *testing.T) { + d := newHealthTestDaemon(t, 1) + + origTTL := predastoreHealthCacheTTL + predastoreHealthCacheTTL = time.Minute + t.Cleanup(func() { predastoreHealthCacheTTL = origTTL }) + + calls := setNodeStatusFn(t, func(context.Context, *pds.Config, pds.NodeID, *x509.CertPool) (pds.Status, error) { + return pds.Status{State: "Leader", Leader: "node-3"}, 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, *calls, "a second poll within the TTL must not re-dial") +} diff --git a/spinifex/gateway/gateway.go b/spinifex/gateway/gateway.go index d426ea76c..79991a39c 100644 --- a/spinifex/gateway/gateway.go +++ b/spinifex/gateway/gateway.go @@ -2,6 +2,7 @@ package gateway import ( "context" + "crypto/x509" "encoding/json" "encoding/xml" "errors" @@ -87,9 +88,14 @@ type GatewayConfig struct { DisableLogging bool `json:"disable_logging"` NATSConn *nats.Conn // Shared NATS connection for service communication Config string // Shared AWS Gateway config for S3 auth - ExpectedNodes int // Number of expected spinifex nodes for multi-node operations - Region string // Region this gateway is running in - InternalSuffix string // Internal DNS suffix for AWS-parity endpoints (e.g. spinifex.internal) + // RootCAs is the cluster CA pool, used to verify predastore meta nodes + // dialed directly for GetStorageStatus. Nil disables that verification, + // so admin storage status reports every meta node unreachable rather than + // dialing with no trust root. + RootCAs *x509.CertPool + ExpectedNodes int // Number of expected spinifex nodes for multi-node operations + Region string // Region this gateway is running in + InternalSuffix string // Internal DNS suffix for AWS-parity endpoints (e.g. spinifex.internal) // RegistryPort is the gateway's advertised port, appended to the ECR // registry host so docker login/tag/push dial the right port. Empty or // "443" renders a port-less host (standard HTTPS parity). diff --git a/spinifex/gateway/spinifex.go b/spinifex/gateway/spinifex.go index d184d6863..fbba9907c 100644 --- a/spinifex/gateway/spinifex.go +++ b/spinifex/gateway/spinifex.go @@ -73,7 +73,7 @@ func (gw *GatewayConfig) Spinifex_Request(w http.ResponseWriter, r *http.Request if gw.NATSConn == nil { return errors.New(awserrors.ErrorServerInternal) } - output, err = gateway_spx.GetStorageStatus(gw.NATSConn) + output, err = gateway_spx.GetStorageStatus(gw.NATSConn, gw.RootCAs) case "PromoteImage": if gw.NATSConn == nil { return errors.New(awserrors.ErrorServerInternal) diff --git a/spinifex/gateway/spx/storage.go b/spinifex/gateway/spx/storage.go index d3666906f..da5661e74 100644 --- a/spinifex/gateway/spx/storage.go +++ b/spinifex/gateway/spx/storage.go @@ -2,15 +2,14 @@ package spx import ( "context" + "crypto/x509" "encoding/json" "fmt" "log/slog" - "net" - "net/http" - "strconv" "sync" "time" + pds "github.com/mulgadc/predastore" "github.com/mulgadc/spinifex/spinifex/types" "github.com/nats-io/nats.go" ) @@ -45,13 +44,17 @@ type StorageEncodingOutput struct { ParityShards int `json:"parity_shards"` } -var storageHTTPClient = &http.Client{ - Timeout: 1 * time.Second, -} +// metaNodeQueryTimeout bounds the whole parallel round of meta node dials, so +// one unreachable node cannot stall the CLI command waiting on the others. +const metaNodeQueryTimeout = 2 * time.Second -// GetStorageStatus fetches predastore topology via NATS, then queries each -// meta node's /status and /health endpoints in parallel. -func GetStorageStatus(nc *nats.Conn) (*StorageStatusOutput, error) { +// GetStorageStatus fetches predastore topology via NATS, then dials each meta +// node's raft status directly and in parallel. rootCAs is the cluster CA +// pool: every meta node in the cluster is queried, not just a local one, so +// verifying each one's identity against the shared CA (rather than a local +// TLS config only the node owning it could read) is what makes every dial +// trusted. +func GetStorageStatus(nc *nats.Conn, rootCAs *x509.CertPool) (*StorageStatusOutput, error) { msg, err := nc.Request("spinifex.storage.config", []byte("{}"), 3*time.Second) if err != nil { return nil, fmt.Errorf("storage config request: %w", err) @@ -64,7 +67,7 @@ func GetStorageStatus(nc *nats.Conn) (*StorageStatusOutput, error) { metaStatuses := make([]MetaNodeStatus, len(cfg.MetaNodes)) var wg sync.WaitGroup - ctx, cancel := context.WithTimeout(context.Background(), 1*time.Second) + ctx, cancel := context.WithTimeout(context.Background(), metaNodeQueryTimeout) defer cancel() for i, meta := range cfg.MetaNodes { @@ -76,7 +79,7 @@ func GetStorageStatus(nc *nats.Conn) (*StorageStatusOutput, error) { wg.Add(1) go func(idx int, host string, port int) { defer wg.Done() - queryMetaNodeStatus(ctx, &metaStatuses[idx], host, port) + queryMetaNodeStatus(ctx, &metaStatuses[idx], rootCAs, host, port) }(i, meta.Host, meta.Port) } wg.Wait() @@ -93,61 +96,65 @@ func GetStorageStatus(nc *nats.Conn) (*StorageStatusOutput, error) { }, nil } -// predastoreStatusResponse matches the predastore s3db.StatusResponse JSON shape. -type predastoreStatusResponse struct { - NodeID string `json:"node_id"` - State string `json:"state"` - Leader string `json:"leader"` - LeaderAddr string `json:"leader_addr"` - Term string `json:"term"` - CommitIdx string `json:"commit_index"` - AppliedIdx string `json:"applied_index"` - IsLeader bool `json:"is_leader"` -} +// queryMetaNodeStatus dials one meta node's raft status directly — the +// opcode-multiplexed RPC predastore's meta nodes actually speak, not the +// HTTPS endpoints they never served — and fills out with what it reports. A +// node that does not answer within ctx leaves out untouched (Healthy stays +// false and every raft field stays blank), matching the CLI's existing +// contract for an unhealthy node. +func queryMetaNodeStatus(ctx context.Context, out *MetaNodeStatus, rootCAs *x509.CertPool, host string, port int) { + nodeID, ok := metaNodeID(out.ID) + if !ok { + slog.Debug("queryMetaNodeStatus: invalid node id", "id", out.ID, "host", host, "port", port) + return + } -func queryMetaNodeStatus(ctx context.Context, out *MetaNodeStatus, host string, port int) { - // Resolve 0.0.0.0 to a routable address for HTTPS. + // Resolve 0.0.0.0 to a routable address. queryHost := host if queryHost == "0.0.0.0" { queryHost = "127.0.0.1" } - healthURL := "https://" + net.JoinHostPort(queryHost, strconv.Itoa(port)) + "/health" - req, err := http.NewRequestWithContext(ctx, http.MethodGet, healthURL, nil) - if err != nil { - return - } - resp, err := storageHTTPClient.Do(req) - if err != nil { - slog.Debug("queryMetaNodeStatus: health check failed", "host", host, "port", port, "err", err) - return - } - resp.Body.Close() - out.Healthy = resp.StatusCode == http.StatusOK - - statusURL := "https://" + net.JoinHostPort(queryHost, strconv.Itoa(port)) + "/status" - req, err = http.NewRequestWithContext(ctx, http.MethodGet, statusURL, nil) - if err != nil { - return - } - resp, err = storageHTTPClient.Do(req) - if err != nil { - slog.Debug("queryMetaNodeStatus: status check failed", "host", host, "port", port, "err", err) - return - } - defer resp.Body.Close() - - var status predastoreStatusResponse - if err := json.NewDecoder(resp.Body).Decode(&status); err != nil { - slog.Debug("queryMetaNodeStatus: failed to decode status", "host", host, "port", port, "err", err) - return + // A single-host, single-node config is all NodeStatus needs to resolve + // and dial this one node: it queries that replica only, never redirects, + // and asks for no local TLS identity of its own. + cfg := &pds.Config{ + Hosts: []pds.HostConfig{ + { + ID: pds.HostID(nodeID), + Addr: queryHost, + Nodes: []pds.NodeConfig{{ID: nodeID, Role: pds.RoleMeta, Port: port}}, + }, + }, } + status, err := pds.NodeStatus(ctx, cfg, nodeID, rootCAs) + if err != nil { + slog.Debug("queryMetaNodeStatus: status probe failed", "id", out.ID, "host", host, "port", port, "err", err) + return + } + applyMetaStatus(out, status) +} + +// metaNodeID converts a meta node's id — plain int on the wire, predastore's +// NodeID (uint64) internally — validating it is never negative first, since +// predastore ids never are and a blind conversion could wrap. +func metaNodeID(id int) (pds.NodeID, bool) { + if id < 0 { + return 0, false + } + return pds.NodeID(id), true //#nosec G115 -- id validated non-negative above +} + +// applyMetaStatus maps a predastore raft Status into a MetaNodeStatus. +// Reaching here means the node answered, so Healthy is unconditionally true. +func applyMetaStatus(out *MetaNodeStatus, status pds.Status) { + out.Healthy = true out.State = status.State out.Leader = status.Leader out.LeaderAddr = status.LeaderAddr out.Term = status.Term - out.CommitIdx = status.CommitIdx - out.AppliedIdx = status.AppliedIdx + out.CommitIdx = status.CommitIndex + out.AppliedIdx = status.AppliedIndex out.IsLeader = status.IsLeader } diff --git a/spinifex/gateway/spx/storage_test.go b/spinifex/gateway/spx/storage_test.go index 28cf4bb26..a7170f474 100644 --- a/spinifex/gateway/spx/storage_test.go +++ b/spinifex/gateway/spx/storage_test.go @@ -1,14 +1,12 @@ package spx import ( - "crypto/tls" + "context" "encoding/json" - "net/http" - "net/http/httptest" - "net/url" - "strconv" "testing" + "time" + pds "github.com/mulgadc/predastore" "github.com/mulgadc/spinifex/spinifex/types" "github.com/nats-io/nats.go" "github.com/stretchr/testify/assert" @@ -23,7 +21,7 @@ func TestGetStorageStatus_Success(t *testing.T) { resp := types.StorageConfigResponse{ Encoding: types.StorageEncoding{DataShards: 2, ParityShards: 1}, MetaNodes: []types.StorageMetaNode{ - {ID: 1, Host: "127.0.0.1", Port: 0}, // port set below + {ID: 1, Host: "127.0.0.1", Port: 1}, // nothing listens on port 1 }, BlobNodes: []types.StorageBlobNode{ {ID: 1, Host: "0.0.0.0", Port: 9991}, @@ -40,7 +38,7 @@ func TestGetStorageStatus_Success(t *testing.T) { defer sub.Unsubscribe() nc.Flush() - out, err := GetStorageStatus(nc) + out, err := GetStorageStatus(nc, nil) require.NoError(t, err) assert.Equal(t, "Reed-Solomon", out.Encoding.Type) @@ -49,75 +47,59 @@ func TestGetStorageStatus_Success(t *testing.T) { assert.Len(t, out.BlobNodes, 2) assert.Len(t, out.Buckets, 1) assert.Equal(t, "predastore", out.Buckets[0].Name) - // Meta node health check will fail (no real predastore running) — expected + // Meta node status probe will fail (no real predastore running) — expected require.Len(t, out.MetaNodes, 1) assert.Equal(t, 1, out.MetaNodes[0].ID) - assert.False(t, out.MetaNodes[0].Healthy) // no predastore server to respond + assert.False(t, out.MetaNodes[0].Healthy) } func TestGetStorageStatus_NoNATSResponse(t *testing.T) { _, nc := startEmbeddedNATS(t) // No subscriber — should timeout - _, err := GetStorageStatus(nc) + _, err := GetStorageStatus(nc, nil) require.Error(t, err) assert.Contains(t, err.Error(), "storage config request") } -func TestQueryMetaNodeStatus_HealthyNode(t *testing.T) { - // Start a mock predastore meta node server - mux := http.NewServeMux() - mux.HandleFunc("/health", func(w http.ResponseWriter, r *http.Request) { - w.Header().Set("Content-Type", "application/json") - w.WriteHeader(http.StatusOK) - json.NewEncoder(w).Encode(map[string]any{"status": "healthy"}) +func TestApplyMetaStatus_MapsRaftFields(t *testing.T) { + out := &MetaNodeStatus{ID: 1, Host: "10.0.0.1", Port: 6100} + applyMetaStatus(out, pds.Status{ + NodeID: "1", + State: "Leader", + Leader: "node-1", + LeaderAddr: "", + Term: "42", + CommitIndex: "1000", + AppliedIndex: "998", + IsLeader: true, }) - mux.HandleFunc("/status", func(w http.ResponseWriter, r *http.Request) { - w.Header().Set("Content-Type", "application/json") - json.NewEncoder(w).Encode(predastoreStatusResponse{ - NodeID: "1", - State: "Leader", - Leader: "1", - LeaderAddr: "127.0.0.1:7660", - Term: "42", - CommitIdx: "1000", - AppliedIdx: "998", - IsLeader: true, - }) - }) - - srv := httptest.NewTLSServer(mux) - defer srv.Close() - - // Use the test server's TLS client - origClient := storageHTTPClient - storageHTTPClient = &http.Client{ - Transport: &http.Transport{ - TLSClientConfig: &tls.Config{InsecureSkipVerify: true}, - }, - } - defer func() { storageHTTPClient = origClient }() - - // Parse host:port from the test server URL - u, err := url.Parse(srv.URL) - require.NoError(t, err) - host := u.Hostname() - port, err := strconv.Atoi(u.Port()) - require.NoError(t, err) - - out := &MetaNodeStatus{ID: 1, Host: host, Port: port} - queryMetaNodeStatus(t.Context(), out, host, port) assert.True(t, out.Healthy) assert.Equal(t, "Leader", out.State) + assert.Equal(t, "node-1", out.Leader) assert.True(t, out.IsLeader) assert.Equal(t, "42", out.Term) assert.Equal(t, "1000", out.CommitIdx) assert.Equal(t, "998", out.AppliedIdx) } -func TestQueryMetaNodeStatus_UnreachableNode(t *testing.T) { - out := &MetaNodeStatus{ID: 1, Host: "127.0.0.1", Port: 1} // port 1 won't respond - queryMetaNodeStatus(t.Context(), out, "127.0.0.1", 1) +func TestApplyMetaStatus_NoLeaderStaysHealthy(t *testing.T) { + // A replica that answers but observes no leader is still a replica that + // answered: Healthy describes reachability, not raft convergence. + out := &MetaNodeStatus{ID: 2} + applyMetaStatus(out, pds.Status{State: "Follower"}) + + assert.True(t, out.Healthy) + assert.Equal(t, "Follower", out.State) + assert.Empty(t, out.Leader) + assert.False(t, out.IsLeader) +} + +func TestQueryMetaNodeStatus_Unreachable(t *testing.T) { + out := &MetaNodeStatus{ID: 1, Host: "127.0.0.1", Port: 1} // nothing listens here + ctx, cancel := context.WithTimeout(context.Background(), 500*time.Millisecond) + defer cancel() + queryMetaNodeStatus(ctx, out, nil, "127.0.0.1", 1) assert.False(t, out.Healthy) assert.Empty(t, out.State) diff --git a/spinifex/services/awsgw/awsgw.go b/spinifex/services/awsgw/awsgw.go index 3ce163b1e..8a29d764d 100644 --- a/spinifex/services/awsgw/awsgw.go +++ b/spinifex/services/awsgw/awsgw.go @@ -3,6 +3,7 @@ package awsgw import ( "context" "crypto/tls" + "crypto/x509" "errors" "fmt" "log/slog" @@ -151,6 +152,16 @@ func launchService(config *config.ClusterConfig) error { } defer natsConn.Close() + // The same cluster CA verifies predastore meta nodes GetStorageStatus + // dials directly, across every host in the cluster. + var rootCAs *x509.CertPool + if nodeConfig.NATS.CACert != "" { + rootCAs, err = utils.LoadCertPool(nodeConfig.NATS.CACert) + if err != nil { + return fmt.Errorf("load cluster CA: %w", err) + } + } + // Append Base dir if config has no leading path if nodeConfig.BaseDir != "" && !strings.HasPrefix(nodeConfig.AWSGW.Config, "/") { nodeConfig.AWSGW.Config = fmt.Sprintf("%s/%s", nodeConfig.BaseDir, nodeConfig.AWSGW.Config) @@ -388,6 +399,7 @@ func launchService(config *config.ClusterConfig) error { Debug: nodeConfig.AWSGW.Debug, DisableLogging: false, NATSConn: natsConn, + RootCAs: rootCAs, Config: nodeConfig.AWSGW.Config, ExpectedNodes: len(config.Nodes), Region: nodeConfig.Region, diff --git a/spinifex/utils/nats.go b/spinifex/utils/nats.go index 56644b58d..57d0eb3ea 100644 --- a/spinifex/utils/nats.go +++ b/spinifex/utils/nats.go @@ -25,6 +25,22 @@ var ( // ErrClusterUnavailable is returned when the NATS connection is not currently connected. var ErrClusterUnavailable = errors.New("cluster unavailable: NATS disconnected") +// LoadCertPool reads a PEM CA certificate from path and returns a pool +// trusting exactly it. Shared by every caller that verifies a peer against +// the cluster CA (e.g. /etc/spinifex/ca.pem) rather than the system trust +// store. +func LoadCertPool(path string) (*x509.CertPool, error) { + caCert, err := os.ReadFile(path) + if err != nil { + return nil, fmt.Errorf("%w %s: %w", ErrCACertRead, path, err) + } + pool := x509.NewCertPool() + if !pool.AppendCertsFromPEM(caCert) { + return nil, fmt.Errorf("%w from %s", ErrCACertParse, path) + } + return pool, nil +} + // natsRetryEscalateAttempt is the threshold at which retry logs escalate from Warn to Error (rate-limited to 1/min). // ~30 attempts at the 60 s backoff cap ≈ ~30 min disconnected, suggesting config error not transient restart. const natsRetryEscalateAttempt = 30 @@ -59,13 +75,9 @@ func ConnectNATS(host, token, caCertPath string, opts ...RetryOption) (*nats.Con } if caCertPath != "" { - caCert, err := os.ReadFile(caCertPath) + pool, err := LoadCertPool(caCertPath) if err != nil { - return nil, fmt.Errorf("%w %s: %w", ErrCACertRead, caCertPath, err) - } - pool := x509.NewCertPool() - if !pool.AppendCertsFromPEM(caCert) { - return nil, fmt.Errorf("%w from %s", ErrCACertParse, caCertPath) + return nil, err } natsOpts = append(natsOpts, nats.Secure(&tls.Config{ RootCAs: pool,