mirror of
https://github.com/mulgadc/spinifex.git
synced 2026-10-15 17:11:30 +00:00
fix(health): probe the local gate when predastore is single-host (#794)
The predastore health probe QUIC-dials the raft meta node, but a single-host install binds no cluster-plane socket — colocated nodes talk over an in-process pipe. The daemon runs in a separate process from predastore, so it can never reach that pipe and the dial is always refused, reporting a permanent false unreachable that gates ansible deploy readiness. On a single-host install, probe predastore's one cross-process listener — the S3 gate — instead. A gate that accepts a connection is a predastore process serving, and on a single voter raft is trivially leader. Multi-host installs keep the raft QUIC probe unchanged.
This commit is contained in:
@@ -5,7 +5,9 @@ import (
|
||||
"crypto/x509"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"net"
|
||||
"path/filepath"
|
||||
"strconv"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
@@ -39,6 +41,15 @@ var predastoreHealthCacheTTL = 5 * time.Second
|
||||
// predastore process.
|
||||
var nodeStatusFn = pds.NodeStatus
|
||||
|
||||
// gateDialFn opens a TCP connection to a predastore S3 gate, indirected so
|
||||
// tests can stub the dial and exercise the single-host verdicts without a live
|
||||
// listener. A bare TCP connect is enough: it completes once the process is
|
||||
// accepting, before any TLS, which is exactly "the predastore process is up".
|
||||
var gateDialFn = func(ctx context.Context, network, addr string) (net.Conn, error) {
|
||||
var d net.Dialer
|
||||
return d.DialContext(ctx, network, addr)
|
||||
}
|
||||
|
||||
// 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
|
||||
@@ -108,9 +119,10 @@ func probePredastore(ctx context.Context, d *Daemon) string {
|
||||
// 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.
|
||||
// that does run a meta node. A single-host install has no reachable meta
|
||||
// socket at all and falls back to probing the local gate — see probeLocalGate.
|
||||
func computePredastoreHealth(ctx context.Context, d *Daemon) string {
|
||||
nodes, cfg, err := localMetaNodes(d)
|
||||
nodes, cfg, hostID, err := localMetaNodes(d)
|
||||
if err != nil {
|
||||
slog.Warn("predastore health probe: could not resolve local meta nodes", "err", err)
|
||||
return predastoreHealthUnreachable
|
||||
@@ -119,6 +131,15 @@ func computePredastoreHealth(ctx context.Context, d *Daemon) string {
|
||||
return predastoreHealthOK
|
||||
}
|
||||
|
||||
// Single-host: the meta node talks over an in-process pipe and binds no
|
||||
// QUIC socket, so this daemon — a separate process — can never dial it.
|
||||
// Probe predastore's one cross-process listener, the S3 gate, instead: a
|
||||
// gate that accepts a connection is a predastore process serving, and on a
|
||||
// single voter raft is trivially leader.
|
||||
if predastoreIsSingleHost(cfg, hostID) {
|
||||
return probeLocalGate(ctx, d, cfg, hostID)
|
||||
}
|
||||
|
||||
rootCAs, err := loadPredastoreTrustRoot(d)
|
||||
if err != nil {
|
||||
slog.Warn("predastore health probe: could not load cluster CA", "err", err)
|
||||
@@ -142,25 +163,82 @@ func computePredastoreHealth(ctx context.Context, d *Daemon) string {
|
||||
return verdict
|
||||
}
|
||||
|
||||
// predastoreIsSingleHost reports whether no predastore node runs off this
|
||||
// host, which is the only condition under which a local node opens a network
|
||||
// socket. When none does, the meta node is pipe-only and unreachable to a
|
||||
// separate process, so the raft QUIC probe cannot apply.
|
||||
func predastoreIsSingleHost(cfg *pds.Config, hostID pds.HostID) bool {
|
||||
for _, h := range cfg.Hosts {
|
||||
if h.ID != hostID && len(h.Nodes) > 0 {
|
||||
return false
|
||||
}
|
||||
}
|
||||
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
|
||||
// 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 {
|
||||
addr, ok := localGateAddr(cfg, hostID)
|
||||
if !ok {
|
||||
return predastoreHealthOK
|
||||
}
|
||||
|
||||
probeCtx, cancel := context.WithTimeout(ctx, predastoreProbeTimeout)
|
||||
defer cancel()
|
||||
|
||||
conn, err := gateDialFn(probeCtx, "tcp", addr)
|
||||
if err != nil {
|
||||
slog.Debug("predastore health probe: gate unreachable", "addr", addr, "err", err)
|
||||
return predastoreHealthUnreachable
|
||||
}
|
||||
_ = conn.Close()
|
||||
return predastoreHealthOK
|
||||
}
|
||||
|
||||
// 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.
|
||||
func localGateAddr(cfg *pds.Config, hostID pds.HostID) (string, bool) {
|
||||
for _, h := range cfg.Hosts {
|
||||
if h.ID != hostID {
|
||||
continue
|
||||
}
|
||||
for _, n := range h.Nodes {
|
||||
if n.Role != pds.RoleGate {
|
||||
continue
|
||||
}
|
||||
host := pds.NodeBindAddr(h, n)
|
||||
if host == "" || host == "0.0.0.0" || host == "::" {
|
||||
host = "127.0.0.1"
|
||||
}
|
||||
return net.JoinHostPort(host, strconv.Itoa(n.Port)), true
|
||||
}
|
||||
}
|
||||
return "", false
|
||||
}
|
||||
|
||||
// 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) {
|
||||
func localMetaNodes(d *Daemon) ([]pds.NodeID, *pds.Config, pds.HostID, error) {
|
||||
if d.config.Predastore.HostID <= 0 {
|
||||
return nil, nil, nil
|
||||
return nil, nil, 0, 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)
|
||||
return nil, nil, 0, fmt.Errorf("load predastore config %s: %w", cfgPath, err)
|
||||
}
|
||||
|
||||
nodes := pds.MetaNodesOnHost(cfg, pds.HostID(d.config.Predastore.HostID))
|
||||
return nodes, cfg, nil
|
||||
hostID := pds.HostID(d.config.Predastore.HostID)
|
||||
return pds.MetaNodesOnHost(cfg, hostID), cfg, hostID, nil
|
||||
}
|
||||
|
||||
// predastoreConfigPath is where the predastore process reads its own config:
|
||||
|
||||
@@ -8,6 +8,7 @@ package daemon
|
||||
import (
|
||||
"context"
|
||||
"crypto/x509"
|
||||
"net"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"testing"
|
||||
@@ -62,13 +63,54 @@ role = "blob"
|
||||
port = 6660
|
||||
`
|
||||
|
||||
// singleHostTOML runs every node in one [[host]] — a gate, a blob and the
|
||||
// meta node — so the meta node is pipe-only and the probe must fall back to
|
||||
// the gate rather than the QUIC raft dial.
|
||||
const singleHostTOML = `
|
||||
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"
|
||||
|
||||
[[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
|
||||
`
|
||||
|
||||
// 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()
|
||||
return newHealthTestDaemonTOML(t, hostID, healthTestTOML)
|
||||
}
|
||||
|
||||
// newHealthTestDaemonTOML is newHealthTestDaemon for an arbitrary predastore
|
||||
// config, so a test can pick single- or multi-host topology.
|
||||
func newHealthTestDaemonTOML(t *testing.T, hostID int, toml string) *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))
|
||||
require.NoError(t, os.WriteFile(filepath.Join(dir, "predastore", "predastore.toml"), []byte(toml), 0o600))
|
||||
|
||||
certPEM, _ := generateTestCert(t)
|
||||
caPath := filepath.Join(dir, "ca.pem")
|
||||
@@ -83,6 +125,20 @@ func newHealthTestDaemon(t *testing.T, hostID int) *Daemon {
|
||||
}
|
||||
}
|
||||
|
||||
// setGateDialFn stubs gateDialFn for the test's lifetime and returns a pointer
|
||||
// to a call counter, mirroring setNodeStatusFn.
|
||||
func setGateDialFn(t *testing.T, fn func(context.Context, string, string) (net.Conn, error)) *int {
|
||||
t.Helper()
|
||||
calls := 0
|
||||
orig := gateDialFn
|
||||
gateDialFn = func(ctx context.Context, network, addr string) (net.Conn, error) {
|
||||
calls++
|
||||
return fn(ctx, network, addr)
|
||||
}
|
||||
t.Cleanup(func() { gateDialFn = 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 {
|
||||
@@ -183,6 +239,58 @@ func TestComputePredastoreHealth_NoCAConfigured(t *testing.T) {
|
||||
assert.Equal(t, predastoreHealthUnreachable, got)
|
||||
}
|
||||
|
||||
func TestComputePredastoreHealth_SingleHostGateOK(t *testing.T) {
|
||||
// Single-host: the meta node is pipe-only, so the probe must dial the gate
|
||||
// and never the raft QUIC endpoint.
|
||||
d := newHealthTestDaemonTOML(t, 1, singleHostTOML)
|
||||
|
||||
metaCalls := setNodeStatusFn(t, func(context.Context, *pds.Config, pds.NodeID, *x509.CertPool) (pds.Status, error) {
|
||||
return pds.Status{}, assert.AnError
|
||||
})
|
||||
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, "single-host must probe the gate")
|
||||
assert.Equal(t, 0, *metaCalls, "single-host must never QUIC-dial the meta node")
|
||||
}
|
||||
|
||||
func TestComputePredastoreHealth_SingleHostGateUnreachable(t *testing.T) {
|
||||
d := newHealthTestDaemonTOML(t, 1, singleHostTOML)
|
||||
setGateDialFn(t, func(context.Context, string, string) (net.Conn, error) {
|
||||
return nil, assert.AnError
|
||||
})
|
||||
|
||||
got := computePredastoreHealth(t.Context(), d)
|
||||
assert.Equal(t, predastoreHealthUnreachable, got)
|
||||
}
|
||||
|
||||
func TestLocalGateAddr(t *testing.T) {
|
||||
// A wildcard gate bind is dialled on the loopback the daemon shares.
|
||||
cfg := &pds.Config{Hosts: []pds.HostConfig{{
|
||||
ID: 1,
|
||||
Nodes: []pds.NodeConfig{
|
||||
{ID: 3, Role: pds.RoleMeta, Port: 7660},
|
||||
{ID: 1, Role: pds.RoleGate, Port: 8443, BindAddr: "0.0.0.0"},
|
||||
},
|
||||
}}}
|
||||
addr, ok := localGateAddr(cfg, 1)
|
||||
require.True(t, ok)
|
||||
assert.Equal(t, "127.0.0.1:8443", addr)
|
||||
|
||||
// A host running a meta node but no gate has no socket to probe.
|
||||
noGate := &pds.Config{Hosts: []pds.HostConfig{{
|
||||
ID: 1,
|
||||
Nodes: []pds.NodeConfig{{ID: 3, Role: pds.RoleMeta, Port: 7660}},
|
||||
}}}
|
||||
_, ok = localGateAddr(noGate, 1)
|
||||
assert.False(t, ok)
|
||||
}
|
||||
|
||||
func TestProbePredastore_CachesWithinTTL(t *testing.T) {
|
||||
d := newHealthTestDaemon(t, 1)
|
||||
|
||||
|
||||
Reference in New Issue
Block a user