mirror of
https://github.com/mulgadc/spinifex.git
synced 2026-10-15 17:11:30 +00:00
fix: read single-host predastore health from /readyz in Service Metrics (#1195)
This commit is contained in:
@@ -2,7 +2,9 @@ package daemon
|
||||
|
||||
import (
|
||||
"log/slog"
|
||||
"net"
|
||||
"os"
|
||||
"strconv"
|
||||
|
||||
"github.com/mulgadc/spinifex/spinifex/types"
|
||||
"github.com/nats-io/nats.go"
|
||||
@@ -26,9 +28,24 @@ type predastoreRS struct {
|
||||
// names the machine and carries no port; the nodes pinned to it supply their
|
||||
// own.
|
||||
type predastoreHost struct {
|
||||
ID int `toml:"id"`
|
||||
Addr string `toml:"addr"`
|
||||
Nodes []predastoreNode `toml:"node"`
|
||||
ID int `toml:"id"`
|
||||
Addr string `toml:"addr"`
|
||||
BindAddr string `toml:"bind_addr"`
|
||||
AdminPort int `toml:"admin_port"`
|
||||
Nodes []predastoreNode `toml:"node"`
|
||||
}
|
||||
|
||||
// adminAddr is where another machine dials this host's admin listener, empty
|
||||
// when it runs none. A wildcard bind is reached at the host's addr.
|
||||
func (h predastoreHost) adminAddr() string {
|
||||
if h.AdminPort == 0 {
|
||||
return ""
|
||||
}
|
||||
host := h.BindAddr
|
||||
if host == "" || host == "0.0.0.0" || host == "::" {
|
||||
host = h.Addr
|
||||
}
|
||||
return net.JoinHostPort(host, strconv.Itoa(h.AdminPort))
|
||||
}
|
||||
|
||||
// predastoreNode is a role pinned to the host it is written under. Its address
|
||||
@@ -79,11 +96,16 @@ func (d *Daemon) handleStorageConfig(msg *nats.Msg) string {
|
||||
|
||||
// A node's address is its host's addr plus its own port: colocated nodes
|
||||
// share a machine but never a port.
|
||||
singleHost := len(cfg.Hosts) == 1
|
||||
for _, h := range cfg.Hosts {
|
||||
for _, n := range h.Nodes {
|
||||
switch n.Role {
|
||||
case predastoreRoleMeta:
|
||||
resp.MetaNodes = append(resp.MetaNodes, types.StorageMetaNode{ID: n.ID, Host: h.Addr, Port: n.Port})
|
||||
meta := types.StorageMetaNode{ID: n.ID, Host: h.Addr, Port: n.Port}
|
||||
if singleHost {
|
||||
meta.AdminAddr = h.adminAddr()
|
||||
}
|
||||
resp.MetaNodes = append(resp.MetaNodes, meta)
|
||||
case predastoreRoleBlob:
|
||||
resp.BlobNodes = append(resp.BlobNodes, types.StorageBlobNode{ID: n.ID, Host: h.Addr, Port: n.Port})
|
||||
case predastoreRoleGate:
|
||||
|
||||
@@ -2,6 +2,7 @@ package daemon
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"testing"
|
||||
@@ -28,6 +29,7 @@ parity = 1
|
||||
id = 1
|
||||
bind_addr = "0.0.0.0"
|
||||
addr = "10.0.0.1"
|
||||
admin_port = 8660
|
||||
data_dir = "/var/lib/spinifex/predastore/cluster"
|
||||
|
||||
[[host.node]]
|
||||
@@ -94,9 +96,8 @@ func storageConfigResponse(t *testing.T, cfg string, reply string) []byte {
|
||||
}
|
||||
|
||||
// TestHandleStorageConfig_SplitsNodesByRole pins what the gateway builds its
|
||||
// storage report from: every node is reported at its host address and its own
|
||||
// port, meta and blob in separate tables, and the gate in neither — it serves
|
||||
// the S3 API and holds no cluster state to report.
|
||||
// storage report from: every node at its host address and own port, meta and
|
||||
// blob apart, no gate, and no admin address since a multi-host cluster has QUIC.
|
||||
func TestHandleStorageConfig_SplitsNodesByRole(t *testing.T) {
|
||||
payload := storageConfigResponse(t, clusterTOML, "test.storage.config.ok")
|
||||
|
||||
@@ -118,6 +119,46 @@ func TestHandleStorageConfig_SplitsNodesByRole(t *testing.T) {
|
||||
assert.Equal(t, "ap-southeast-2", resp.Buckets[0].Region)
|
||||
}
|
||||
|
||||
// singleHostAdminTOML is the default install: every node on one host, which
|
||||
// opens no QUIC socket, so raft status is only on its admin port.
|
||||
const singleHostAdminTOML = `
|
||||
[rs]
|
||||
data = 1
|
||||
parity = 0
|
||||
|
||||
[[host]]
|
||||
id = 1
|
||||
addr = "10.0.0.1"
|
||||
bind_addr = "%s"
|
||||
admin_port = 8660
|
||||
|
||||
[[host.node]]
|
||||
id = 3
|
||||
role = "meta"
|
||||
port = 7660
|
||||
`
|
||||
|
||||
// TestHandleStorageConfig_SingleHostNamesTheAdminListener pins where the
|
||||
// gateway reads a single-host node's health: the host's bind address when it
|
||||
// names one, its addr when it binds a wildcard.
|
||||
func TestHandleStorageConfig_SingleHostNamesTheAdminListener(t *testing.T) {
|
||||
for i, tc := range []struct{ bind, want string }{
|
||||
{"127.0.0.1", "127.0.0.1:8660"},
|
||||
{"0.0.0.0", "10.0.0.1:8660"},
|
||||
{"", "10.0.0.1:8660"},
|
||||
} {
|
||||
t.Run("bind="+tc.bind, func(t *testing.T) {
|
||||
payload := storageConfigResponse(t, fmt.Sprintf(singleHostAdminTOML, tc.bind),
|
||||
fmt.Sprintf("test.storage.config.single.%d", i))
|
||||
|
||||
var resp types.StorageConfigResponse
|
||||
require.NoError(t, json.Unmarshal(payload, &resp))
|
||||
require.Len(t, resp.MetaNodes, 1)
|
||||
assert.Equal(t, tc.want, resp.MetaNodes[0].AdminAddr)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// TestHandleStorageConfig_ReportsAMissingConfig covers the node that has no
|
||||
// predastore config: the report is an error rather than an empty topology,
|
||||
// which would read as a cluster with no storage at all.
|
||||
|
||||
@@ -6,6 +6,8 @@ import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"net"
|
||||
"net/http"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
@@ -53,7 +55,8 @@ const metaNodeQueryTimeout = 2 * time.Second
|
||||
// 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.
|
||||
// trusted. A single-host cluster opens no QUIC socket, so its nodes are read
|
||||
// from their host's /readyz instead.
|
||||
func GetStorageStatus(nc *nats.Conn, rootCAs *x509.CertPool) (*StorageStatusOutput, error) {
|
||||
msg, err := nc.Request("spinifex.storage.config", []byte("{}"), 3*time.Second)
|
||||
if err != nil {
|
||||
@@ -77,10 +80,14 @@ func GetStorageStatus(nc *nats.Conn, rootCAs *x509.CertPool) (*StorageStatusOutp
|
||||
Port: meta.Port,
|
||||
}
|
||||
wg.Add(1)
|
||||
go func(idx int, host string, port int) {
|
||||
go func(idx int, meta types.StorageMetaNode) {
|
||||
defer wg.Done()
|
||||
queryMetaNodeStatus(ctx, &metaStatuses[idx], rootCAs, host, port)
|
||||
}(i, meta.Host, meta.Port)
|
||||
if meta.AdminAddr != "" {
|
||||
queryMetaNodeReadyz(ctx, &metaStatuses[idx], meta.AdminAddr)
|
||||
return
|
||||
}
|
||||
queryMetaNodeStatus(ctx, &metaStatuses[idx], rootCAs, meta.Host, meta.Port)
|
||||
}(i, meta)
|
||||
}
|
||||
wg.Wait()
|
||||
|
||||
@@ -136,6 +143,44 @@ func queryMetaNodeStatus(ctx context.Context, out *MetaNodeStatus, rootCAs *x509
|
||||
applyMetaStatus(out, status)
|
||||
}
|
||||
|
||||
// queryMetaNodeReadyz marks a meta node healthy when its host's /readyz reports
|
||||
// that it observes a raft leader. It carries no raft detail, so every raft
|
||||
// field stays blank; a node that does not answer within ctx stays unhealthy.
|
||||
func queryMetaNodeReadyz(ctx context.Context, out *MetaNodeStatus, adminAddr string) {
|
||||
host, port, err := net.SplitHostPort(adminAddr)
|
||||
if err != nil {
|
||||
slog.Debug("queryMetaNodeReadyz: invalid admin address", "id", out.ID, "admin_addr", adminAddr, "err", err)
|
||||
return
|
||||
}
|
||||
if host == "0.0.0.0" {
|
||||
host = "127.0.0.1"
|
||||
}
|
||||
url := "http://" + net.JoinHostPort(host, port) + "/readyz"
|
||||
|
||||
req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
|
||||
if err != nil {
|
||||
slog.Debug("queryMetaNodeReadyz: build request failed", "id", out.ID, "url", url, "err", err)
|
||||
return
|
||||
}
|
||||
resp, err := http.DefaultClient.Do(req)
|
||||
if err != nil {
|
||||
slog.Debug("queryMetaNodeReadyz: probe failed", "id", out.ID, "url", url, "err", err)
|
||||
return
|
||||
}
|
||||
defer func() { _ = resp.Body.Close() }()
|
||||
|
||||
// A 503 still names each check, and a blob failure there says nothing
|
||||
// about the meta node, so only the leader check is read.
|
||||
var body struct {
|
||||
Checks map[string]string `json:"checks"`
|
||||
}
|
||||
if err := json.NewDecoder(resp.Body).Decode(&body); err != nil {
|
||||
slog.Debug("queryMetaNodeReadyz: decode failed", "id", out.ID, "url", url, "status", resp.StatusCode, "err", err)
|
||||
return
|
||||
}
|
||||
out.Healthy = body.Checks["meta_leader"] == "ok"
|
||||
}
|
||||
|
||||
// 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.
|
||||
|
||||
@@ -3,6 +3,8 @@ package spx
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
@@ -104,3 +106,89 @@ func TestQueryMetaNodeStatus_Unreachable(t *testing.T) {
|
||||
assert.False(t, out.Healthy)
|
||||
assert.Empty(t, out.State)
|
||||
}
|
||||
|
||||
// readyzServer stands in for a predastore admin listener whose /readyz
|
||||
// answers code with body.
|
||||
func readyzServer(t *testing.T, code int, body string) string {
|
||||
t.Helper()
|
||||
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
if r.URL.Path != "/readyz" {
|
||||
http.NotFound(w, r)
|
||||
return
|
||||
}
|
||||
w.WriteHeader(code)
|
||||
_, _ = w.Write([]byte(body))
|
||||
}))
|
||||
t.Cleanup(srv.Close)
|
||||
return srv.Listener.Addr().String()
|
||||
}
|
||||
|
||||
// serveStorageConfig answers spinifex.storage.config with resp, as the daemon
|
||||
// would, and returns the gateway's report.
|
||||
func serveStorageConfig(t *testing.T, resp types.StorageConfigResponse) *StorageStatusOutput {
|
||||
t.Helper()
|
||||
_, nc := startEmbeddedNATS(t)
|
||||
sub, err := nc.Subscribe("spinifex.storage.config", func(msg *nats.Msg) {
|
||||
data, _ := json.Marshal(resp)
|
||||
_ = msg.Respond(data)
|
||||
})
|
||||
require.NoError(t, err)
|
||||
t.Cleanup(func() { _ = sub.Unsubscribe() })
|
||||
require.NoError(t, nc.Flush())
|
||||
|
||||
out, err := GetStorageStatus(nc, nil)
|
||||
require.NoError(t, err)
|
||||
require.Len(t, out.MetaNodes, 1)
|
||||
return out
|
||||
}
|
||||
|
||||
// singleHostMeta is the default install's meta node: nothing listens on its
|
||||
// QUIC port, so its health can only come from the admin listener at addr.
|
||||
func singleHostMeta(addr string) types.StorageConfigResponse {
|
||||
return types.StorageConfigResponse{
|
||||
MetaNodes: []types.StorageMetaNode{{ID: 3, Host: "127.0.0.1", Port: 1, AdminAddr: addr}},
|
||||
}
|
||||
}
|
||||
|
||||
func TestGetStorageStatus_SingleHostReadsReadyz(t *testing.T) {
|
||||
addr := readyzServer(t, http.StatusOK,
|
||||
`{"checks":{"blob_nodes":"ok","meta_leader":"ok","meta_reachable":"ok"},"status":"ready"}`)
|
||||
|
||||
node := serveStorageConfig(t, singleHostMeta(addr)).MetaNodes[0]
|
||||
|
||||
assert.True(t, node.Healthy)
|
||||
assert.Empty(t, node.State, "/readyz carries no raft detail to report")
|
||||
assert.Empty(t, node.Leader)
|
||||
}
|
||||
|
||||
// /readyz answers 503 when any check fails; a blob failure must not mark a
|
||||
// meta node that observes a leader unreachable.
|
||||
func TestGetStorageStatus_ReadyzBlobFailureKeepsMetaHealthy(t *testing.T) {
|
||||
addr := readyzServer(t, http.StatusServiceUnavailable,
|
||||
`{"checks":{"blob_nodes":"fewer than 1 blob nodes answered","meta_leader":"ok"},"status":"unready"}`)
|
||||
|
||||
assert.True(t, serveStorageConfig(t, singleHostMeta(addr)).MetaNodes[0].Healthy)
|
||||
}
|
||||
|
||||
func TestGetStorageStatus_ReadyzWithoutLeaderStaysUnhealthy(t *testing.T) {
|
||||
addr := readyzServer(t, http.StatusServiceUnavailable,
|
||||
`{"checks":{"meta_leader":"no raft leader observed"},"status":"unready"}`)
|
||||
|
||||
assert.False(t, serveStorageConfig(t, singleHostMeta(addr)).MetaNodes[0].Healthy)
|
||||
}
|
||||
|
||||
// Before the sampler's first cycle /readyz names no checks at all.
|
||||
func TestGetStorageStatus_ReadyzBeforeFirstSampleStaysUnhealthy(t *testing.T) {
|
||||
addr := readyzServer(t, http.StatusServiceUnavailable, `{"checks":{},"status":"unready"}`)
|
||||
|
||||
assert.False(t, serveStorageConfig(t, singleHostMeta(addr)).MetaNodes[0].Healthy)
|
||||
}
|
||||
|
||||
func TestQueryMetaNodeReadyz_Unreachable(t *testing.T) {
|
||||
out := &MetaNodeStatus{ID: 3}
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 500*time.Millisecond)
|
||||
defer cancel()
|
||||
queryMetaNodeReadyz(ctx, out, "127.0.0.1:1")
|
||||
|
||||
assert.False(t, out.Healthy)
|
||||
}
|
||||
|
||||
+2
-2
File diff suppressed because one or more lines are too long
+1
-1
File diff suppressed because one or more lines are too long
+1
-1
@@ -7,7 +7,7 @@
|
||||
<link rel="icon" href="/favicon.svg" type="image/svg+xml" />
|
||||
<meta name="theme-color" content="#000000" />
|
||||
<meta name="description" content="Spinifex Control Panel" />
|
||||
<script type="module" crossorigin src="/assets/index-Bp3ivx_J.js"></script>
|
||||
<script type="module" crossorigin src="/assets/index-NKm_rfBO.js"></script>
|
||||
<link rel="modulepreload" crossorigin href="/assets/compiler-runtime-D_vZfn79.js">
|
||||
<link rel="modulepreload" crossorigin href="/assets/removable-Dwr9otKB.js">
|
||||
<link rel="modulepreload" crossorigin href="/assets/query-BPr_oSEg.js">
|
||||
|
||||
+15
@@ -139,6 +139,21 @@ describe("service metrics cluster health", () => {
|
||||
expect(within(metaRow(2)).getByText("Unreachable")).toBeInTheDocument()
|
||||
})
|
||||
|
||||
it("reports Healthy for a single-host node vouched for by /readyz", async () => {
|
||||
await renderPage([
|
||||
metaNode(3, {
|
||||
state: undefined,
|
||||
leader: undefined,
|
||||
term: undefined,
|
||||
commit_index: undefined,
|
||||
applied_index: undefined,
|
||||
}),
|
||||
])
|
||||
|
||||
expect(clusterHealth()).toBe("Healthy")
|
||||
expect(within(metaRow(0)).getByText("Healthy")).toBeInTheDocument()
|
||||
})
|
||||
|
||||
it("reports Unknown when the topology has no meta nodes", async () => {
|
||||
await renderPage([])
|
||||
|
||||
|
||||
+6
-9
@@ -122,18 +122,15 @@ export function ServiceMetricsPage() {
|
||||
)
|
||||
}
|
||||
|
||||
// hasQuorum reads the raft view each meta node reports, not just whether it
|
||||
// answered: a replica that has lost quorum still replies, so reachability
|
||||
// alone would call a leaderless cluster healthy.
|
||||
// hasQuorum reads the raft view each meta node reports, since a leaderless
|
||||
// replica still answers. A healthy node with no raft state was vouched for by
|
||||
// its host's /readyz leader check, so it has no leader to compare.
|
||||
function hasQuorum(nodes: MetaNodeStatus[]) {
|
||||
if (nodes.length === 0) {
|
||||
if (nodes.length === 0 || !nodes.every((n) => n.healthy)) {
|
||||
return false
|
||||
}
|
||||
const leader = nodes[0]?.leader
|
||||
if (!leader) {
|
||||
return false
|
||||
}
|
||||
return nodes.every((n) => n.healthy && n.leader === leader)
|
||||
const leaders = new Set(nodes.filter((n) => n.state).map((n) => n.leader))
|
||||
return leaders.size <= 1 && !leaders.has("") && !leaders.has(undefined)
|
||||
}
|
||||
|
||||
function ClusterHealthBadge({
|
||||
|
||||
@@ -21,6 +21,9 @@ type StorageMetaNode struct {
|
||||
ID int `json:"id"`
|
||||
Host string `json:"host"`
|
||||
Port int `json:"port"`
|
||||
// AdminAddr is set only on a single-host cluster, which opens no QUIC socket:
|
||||
// its host's admin listener, whose /readyz answers for the node's health.
|
||||
AdminAddr string `json:"admin_addr,omitempty"`
|
||||
}
|
||||
|
||||
// StorageBlobNode describes a predastore blob node, which holds erasure-coded
|
||||
|
||||
Reference in New Issue
Block a user