From 455c97a2e77b3fcf6de6722ee04d6d1d000a79bf Mon Sep 17 00:00:00 2001 From: Ben Duncan <5178243+benduncan@users.noreply.github.com> Date: Mon, 14 Sep 2026 15:39:03 -0500 Subject: [PATCH] =?UTF-8?q?fix:=20prod=20review=20of=2014=20September=20?= =?UTF-8?q?=E2=80=94=20IPsec=20follower=20reads,=20stopped-instance=20EIP?= =?UTF-8?q?=20probes,=20audit=20client=20IP=20and=20JetStream=20formation?= =?UTF-8?q?=20divergence=20(#1029)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix: read NB_Global.ipsec through the NB remote list so followers and compute nodes answer quietly * fix: stop probing and nudging the EIP datapath of a stopped instance's port * fix: log the real client IP, request ID and duration_ms in the awsgw access line * test: mark the UI proxy header test as in-package * fix: discard pre-formation JetStream data when forming a multi-node cluster A node that ran as its own single-node cluster carries KV streams into formation. nats-server adopts a local stream with the same name as the clustered one, and raft catch-up only fetches sequences beyond a follower's last, so the conflicting prefix is never repaired and replicas diverge permanently. - spx admin join removes the local JetStream store once formation succeeds. A store holding streams now needs --force, and the identity-discard warning names it. - spx admin init --nodes >= 2 refuses a store that holds streams unless --discard-jetstream is given, and removes it once every node has joined. Keeping it as the seed is unsafe: stream leadership ignores who holds data, and unassigned local streams are deleted as orphans. - Neither removes a store while `spx service nats` runs, --force included. - install-node.sh passes --discard-jetstream and waits for every `spx service` process to exit before formation. * feat: add spx admin kv digest for per-replica JetStream consistency checks Replicas of a stream can hold different content while raft reports them current, and nothing in the cluster detects it. A direct get over NATS is no way to inspect one replica: a client's request goes to a random member of the direct-get queue group, routed replicas included. kv digest copies this node's $G streams tree (without consumer state) to a temporary directory, opens the copy in a private in-process server with no listener, and prints first/last sequence, message count and a SHA-256 over every stored message per stream. Running it on each node and diffing the output shows which streams diverge; --seqs locates the first differing sequence. The live store is only ever read. * fix: let spx admin kv digest read replicated streams A replicated stream's meta.inf carries num_replicas > 1, which a standalone server refuses to recover, so the digest could not read any production store. Rewrite the copy's config to one replica with no expiry and a recomputed checksum, and test it against a three-node cluster and an expired bucket. * test: use assert.ErrorIs for the removed JetStream store * test: mark the JetStream store and digest tests as in-package * fix: stop the process-group test misreading a reaped child as alive The test polled until the child was gone, then checked once more. A task being reaped shows state X for an instant, which processRunning counted as running, so that second look failed about 1 run in 50 under -race. * fix: wait for the recovery retry loop before removing a test's data dir restoreInstances starts the retry loop untracked, and it rewrites the local state file when it ends. A write landing during t.TempDir cleanup failed TestRestoreInstances_* with RemoveAll: directory not empty. --- cmd/spinifex/cmd/admin.go | 66 +++- cmd/spinifex/cmd/admin_kv.go | 371 ++++++++++++++++++ cmd/spinifex/cmd/admin_kv_test.go | 294 ++++++++++++++ cmd/spinifex/cmd/admin_test.go | 28 +- cmd/spinifex/cmd/jetstream_store.go | 139 +++++++ cmd/spinifex/cmd/jetstream_store_test.go | 140 +++++++ docs/COMMANDS.md | 5 +- docs/install/install-multi-node/README.md | 6 +- go.mod | 2 +- scripts/install-node.sh | 21 +- spinifex/daemon/instance_records_vpcd_test.go | 17 + spinifex/gateway/requestaudit.go | 22 +- spinifex/gateway/requestaudit_test.go | 86 +++- spinifex/network/host/ipsec.go | 45 ++- spinifex/network/host/ipsec_enable.go | 47 ++- spinifex/network/host/ipsec_enable_test.go | 80 +++- spinifex/network/host/mac_binding.go | 2 + spinifex/network/reconcile/apply.go | 57 +++ .../reconcile/apply_guest_port_idle_test.go | 132 +++++++ spinifex/network/reconcile/intent.go | 79 +++- spinifex/network/reconcile/intent_test.go | 64 +++ spinifex/network/reconcile/reconcile.go | 2 + spinifex/services/spinifexui/proxy.go | 6 + .../services/spinifexui/proxy_realip_test.go | 59 +++ spinifex/utils/utils.go | 17 + spinifex/utils/utils_test.go | 36 ++ 26 files changed, 1745 insertions(+), 78 deletions(-) create mode 100644 cmd/spinifex/cmd/admin_kv.go create mode 100644 cmd/spinifex/cmd/admin_kv_test.go create mode 100644 cmd/spinifex/cmd/jetstream_store.go create mode 100644 cmd/spinifex/cmd/jetstream_store_test.go create mode 100644 spinifex/daemon/instance_records_vpcd_test.go create mode 100644 spinifex/network/reconcile/apply_guest_port_idle_test.go create mode 100644 spinifex/services/spinifexui/proxy_realip_test.go diff --git a/cmd/spinifex/cmd/admin.go b/cmd/spinifex/cmd/admin.go index 48c71d05a..31e24b59c 100644 --- a/cmd/spinifex/cmd/admin.go +++ b/cmd/spinifex/cmd/admin.go @@ -358,6 +358,7 @@ func init() { // Flags for admin init adminInitCmd.Flags().Bool("force", false, "Force re-initialization (overwrites existing config)") + adminInitCmd.Flags().Bool("discard-jetstream", false, "With --nodes >= 2, remove this node's existing JetStream streams once every node has joined") adminInitCmd.Flags().String("region", "ap-southeast-2", "Mulga region to create") adminInitCmd.Flags().String("az", "ap-southeast-2a", "Mulga AZ to create") adminInitCmd.Flags().String("node", "node1", "Node name, increment for additional nodes (default, node1)") @@ -1313,6 +1314,15 @@ func runAdminInit(cmd *cobra.Command, args []string) { os.Exit(0) } + if spxRoot == "" { + spxRoot = DefaultDataDir() + } + discardJetStream, _ := cmd.Flags().GetBool("discard-jetstream") + if err := checkInitJetStreamStore(jetStreamStoreDir(spxRoot), nodes, discardJetStream); err != nil { + fmt.Fprintf(os.Stderr, "❌ Error: %v\n", err) + os.Exit(1) + } + // Preserve the previously-captured operator email across --force re-inits // when --email is omitted (e.g. reset-dev-env.sh workflows). Without this // a reset would silently blank the address. @@ -1845,6 +1855,17 @@ func runAdminInitMultiNode(cmd *cobra.Command, accessKey, secretKey, accountID, fmt.Printf("✅ All %d nodes joined!\n", expectedNodes) + // The pre-start check refused a store with streams unless discard was asked + // for, so anything left here is removed now that the cluster is committed. + if discard, _ := cmd.Flags().GetBool("discard-jetstream"); discard { + storeDir := jetStreamStoreDir(spxRoot) + if err := discardJetStreamStore(storeDir); err != nil { + fmt.Fprintf(os.Stderr, "❌ Error: %v\n", err) + os.Exit(1) + } + fmt.Printf("✅ Discarded this node's pre-formation JetStream store: %s\n", storeDir) + } + // Build cluster topology from formation data allNodes := fs.Nodes() clusterRoutes := formation.BuildClusterRoutes(allNodes) @@ -2002,19 +2023,27 @@ func joinRetryable(err error, statusCode int) bool { } // checkJoinPreconditions rejects a join that would silently destroy this node's -// existing cluster identity. Joining adopts the primary's CA and master key, -// overwriting whatever is here: correct on a freshly installed node, and on one -// that has been in service it orphans every fragment and volume sealed under the -// old key. runAdminInit guards the same way before re-initializing. -func checkJoinPreconditions(configDir string, force bool) error { +// identity or JetStream data; --force accepts that loss. A running NATS service +// is refused even with --force, because the store cannot be removed under it. +func checkJoinPreconditions(configDir, storeDir string, force bool) error { + if err := checkStoreDiscardable(); err != nil { + return err + } if force { return nil } tomlPath := filepath.Join(configDir, "spinifex.toml") - if !admin.FileExists(tomlPath) { - return nil + if admin.FileExists(tomlPath) { + return fmt.Errorf("this node is already initialized: %s", tomlPath) } - return fmt.Errorf("this node is already initialized: %s", tomlPath) + streams, err := localStreams(storeDir) + if err != nil { + return err + } + if len(streams) > 0 { + return fmt.Errorf("this node's JetStream store %s holds %d stream(s)", storeDir, len(streams)) + } + return nil } // joinDiscardsIdentityMsg spells out what a forced join throws away. Kept out of @@ -2023,6 +2052,8 @@ const joinDiscardsIdentityMsg = `Joining will discard this node's own cluster id - CA certificate and key - master key, and any data sealed under it - viperblock key, and any volumes encrypted under it + - JetStream store (/nats/jetstream): every stream and KV bucket + this node holds, so it takes the cluster's copy instead of diverging from it That is safe on a freshly installed node — an ISO install initializes a single-node cluster at first boot, and nothing has been sealed under these keys @@ -2103,7 +2134,11 @@ func runAdminJoin(cmd *cobra.Command, args []string) { // Checked before any network call so a node that will not join says so // immediately. Unlike init this exits non-zero: a node that did not join // must not look like success to a provisioning script. - if err := checkJoinPreconditions(configDir, force); err != nil { + if dataDir == "" { + dataDir = DefaultDataDir() + } + storeDir := jetStreamStoreDir(dataDir) + if err := checkJoinPreconditions(configDir, storeDir, force); err != nil { fmt.Fprintf(os.Stderr, "⚠️ %v\n\n%s\n", err, joinDiscardsIdentityMsg) os.Exit(1) } @@ -2113,11 +2148,6 @@ func runAdminJoin(cmd *cobra.Command, args []string) { clusterBind = bindIP } - // Set default data directory - if dataDir == "" { - dataDir = DefaultDataDir() - } - fmt.Println("🚀 Joining Spinifex cluster...") fmt.Printf("Node: %s\n", node) fmt.Printf("Leader: %s\n", leaderHost) @@ -2298,6 +2328,14 @@ func runAdminJoin(cmd *cobra.Command, args []string) { os.Exit(1) } + // Removed only once formation has succeeded, so a failed join leaves the + // node's single-node cluster intact. + if err := discardJetStreamStore(storeDir); err != nil { + fmt.Fprintf(os.Stderr, "❌ Error: %v\n", err) + os.Exit(1) + } + fmt.Printf("✅ Discarded this node's pre-formation JetStream store: %s\n", storeDir) + // Set up config directory if configDir == "" { configDir = DefaultConfigDir() diff --git a/cmd/spinifex/cmd/admin_kv.go b/cmd/spinifex/cmd/admin_kv.go new file mode 100644 index 000000000..29f3b65a7 --- /dev/null +++ b/cmd/spinifex/cmd/admin_kv.go @@ -0,0 +1,371 @@ +package cmd + +import ( + "bytes" + "context" + "crypto/sha256" + "encoding/binary" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "io" + "io/fs" + "os" + "path/filepath" + "slices" + "sort" + "strings" + "time" + + "github.com/minio/highwayhash" + "github.com/nats-io/nats-server/v2/server" + "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" + "github.com/spf13/cobra" +) + +var kvCmd = &cobra.Command{ + Use: "kv", + Short: "Inspect JetStream KV state", +} + +var kvDigestCmd = &cobra.Command{ + Use: "digest", + Short: "Print a digest of every stream held by this node's JetStream replica", + Long: `Read this node's own copy of every JetStream stream and print one line per +stream: name, first and last sequence, message count, and a SHA-256 over every +stored message (sequence, subject, timestamp, headers, data). + +Run it on every node and compare the output. Replicas of one stream must +produce identical lines; any difference means the replicas have diverged, and +--seqs prints a hash per sequence to find where. + +It never touches the live store. A direct get over NATS may be answered by any +replica, so instead the streams tree is copied to --work-dir and read by a +private in-process server with no listener, which is removed afterwards. The +copy's stream config is set to one replica with no expiry so it can be loaded. + +Writes landing during the copy, or applied but not yet flushed to disk, can +make a replica's tail differ, so gate API writes and re-run before calling a +tail difference divergence.`, + Run: runKVDigest, +} + +func init() { + adminCmd.AddCommand(kvCmd) + kvCmd.AddCommand(kvDigestCmd) + kvDigestCmd.Flags().String("store-dir", "", "JetStream store to read (default /nats/jetstream)") + kvDigestCmd.Flags().String("work-dir", os.TempDir(), "Directory to hold the temporary copy") + kvDigestCmd.Flags().StringSlice("stream", nil, "Only these streams (repeatable)") + kvDigestCmd.Flags().Bool("seqs", false, "Also print a hash per sequence") +} + +// streamDigest is one stream's content as held by one replica. +type streamDigest struct { + Name string + FirstSeq uint64 + LastSeq uint64 + Msgs uint64 + Digest string + Seqs []seqDigest +} + +// seqDigest is the hash of one stored sequence, or "deleted" when the replica +// has no message at that sequence. +type seqDigest struct { + Seq uint64 + Hash string +} + +func runKVDigest(cmd *cobra.Command, _ []string) { + storeDir, _ := cmd.Flags().GetString("store-dir") + if storeDir == "" { + spxRoot, _ := cmd.Flags().GetString("spinifex-dir") + if spxRoot == "" { + spxRoot = DefaultDataDir() + } + storeDir = jetStreamStoreDir(spxRoot) + } + workParent, _ := cmd.Flags().GetString("work-dir") + only, _ := cmd.Flags().GetStringSlice("stream") + withSeqs, _ := cmd.Flags().GetBool("seqs") + + work, err := os.MkdirTemp(workParent, "spx-kv-digest-") + if err != nil { + fmt.Fprintf(os.Stderr, "kv digest: %v\n", err) + os.Exit(1) + } + defer os.RemoveAll(work) + + copiedAt := time.Now().UTC() + skipped, err := copyStreamTree(storeDir, filepath.Join(work, "jetstream")) + if err != nil { + fmt.Fprintf(os.Stderr, "kv digest: %v\n", err) + os.RemoveAll(work) + os.Exit(1) + } + + ctx, cancel := context.WithTimeout(cmd.Context(), 10*time.Minute) + defer cancel() + digests, err := digestStore(ctx, work, only, withSeqs) + if err != nil { + fmt.Fprintf(os.Stderr, "kv digest: %v\n", err) + cancel() + os.RemoveAll(work) + os.Exit(1) + } + + host, _ := os.Hostname() + writeDigests(cmd.OutOrStdout(), host, storeDir, copiedAt, skipped, digests) +} + +// writeDigests prints the report. Only the "#" lines carry node-specific +// values, so `grep -v '^#'` output from two replicas diffs cleanly. +func writeDigests(w io.Writer, host, storeDir string, copiedAt time.Time, skipped []string, digests []streamDigest) { + fmt.Fprintf(w, "# host=%s store=%s copied_at=%s\n", host, storeDir, copiedAt.Format(time.RFC3339)) + for _, s := range skipped { + fmt.Fprintf(w, "# skipped account stream (not read): %s\n", s) + } + for _, d := range digests { + fmt.Fprintf(w, "%s\tfirst=%d\tlast=%d\tmsgs=%d\tsha256=%s\n", d.Name, d.FirstSeq, d.LastSeq, d.Msgs, d.Digest) + for _, s := range d.Seqs { + fmt.Fprintf(w, "%s\tseq=%d\t%s\n", d.Name, s.Seq, s.Hash) + } + } +} + +// globalAccountDir is the account Spinifex's token-authenticated clients use. +// Streams under any other account are listed but not read. +const globalAccountDir = "$G" + +// copyStreamTree copies /$G/streams to /$G/streams, leaving out +// consumer state (obs), and returns streams found under other accounts. +func copyStreamTree(storeDir, dst string) (skipped []string, err error) { + streams, err := localStreams(storeDir) + if err != nil { + return nil, err + } + if len(streams) == 0 { + return nil, fmt.Errorf("no streams found under %s", storeDir) + } + for _, s := range streams { + if !strings.HasPrefix(s, globalAccountDir+"/") { + skipped = append(skipped, s) + } + } + + src := filepath.Join(storeDir, globalAccountDir, "streams") + out := filepath.Join(dst, globalAccountDir, "streams") + err = filepath.WalkDir(src, func(path string, d fs.DirEntry, walkErr error) error { + if walkErr != nil { + return walkErr + } + rel, err := filepath.Rel(src, path) + if err != nil { + return err + } + // /obs holds consumers, which are not stream content. + if parts := strings.Split(rel, string(filepath.Separator)); d.IsDir() && len(parts) == 2 && parts[1] == "obs" { + return filepath.SkipDir + } + target := filepath.Join(out, rel) + if d.IsDir() { + return os.MkdirAll(target, 0o700) + } + if !d.Type().IsRegular() { + return nil + } + return copyFile(path, target) + }) + if err != nil { + return nil, fmt.Errorf("copy streams from %s: %w", src, err) + } + entries, err := os.ReadDir(out) + if err != nil { + return nil, err + } + for _, e := range entries { + if e.IsDir() { + if err := rewriteStreamMetaStandalone(filepath.Join(out, e.Name())); err != nil { + return nil, err + } + } + } + return skipped, nil +} + +// rewriteStreamMetaStandalone makes a copied stream loadable by a standalone +// server, which refuses replicas > 1 and would expire aged messages on +// recovery. meta.sum is highwayhash64 keyed by sha256 of the directory name. +func rewriteStreamMetaStandalone(streamDir string) error { + metaPath := filepath.Join(streamDir, "meta.inf") + raw, err := os.ReadFile(metaPath) + if errors.Is(err, os.ErrNotExist) { + return nil + } + if err != nil { + return err + } + dec := json.NewDecoder(bytes.NewReader(raw)) + dec.UseNumber() + var meta map[string]any + if err := dec.Decode(&meta); err != nil { + return fmt.Errorf("parse %s: %w", metaPath, err) + } + meta["num_replicas"] = 1 + meta["max_age"] = 0 + delete(meta, "allow_msg_ttl") + delete(meta, "subject_delete_marker_ttl") + delete(meta, "placement") + out, err := json.Marshal(meta) + if err != nil { + return err + } + key := sha256.Sum256([]byte(filepath.Base(streamDir))) + hh, err := highwayhash.New64(key[:]) + if err != nil { + return err + } + hh.Write(out) + if err := os.WriteFile(metaPath, out, 0o600); err != nil { + return err + } + return os.WriteFile(filepath.Join(streamDir, "meta.sum"), []byte(hex.EncodeToString(hh.Sum(nil))), 0o600) +} + +func copyFile(src, dst string) error { + in, err := os.Open(src) + if err != nil { + return err + } + defer in.Close() + out, err := os.OpenFile(dst, os.O_CREATE|os.O_WRONLY|os.O_TRUNC, 0o600) + if err != nil { + return err + } + if _, err := io.Copy(out, in); err != nil { + out.Close() + return err + } + return out.Close() +} + +// digestStore opens storeParent (holding jetstream/) in a standalone server +// that accepts in-process connections only, and digests each stream. +func digestStore(ctx context.Context, storeParent string, only []string, withSeqs bool) ([]streamDigest, error) { + ns, err := server.NewServer(&server.Options{ + ServerName: "spx-kv-digest", + JetStream: true, + StoreDir: storeParent, + DontListen: true, + NoLog: true, + NoSigs: true, + }) + if err != nil { + return nil, fmt.Errorf("create reader server: %w", err) + } + ns.Start() + defer func() { + ns.Shutdown() + ns.WaitForShutdown() + }() + if !ns.ReadyForConnections(30 * time.Second) { + return nil, errors.New("reader server did not become ready") + } + + nc, err := nats.Connect("", nats.InProcessServer(ns)) + if err != nil { + return nil, fmt.Errorf("connect to reader server: %w", err) + } + defer nc.Close() + js, err := jetstream.New(nc) + if err != nil { + return nil, err + } + + var names []string + lister := js.StreamNames(ctx) + for name := range lister.Name() { + if len(only) == 0 || slices.Contains(only, name) { + names = append(names, name) + } + } + if err := lister.Err(); err != nil { + return nil, fmt.Errorf("list streams: %w", err) + } + sort.Strings(names) + + digests := make([]streamDigest, 0, len(names)) + for _, name := range names { + d, err := digestStream(ctx, js, name, withSeqs) + if err != nil { + return nil, err + } + digests = append(digests, d) + } + return digests, nil +} + +func digestStream(ctx context.Context, js jetstream.JetStream, name string, withSeqs bool) (streamDigest, error) { + s, err := js.Stream(ctx, name) + if err != nil { + return streamDigest{}, fmt.Errorf("open stream %s: %w", name, err) + } + info, err := s.Info(ctx) + if err != nil { + return streamDigest{}, fmt.Errorf("stream info %s: %w", name, err) + } + st := info.State + d := streamDigest{Name: name, FirstSeq: st.FirstSeq, LastSeq: st.LastSeq, Msgs: st.Msgs} + + whole := sha256.New() + for seq := st.FirstSeq; st.Msgs > 0 && seq <= st.LastSeq; seq++ { + msg, err := s.GetMsg(ctx, seq) + var h string + switch { + case errors.Is(err, jetstream.ErrMsgNotFound): + h = "deleted" + case err != nil: + return streamDigest{}, fmt.Errorf("read %s seq %d: %w", name, seq, err) + default: + h = hashStoredMsg(msg) + } + fmt.Fprintf(whole, "%d:%s\n", seq, h) + if withSeqs { + d.Seqs = append(d.Seqs, seqDigest{Seq: seq, Hash: h}) + } + } + d.Digest = hex.EncodeToString(whole.Sum(nil)) + return d, nil +} + +// hashStoredMsg hashes everything a replica stores for one sequence. Header +// keys are sorted so the hash does not depend on map iteration order. +func hashStoredMsg(m *jetstream.RawStreamMsg) string { + h := sha256.New() + var n [8]byte + field := func(b []byte) { + binary.BigEndian.PutUint64(n[:], uint64(len(b))) + h.Write(n[:]) + h.Write(b) + } + binary.BigEndian.PutUint64(n[:], m.Sequence) + h.Write(n[:]) + field([]byte(m.Subject)) + binary.BigEndian.PutUint64(n[:], uint64(m.Time.UnixNano())) + h.Write(n[:]) + keys := make([]string, 0, len(m.Header)) + for k := range m.Header { + keys = append(keys, k) + } + sort.Strings(keys) + for _, k := range keys { + field([]byte(k)) + for _, v := range m.Header[k] { + field([]byte(v)) + } + } + field(m.Data) + return hex.EncodeToString(h.Sum(nil))[:16] +} diff --git a/cmd/spinifex/cmd/admin_kv_test.go b/cmd/spinifex/cmd/admin_kv_test.go new file mode 100644 index 000000000..9457b4633 --- /dev/null +++ b/cmd/spinifex/cmd/admin_kv_test.go @@ -0,0 +1,294 @@ +//test:in-package: exercises the unexported digest helpers against a real clustered +// store, which the cobra command only reaches through a live node. + +package cmd + +import ( + "bytes" + "context" + "fmt" + "net" + "os" + "path/filepath" + "strings" + "testing" + "time" + + "github.com/nats-io/nats-server/v2/server" + "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// kvServer runs a standalone JetStream server on storeParent and returns a KV +// bucket on it plus a stop function. +func kvServer(t *testing.T, storeParent, bucket string) (jetstream.KeyValue, func()) { + t.Helper() + ns, err := server.NewServer(&server.Options{ + ServerName: "kv-test", JetStream: true, StoreDir: storeParent, + DontListen: true, NoLog: true, NoSigs: true, + }) + require.NoError(t, err) + ns.Start() + require.True(t, ns.ReadyForConnections(10*time.Second)) + nc, err := nats.Connect("", nats.InProcessServer(ns)) + require.NoError(t, err) + js, err := jetstream.New(nc) + require.NoError(t, err) + kv, err := js.CreateKeyValue(context.Background(), jetstream.KeyValueConfig{Bucket: bucket, History: 5}) + require.NoError(t, err) + return kv, func() { + nc.Close() + ns.Shutdown() + ns.WaitForShutdown() + } +} + +func digestOf(t *testing.T, storeDir string, withSeqs bool) []streamDigest { + t.Helper() + work := t.TempDir() + _, err := copyStreamTree(storeDir, filepath.Join(work, "jetstream")) + require.NoError(t, err) + d, err := digestStore(context.Background(), work, nil, withSeqs) + require.NoError(t, err) + return d +} + +func TestKVDigest(t *testing.T) { + ctx := context.Background() + parent := t.TempDir() + kv, stop := kvServer(t, parent, "test") + for _, kvp := range [][2]string{{"a", "1"}, {"b", "2"}, {"a", "3"}} { + _, err := kv.PutString(ctx, kvp[0], kvp[1]) + require.NoError(t, err) + } + require.NoError(t, kv.Delete(ctx, "b")) + store := filepath.Join(parent, "jetstream") + + // Copied while the server is live, exactly as it runs on a node. + live := digestOf(t, store, true) + stop() + + require.Len(t, live, 1) + assert.Equal(t, "KV_test", live[0].Name) + assert.Equal(t, uint64(1), live[0].FirstSeq) + assert.Equal(t, uint64(4), live[0].LastSeq) + assert.Equal(t, uint64(4), live[0].Msgs) + require.Len(t, live[0].Seqs, 4) + + t.Run("stable across reads and a stopped server", func(t *testing.T) { + again := digestOf(t, store, false) + assert.Equal(t, live[0].Digest, again[0].Digest) + }) + + t.Run("source store is left untouched", func(t *testing.T) { + _, err := os.Stat(filepath.Join(store, "$G", "streams", "KV_test")) + assert.NoError(t, err) + }) + + // Same keys and values written separately differ in timestamps, which is + // exactly what separates two nodes' pre-formation writes. + t.Run("independently written store differs", func(t *testing.T) { + other := t.TempDir() + kv2, stop2 := kvServer(t, other, "test") + for _, kvp := range [][2]string{{"a", "1"}, {"b", "2"}, {"a", "3"}} { + _, err := kv2.PutString(ctx, kvp[0], kvp[1]) + require.NoError(t, err) + } + require.NoError(t, kv2.Delete(ctx, "b")) + stop2() + d := digestOf(t, filepath.Join(other, "jetstream"), false) + assert.NotEqual(t, live[0].Digest, d[0].Digest) + }) + + t.Run("report lines", func(t *testing.T) { + var buf bytes.Buffer + writeDigests(&buf, "node1", store, time.Unix(0, 0), []string{"ACC/S"}, live) + out := buf.String() + assert.Contains(t, out, "# host=node1 ") + assert.Contains(t, out, "# skipped account stream (not read): ACC/S") + assert.Contains(t, out, "KV_test\tfirst=1\tlast=4\tmsgs=4\tsha256="+live[0].Digest) + assert.Equal(t, 4, strings.Count(out, "KV_test\tseq=")) + }) +} + +func TestKVDigestStreamFilterAndDeletedSeq(t *testing.T) { + ctx := context.Background() + parent := t.TempDir() + // The KV bucket is only there to be excluded by the stream filter. + _, stop := kvServer(t, parent, "other") + stop() + + // KV streams deny message deletes, so an interior gap needs a plain stream. + ns, err := server.NewServer(&server.Options{JetStream: true, StoreDir: parent, DontListen: true, NoLog: true, NoSigs: true}) + require.NoError(t, err) + ns.Start() + require.True(t, ns.ReadyForConnections(10*time.Second)) + nc, err := nats.Connect("", nats.InProcessServer(ns)) + require.NoError(t, err) + js, err := jetstream.New(nc) + require.NoError(t, err) + s, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "ONE", Subjects: []string{"one.>"}}) + require.NoError(t, err) + for _, v := range []string{"v1", "v2", "v3"} { + _, err = js.Publish(ctx, "one.k", []byte(v)) + require.NoError(t, err) + } + require.NoError(t, s.DeleteMsg(ctx, 2)) + nc.Close() + ns.Shutdown() + ns.WaitForShutdown() + + work := t.TempDir() + _, err = copyStreamTree(filepath.Join(parent, "jetstream"), filepath.Join(work, "jetstream")) + require.NoError(t, err) + d, err := digestStore(ctx, work, []string{"ONE"}, true) + require.NoError(t, err) + require.Len(t, d, 1) + assert.Equal(t, "ONE", d[0].Name) + assert.Equal(t, uint64(2), d[0].Msgs) + assert.Equal(t, []seqDigest{{1, d[0].Seqs[0].Hash}, {2, "deleted"}, {3, d[0].Seqs[2].Hash}}, d[0].Seqs) +} + +func freePort(t *testing.T) int { + t.Helper() + l, err := net.Listen("tcp", "127.0.0.1:0") + require.NoError(t, err) + defer l.Close() + return l.Addr().(*net.TCPAddr).Port +} + +// A replicated stream's meta.inf carries num_replicas > 1, which a standalone +// server refuses to load; this is the case every production node presents. +func TestKVDigestClusteredReplicas(t *testing.T) { + ctx := context.Background() + const n = 3 + ports := make([]int, n) + routes := make([]string, n) + for i := range ports { + ports[i] = freePort(t) + routes[i] = fmt.Sprintf("nats://127.0.0.1:%d", ports[i]) + } + stores := make([]string, n) + servers := make([]*server.Server, n) + for i := range servers { + stores[i] = t.TempDir() + ns, err := server.NewServer(&server.Options{ + ServerName: fmt.Sprintf("s%d", i+1), JetStream: true, StoreDir: stores[i], + Host: "127.0.0.1", Port: -1, NoLog: true, NoSigs: true, + Cluster: server.ClusterOpts{Name: "digest", Host: "127.0.0.1", Port: ports[i]}, + Routes: server.RoutesFromStr(strings.Join(routes, ",")), + }) + require.NoError(t, err) + ns.Start() + servers[i] = ns + } + t.Cleanup(func() { + for _, ns := range servers { + ns.Shutdown() + ns.WaitForShutdown() + } + }) + + nc, err := nats.Connect("", nats.InProcessServer(servers[0])) + require.NoError(t, err) + defer nc.Close() + js, err := jetstream.New(nc) + require.NoError(t, err) + + var kv jetstream.KeyValue + require.Eventually(t, func() bool { + kv, err = js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: "rep", History: 3, Replicas: n, TTL: time.Hour}) + return err == nil + }, 30*time.Second, 250*time.Millisecond, "clustered KV create: %v", err) + for i := range 5 { + _, err := kv.PutString(ctx, fmt.Sprintf("k%d", i%2), fmt.Sprintf("v%d", i)) + require.NoError(t, err) + } + s, err := js.Stream(ctx, "KV_rep") + require.NoError(t, err) + require.Eventually(t, func() bool { + info, err := s.Info(ctx) + if err != nil || info.Cluster == nil || len(info.Cluster.Replicas) != n-1 { + return false + } + for _, r := range info.Cluster.Replicas { + if !r.Current || r.Lag != 0 { + return false + } + } + return true + }, 30*time.Second, 250*time.Millisecond) + + // Current means applied, not flushed; shutdown flushes every block to disk. + nc.Close() + for _, ns := range servers { + ns.Shutdown() + ns.WaitForShutdown() + } + + var first string + for i, store := range stores { + d := digestOf(t, filepath.Join(store, "jetstream"), false) + require.Len(t, d, 1, "replica %d", i+1) + assert.Equal(t, uint64(5), d[0].Msgs, "replica %d", i+1) + if i == 0 { + first = d[0].Digest + continue + } + assert.Equal(t, first, d[0].Digest, "replica %d matches replica 1", i+1) + } +} + +// Recovery applies MaxAge, so without the rewrite an expired bucket copy +// would read as empty and mask whatever the replica holds. +func TestKVDigestIgnoresExpiry(t *testing.T) { + ctx := context.Background() + parent := t.TempDir() + ns, err := server.NewServer(&server.Options{JetStream: true, StoreDir: parent, DontListen: true, NoLog: true, NoSigs: true}) + require.NoError(t, err) + ns.Start() + require.True(t, ns.ReadyForConnections(10*time.Second)) + nc, err := nats.Connect("", nats.InProcessServer(ns)) + require.NoError(t, err) + js, err := jetstream.New(nc) + require.NoError(t, err) + kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: "ttl", TTL: time.Second}) + require.NoError(t, err) + _, err = kv.PutString(ctx, "k", "v") + require.NoError(t, err) + + work := t.TempDir() + _, err = copyStreamTree(filepath.Join(parent, "jetstream"), filepath.Join(work, "jetstream")) + require.NoError(t, err) + nc.Close() + ns.Shutdown() + ns.WaitForShutdown() + + time.Sleep(1500 * time.Millisecond) + d, err := digestStore(ctx, work, nil, false) + require.NoError(t, err) + require.Len(t, d, 1) + assert.Equal(t, uint64(1), d[0].Msgs) +} + +func TestCopyStreamTree(t *testing.T) { + store := seedStore(t, "KV_a") + require.NoError(t, os.MkdirAll(filepath.Join(store, "$G", "streams", "KV_a", "obs", "c1"), 0o755)) + require.NoError(t, os.WriteFile(filepath.Join(store, "$G", "streams", "KV_a", "obs", "c1", "o.dat"), []byte("x"), 0o600)) + require.NoError(t, os.MkdirAll(filepath.Join(store, "ACME", "streams", "ORDERS"), 0o755)) + + dst := filepath.Join(t.TempDir(), "jetstream") + skipped, err := copyStreamTree(store, dst) + require.NoError(t, err) + assert.Equal(t, []string{"ACME/ORDERS"}, skipped) + + _, err = os.Stat(filepath.Join(dst, "$G", "streams", "KV_a", "msgs", "1.blk")) + assert.NoError(t, err, "message blocks are copied") + _, err = os.Stat(filepath.Join(dst, "$G", "streams", "KV_a", "obs")) + assert.True(t, os.IsNotExist(err), "consumer state is left out") + + _, err = copyStreamTree(filepath.Join(t.TempDir(), "absent"), dst) + assert.Error(t, err) +} diff --git a/cmd/spinifex/cmd/admin_test.go b/cmd/spinifex/cmd/admin_test.go index e095eab35..0b3f73e22 100644 --- a/cmd/spinifex/cmd/admin_test.go +++ b/cmd/spinifex/cmd/admin_test.go @@ -1289,6 +1289,9 @@ func TestAMIVolumeSizeGiB(t *testing.T) { } func TestCheckJoinPreconditions(t *testing.T) { + withFakeProc(t, nil) + noStore := filepath.Join(t.TempDir(), "nats", "jetstream") + writeToml := func(t *testing.T) string { t.Helper() dir := t.TempDir() @@ -1299,13 +1302,13 @@ func TestCheckJoinPreconditions(t *testing.T) { } t.Run("uninitialized node joins", func(t *testing.T) { - if err := checkJoinPreconditions(t.TempDir(), false); err != nil { + if err := checkJoinPreconditions(t.TempDir(), noStore, false); err != nil { t.Fatalf("want nil, got %v", err) } }) t.Run("initialized node is refused", func(t *testing.T) { - err := checkJoinPreconditions(writeToml(t), false) + err := checkJoinPreconditions(writeToml(t), noStore, false) if err == nil { t.Fatal("want refusal, got nil") } @@ -1317,7 +1320,7 @@ func TestCheckJoinPreconditions(t *testing.T) { // The guidance is the whole point of the guard: an operator must be able to // tell what a forced join would destroy without reading the source. t.Run("guidance names what is discarded", func(t *testing.T) { - for _, want := range []string{"CA certificate", "master key", "viperblock key", "--force"} { + for _, want := range []string{"CA certificate", "master key", "viperblock key", "JetStream store", "--force"} { if !strings.Contains(joinDiscardsIdentityMsg, want) { t.Errorf("guidance missing %q", want) } @@ -1325,10 +1328,27 @@ func TestCheckJoinPreconditions(t *testing.T) { }) t.Run("force overrides", func(t *testing.T) { - if err := checkJoinPreconditions(writeToml(t), true); err != nil { + if err := checkJoinPreconditions(writeToml(t), noStore, true); err != nil { t.Fatalf("want nil with force, got %v", err) } }) + + // A store with streams and no config is still data the join would destroy. + t.Run("uninitialized node with streams is refused", func(t *testing.T) { + err := checkJoinPreconditions(t.TempDir(), seedStore(t, "KV_a"), false) + if err == nil || !strings.Contains(err.Error(), "JetStream store") { + t.Fatalf("want JetStream refusal, got %v", err) + } + }) + + // --force accepts the loss of data, never a wipe under a live server. + t.Run("running NATS is refused even with force", func(t *testing.T) { + withFakeProc(t, map[string]string{"4242": "/usr/local/bin/spx\x00service\x00nats\x00start\x00"}) + err := checkJoinPreconditions(writeToml(t), seedStore(t, "KV_a"), true) + if !errors.Is(err, errNATSRunning) { + t.Fatalf("want errNATSRunning, got %v", err) + } + }) } func TestJoinRetryable(t *testing.T) { diff --git a/cmd/spinifex/cmd/jetstream_store.go b/cmd/spinifex/cmd/jetstream_store.go new file mode 100644 index 000000000..428b38839 --- /dev/null +++ b/cmd/spinifex/cmd/jetstream_store.go @@ -0,0 +1,139 @@ +package cmd + +import ( + "bytes" + "errors" + "fmt" + "os" + "path/filepath" + "sort" + "strings" +) + +// jetStreamStoreDir is where the NATS service keeps its JetStream data under a +// node's data directory. The nats.conf template sets store_dir to +// /nats/, and nats-server appends "jetstream" itself. +func jetStreamStoreDir(dataDir string) string { + return filepath.Join(dataDir, "nats", "jetstream") +} + +// localStreams lists the streams a JetStream store holds on disk, as +// "/". A missing store holds nothing. +func localStreams(storeDir string) ([]string, error) { + accounts, err := os.ReadDir(storeDir) + if errors.Is(err, os.ErrNotExist) { + return nil, nil + } + if err != nil { + return nil, fmt.Errorf("read JetStream store %s: %w", storeDir, err) + } + var streams []string + for _, acc := range accounts { + if !acc.IsDir() { + continue + } + entries, err := os.ReadDir(filepath.Join(storeDir, acc.Name(), "streams")) + if errors.Is(err, os.ErrNotExist) { + continue + } + if err != nil { + return nil, fmt.Errorf("read streams for account %s: %w", acc.Name(), err) + } + for _, e := range entries { + if e.IsDir() { + streams = append(streams, acc.Name()+"/"+e.Name()) + } + } + } + sort.Strings(streams) + return streams, nil +} + +// procRoot is overridable in tests so the process scan can run against a fake +// /proc tree. +var procRoot = "/proc" + +// natsRunningLocally reports whether this host runs the Spinifex NATS service, +// found as a process whose command line is `spx service nats ...`. Removing a +// store under a live server races JetStream, which rewrites its index files. +func natsRunningLocally() (bool, error) { + entries, err := os.ReadDir(procRoot) + if err != nil { + return false, fmt.Errorf("scan processes: %w", err) + } + for _, e := range entries { + if !e.IsDir() || strings.TrimLeft(e.Name(), "0123456789") != "" { + continue + } + raw, err := os.ReadFile(filepath.Join(procRoot, e.Name(), "cmdline")) + if err != nil || len(raw) == 0 { + continue + } + if isSpxNATSService(bytes.Split(bytes.TrimRight(raw, "\x00"), []byte{0})) { + return true, nil + } + } + return false, nil +} + +// isSpxNATSService matches `spx service nats` with any leading path on spx. +func isSpxNATSService(argv [][]byte) bool { + return len(argv) >= 3 && + filepath.Base(string(argv[0])) == "spx" && + string(argv[1]) == "service" && + string(argv[2]) == "nats" +} + +// errNATSRunning is returned whenever a store would be discarded with the NATS +// service still up. No flag overrides it: the wipe would race the live server. +var errNATSRunning = errors.New("the NATS service is still running on this node; stop it first (sudo systemctl stop spinifex.target, then confirm `pgrep -af 'spx service'` prints nothing)") + +// checkStoreDiscardable fails if the NATS service is running locally. +func checkStoreDiscardable() error { + running, err := natsRunningLocally() + if err != nil { + return err + } + if running { + return errNATSRunning + } + return nil +} + +// discardJetStreamStore removes a node's JetStream store before it enters a +// multi-node cluster. nats-server adopts a same-named local stream and raft +// catch-up never repairs the prefix it already holds. +func discardJetStreamStore(storeDir string) error { + if err := checkStoreDiscardable(); err != nil { + return err + } + if err := os.RemoveAll(storeDir); err != nil { + return fmt.Errorf("remove JetStream store %s: %w", storeDir, err) + } + if _, err := os.Stat(storeDir); !errors.Is(err, os.ErrNotExist) { + return fmt.Errorf("JetStream store %s still present after removal", storeDir) + } + return nil +} + +// checkInitJetStreamStore refuses a multi-node init over a store holding streams +// unless discard was asked for. Keeping it as a seed is not safe: leadership +// ignores who holds data, and unassigned local streams are deleted as orphans. +func checkInitJetStreamStore(storeDir string, nodes int, discard bool) error { + if nodes < 2 { + return nil + } + streams, err := localStreams(storeDir) + if err != nil { + return err + } + if len(streams) == 0 { + return nil + } + if !discard { + return fmt.Errorf("this node's JetStream store %s holds %d stream(s) from before formation; "+ + "a multi-node cluster cannot adopt them consistently. "+ + "Re-run with --discard-jetstream to remove them once every node has joined", storeDir, len(streams)) + } + return checkStoreDiscardable() +} diff --git a/cmd/spinifex/cmd/jetstream_store_test.go b/cmd/spinifex/cmd/jetstream_store_test.go new file mode 100644 index 000000000..6c890c334 --- /dev/null +++ b/cmd/spinifex/cmd/jetstream_store_test.go @@ -0,0 +1,140 @@ +//test:in-package: swaps the unexported procRoot for a fake /proc tree and asserts on +// the store helpers the init and join commands call. + +package cmd + +import ( + "os" + "path/filepath" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// withFakeProc points the process scan at a temp tree holding the given +// pid -> cmdline entries (NUL-separated argv) for the duration of the test. +func withFakeProc(t *testing.T, procs map[string]string) { + t.Helper() + root := t.TempDir() + for pid, cmdline := range procs { + dir := filepath.Join(root, pid) + require.NoError(t, os.MkdirAll(dir, 0o755)) + require.NoError(t, os.WriteFile(filepath.Join(dir, "cmdline"), []byte(cmdline), 0o644)) + } + prev := procRoot + procRoot = root + t.Cleanup(func() { procRoot = prev }) +} + +// seedStore builds a JetStream store with the named streams under account $G, +// laid out the way nats-server writes it, and returns the store path. +func seedStore(t *testing.T, streams ...string) string { + t.Helper() + store := filepath.Join(t.TempDir(), "nats", "jetstream") + for _, s := range streams { + dir := filepath.Join(store, "$G", "streams", s, "msgs") + require.NoError(t, os.MkdirAll(dir, 0o755)) + require.NoError(t, os.WriteFile(filepath.Join(dir, "1.blk"), []byte("x"), 0o600)) + } + return store +} + +func TestJetStreamStoreDir(t *testing.T) { + assert.Equal(t, "/var/lib/spinifex/nats/jetstream", jetStreamStoreDir("/var/lib/spinifex")) +} + +func TestLocalStreams(t *testing.T) { + t.Run("missing store holds nothing", func(t *testing.T) { + got, err := localStreams(filepath.Join(t.TempDir(), "absent")) + require.NoError(t, err) + assert.Empty(t, got) + }) + + t.Run("lists streams across accounts, sorted", func(t *testing.T) { + store := seedStore(t, "KV_b", "KV_a") + require.NoError(t, os.MkdirAll(filepath.Join(store, "$SYS", "_js_", "_meta_"), 0o755)) + require.NoError(t, os.WriteFile(filepath.Join(store, "$G", "streams", "stray-file"), nil, 0o600)) + + got, err := localStreams(store) + require.NoError(t, err) + assert.Equal(t, []string{"$G/KV_a", "$G/KV_b"}, got) + }) +} + +func TestNATSRunningLocally(t *testing.T) { + tests := []struct { + name string + procs map[string]string + want bool + }{ + {"no processes", nil, false}, + {"installed service", map[string]string{"10": "/usr/local/bin/spx\x00service\x00nats\x00start\x00"}, true}, + {"bare spx on PATH", map[string]string{"10": "spx\x00service\x00nats\x00start\x00"}, true}, + // The join itself is an spx process and must not count. + {"spx admin join", map[string]string{"10": "spx\x00admin\x00join\x00--force\x00"}, false}, + {"other spx service", map[string]string{"10": "spx\x00service\x00predastore\x00start\x00"}, false}, + {"nats-server binary", map[string]string{"10": "nats-server\x00service\x00nats\x00"}, false}, + {"kernel thread, empty cmdline", map[string]string{"2": ""}, false}, + {"non-pid entry ignored", map[string]string{"self": "spx\x00service\x00nats\x00"}, false}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + withFakeProc(t, tt.procs) + got, err := natsRunningLocally() + require.NoError(t, err) + assert.Equal(t, tt.want, got) + }) + } +} + +func TestDiscardJetStreamStore(t *testing.T) { + t.Run("removes the store", func(t *testing.T) { + withFakeProc(t, nil) + store := seedStore(t, "KV_a") + require.NoError(t, discardJetStreamStore(store)) + _, err := os.Stat(store) + assert.ErrorIs(t, err, os.ErrNotExist) + }) + + t.Run("missing store is not an error", func(t *testing.T) { + withFakeProc(t, nil) + assert.NoError(t, discardJetStreamStore(filepath.Join(t.TempDir(), "absent"))) + }) + + t.Run("refuses under a running NATS and leaves the store", func(t *testing.T) { + withFakeProc(t, map[string]string{"7": "spx\x00service\x00nats\x00start\x00"}) + store := seedStore(t, "KV_a") + assert.ErrorIs(t, discardJetStreamStore(store), errNATSRunning) + _, err := os.Stat(store) + assert.NoError(t, err) + }) +} + +func TestCheckInitJetStreamStore(t *testing.T) { + withFakeProc(t, nil) + + t.Run("single-node init keeps its store", func(t *testing.T) { + assert.NoError(t, checkInitJetStreamStore(seedStore(t, "KV_a"), 1, false)) + }) + + t.Run("multi-node init over an empty store", func(t *testing.T) { + assert.NoError(t, checkInitJetStreamStore(filepath.Join(t.TempDir(), "absent"), 3, false)) + }) + + t.Run("multi-node init over streams is refused without the flag", func(t *testing.T) { + err := checkInitJetStreamStore(seedStore(t, "KV_a", "KV_b"), 4, false) + require.Error(t, err) + assert.Contains(t, err.Error(), "2 stream(s)") + assert.Contains(t, err.Error(), "--discard-jetstream") + }) + + t.Run("multi-node init over streams proceeds with the flag", func(t *testing.T) { + assert.NoError(t, checkInitJetStreamStore(seedStore(t, "KV_a"), 4, true)) + }) + + t.Run("flag does not override a running NATS", func(t *testing.T) { + withFakeProc(t, map[string]string{"7": "spx\x00service\x00nats\x00start\x00"}) + assert.ErrorIs(t, checkInitJetStreamStore(seedStore(t, "KV_a"), 4, true), errNATSRunning) + }) +} diff --git a/docs/COMMANDS.md b/docs/COMMANDS.md index 6701667ea..e38250339 100644 --- a/docs/COMMANDS.md +++ b/docs/COMMANDS.md @@ -95,8 +95,8 @@ Operational commands for inspecting cluster state. These fan out NATS requests t | Command | Flags | Description | |---------|-------|-------------| -| `spx admin init` | `--nodes`, `--node`, `--bind`, `--advertise` (external IP off-host clients dial; auto-detected from the default route), `--port`, `--region`, `--az`, `--cluster-name`, `--cluster-bind`, `--cluster-routes`, `--predastore-nodes`, `--predastore-compaction-interval` (seconds; test clusters set a short one), `--services`, `--formation-timeout`, `--token-ttl`, `--force`, `--skip-host-dns`, `--email` (operator address for update and security notices), `--no-telemetry`, `--gpu-passthrough`, `--ipsec` (default true; encrypt intra-AZ Geneve via OVN native IPsec cluster-wide — disable only for a trusted single-rack lab), `--host` (leader to join; falls back to multicast discovery), `--external-mode` (`pool` \| `nat` routed NAT for non-bridgeable uplinks, single-node; pair with `setup-ovn.sh --nat-uplink`; note: system instances (ECS/EKS/load-balancer agents) are unsupported in nat v1 — they require `pool` mode), `--external-iface` (WAN NIC for br-external, auto-detected), `--external-source` (`dhcp` \| `static`), `--external-pool` (`start-end`), `--external-prefix-len` (default 24), `--external-gateway` (auto-detected), `--external-bind-bridge` (default `br-wan` when `--external-source=dhcp`), `--gateway-ip` (OVN gateway router's external IP for SNAT; defaults to the pool's range start, required for `nat` mode without DHCP) | Generates root IAM credentials (AKIA-prefixed access key + secret) → creates master.key (AES-256, 32 bytes, 0600) → writes bootstrap.json (consumed on first start) → generates CA + server TLS certificates → generates join token (written to `join-token` file, displayed in join command) → creates NATS config with auth token → writes spinifex.toml, awsgw.toml, predastore.toml → configures AWS CLI `spx` profile → creates directory structure under `~/spinifex/` → points the host resolver at this node's local northstar so the Spinifex zones (ELB/EKS names) resolve from the node itself (opt out with `--skip-host-dns`). `--force` on an already-initialized node is idempotent for crypto: it preserves the existing keys (master.key, predastore/viperblock encryption keys), system + admin credentials, and CA, and refreshes only the config files and the CA-signed server certificate (so a changed bind IP / SANs is picked up without breaking already-joined nodes or CA-baked AMIs). A genuine clean slate requires removing the data dirs first. | -| `spx admin join` | `--host` (required), `--node` (required), `--token` (required), `--bind`, `--advertise` (defaults to `--bind`, or the auto-detected WAN IP when `--bind` is 0.0.0.0), `--port`, `--region`, `--az`, `--cluster-bind`, `--cluster-routes`, `--data-dir`, `--services`, `--skip-host-dns`, `--email`, `--no-telemetry`, `--predastore-compaction-interval`, `--join-timeout` (how long to keep retrying while the formation server is unreachable, default 20m), `--force` (join even though this node is already initialized, **discarding its own CA and master key**) | Connects to leader node with join token (Authorization: Bearer header) → retrieves cluster configuration → configures local node to join cluster and participate in distributed operations → points the host resolver at this node's local northstar so the Spinifex zones resolve from the node itself (opt out with `--skip-host-dns`) | +| `spx admin init` | `--nodes`, `--node`, `--bind`, `--advertise` (external IP off-host clients dial; auto-detected from the default route), `--port`, `--region`, `--az`, `--cluster-name`, `--cluster-bind`, `--cluster-routes`, `--predastore-nodes`, `--predastore-compaction-interval` (seconds; test clusters set a short one), `--services`, `--formation-timeout`, `--token-ttl`, `--force`, `--discard-jetstream` (with `--nodes` >= 2, remove this node's existing JetStream streams once every node has joined; without it a multi-node init over a store that holds streams is refused, and it never proceeds while NATS is running), `--skip-host-dns`, `--email` (operator address for update and security notices), `--no-telemetry`, `--gpu-passthrough`, `--ipsec` (default true; encrypt intra-AZ Geneve via OVN native IPsec cluster-wide — disable only for a trusted single-rack lab), `--host` (leader to join; falls back to multicast discovery), `--external-mode` (`pool` \| `nat` routed NAT for non-bridgeable uplinks, single-node; pair with `setup-ovn.sh --nat-uplink`; note: system instances (ECS/EKS/load-balancer agents) are unsupported in nat v1 — they require `pool` mode), `--external-iface` (WAN NIC for br-external, auto-detected), `--external-source` (`dhcp` \| `static`), `--external-pool` (`start-end`), `--external-prefix-len` (default 24), `--external-gateway` (auto-detected), `--external-bind-bridge` (default `br-wan` when `--external-source=dhcp`), `--gateway-ip` (OVN gateway router's external IP for SNAT; defaults to the pool's range start, required for `nat` mode without DHCP) | Generates root IAM credentials (AKIA-prefixed access key + secret) → creates master.key (AES-256, 32 bytes, 0600) → writes bootstrap.json (consumed on first start) → generates CA + server TLS certificates → generates join token (written to `join-token` file, displayed in join command) → creates NATS config with auth token → writes spinifex.toml, awsgw.toml, predastore.toml → configures AWS CLI `spx` profile → creates directory structure under `~/spinifex/` → points the host resolver at this node's local northstar so the Spinifex zones (ELB/EKS names) resolve from the node itself (opt out with `--skip-host-dns`). `--force` on an already-initialized node is idempotent for crypto: it preserves the existing keys (master.key, predastore/viperblock encryption keys), system + admin credentials, and CA, and refreshes only the config files and the CA-signed server certificate (so a changed bind IP / SANs is picked up without breaking already-joined nodes or CA-baked AMIs). A genuine clean slate requires removing the data dirs first. | +| `spx admin join` | `--host` (required), `--node` (required), `--token` (required), `--bind`, `--advertise` (defaults to `--bind`, or the auto-detected WAN IP when `--bind` is 0.0.0.0), `--port`, `--region`, `--az`, `--cluster-bind`, `--cluster-routes`, `--data-dir`, `--services`, `--skip-host-dns`, `--email`, `--no-telemetry`, `--predastore-compaction-interval`, `--join-timeout` (how long to keep retrying while the formation server is unreachable, default 20m), `--force` (join even though this node is already initialized or holds JetStream streams, **discarding its own CA, master key and JetStream store**; the store is removed only after formation succeeds, and the join is refused while NATS is running even with `--force`) | Connects to leader node with join token (Authorization: Bearer header) → retrieves cluster configuration → configures local node to join cluster and participate in distributed operations → points the host resolver at this node's local northstar so the Spinifex zones resolve from the node itself (opt out with `--skip-host-dns`) | ### Version @@ -108,6 +108,7 @@ Operational commands for inspecting cluster state. These fan out NATS requests t | Command | Flags | Description | |---------|-------|-------------| +| `spx admin kv digest` | `--store-dir` (default `/nats/jetstream`), `--work-dir` (holds the temporary copy, default the system temp dir), `--stream` (repeatable filter), `--seqs` (also print a hash per sequence) | Read-only per-replica consistency check. Copies this node's `$G` streams tree (without consumer state) to a temporary directory, opens the copy in a private in-process NATS server with no listener, and prints one line per stream with first/last sequence, message count and a SHA-256 over every stored message (sequence, subject, timestamp, headers, data). Run it on every node with `sudo` and diff the output with the `#` lines removed: replicas of one stream must match. A direct get over NATS is not used because any replica may answer it. Writes landing during the copy can change a replica's tail, so gate API writes or re-run before calling a tail difference divergence. | | `spx admin cluster shutdown` | `--force` (shutdown even if nodes don't respond), `--timeout` (max wait per phase, default 120s), `--dry-run` (print phase plan without executing) | Performs coordinated, phased shutdown of entire cluster. Phases execute in order: GATE (stop API/UI) → DRAIN (stop VMs) → STORAGE (stop viperblock) → PERSIST (stop predastore) → INFRA (stop NATS/daemon). Each phase waits for all nodes to ACK before proceeding. Uses JetStream state tracking. | | `spx admin cluster drain-dhcp` | `--timeout` (reply-collection window, default 30s) | Asks each vpcd to DHCPRELEASE every external-pool DHCP lease it currently holds, returning them to the upstream DHCP server. Run on teardown before stopping services — an env reset otherwise strands held leases upstream until their TTL expires, eventually exhausting the upstream scope. Best-effort: warns and exits 0 if the cluster is already down. | | `spx admin node drain --local` | `--local` (drain the local node only, required), `--timeout` (max wait per phase, default 120s), `--unless-restarting` (drain on a real host shutdown or a plain target stop, and skip only when a restart of the stack is already queued; unset, it always drains) | Runs the GATE and DRAIN phases against the local node only: powers down its guests via QMP and unmounts their volumes (flushing the viperblock WAL) while every service is still up. STORAGE/PERSIST/INFRA are left to systemd's ordered teardown. Wired as the `ExecStop` of `spinifex-shutdown.service`, so `systemctl stop spinifex.target` and host reboot/poweroff drain guests before any storage service stops. | diff --git a/docs/install/install-multi-node/README.md b/docs/install/install-multi-node/README.md index 09f103bd5..b3348dbe8 100644 --- a/docs/install/install-multi-node/README.md +++ b/docs/install/install-multi-node/README.md @@ -235,7 +235,9 @@ sudo ovn-sbctl show Run init and join **concurrently** — init blocks until all nodes have joined. -`--force` is in every command below so the sequence is identical whichever way you installed. It does the work on ISO-installed servers, which arrive as their own single-node cluster: joining replaces that server's CA and master key with server 1's, and `--force` is the confirmation. On `spx admin init` it is idempotent — existing keys, credentials and CA are preserved, and only the config files and server certificate are refreshed. On a freshly installed server there is nothing to lose either way. +`--force` is in every command below so the sequence is identical whichever way you installed. It does the work on ISO-installed servers, which arrive as their own single-node cluster: joining replaces that server's CA and master key with server 1's and removes its JetStream store, and `--force` is the confirmation. On `spx admin init` it is idempotent — existing keys, credentials and CA are preserved, and only the config files and server certificate are refreshed. On a freshly installed server there is nothing to lose either way. + +`--discard-jetstream` on server 1 removes the streams its single-node cluster wrote, once every server has joined. Without it init refuses to form a cluster over them, because NATS would adopt each server's own copy as a replica and those copies never converge. Stop `spinifex.target` on every server first and confirm `pgrep -af 'spx service'` prints nothing: init and join both refuse while NATS is still running. > [!WARNING] > Do not point these commands at a server that has already been in service. Joining discards its master key, orphaning every volume and fragment sealed under it. That is what `--force` overrides, and it is unrecoverable. @@ -243,7 +245,7 @@ Run init and join **concurrently** — init blocks until all nodes have joined. **Server 1 — initialize:** ```bash -sudo spx admin init --force \ +sudo spx admin init --force --discard-jetstream \ --node node1 --nodes 3 \ --bind $SPINIFEX_NODE1 --cluster-bind $SPINIFEX_NODE1 \ --port 4432 --region $AWS_REGION --az $AWS_AZ diff --git a/go.mod b/go.mod index 1e068982c..d68b0c730 100644 --- a/go.mod +++ b/go.mod @@ -21,6 +21,7 @@ require ( github.com/jackc/pgx/v5 v5.11.0 github.com/klauspost/cpuid/v2 v2.4.0 github.com/miekg/dns v1.1.73 + github.com/minio/highwayhash v1.0.4 github.com/mulgadc/bluebottle v1.19.0 github.com/mulgadc/northstar v1.19.0 github.com/mulgadc/predastore v1.19.0 @@ -146,7 +147,6 @@ require ( github.com/mdlayher/packet v1.1.2 // indirect github.com/mdlayher/socket v0.5.1 // indirect github.com/minio/crc64nvme v1.1.1 // indirect - github.com/minio/highwayhash v1.0.4 // indirect github.com/moby/locker v1.0.1 // indirect github.com/moby/sys/capability v0.4.0 // indirect github.com/moby/sys/mountinfo v0.7.2 // indirect diff --git a/scripts/install-node.sh b/scripts/install-node.sh index 7431e4e35..29a28c959 100755 --- a/scripts/install-node.sh +++ b/scripts/install-node.sh @@ -368,6 +368,22 @@ for host in "${HOSTS[@]}"; do on "$host" "sudo systemctl stop spinifex.target" || fail "$host: could not stop spinifex.target" done +# Init and join remove each node's JetStream store and refuse while NATS still +# runs. A stopped target is not proof of that: its storage units can outlive it. +# Matched on the service argv so the ssh shell running this cannot match itself. +for host in "${HOSTS[@]}"; do + $DRY_RUN && { echo " [dry-run] $host: wait for every 'spx service' process to exit"; continue; } + on "$host" "for u in spinifex-predastore spinifex-viperblock spinifex-nats; do + systemctl is-active --quiet \$u && sudo systemctl stop \$u + done + for _ in \$(seq 1 45); do + pgrep -f '^(/usr/local/bin/)?spx service ' >/dev/null || exit 0 + sleep 2 + done + pgrep -af '^(/usr/local/bin/)?spx service ' >&2 + exit 1" || fail "$host: spx services still running 90s after stopping spinifex.target" +done + # --- Disarm the host firewall ---------------------------------------------- # # The policy lives in the kernel and outlives the services, so stopping the @@ -529,6 +545,7 @@ done # target. init_args=( --force + --discard-jetstream --node "${NODE_NAMES[0]}" --nodes "$N" --bind "${LAN_IPS[0]}" @@ -577,8 +594,8 @@ else fi # --force: each joining node arrived from the ISO already initialized, with its -# own CA and master key. Joining replaces them with the init node's, and -# `spx admin join` refuses to do that silently. +# own CA, master key and JetStream store. Joining replaces the keys with the init +# node's and removes the store, and `spx admin join` refuses to do that silently. # # Every join must run concurrently. `spx admin join` registers and then blocks # until the formation server has all N nodes, so joining one at a time deadlocks diff --git a/spinifex/daemon/instance_records_vpcd_test.go b/spinifex/daemon/instance_records_vpcd_test.go new file mode 100644 index 000000000..bfca81aba --- /dev/null +++ b/spinifex/daemon/instance_records_vpcd_test.go @@ -0,0 +1,17 @@ +package daemon_test + +import ( + "testing" + + "github.com/mulgadc/spinifex/spinifex/daemon" + "github.com/mulgadc/spinifex/spinifex/network/reconcile" + "github.com/stretchr/testify/assert" +) + +// vpcd reads instance records to tell a stopped guest's port from a broken one, +// and cannot import this package to learn where they live. If these drift apart +// it silently probes every stopped instance's port again. +func TestInstanceRecordLocationMatchesVPCD(t *testing.T) { + assert.Equal(t, daemon.InstanceStateBucket, reconcile.InstanceRecordBucket) + assert.Equal(t, daemon.InstanceRecordPrefix, reconcile.InstanceRecordPrefix) +} diff --git a/spinifex/gateway/requestaudit.go b/spinifex/gateway/requestaudit.go index 7e1a6c6eb..242437e76 100644 --- a/spinifex/gateway/requestaudit.go +++ b/spinifex/gateway/requestaudit.go @@ -4,12 +4,14 @@ import ( "context" "log/slog" "net/http" + "regexp" "sync" "time" "go.opentelemetry.io/otel/attribute" "go.opentelemetry.io/otel/trace" + "github.com/mulgadc/spinifex/spinifex/otelsetup" "github.com/mulgadc/spinifex/spinifex/utils" ) @@ -26,6 +28,7 @@ const ctxAudit contextKey = "gateway.audit" type requestAudit struct { mu sync.Mutex clientIP string + requestID string accessKeyID string accountID string region string @@ -40,13 +43,27 @@ type requestAudit struct { // logging happens to be enabled. func requestAuditMiddleware(next http.Handler) http.Handler { return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - audit := &requestAudit{clientIP: utils.ClientIP(r.RemoteAddr)} + audit := &requestAudit{clientIP: utils.RequestClientIP(r), requestID: requestIDFrom(r)} ctx := context.WithValue(r.Context(), ctxAudit, audit) next.ServeHTTP(w, r.WithContext(ctx)) audit.annotate(ctx) }) } +// requestIDPattern is what a correlation ID may look like. nginx's $request_id is +// 32 hex characters; anything else a client sends is dropped rather than logged. +var requestIDPattern = regexp.MustCompile(`^[A-Za-z0-9._:-]{1,128}$`) + +// requestIDFrom returns the X-Request-ID the edge stamped, so an awsgw line can +// be joined to the edge access-log line that produced it. +func requestIDFrom(r *http.Request) string { + id := r.Header.Get("X-Request-ID") + if !requestIDPattern.MatchString(id) { + return "" + } + return id +} + // auditFrom returns the request's audit record, or nil when there is none. Every // method below is nil-safe, so a handler called directly by a test behaves // exactly as it did before this record existed. @@ -164,12 +181,13 @@ func (a *requestAudit) pairs() []auditField { {"action", "aws.action", a.action}, {"principalType", "aws.principal_type", a.principalType}, {"authError", "aws.auth_error", a.authError}, + {"requestID", "http.request.header.x-request-id", a.requestID}, } } // logRequest emits the access log line for a finished request, carrying the // audit fields alongside the HTTP result. func logRequest(r *http.Request, status int, duration time.Duration) { - attrs := []any{"method", r.Method, "path", r.URL.Path, "status", status, "duration", duration} + attrs := []any{"method", r.Method, "path", r.URL.Path, "status", status, "duration_ms", otelsetup.Millis(duration)} slog.InfoContext(r.Context(), "request", append(attrs, auditFrom(r.Context()).fields()...)...) } diff --git a/spinifex/gateway/requestaudit_test.go b/spinifex/gateway/requestaudit_test.go index 247a6dd77..58526928f 100644 --- a/spinifex/gateway/requestaudit_test.go +++ b/spinifex/gateway/requestaudit_test.go @@ -1,10 +1,14 @@ package gateway import ( + "bytes" "context" "fmt" + "log/slog" + "net" "net/http" "net/http/httptest" + "strings" "testing" "time" @@ -62,7 +66,7 @@ func TestRequestAuditCarriesNoSecret(t *testing.T) { } assert.Equal(t, []string{ "sourceIP", "accessKeyID", "accountID", "region", "service", "action", - "principalType", "authError", + "principalType", "authError", "requestID", }, keys) } @@ -170,3 +174,83 @@ func TestRequestAuditRecordsRateLimitLockout(t *testing.T) { assert.Equal(t, "10.15.8.13", audit().clientIP) assert.Equal(t, awserrors.ErrorRequestLimitExceeded, audit().authError) } + +// sourceIP must not be something a direct caller can choose. X-Real-IP is the +// edge's word only on a loopback connection; from anywhere else both it and +// X-Forwarded-For are ignored. +func TestRequestAuditClientIPIgnoresSpoofedHeadersFromNonLoopback(t *testing.T) { + handler, audit := auditRouter(t, nil) + + for _, remote := range []string{"203.0.113.9:5555", "10.2.0.3:5555", "[2001:db8::9]:5555"} { + req := httptest.NewRequest(http.MethodGet, "/", nil) + req.RemoteAddr = remote + req.Header.Set("X-Real-IP", "127.0.0.1") + req.Header.Set("X-Forwarded-For", "198.51.100.1") + handler.ServeHTTP(httptest.NewRecorder(), req) + + require.NotNil(t, audit()) + host, _, err := net.SplitHostPort(remote) + require.NoError(t, err) + assert.Equal(t, host, audit().clientIP, "spoofed headers from %s must not be logged as the source", remote) + } +} + +// Through the edge (or the console proxy) the connection is loopback, and the +// edge's X-Real-IP is the real client. +func TestRequestAuditClientIPTrustsRealIPFromLoopback(t *testing.T) { + handler, audit := auditRouter(t, nil) + + req := httptest.NewRequest(http.MethodGet, "/", nil) + req.RemoteAddr = "127.0.0.1:40000" + req.Header.Set("X-Real-IP", "198.51.100.7") + req.Header.Set("X-Forwarded-For", "203.0.113.1") + handler.ServeHTTP(httptest.NewRecorder(), req) + + require.NotNil(t, audit()) + assert.Equal(t, "198.51.100.7", audit().clientIP) +} + +// The edge's request ID joins an awsgw line to its access-log line; a value that +// does not look like an ID is dropped rather than written into the log. +func TestRequestAuditRecordsRequestID(t *testing.T) { + handler, audit := auditRouter(t, nil) + + cases := map[string]string{ + "4f1c2b7e9a0d4e6f8b3c5a7d9e1f2a3b": "4f1c2b7e9a0d4e6f8b3c5a7d9e1f2a3b", + "": "", + "has space": "", + strings.Repeat("a", 129): "", + } + for header, want := range cases { + req := httptest.NewRequest(http.MethodGet, "/", nil) + req.RemoteAddr = "127.0.0.1:40000" + if header != "" { + req.Header.Set("X-Request-ID", header) + } + handler.ServeHTTP(httptest.NewRecorder(), req) + + require.NotNil(t, audit()) + assert.Equal(t, want, audit().requestID, "X-Request-ID %q", header) + } +} + +// slog renders a bare time.Duration as unitless nanoseconds, so the access line +// carries duration_ms instead. +func TestLogRequestDurationInMilliseconds(t *testing.T) { + buf := &bytes.Buffer{} + prev := slog.Default() + slog.SetDefault(slog.New(slog.NewTextHandler(buf, &slog.HandlerOptions{Level: slog.LevelInfo}))) + t.Cleanup(func() { slog.SetDefault(prev) }) + + req := httptest.NewRequest(http.MethodGet, "/", nil) + audit := &requestAudit{clientIP: "198.51.100.7", requestID: "abc123"} + req = req.WithContext(context.WithValue(req.Context(), ctxAudit, audit)) + + logRequest(req, http.StatusOK, 1500*time.Millisecond) + + out := buf.String() + assert.Contains(t, out, "duration_ms=1500") + assert.NotContains(t, out, "duration=") + assert.Contains(t, out, "sourceIP=198.51.100.7") + assert.Contains(t, out, "requestID=abc123") +} diff --git a/spinifex/network/host/ipsec.go b/spinifex/network/host/ipsec.go index dfbd04e0e..d13f02cab 100644 --- a/spinifex/network/host/ipsec.go +++ b/spinifex/network/host/ipsec.go @@ -37,28 +37,41 @@ func EnableIPSecEncapsulation() error { return nil } -// errNBUnreachable marks the one failure that means "this node has no local NB -// DB" rather than "the read broke". A permission change on the socket, a missing -// binary or a timed-out transaction are all real faults on a management node, -// and lumping them in with this would silence them. -var errNBUnreachable = errors.New("no local OVN NB DB") +// errNBUnreachable marks the one failure that means "no NB DB answered" rather +// than "the read broke". A permission change on the socket, a missing binary or +// a timed-out transaction are real faults, and lumping them in would silence them. +var errNBUnreachable = errors.New("no OVN NB DB reachable") // nbUnreachablePatterns are what ovsdb's client library prints when it cannot -// open or complete a handshake on the socket. +// open or complete a handshake with any of its remotes. var nbUnreachablePatterns = []string{ "database connection failed", "connection refused", "no such file or directory", } -// GetNBGlobalIPSec reads NB_Global.ipsec from the local OVN NB DB. The error -// doubles as the reachability answer: a present socket file says nothing about -// whether the database behind it accepts connections yet. +// nbctlArgs targets nbAddr (the cluster remote list; empty is the local socket). +// Reads accept any connected member, so a raft follower answers instead of +// refusing; writes stay leader-only so ovn-nbctl walks the remotes to the leader. +func nbctlArgs(nbAddr string, leaderOnly bool, cmd ...string) []string { + var args []string + if nbAddr != "" { + args = append(args, "--db="+nbAddr) + } + if !leaderOnly { + args = append(args, "--no-leader-only") + } + return append(append(args, "--timeout=5"), cmd...) +} + +// GetNBGlobalIPSec reads NB_Global.ipsec. The error doubles as the reachability +// answer: a present socket file says nothing about whether the database behind +// it accepts connections yet. // // Reads stdout alone. ovn-nbctl writes vlog lines to stderr on a successful run, // and folding those into the value parses a live "true" as false. -func GetNBGlobalIPSec() (bool, error) { - cmd := utils.SudoCommand("ovn-nbctl", "--timeout=5", "get", "NB_Global", ".", "ipsec") +func GetNBGlobalIPSec(nbAddr string) (bool, error) { + cmd := utils.SudoCommand("ovn-nbctl", nbctlArgs(nbAddr, false, "get", "NB_Global", ".", "ipsec")...) var stderr bytes.Buffer cmd.Stderr = &stderr out, err := cmd.Output() @@ -83,15 +96,15 @@ func isNBUnreachable(msg string) bool { return false } -// SetNBGlobalIPSec writes NB_Global.ipsec on the local OVN NB DB, triggering -// ovn-controller to add options:remote_name to Geneve tunnels for strongSwan. -// Only the management node has a reachable NB DB; callers gate on that. -func SetNBGlobalIPSec(enable bool) error { +// SetNBGlobalIPSec writes NB_Global.ipsec, triggering ovn-controller to add +// options:remote_name to Geneve tunnels for strongSwan. Callers elect a single +// writer, so this never races another node's write. +func SetNBGlobalIPSec(nbAddr string, enable bool) error { val := "false" if enable { val = "true" } - out, err := utils.SudoCommand("ovn-nbctl", "--timeout=5", "set", "NB_Global", ".", "ipsec="+val).CombinedOutput() + out, err := utils.SudoCommand("ovn-nbctl", nbctlArgs(nbAddr, true, "set", "NB_Global", ".", "ipsec="+val)...).CombinedOutput() if err != nil { return fmt.Errorf("set NB_Global ipsec=%s: %s: %w", val, strings.TrimSpace(string(out)), err) } diff --git a/spinifex/network/host/ipsec_enable.go b/spinifex/network/host/ipsec_enable.go index ec2edd4fd..d77f2d3f3 100644 --- a/spinifex/network/host/ipsec_enable.go +++ b/spinifex/network/host/ipsec_enable.go @@ -36,12 +36,21 @@ type IPSecNodeStatus struct { // Ready reports that this node's local IPsec configuration is complete. Ready bool - // NBReachable reports that this node can read and write NB_Global. Only - // management nodes can, and which nodes those are is not in the config, so - // they say so here and every node elects the same writer from the answer. + // NBReachable reports that this node read NB_Global this pass, over its + // configured NB remotes or its local socket. Reachability can differ per node, + // so each says so here and every node elects the same writer from the answer. NBReachable bool } +// nbAddrFor returns this node's configured NB remote list, or "" for the local +// socket. The list reaches the cluster from a node that runs no NB DB at all. +func nbAddrFor(clusterConfig *config.ClusterConfig) string { + if clusterConfig == nil { + return "" + } + return clusterConfig.Nodes[clusterConfig.Node].VPCD.OVNNBAddr +} + // IPSecBarrier carries each node's IPsec state across the cluster. // NB_Global.ipsec is cluster-wide, so asserting it from one node's local // knowledge is what lets a partially configured mesh black-hole guest traffic. @@ -156,7 +165,8 @@ func DisableOVNIPSec(ctx context.Context, clusterConfig *config.ClusterConfig, b } } - current, err := GetNBGlobalIPSec() + nbAddr := nbAddrFor(clusterConfig) + current, err := GetNBGlobalIPSec(nbAddr) if err != nil { return skipUnlessNBReadable(err) } @@ -164,15 +174,15 @@ func DisableOVNIPSec(ctx context.Context, clusterConfig *config.ClusterConfig, b return nil } slog.Info("ipsec: switched off for this cluster, releasing the cluster-wide encryption requirement") - return SetNBGlobalIPSec(false) + return SetNBGlobalIPSec(nbAddr, false) } -// skipUnlessNBReadable turns "this node has no local NB DB" into a quiet skip -// and leaves every other failure — a socket it may not open, a missing binary, -// a timed-out transaction — as a real error, so the pass retries and says so. +// skipUnlessNBReadable turns "no NB DB answered" into a quiet skip and leaves +// every other failure — a socket it may not open, a missing binary, a timed-out +// transaction — as a real error, so the pass retries and says so. func skipUnlessNBReadable(err error) error { if errors.Is(err, errNBUnreachable) { - slog.Debug("ipsec: no local OVN NB DB, leaving NB_Global to a management node", "err", err) + slog.Debug("ipsec: no OVN NB DB reachable, leaving NB_Global to a node that can read it", "err", err) return nil } return err @@ -232,7 +242,8 @@ func EnableOVNIPSec(ctx context.Context, configPath string, clusterConfig *confi // Probed before the local half, because the answer is published alongside it: // which nodes can write NB_Global is not in the config, so they have to say. - current, nbErr := GetNBGlobalIPSec() + nbAddr := nbAddrFor(clusterConfig) + current, nbErr := GetNBGlobalIPSec(nbAddr) if nbErr != nil && !errors.Is(nbErr, errNBUnreachable) { return nbErr } @@ -256,7 +267,7 @@ func EnableOVNIPSec(ctx context.Context, configPath string, clusterConfig *confi if nbErr != nil { return skipUnlessNBReadable(nbErr) } - return reconcileNBGlobalIPSec(ctx, clusterConfig, barrier, current) + return reconcileNBGlobalIPSec(ctx, clusterConfig, barrier, nbAddr, current) } // configureLocalIPSec does this node's own half: point OVS at the peer cert and @@ -298,12 +309,12 @@ func publishStatus(ctx context.Context, clusterConfig *config.ClusterConfig, bar // and ovs-monitor-ipsec materialises no strongSwan connections; with it, a // chassis that has not finished its own setup silently drops guest traffic. So // it tracks the slowest live chassis, not this one. -func reconcileNBGlobalIPSec(ctx context.Context, clusterConfig *config.ClusterConfig, barrier IPSecBarrier, current bool) error { +func reconcileNBGlobalIPSec(ctx context.Context, clusterConfig *config.ClusterConfig, barrier IPSecBarrier, nbAddr string, current bool) error { if barrier == nil || clusterConfig == nil { if current { return nil } - return SetNBGlobalIPSec(true) + return SetNBGlobalIPSec(nbAddr, true) } cluster, err := barrier.Cluster(ctx, nodeNames(clusterConfig)) @@ -313,7 +324,7 @@ func reconcileNBGlobalIPSec(ctx context.Context, clusterConfig *config.ClusterCo return fmt.Errorf("read IPsec cluster state: %w", err) } - // One writer. Every management node can reach the NB DB, so without this they + // One writer. Every node that reads NB_Global can also write it, so without this they // race on one cluster-global row from snapshots taken at different instants, // and two that disagree for a single pass flip the flag — tearing down and // rebuilding every strongSwan connection in the cluster on each flip. @@ -329,7 +340,7 @@ func reconcileNBGlobalIPSec(ctx context.Context, clusterConfig *config.ClusterCo return nil case len(pending) == 0: slog.Info("ipsec: every live chassis reports a complete configuration, requiring encryption cluster-wide") - return SetNBGlobalIPSec(true) + return SetNBGlobalIPSec(nbAddr, true) case !current: slog.Info("ipsec: holding encryption off until every chassis is configured", "pending", pending) return nil @@ -338,7 +349,7 @@ func reconcileNBGlobalIPSec(ctx context.Context, clusterConfig *config.ClusterCo // Plaintext Geneve is where the cluster sat before IPsec was asked for, and it // is both recoverable and visible. A black hole is neither. slog.Error("ipsec: retracting cluster-wide encryption, chassis are unconfigured and cross-chassis guest traffic would black-hole", "pending", pending) - if err := SetNBGlobalIPSec(false); err != nil { + if err := SetNBGlobalIPSec(nbAddr, false); err != nil { slog.Error("ipsec: NB_Global still demands encryption that unconfigured chassis cannot perform, cross-chassis guest traffic is being dropped now", "pending", pending, "err", err) return err @@ -347,8 +358,8 @@ func reconcileNBGlobalIPSec(ctx context.Context, clusterConfig *config.ClusterCo } // nbGlobalWriter picks the one node that may write NB_Global this pass: the -// first live management node in name order. Every node computes it from the same -// published set, so they agree without a lock. +// first live node in name order that could read it. Every node computes it from +// the same published set, so they agree without a lock. func nbGlobalWriter(cluster map[string]IPSecNodeStatus) string { var writer string for _, node := range slices.Sorted(maps.Keys(cluster)) { diff --git a/spinifex/network/host/ipsec_enable_test.go b/spinifex/network/host/ipsec_enable_test.go index c3643b7fb..39c01faa2 100644 --- a/spinifex/network/host/ipsec_enable_test.go +++ b/spinifex/network/host/ipsec_enable_test.go @@ -33,15 +33,15 @@ type recordingSudo struct { activeOutput string enabledOutput map[string]string activePerUnit map[string]string - // nbUnreachable stands in for a node whose ovn-central is not answering, or - // which is not a management node at all. The socket file exists in both - // cases, which is why its presence was never the right question. + // nbUnreachable stands in for a node that no NB DB answered, over its + // remotes or its local socket. A socket file can exist either way, which is + // why its presence was never the right question. nbUnreachable bool // nbIPSec is what NB_Global.ipsec currently reads as. nbIPSec string - // nbError is a failure that is not "there is no local NB DB here" — a - // permission change on the socket, a missing binary, a timed-out - // transaction. On a management node each of those is a real fault. + // nbError is a failure that is not "no NB DB answered" — a permission + // change on the socket, a missing binary, a timed-out transaction. Each of + // those is a real fault. nbError string } @@ -122,7 +122,7 @@ func (f *fakeBarrier) Cluster(_ context.Context, _ []string) (map[string]IPSecNo } // allReady is the steady state: every chassis has finished its local setup and -// node1 is the management node that owns the NB_Global write. +// node1 is the first node that reads NB_Global, so it owns the write. func allReady() *fakeBarrier { return &fakeBarrier{cluster: map[string]IPSecNodeStatus{ "node1": {Ready: true, NBReachable: true}, @@ -521,7 +521,7 @@ func (f *flakyBarrier) succeeded() bool { return f.calls > f.failFor } -// Every management node can reach the NB DB, so without a single elected writer +// Every node that reads NB_Global can write it, so without a single elected writer // two of them disagreeing for one pass flip the flag and rebuild every // strongSwan connection in the cluster. func TestEnableOVNIPSec_OnlyTheElectedWriterTouchesNBGlobal(t *testing.T) { @@ -542,8 +542,8 @@ func TestEnableOVNIPSec_OnlyTheElectedWriterTouchesNBGlobal(t *testing.T) { "a node that is not the elected writer must not race the flag") } -// The elected writer is the first live management node, not merely the first -// node: a hypervisor cannot write the flag and must not veto the one that can. +// The elected writer is the first live node that could read NB_Global, not +// merely the first node: one that cannot reach NB must not veto one that can. func TestNBGlobalWriter_SkipsNodesWithoutANBDB(t *testing.T) { writer := nbGlobalWriter(map[string]IPSecNodeStatus{ "node1": {Ready: true}, @@ -606,8 +606,8 @@ func TestReconcileOVNIPSec_DisabledIsANoOpWhenAlreadyReleased(t *testing.T) { assert.Empty(t, recorder.nbctlWrites()) } -// A management node whose NB read fails for any reason other than "there is no -// local DB here" has a real fault. Swallowing it was how the node that holds +// A node whose NB read fails for any reason other than "no NB DB answered" has +// a real fault. Swallowing it was how the node that holds // the flag could go blind and never retract. func TestEnableOVNIPSec_NBReadFailureFailsThePass(t *testing.T) { recorder := &recordingSudo{nbError: "ovn-nbctl: unix:/var/run/ovn/ovnnb_db.sock: permission denied"} @@ -621,7 +621,7 @@ func TestEnableOVNIPSec_NBReadFailureFailsThePass(t *testing.T) { assert.Empty(t, recorder.nbctlWrites()) } -// A node with no local NB DB is the ordinary hypervisor case, not a fault. +// No NB DB answering is skipped quietly, not a fault, and the node stands down. func TestEnableOVNIPSec_NoLocalNBDBIsNotAnError(t *testing.T) { recorder := &recordingSudo{nbUnreachable: true} t.Cleanup(utils.SetSudoCommandForTest(recorder.stub)) @@ -645,7 +645,59 @@ func TestGetNBGlobalIPSec_IgnoresStderrChatter(t *testing.T) { return exec.Command("sh", "-c", `echo "ovsdb-idl|WARN|reconnecting" >&2; echo true`) })) - value, err := GetNBGlobalIPSec() + value, err := GetNBGlobalIPSec("") require.NoError(t, err) assert.True(t, value) } + +// nbctlReads returns the ovn-nbctl invocations that read NB_Global. +func (r *recordingSudo) nbctlReads() [][]string { + var out [][]string + for _, run := range r.runs { + if run[0] == "ovn-nbctl" && slices.Contains(run, "get") { + out = append(out, run) + } + } + return out +} + +// A raft follower refuses a leader-only read on its own socket, which logged an +// ERR every minute and read as "no NB DB here". The read must accept any +// connected member; the write must still reach the leader. +func TestNBctlArgs_ReadsAcceptFollowersWritesStayLeaderOnly(t *testing.T) { + const nb = "tcp:10.0.0.1:6641,tcp:10.0.0.2:6641" + + assert.Equal(t, []string{"--no-leader-only", "--timeout=5", "get", "NB_Global", ".", "ipsec"}, + nbctlArgs("", false, "get", "NB_Global", ".", "ipsec")) + assert.Equal(t, []string{"--db=" + nb, "--no-leader-only", "--timeout=5", "get", "NB_Global", ".", "ipsec"}, + nbctlArgs(nb, false, "get", "NB_Global", ".", "ipsec")) + assert.Equal(t, []string{"--db=" + nb, "--timeout=5", "set", "NB_Global", ".", "ipsec=false"}, + nbctlArgs(nb, true, "set", "NB_Global", ".", "ipsec=false")) +} + +// A node that runs no NB DB reaches the cluster through its configured remotes, +// so both the read and the retraction carry the list rather than the local socket. +func TestReconcileOVNIPSec_UsesTheConfiguredNBRemotes(t *testing.T) { + const nb = "tcp:10.0.0.1:6641,tcp:10.0.0.2:6641" + recorder := &recordingSudo{nbIPSec: "true"} + t.Cleanup(utils.SetSudoCommandForTest(recorder.stub)) + + cfg := multiNodeIPSecConfig(false) + cfg.Nodes["node1"] = config.Config{VPCD: config.VPCDConfig{OVNNBAddr: nb}} + + require.NoError(t, ReconcileOVNIPSec(t.Context(), "/etc/spinifex/spinifex.toml", cfg, allReady())) + + assert.Equal(t, [][]string{{"ovn-nbctl", "--db=" + nb, "--no-leader-only", "--timeout=5", "get", "NB_Global", ".", "ipsec"}}, + recorder.nbctlReads()) + assert.Equal(t, [][]string{{"ovn-nbctl", "--db=" + nb, "--timeout=5", "set", "NB_Global", ".", "ipsec=false"}}, + recorder.nbctlWrites()) +} + +// Only "nothing answered" is unreachable. The local socket being absent is that +// case; a follower that answers is not, and never produces this output. +func TestIsNBUnreachable(t *testing.T) { + assert.True(t, isNBUnreachable("ovn-nbctl: unix:/var/run/ovn/ovnnb_db.sock: database connection failed (No such file or directory)")) + assert.True(t, isNBUnreachable("ovn-nbctl: tcp:10.0.0.1:6641: database connection failed (Connection refused)")) + assert.False(t, isNBUnreachable("ovn-nbctl: unix:/var/run/ovn/ovnnb_db.sock: permission denied")) + assert.False(t, isNBUnreachable("ovn-nbctl: transaction timed out")) +} diff --git a/spinifex/network/host/mac_binding.go b/spinifex/network/host/mac_binding.go index 5cbbbde0c..b34536c64 100644 --- a/spinifex/network/host/mac_binding.go +++ b/spinifex/network/host/mac_binding.go @@ -58,6 +58,8 @@ func SeedNexthopMAC(ctx context.Context, runner Runner, nbAddr, lrpName, nexthop } } + // Writes with --no-leader-only and relies on a follower forwarding the + // transaction. NB_Global.ipsec writes stay leader-only instead (ipsec.go). var db []string if nbAddr != "" { db = []string{"--db=" + nbAddr, "--no-leader-only"} diff --git a/spinifex/network/reconcile/apply.go b/spinifex/network/reconcile/apply.go index 104a8e2ea..5a40ff71d 100644 --- a/spinifex/network/reconcile/apply.go +++ b/spinifex/network/reconcile/apply.go @@ -707,6 +707,9 @@ func (r *reconciler) applyEIPs(ctx context.Context, intent IntentState, _ Actual sem := make(chan struct{}, guestPortDatapathConcurrency) var wg sync.WaitGroup for _, spec := range specs { + if r.guestPortIdle(spec, intent.IdlePorts) { + continue + } select { case <-ctx.Done(): wg.Wait() @@ -719,6 +722,60 @@ func (r *reconciler) applyEIPs(ctx context.Context, intent IntentState, _ Actual }) } wg.Wait() + r.forgetIdlePorts(specs) +} + +// guestPortIdle reports whether spec's port belongs to a stopped or terminated +// instance, logging at Info only when that changes. Entering idle drops any held +// backoff, so the port is probed on the first pass after the instance starts. +func (r *reconciler) guestPortIdle(spec policy.EIPSpec, idle map[string]string) bool { + if spec.PortName == "" { + return false + } + instanceID, isIdle := idle[spec.PortName] + + r.portBackoffMu.Lock() + _, wasIdle := r.idlePorts[spec.PortName] + switch { + case isIdle && !wasIdle: + if r.idlePorts == nil { + r.idlePorts = map[string]struct{}{} + } + r.idlePorts[spec.PortName] = struct{}{} + delete(r.portBackoff, spec.PortName) + case !isIdle && wasIdle: + delete(r.idlePorts, spec.PortName) + } + r.portBackoffMu.Unlock() + + logKV := []any{"vpc_id", spec.VPCID, "lsp", spec.PortName, "eni_id", eniIDFromPort(spec.PortName), + "external_ip", spec.ExternalIP, "logical_ip", spec.LogicalIP} + switch { + case isIdle && !wasIdle: + slog.Info("reconcile/apply: guest port's instance is not running; not probing its datapath", + append(logKV, "instance_id", instanceID)...) + case isIdle: + slog.Debug("reconcile/apply: guest port's instance still not running; not probing", logKV...) + case wasIdle: + slog.Info("reconcile/apply: guest port's instance is no longer stopped; probing its datapath again", logKV...) + } + return isIdle +} + +// forgetIdlePorts drops idle marks for ports that no longer carry a public IP, so +// the set tracks live EIPs rather than every port ever seen. +func (r *reconciler) forgetIdlePorts(specs []policy.EIPSpec) { + live := make(map[string]struct{}, len(specs)) + for _, spec := range specs { + live[spec.PortName] = struct{}{} + } + r.portBackoffMu.Lock() + defer r.portBackoffMu.Unlock() + for lsp := range r.idlePorts { + if _, ok := live[lsp]; !ok { + delete(r.idlePorts, lsp) + } + } } // floatingIPSpecs is every dnat_and_snat the datapath must carry: user EIPs diff --git a/spinifex/network/reconcile/apply_guest_port_idle_test.go b/spinifex/network/reconcile/apply_guest_port_idle_test.go new file mode 100644 index 000000000..ba06f5c38 --- /dev/null +++ b/spinifex/network/reconcile/apply_guest_port_idle_test.go @@ -0,0 +1,132 @@ +package reconcile + +//test:in-package — drives applyEIPs and reads the unexported idle and backoff +//state the skip is recorded in. + +import ( + "bytes" + "context" + "log/slog" + "strings" + "testing" + "time" + + "github.com/mulgadc/spinifex/spinifex/network/policy" +) + +// captureReconcileLogs swaps the default logger for one writing to a buffer at +// Info, so a test can count what an operator would see in the journal. +func captureReconcileLogs(t *testing.T) *bytes.Buffer { + t.Helper() + buf := &bytes.Buffer{} + prev := slog.Default() + slog.SetDefault(slog.New(slog.NewTextHandler(buf, &slog.HandlerOptions{Level: slog.LevelInfo}))) + t.Cleanup(func() { slog.SetDefault(prev) }) + return buf +} + +func idleEIPIntent(idle bool) IntentState { + intent := IntentState{EIPs: map[string]policy.EIPSpec{ + "10.0.0.4": {VPCID: "vpc-a", ExternalIP: "192.0.2.46", LogicalIP: "10.0.0.4", PortName: "port-eni-1"}, + }} + if idle { + intent.IdlePorts = map[string]string{"port-eni-1": "i-stopped"} + } + return intent +} + +// A customer-stopped instance keeps its ENI and EIP but has no tap, so its port +// can never bind. Passes must keep the DNAT row and never probe, nudge or file +// an ERROR, and say so once rather than every pass. +func TestApplyEIPs_StoppedInstancePortIsNotProbed(t *testing.T) { + withFastGuestPortBounds(t) + logs := captureReconcileLogs(t) + + f := &fakeClaimVerifier{guestUpAfter: -1} + nat := &countingNAT{} + r := &reconciler{gwClaim: f, nat: nat, ovn: ovnWithGuestLSP(t, "port-eni-1")} + + for range 3 { + r.applyEIPs(context.Background(), idleEIPIntent(true), ActualState{}, &passResult{}) + } + + if got := nat.count(); got != 3 { + t.Errorf("AddEIP called %d times, want 3: the DNAT row must still be asserted", got) + } + if f.guestChecks != 0 || f.nudges != 0 || f.sbChecks != 0 { + t.Errorf("probes=%d nudges=%d sb_checks=%d, want all 0 for a stopped instance's port", + f.guestChecks, f.nudges, f.sbChecks) + } + out := logs.String() + if n := strings.Count(out, "guest port's instance is not running"); n != 1 { + t.Errorf("idle notice logged %d times over 3 passes, want 1:\n%s", n, out) + } + for _, noisy := range []string{"level=WARN", "level=ERROR"} { + if strings.Contains(out, noisy) { + t.Errorf("stopped instance's port produced %s:\n%s", noisy, out) + } + } +} + +// Starting the instance must converge on the first pass without a nudge, and +// the resume is logged once. +func TestApplyEIPs_StartedInstanceConvergesWithoutNudge(t *testing.T) { + withFastGuestPortBounds(t) + logs := captureReconcileLogs(t) + + f := &fakeClaimVerifier{guestUpAfter: 0} + r := &reconciler{gwClaim: f, nat: &countingNAT{}, ovn: ovnWithGuestLSP(t, "port-eni-1")} + + r.applyEIPs(context.Background(), idleEIPIntent(true), ActualState{}, &passResult{}) + r.applyEIPs(context.Background(), idleEIPIntent(false), ActualState{}, &passResult{}) + r.applyEIPs(context.Background(), idleEIPIntent(false), ActualState{}, &passResult{}) + + if f.guestChecks != 2 { + t.Errorf("probes = %d, want 2: one per pass once the instance is running", f.guestChecks) + } + if f.nudges != 0 { + t.Errorf("nudges = %d, want 0: a started instance's bound port needs no recompute", f.nudges) + } + out := logs.String() + if n := strings.Count(out, "no longer stopped; probing its datapath again"); n != 1 { + t.Errorf("resume notice logged %d times, want 1:\n%s", n, out) + } + if strings.Contains(out, "did not converge") { + t.Errorf("started instance logged a convergence failure:\n%s", out) + } +} + +// Failures filed while the instance was stopping must not hold back the probe +// once it starts: backoff caps at 30 minutes, which would hide a real outage. +func TestApplyEIPs_IdleClearsHeldBackoff(t *testing.T) { + withFastGuestPortBounds(t) + withFastGuestPortBackoff(t, time.Hour, time.Hour) + captureReconcileLogs(t) + + f := &fakeClaimVerifier{guestUpAfter: 0} + r := &reconciler{gwClaim: f, nat: &countingNAT{}, ovn: ovnWithGuestLSP(t, "port-eni-1")} + r.recordPortFailure("port-eni-1") + + r.applyEIPs(context.Background(), idleEIPIntent(true), ActualState{}, &passResult{}) + r.applyEIPs(context.Background(), idleEIPIntent(false), ActualState{}, &passResult{}) + + if f.guestChecks != 1 { + t.Errorf("probes = %d, want 1: the backoff from before the stop must not survive it", f.guestChecks) + } +} + +// An EIP that is released while its instance is stopped must not leave its idle +// mark behind to swallow a later notice for the same port. +func TestApplyEIPs_ForgetsIdleMarkWhenEIPGoes(t *testing.T) { + withFastGuestPortBounds(t) + captureReconcileLogs(t) + + r := &reconciler{gwClaim: &fakeClaimVerifier{guestUpAfter: 0}, nat: &countingNAT{}, ovn: ovnWithGuestLSP(t, "port-eni-1")} + + r.applyEIPs(context.Background(), idleEIPIntent(true), ActualState{}, &passResult{}) + r.applyEIPs(context.Background(), IntentState{}, ActualState{}, &passResult{}) + + if len(r.idlePorts) != 0 { + t.Errorf("idle marks = %v, want none once the port carries no public IP", r.idlePorts) + } +} diff --git a/spinifex/network/reconcile/intent.go b/spinifex/network/reconcile/intent.go index b8a4d6584..ee0a1a21e 100644 --- a/spinifex/network/reconcile/intent.go +++ b/spinifex/network/reconcile/intent.go @@ -19,6 +19,7 @@ import ( "github.com/mulgadc/spinifex/spinifex/network/policy" "github.com/mulgadc/spinifex/spinifex/network/topology" "github.com/mulgadc/spinifex/spinifex/utils" + "github.com/mulgadc/spinifex/spinifex/vm" "github.com/nats-io/nats.go/jetstream" ) @@ -35,8 +36,18 @@ type IntentState struct { IGWRoutes map[string]SubnetEgressIntent // per (subnet, dest) IGW egress reroute NATGWRoutes map[string]SubnetEgressIntent // per (subnet, dest) NATGW egress reroute DropGates map[string]SubnetEgressIntent // per (subnet, dest=0.0.0.0/0) drop policy + // IdlePorts maps a public-IP guest LSP to its instance when that instance is + // stopped or terminated. Its port has no tap, so the datapath probe skips it. + IdlePorts map[string]string } +// The instance record space the daemon owns. Duplicated rather than imported +// because daemon imports this package; a daemon test pins the two together. +const ( + InstanceRecordBucket = "spinifex-instance-state" + InstanceRecordPrefix = "i." +) + // SubnetEgressIntent is a per-subnet default-route policy entry for the VPC LR, // keyed by (subnet, destCIDR) pointing at an IGW or NAT gateway. type SubnetEgressIntent struct { @@ -64,6 +75,7 @@ func LoadIntentFromKV(ctx context.Context, js jetstream.JetStream, localAZ strin IGWRoutes: make(map[string]SubnetEgressIntent), NATGWRoutes: make(map[string]SubnetEgressIntent), DropGates: make(map[string]SubnetEgressIntent), + IdlePorts: make(map[string]string), } localVPCs, err := loadVPCs(ctx, js, localAZ, intent.VPCs) @@ -76,7 +88,8 @@ func LoadIntentFromKV(ctx context.Context, js jetstream.JetStream, localAZ strin if err := loadSGs(ctx, js, localVPCs, intent.SGs); err != nil { return IntentState{}, err } - if err := loadPorts(ctx, js, localVPCs, intent.Ports); err != nil { + portInstances := make(map[string]string) + if err := loadPorts(ctx, js, localVPCs, intent.Ports, portInstances); err != nil { return IntentState{}, err } if err := loadIGWs(ctx, js, localVPCs, intent.IGWs); err != nil { @@ -85,6 +98,7 @@ func LoadIntentFromKV(ctx context.Context, js jetstream.JetStream, localAZ strin if err := loadEIPs(ctx, js, localVPCs, intent.EIPs); err != nil { return IntentState{}, err } + loadIdlePorts(ctx, js, intent, portInstances) routeTables, err := loadRouteTables(ctx, js, localVPCs) if err != nil { return IntentState{}, err @@ -348,7 +362,7 @@ func loadSGs(ctx context.Context, js jetstream.JetStream, localVPCs map[string]s return nil } -func loadPorts(ctx context.Context, js jetstream.JetStream, localVPCs map[string]struct{}, out map[string]topology.PortSpec) error { +func loadPorts(ctx context.Context, js jetstream.JetStream, localVPCs map[string]struct{}, out map[string]topology.PortSpec, instances map[string]string) error { kv, err := js.KeyValue(ctx, handlers_ec2_vpc.KVBucketENIs) if err != nil { slog.Debug("reconcile/intent: ENI bucket not available, skipping", "err", err) @@ -401,10 +415,71 @@ func loadPorts(ctx context.Context, js jetstream.JetStream, localVPCs map[string PublicIP: publicIP, SuppressDHCP: rec.SuppressDHCP, } + if rec.InstanceId != "" { + instances[rec.NetworkInterfaceId] = rec.InstanceId + } } return nil } +// loadIdlePorts fills intent.IdlePorts for the ports the EIP datapath probe would +// visit. Anything it cannot read is left out, so the port is probed as before. +func loadIdlePorts(ctx context.Context, js jetstream.JetStream, intent IntentState, instances map[string]string) { + candidates := make(map[string]struct{}, len(intent.EIPs)) + for id, p := range intent.Ports { + if p.PublicIP.IsValid() { + candidates[topology.Port(id)] = struct{}{} + } + } + for _, e := range intent.EIPs { + if e.PortName != "" { + candidates[e.PortName] = struct{}{} + } + } + if len(candidates) == 0 { + return + } + + kv, err := js.KeyValue(ctx, InstanceRecordBucket) + if err != nil { + slog.Debug("reconcile/intent: instance record bucket not available, probing every guest port", "err", err) + return + } + for lsp := range candidates { + instanceID := instances[eniIDFromPort(lsp)] + if instanceID == "" { + continue + } + entry, err := kv.Get(ctx, InstanceRecordPrefix+instanceID) + if err != nil { + if !errors.Is(err, jetstream.ErrKeyNotFound) { + slog.Warn("reconcile/intent: instance record read failed", "instance_id", instanceID, "err", err) + } + continue + } + var rec vm.InstanceRecord + if err := json.Unmarshal(entry.Value(), &rec); err != nil { + slog.Warn("reconcile/intent: instance record unmarshal failed", "instance_id", instanceID, "err", err) + continue + } + if instanceHasNoTap(&rec) { + intent.IdlePorts[lsp] = instanceID + } + } +} + +// instanceHasNoTap reports an instance nothing will relaunch. A DRAIN also leaves +// StateStopped, but restore relaunches those, so desired state has to agree. +func instanceHasNoTap(rec *vm.InstanceRecord) bool { + switch rec.Status.Status { + case vm.StateTerminated: + return true + case vm.StateStopped: + return rec.Spec.DesiredState == vm.DesiredStopped + } + return false +} + func loadIGWs(ctx context.Context, js jetstream.JetStream, localVPCs map[string]struct{}, out map[string]external.IGWSpec) error { kv, err := js.KeyValue(ctx, handlers_ec2_igw.KVBucketIGW) if err != nil { diff --git a/spinifex/network/reconcile/intent_test.go b/spinifex/network/reconcile/intent_test.go index e593c823f..c1ac7bf27 100644 --- a/spinifex/network/reconcile/intent_test.go +++ b/spinifex/network/reconcile/intent_test.go @@ -15,6 +15,7 @@ import ( handlers_ec2_vpc "github.com/mulgadc/spinifex/spinifex/handlers/ec2/vpc" "github.com/mulgadc/spinifex/spinifex/network/policy" "github.com/mulgadc/spinifex/spinifex/testutil" + "github.com/mulgadc/spinifex/spinifex/vm" "github.com/nats-io/nats.go/jetstream" ) @@ -556,6 +557,69 @@ func TestLoadIntentFromKV_SGRuleFieldMapping(t *testing.T) { } } +// Only an instance nothing will relaunch marks its public-IP port idle. A DRAIN +// stop is relaunched by restore, and a missing record is unknown, so both stay +// probed; a port with no public IP is never read at all. +func TestLoadIntentFromKV_IdlePortsFromInstanceState(t *testing.T) { + js := startKV(t) + + testutil.SeedKV(t, js, handlers_ec2_vpc.KVBucketVPCs, map[string][]byte{ + "acct/vpc-a": mustJSON(t, handlers_ec2_vpc.VPCRecord{ + VpcId: "vpc-a", CidrBlock: "10.0.0.0/16", AZ: "us-east-1a", CreatedAt: time.Now(), + }), + }) + eni := func(id, ip, publicIP, instanceID string) []byte { + return mustJSON(t, handlers_ec2_vpc.ENIRecord{ + NetworkInterfaceId: id, SubnetId: "subnet-a", VpcId: "vpc-a", + PrivateIpAddress: ip, MacAddress: "02:00:00:00:00:" + ip[len(ip)-2:], + PublicIpAddress: publicIP, InstanceId: instanceID, Status: "in-use", CreatedAt: time.Now(), + }) + } + testutil.SeedKV(t, js, handlers_ec2_vpc.KVBucketENIs, map[string][]byte{ + "acct/eni-stopped": eni("eni-stopped", "10.0.1.11", "192.0.2.11", "i-stopped"), + "acct/eni-term": eni("eni-term", "10.0.1.12", "192.0.2.12", "i-term"), + "acct/eni-drain": eni("eni-drain", "10.0.1.13", "192.0.2.13", "i-drain"), + "acct/eni-running": eni("eni-running", "10.0.1.14", "192.0.2.14", "i-running"), + "acct/eni-gone": eni("eni-gone", "10.0.1.15", "192.0.2.15", "i-gone"), + "acct/eni-eip": eni("eni-eip", "10.0.1.16", "", "i-eip"), + "acct/eni-private": eni("eni-private", "10.0.1.17", "", "i-private"), + }) + testutil.SeedKV(t, js, handlers_ec2_eip.KVBucketEIPs, map[string][]byte{ + "acct/eipassoc-a": mustJSON(t, handlers_ec2_eip.EIPRecord{ + AllocationId: "eipalloc-a", PublicIp: "203.0.113.16", PrivateIp: "10.0.1.16", + VpcId: "vpc-a", ENIId: "eni-eip", State: "associated", + }), + }) + record := func(status vm.InstanceState, desired vm.DesiredState) []byte { + return mustJSON(t, &vm.InstanceRecord{ + Spec: vm.InstanceSpec{DesiredState: desired}, + Status: vm.InstanceStatus{Status: status}, + }) + } + testutil.SeedKV(t, js, InstanceRecordBucket, map[string][]byte{ + InstanceRecordPrefix + "i-stopped": record(vm.StateStopped, vm.DesiredStopped), + InstanceRecordPrefix + "i-term": record(vm.StateTerminated, vm.DesiredRunning), + InstanceRecordPrefix + "i-drain": record(vm.StateStopped, vm.DesiredRunning), + InstanceRecordPrefix + "i-running": record(vm.StateRunning, vm.DesiredRunning), + InstanceRecordPrefix + "i-eip": record(vm.StateStopped, vm.DesiredStopped), + InstanceRecordPrefix + "i-private": record(vm.StateStopped, vm.DesiredStopped), + }) + + intent, err := LoadIntentFromKV(context.Background(), js, "us-east-1a") + if err != nil { + t.Fatalf("LoadIntentFromKV: %v", err) + } + + want := map[string]string{ + "port-eni-stopped": "i-stopped", + "port-eni-term": "i-term", + "port-eni-eip": "i-eip", + } + if !reflect.DeepEqual(intent.IdlePorts, want) { + t.Errorf("IdlePorts = %v, want %v", intent.IdlePorts, want) + } +} + func mustJSON(t *testing.T, v any) []byte { t.Helper() b, err := json.Marshal(v) diff --git a/spinifex/network/reconcile/reconcile.go b/spinifex/network/reconcile/reconcile.go index 01b1d2c1a..a65fb527d 100644 --- a/spinifex/network/reconcile/reconcile.go +++ b/spinifex/network/reconcile/reconcile.go @@ -154,6 +154,8 @@ type reconciler struct { // is gone stops paying the full nudge sequence every cycle. portBackoffMu sync.Mutex portBackoff map[string]portBackoffState + // Guest ports last seen idle, so the skip is logged once per state change. + idlePorts map[string]struct{} // Serialises sb-cluster-state-reset. It is cluster-wide and re-syncs every // chassis, so concurrent guest-port probes must not each fire one. diff --git a/spinifex/services/spinifexui/proxy.go b/spinifex/services/spinifexui/proxy.go index cdfb5b7cf..05f887bb4 100644 --- a/spinifex/services/spinifexui/proxy.go +++ b/spinifex/services/spinifexui/proxy.go @@ -14,6 +14,7 @@ import ( "time" "github.com/mulgadc/bluebottle/pkg/tlsconfig" + "github.com/mulgadc/spinifex/spinifex/utils" "go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp" ) @@ -59,6 +60,11 @@ func newReverseProxy(backendHost, pathPrefix string, transport http.RoundTripper pr.SetURL(target) pr.Out.Host = target.Host + // The backend trusts X-Real-IP from loopback, which this hop always is. + // The UI also listens publicly, so re-derive it rather than forward a + // client's own claim. X-Request-ID passes through unchanged. + pr.Out.Header.Set("X-Real-IP", utils.RequestClientIP(pr.In)) + // Strip the prefix from both path forms. RawPath must survive: a // client that percent-encodes a segment (a ':' in a bedrock // modelId) signs it, so dropping it re-encodes and breaks SigV4. diff --git a/spinifex/services/spinifexui/proxy_realip_test.go b/spinifex/services/spinifexui/proxy_realip_test.go new file mode 100644 index 000000000..bd2eb153f --- /dev/null +++ b/spinifex/services/spinifexui/proxy_realip_test.go @@ -0,0 +1,59 @@ +//test:in-package: drives the unexported newReverseProxy, which the package exports only through +// the full UI server; building that here would test far more than the header rule. + +package spinifexui + +import ( + "net/http" + "net/http/httptest" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// proxyHeaders sends one request through the awsgw proxy from remoteAddr and +// returns the headers the backend received. +func proxyHeaders(t *testing.T, remoteAddr string, set map[string]string) http.Header { + t.Helper() + var got http.Header + backend := httptest.NewTLSServer(http.HandlerFunc(func(_ http.ResponseWriter, r *http.Request) { + got = r.Header.Clone() + })) + t.Cleanup(backend.Close) + + transport, ok := backend.Client().Transport.(*http.Transport) + require.True(t, ok, "expected *http.Transport") + proxy := newReverseProxy(backend.Listener.Addr().String(), "/proxy/awsgw", transport) + + req := httptest.NewRequest(http.MethodPost, "/proxy/awsgw/", nil) + req.RemoteAddr = remoteAddr + for k, v := range set { + req.Header.Set(k, v) + } + proxy.ServeHTTP(httptest.NewRecorder(), req) + require.NotNil(t, got, "backend was not reached") + return got +} + +// Behind nginx the UI's own connection is loopback, so the edge's X-Real-IP and +// X-Request-ID reach awsgw unchanged. +func TestNewReverseProxy_PassesEdgeHeadersFromLoopback(t *testing.T) { + got := proxyHeaders(t, "127.0.0.1:50000", map[string]string{ + "X-Real-IP": "198.51.100.7", + "X-Request-ID": "4f1c2b7e9a0d4e6f8b3c5a7d9e1f2a3b", + }) + assert.Equal(t, "198.51.100.7", got.Get("X-Real-IP")) + assert.Equal(t, "4f1c2b7e9a0d4e6f8b3c5a7d9e1f2a3b", got.Get("X-Request-ID")) +} + +// The UI listens publicly. A client talking to it directly must not be able to +// hand awsgw a forged X-Real-IP over the loopback hop, where awsgw trusts it. +func TestNewReverseProxy_OverwritesSpoofedRealIPFromPublicClient(t *testing.T) { + got := proxyHeaders(t, "203.0.113.9:50000", map[string]string{ + "X-Real-IP": "10.0.0.1", + "X-Forwarded-For": "127.0.0.1", + }) + assert.Equal(t, "203.0.113.9", got.Get("X-Real-IP")) + assert.Empty(t, got.Get("X-Forwarded-For"), "Rewrite drops inbound X-Forwarded-For") +} diff --git a/spinifex/utils/utils.go b/spinifex/utils/utils.go index 9bf8f51ad..a01f64a9d 100644 --- a/spinifex/utils/utils.go +++ b/spinifex/utils/utils.go @@ -14,9 +14,11 @@ import ( "log/slog" "net" "net/http" + "net/netip" "os" "os/signal" "reflect" + "strings" "time" "github.com/aws/aws-sdk-go/aws" @@ -393,6 +395,21 @@ func HashMAC(id string) string { return net.HardwareAddr(b).String() } +// RequestClientIP is the caller's address: X-Real-IP when the connection is from +// loopback, where only a local proxy (the nginx edge, the console) can have set +// it, and RemoteAddr otherwise, so a direct client cannot choose what is logged. +func RequestClientIP(r *http.Request) string { + ip := ClientIP(r.RemoteAddr) + if addr, err := netip.ParseAddr(ip); err != nil || !addr.IsLoopback() { + return ip + } + realIP, err := netip.ParseAddr(strings.TrimSpace(r.Header.Get("X-Real-IP"))) + if err != nil { + return ip + } + return realIP.String() +} + // ClientIP returns the IP from a RemoteAddr, stripping the port. Handles both // IPv4 and IPv6, and tolerates a RemoteAddr that is already a bare IP (no port). func ClientIP(remoteAddr string) string { diff --git a/spinifex/utils/utils_test.go b/spinifex/utils/utils_test.go index a66721d31..9ffe786cf 100644 --- a/spinifex/utils/utils_test.go +++ b/spinifex/utils/utils_test.go @@ -1807,3 +1807,39 @@ func TestWithRequestID_PointerToStruct(t *testing.T) { require.Equal(t, "req-456", v.FieldByName("RequestId").String()) require.Equal(t, []string{"b"}, v.FieldByName("Names").Interface()) } + +// X-Real-IP is honoured only on a loopback connection, where the local edge set +// it. From anywhere else it, and X-Forwarded-For, are the client's own claim. +func TestRequestClientIP(t *testing.T) { + tests := []struct { + name string + remoteAddr string + realIP string + xff string + expected string + }{ + {"loopback trusts X-Real-IP", "127.0.0.1:40000", "198.51.100.7", "", "198.51.100.7"}, + {"IPv6 loopback trusts X-Real-IP", "[::1]:40000", "2001:db8::7", "", "2001:db8::7"}, + {"4in6 loopback trusts X-Real-IP", "[::ffff:127.0.0.1]:40000", "198.51.100.7", "", "198.51.100.7"}, + {"loopback without header", "127.0.0.1:40000", "", "", "127.0.0.1"}, + {"loopback with garbage header", "127.0.0.1:40000", "not-an-ip", "", "127.0.0.1"}, + {"loopback ignores X-Forwarded-For", "127.0.0.1:40000", "", "203.0.113.1", "127.0.0.1"}, + {"spoofed X-Real-IP from public client", "203.0.113.9:5555", "10.0.0.1", "", "203.0.113.9"}, + {"spoofed X-Forwarded-For from public client", "203.0.113.9:5555", "", "127.0.0.1", "203.0.113.9"}, + {"spoofed both from private client", "10.2.0.3:5555", "127.0.0.1", "127.0.0.1", "10.2.0.3"}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + r := httptest.NewRequest(http.MethodGet, "/", nil) + r.RemoteAddr = tt.remoteAddr + if tt.realIP != "" { + r.Header.Set("X-Real-IP", tt.realIP) + } + if tt.xff != "" { + r.Header.Set("X-Forwarded-For", tt.xff) + } + assert.Equal(t, tt.expected, RequestClientIP(r)) + }) + } +}