diff --git a/cmd/spinifex/cmd/admin.go b/cmd/spinifex/cmd/admin.go index 3ae5a71d0..33803d13f 100644 --- a/cmd/spinifex/cmd/admin.go +++ b/cmd/spinifex/cmd/admin.go @@ -1524,7 +1524,7 @@ func runAdminInit(cmd *cobra.Command, args []string) { northstarConfigPath := filepath.Join(dirs.Northstar, "northstar.toml") // Parse multi-node predastore configuration (legacy flag-based approach for single-node) - var predastoreNodeID int + var predastoreHostID int if predastoreNodesStr != "" { ips := strings.Split(predastoreNodesStr, ",") if len(ips) < 2 { @@ -1545,14 +1545,16 @@ func runAdminInit(cmd *cobra.Command, args []string) { }) } - predastoreNodeID = admin.FindNodeIDByIP(predastoreNodes, bindIP) - if predastoreNodeID == 0 { + // One machine is one predastore host, so the node list index that + // matches this bind IP is this node's host ID. + predastoreHostID = admin.FindNodeIDByIP(predastoreNodes, bindIP) + if predastoreHostID == 0 { fmt.Fprintf(os.Stderr, "❌ Error: --bind IP %s not found in --predastore-nodes list\n", bindIP) os.Exit(1) } // Generate multi-node predastore.toml - predastoreContent, err := admin.GenerateMultiNodePredastoreConfig(predastoreMultiNodeTemplate, predastoreNodes, accessKey, secretKey, region, natsToken, configDir, bindIP, compactionInterval, northstarCreds) + predastoreContent, err := admin.GenerateMultiNodePredastoreConfig(predastoreMultiNodeTemplate, predastoreNodes, accessKey, secretKey, region, natsToken, configDir, spxRoot, bindIP, compactionInterval, northstarCreds) if err != nil { fmt.Fprintf(os.Stderr, "Error generating multi-node predastore config: %v\n", err) os.Exit(1) @@ -1563,7 +1565,7 @@ func runAdminInit(cmd *cobra.Command, args []string) { fmt.Fprintf(os.Stderr, "Error writing predastore config: %v\n", err) os.Exit(1) } - fmt.Printf("✅ Created: multi-node predastore.toml (node ID: %d)\n", predastoreNodeID) + fmt.Printf("✅ Created: multi-node predastore.toml (host ID: %d)\n", predastoreHostID) } // Pre-generate default VPC/subnet/IGW IDs for bootstrap config. @@ -1593,7 +1595,7 @@ func runAdminInit(cmd *cobra.Command, args []string) { ClusterRoutes: clusterRoutes, ClusterName: clusterName, - PredastoreNodeID: predastoreNodeID, + PredastoreHostID: predastoreHostID, CompactionIntervalSeconds: compactionInterval, Services: services, @@ -1840,10 +1842,10 @@ func runAdminInitMultiNode(cmd *cobra.Command, accessKey, secretKey, accountID, northstarConfigPath := filepath.Join(dirs.Northstar, "northstar.toml") // Generate multi-node predastore config - var predastoreNodeID int + var predastoreHostID int hasPredastoreConfig := len(predastoreNodes) >= 2 if hasPredastoreConfig { - predastoreContent, err := admin.GenerateMultiNodePredastoreConfig(predastoreMultiNodeTemplate, predastoreNodes, accessKey, secretKey, region, natsToken, configDir, bindIP, compactionInterval, northstarCreds) + predastoreContent, err := admin.GenerateMultiNodePredastoreConfig(predastoreMultiNodeTemplate, predastoreNodes, accessKey, secretKey, region, natsToken, configDir, spxRoot, bindIP, compactionInterval, northstarCreds) if err != nil { fmt.Fprintf(os.Stderr, "Error generating multi-node predastore config: %v\n", err) os.Exit(1) @@ -1855,8 +1857,8 @@ func runAdminInitMultiNode(cmd *cobra.Command, accessKey, secretKey, accountID, os.Exit(1) } - predastoreNodeID = admin.FindNodeIDByIP(predastoreNodes, bindIP) - fmt.Printf("✅ Created: multi-node predastore.toml (node ID: %d)\n", predastoreNodeID) + predastoreHostID = admin.FindNodeIDByIP(predastoreNodes, bindIP) + fmt.Printf("✅ Created: multi-node predastore.toml (host ID: %d)\n", predastoreHostID) } spinifexTomlPath := filepath.Join(configDir, "spinifex.toml") @@ -1880,7 +1882,7 @@ func runAdminInitMultiNode(cmd *cobra.Command, accessKey, secretKey, accountID, ClusterRoutes: clusterRoutes, ClusterName: clusterName, - PredastoreNodeID: predastoreNodeID, + PredastoreHostID: predastoreHostID, CompactionIntervalSeconds: compactionInterval, Services: services, RemoteNodes: buildRemoteNodes(allNodes, node, northstarConfigPath), @@ -2390,11 +2392,11 @@ func runAdminJoin(cmd *cobra.Command, args []string) { northstarCreds, northstarConfigPath := northstarFromFormation(creds, dirs) // Generate multi-node predastore config - var predastoreNodeID int + var predastoreHostID int hasPredastoreConfig := len(predastoreNodes) >= 2 if hasPredastoreConfig { - predastoreContent, err := admin.GenerateMultiNodePredastoreConfig(predastoreMultiNodeTemplate, predastoreNodes, creds.AccessKey, creds.SecretKey, creds.Region, creds.NatsToken, configDir, bindIP, compactionInterval, + predastoreContent, err := admin.GenerateMultiNodePredastoreConfig(predastoreMultiNodeTemplate, predastoreNodes, creds.AccessKey, creds.SecretKey, creds.Region, creds.NatsToken, configDir, dataDir, bindIP, compactionInterval, northstarCreds) if err != nil { fmt.Fprintf(os.Stderr, "Error generating multi-node predastore config: %v\n", err) @@ -2407,12 +2409,12 @@ func runAdminJoin(cmd *cobra.Command, args []string) { os.Exit(1) } - predastoreNodeID = admin.FindNodeIDByIP(predastoreNodes, bindIP) - if predastoreNodeID == 0 { + predastoreHostID = admin.FindNodeIDByIP(predastoreNodes, bindIP) + if predastoreHostID == 0 { fmt.Fprintf(os.Stderr, "❌ Error: bind IP %s not found in predastore node list\n", bindIP) os.Exit(1) } - fmt.Printf("✅ Created: multi-node predastore.toml (node ID: %d)\n", predastoreNodeID) + fmt.Printf("✅ Created: multi-node predastore.toml (host ID: %d)\n", predastoreHostID) } spinifexTomlPath := filepath.Join(configDir, "spinifex.toml") @@ -2436,7 +2438,7 @@ func runAdminJoin(cmd *cobra.Command, args []string) { ClusterRoutes: clusterRoutes, ClusterName: creds.ClusterName, - PredastoreNodeID: predastoreNodeID, + PredastoreHostID: predastoreHostID, CompactionIntervalSeconds: compactionInterval, Services: services, RemoteNodes: buildRemoteNodes(statusResp.Nodes, node, northstarConfigPath), diff --git a/cmd/spinifex/cmd/admin_test.go b/cmd/spinifex/cmd/admin_test.go index fae7f8632..1a6ef792f 100644 --- a/cmd/spinifex/cmd/admin_test.go +++ b/cmd/spinifex/cmd/admin_test.go @@ -316,7 +316,7 @@ func TestPredastoreMultinodeTemplate_NatsURLLoopbackShim(t *testing.T) { {ID: 3, Host: "10.0.0.3"}, } content, err := admin.GenerateMultiNodePredastoreConfig( - predastoreMultiNodeTemplate, nodes, "AK", "SK", "ap-southeast-2", "token", "/tmp", "0.0.0.0", 0, admin.NorthstarCredentials{}, + predastoreMultiNodeTemplate, nodes, "AK", "SK", "ap-southeast-2", "token", "/tmp", "/var/lib/spinifex", "0.0.0.0", 0, admin.NorthstarCredentials{}, ) require.NoError(t, err) assert.Contains(t, content, `nats_url = "nats://localhost:4222"`) @@ -324,7 +324,7 @@ func TestPredastoreMultinodeTemplate_NatsURLLoopbackShim(t *testing.T) { // Specific bind IP → stays as-is. content2, err := admin.GenerateMultiNodePredastoreConfig( - predastoreMultiNodeTemplate, nodes, "AK", "SK", "ap-southeast-2", "token", "/tmp", "10.11.12.1", 0, admin.NorthstarCredentials{}, + predastoreMultiNodeTemplate, nodes, "AK", "SK", "ap-southeast-2", "token", "/tmp", "/var/lib/spinifex", "10.11.12.1", 0, admin.NorthstarCredentials{}, ) require.NoError(t, err) assert.Contains(t, content2, `nats_url = "nats://10.11.12.1:4222"`) @@ -813,7 +813,7 @@ func (c predastoreBucketAuth) hasBucket(name string) bool { // them. func TestPredastoreMultinodeTemplate_NorthstarProvisioned(t *testing.T) { content, err := admin.GenerateMultiNodePredastoreConfig( - predastoreMultiNodeTemplate, predastoreMultinodeNodes(), "AK", "SK", "ap-southeast-2", "tok", "/cfg", "10.0.0.1", 0, + predastoreMultiNodeTemplate, predastoreMultinodeNodes(), "AK", "SK", "ap-southeast-2", "tok", "/cfg", "/var/lib/spinifex", "10.0.0.1", 0, admin.NorthstarCredentials{AccessKey: "NSAK", SecretKey: "NSSK", Bucket: admin.NorthstarBucketName}, ) require.NoError(t, err) @@ -857,7 +857,7 @@ func TestPredastoreMultinodeTemplate_NorthstarProvisioned(t *testing.T) { // zone bucket. func TestPredastoreMultinodeTemplate_NorthstarOmittedWithoutCredentials(t *testing.T) { content, err := admin.GenerateMultiNodePredastoreConfig( - predastoreMultiNodeTemplate, predastoreMultinodeNodes(), "AK", "SK", "ap-southeast-2", "tok", "/cfg", "10.0.0.1", 0, + predastoreMultiNodeTemplate, predastoreMultinodeNodes(), "AK", "SK", "ap-southeast-2", "tok", "/cfg", "/var/lib/spinifex", "10.0.0.1", 0, admin.NorthstarCredentials{}, ) require.NoError(t, err) @@ -1067,7 +1067,7 @@ func TestPredastoreTemplates_UnsetCompactionEmitsNoBlock(t *testing.T) { "unset single-node tail changed: %q", single[len(single)-40:]) multi, err := admin.GenerateMultiNodePredastoreConfig( - predastoreMultiNodeTemplate, predastoreMultinodeNodes(), "AK", "SK", "ap-southeast-2", "tok", "/cfg", "10.0.0.1", 0, admin.NorthstarCredentials{}, + predastoreMultiNodeTemplate, predastoreMultinodeNodes(), "AK", "SK", "ap-southeast-2", "tok", "/cfg", "/var/lib/spinifex", "10.0.0.1", 0, admin.NorthstarCredentials{}, ) require.NoError(t, err) assert.NotContains(t, multi, "[compaction]") @@ -1088,7 +1088,7 @@ func TestPredastoreTemplates_SetCompactionEmitsParseableBlock(t *testing.T) { assert.Equal(t, interval, singleCfg.Compaction.IntervalSeconds) multi, err := admin.GenerateMultiNodePredastoreConfig( - predastoreMultiNodeTemplate, predastoreMultinodeNodes(), "AK", "SK", "ap-southeast-2", "tok", "/cfg", "10.0.0.1", interval, admin.NorthstarCredentials{}, + predastoreMultiNodeTemplate, predastoreMultinodeNodes(), "AK", "SK", "ap-southeast-2", "tok", "/cfg", "/var/lib/spinifex", "10.0.0.1", interval, admin.NorthstarCredentials{}, ) require.NoError(t, err) var multiCfg compactionConfig diff --git a/cmd/spinifex/cmd/admin_upgrade.go b/cmd/spinifex/cmd/admin_upgrade.go index 4c84ce17c..3455e0904 100644 --- a/cmd/spinifex/cmd/admin_upgrade.go +++ b/cmd/spinifex/cmd/admin_upgrade.go @@ -7,6 +7,11 @@ import ( "strings" "github.com/mulgadc/spinifex/spinifex/migrate" + + // Config migrations that need more than the framework register themselves + // here rather than in migrate, which every handler imports. + _ "github.com/mulgadc/spinifex/spinifex/migrate/predastoretopology" + "github.com/spf13/cobra" ) diff --git a/cmd/spinifex/cmd/admin_upgrade_test.go b/cmd/spinifex/cmd/admin_upgrade_test.go new file mode 100644 index 000000000..896cc9fa1 --- /dev/null +++ b/cmd/spinifex/cmd/admin_upgrade_test.go @@ -0,0 +1,39 @@ +package cmd + +import ( + "os" + "path/filepath" + "testing" + + "github.com/mulgadc/spinifex/spinifex/migrate" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// TestUpgradeSeesEveryConfigTarget pins the registrations `spx admin upgrade` +// depends on. A migration whose package is no longer linked here would leave +// the command reporting nothing pending on an install that needs migrating. +func TestUpgradeSeesEveryConfigTarget(t *testing.T) { + configDir := t.TempDir() + + // Both targets must exist on disk to be reported at all. + require.NoError(t, os.MkdirAll(filepath.Join(configDir, "predastore"), 0750)) + require.NoError(t, os.WriteFile(filepath.Join(configDir, "spinifex.toml"), []byte("version = \"3\"\n"), 0640)) + require.NoError(t, os.WriteFile(filepath.Join(configDir, "predastore", "predastore.toml"), []byte("version = \"4\"\n"), 0640)) + + versions, err := migrate.DefaultRegistry.ConfigVersions(configDir) + require.NoError(t, err) + assert.Equal(t, map[string]int{"spinifex.toml": 3, "predastore.toml": 4}, versions) + + pending, err := migrate.DefaultRegistry.PendingConfig(configDir) + require.NoError(t, err) + + targets := make(map[string][2]int, len(pending)) + for _, p := range pending { + targets[p.Target] = [2]int{p.FromVersion, p.ToVersion} + } + assert.Equal(t, map[string][2]int{ + "spinifex.toml": {3, 4}, + "predastore.toml": {4, 5}, + }, targets) +} diff --git a/cmd/spinifex/cmd/predastore_bind_test.go b/cmd/spinifex/cmd/predastore_bind_test.go index cc010eb21..b278e10a0 100644 --- a/cmd/spinifex/cmd/predastore_bind_test.go +++ b/cmd/spinifex/cmd/predastore_bind_test.go @@ -28,7 +28,7 @@ func writeSpinifexToml(t *testing.T, body string) string { return path } -func TestDerivePredastoreBind_NodeIDConfigured(t *testing.T) { +func TestDerivePredastoreBind_HostIDConfigured(t *testing.T) { resetGlobalViper(t) path := writeSpinifexToml(t, ` @@ -45,7 +45,7 @@ az = "us-east-1a" host = "0.0.0.0:8443" bucket = "predastore" region = "us-east-1" -node_id = 3 +host_id = 3 `) clusterConfig, err := config.LoadConfig(path) @@ -55,16 +55,14 @@ node_id = 3 require.NoError(t, err) assert.Equal(t, "0.0.0.0", bind.Host) assert.Equal(t, 8443, bind.Port) - assert.Equal(t, 3, bind.NodeID) + assert.Equal(t, 3, bind.HostID) } -// TestDerivePredastoreBind_NodeIDAbsentDefaultsToColocated covers the -// co-located NODE_ID=-1 rule: single-node spinifex.toml templates omit -// node_id entirely, since every configured DB peer runs in this one process. -// predastore-start.sh defaulted to -1 for exactly that reason, and predastore -// itself rejects node_id=0, so an absent key must never silently resolve to -// the Go zero value instead. -func TestDerivePredastoreBind_NodeIDAbsentDefaultsToColocated(t *testing.T) { +// TestDerivePredastoreBind_HostIDAbsentRunsWholeTopology covers the +// single-node rule: those spinifex.toml templates omit host_id entirely, +// pinning every node of the topology to one host, which only a +// whole-topology process can serve. +func TestDerivePredastoreBind_HostIDAbsentRunsWholeTopology(t *testing.T) { resetGlobalViper(t) path := writeSpinifexToml(t, ` @@ -90,7 +88,7 @@ region = "us-east-1" require.NoError(t, err) assert.Equal(t, "0.0.0.0", bind.Host) assert.Equal(t, 8443, bind.Port) - assert.Equal(t, -1, bind.NodeID) + assert.Equal(t, 0, bind.HostID) } // TestDerivePredastoreBind_BindHostSurvivesLoadConfigNormalization asserts diff --git a/cmd/spinifex/cmd/service.go b/cmd/spinifex/cmd/service.go index f7f2b9760..d256d577b 100644 --- a/cmd/spinifex/cmd/service.go +++ b/cmd/spinifex/cmd/service.go @@ -89,12 +89,12 @@ var spinifexUICmd = &cobra.Command{ Short: "Manage the spinifex-ui service", } -// predastoreBind is the local predastore bind host, port and node_id derived +// predastoreBind is the local predastore bind host, port and host_id derived // directly from spinifex.toml. type predastoreBind struct { Host string Port int - NodeID int + HostID int } // derivePredastoreBind reads this node's [nodes..predastore] section @@ -104,9 +104,9 @@ type predastoreBind struct { // normalization for callers that DIAL predastore, not for the address // predastore itself binds to. // -// node_id defaults to -1 (co-located: every configured DB peer runs in this -// one process) when spinifex.toml omits the key. predastore rejects -// node_id=0, so an absent key must never silently resolve to that. +// host_id defaults to 0 when spinifex.toml omits the key, which runs the whole +// predastore topology in this one process — the single-node deployment. Only a +// multi-node config names a host (>= 1), selecting just that host's nodes. func derivePredastoreBind(clusterConfig *config.ClusterConfig) (predastoreBind, error) { node := clusterConfig.Node bindKey := "nodes." + node + ".predastore.host" @@ -124,12 +124,7 @@ func derivePredastoreBind(clusterConfig *config.ClusterConfig) (predastoreBind, return predastoreBind{}, fmt.Errorf("parse nodes.%s.predastore.host port %q: %w", node, portStr, err) } - nodeID := -1 - if viper.IsSet("nodes." + node + ".predastore.node_id") { - nodeID = clusterConfig.Nodes[node].Predastore.NodeID - } - - return predastoreBind{Host: host, Port: port, NodeID: nodeID}, nil + return predastoreBind{Host: host, Port: port, HostID: clusterConfig.Nodes[node].Predastore.HostID}, nil } var predastoreStartCmd = &cobra.Command{ @@ -144,9 +139,9 @@ var predastoreStartCmd = &cobra.Command{ host := viper.GetString("predastore-host") basePath := viper.GetString("predastore-base-path") debug := viper.GetBool("predastore-debug") - nodeID := viper.GetInt("predastore-node-id") + hostID := viper.GetInt("predastore-host-id") - // Derive bind host/port/node-id from spinifex.toml when its path is + // Derive bind host/port/host-id from spinifex.toml when its path is // known and the caller hasn't explicitly overridden them — replaces // predastore-start.sh, which used to do this derivation and exec us. if cfgFile := viper.GetString("config"); cfgFile != "" { @@ -166,8 +161,8 @@ var predastoreStartCmd = &cobra.Command{ if !viper.IsSet("predastore-port") { port = bind.Port } - if !viper.IsSet("predastore-node-id") { - nodeID = bind.NodeID + if !viper.IsSet("predastore-host-id") { + hostID = bind.HostID } } @@ -221,7 +216,7 @@ var predastoreStartCmd = &cobra.Command{ EncryptionKeyFile: encryptionKeyFile, - NodeID: nodeID, + HostID: hostID, PprofEnabled: pprofEnabled, PprofOutputPath: pprofOutput, @@ -1034,7 +1029,6 @@ func runNorthstarStart(options northstarStartOptions, deps northstarStartDepende svc, err := deps.newService(&northstar.Config{ ConfigPath: configPath, BasePath: baseDir, - NodeID: nodeConfig.Predastore.NodeID, NatsHost: nodeConfig.NATS.Host, NatsToken: nodeConfig.NATS.ACL.Token, NatsCACert: nodeConfig.NATS.CACert, @@ -1182,7 +1176,7 @@ func bindPredastoreNamespacedEnv() { } // bindPredastoreCollisionEnv namespaces predastore's port, debug, tls-cert, -// tls-key, encryption-key-file and node-id keys, which nats and awsgw also +// tls-key, encryption-key-file and host-id keys, which nats and awsgw also // bind bare. Each derived env name now matches its own BindEnv target. func bindPredastoreCollisionEnv() { viper.BindEnv("predastore-port", "SPINIFEX_PREDASTORE_PORT") @@ -1200,8 +1194,8 @@ func bindPredastoreCollisionEnv() { viper.BindEnv("predastore-encryption-key-file", "SPINIFEX_PREDASTORE_ENCRYPTION_KEY_FILE") viper.BindPFlag("predastore-encryption-key-file", predastoreCmd.PersistentFlags().Lookup("encryption-key-file")) - viper.BindEnv("predastore-node-id", "SPINIFEX_PREDASTORE_NODE_ID") - viper.BindPFlag("predastore-node-id", predastoreCmd.PersistentFlags().Lookup("node-id")) + viper.BindEnv("predastore-host-id", "SPINIFEX_PREDASTORE_HOST_ID") + viper.BindPFlag("predastore-host-id", predastoreCmd.PersistentFlags().Lookup("host-id")) } // bindNatsCollisionEnv namespaces nats's port, host and debug keys, which @@ -1281,12 +1275,13 @@ func init() { // Predastore at-rest encryption master key (per node; mode 0600) predastoreCmd.PersistentFlags().String("encryption-key-file", "", "Path to this node's 32-byte AES-256 master key file (required)") - // Predastore Node ID. Default -1 is dev mode (launch every configured - // QUIC node in-process). Production deployments set this to the node's - // real ID (>= 1) via SPINIFEX_PREDASTORE_NODE_ID or --node-id. - predastoreCmd.PersistentFlags().Int("node-id", -1, "Predastore (S3) node ID (-1 = dev mode, >= 1 = production)") + // Predastore host ID: which [[host]] of the predastore topology this + // process is. Default 0 runs every node of the topology in this process, + // which is the single-node deployment. Multi-node deployments set the + // host's real ID (>= 1) via SPINIFEX_PREDASTORE_HOST_ID or --host-id. + predastoreCmd.PersistentFlags().Int("host-id", 0, "Predastore cluster host ID (0 = run the whole topology in this process, >= 1 = this host only)") - // Namespaced viper keys for port/debug/tls-cert/tls-key/encryption-key-file/node-id + // Namespaced viper keys for port/debug/tls-cert/tls-key/encryption-key-file/host-id // (see bindPredastoreCollisionEnv doc comment) bindPredastoreCollisionEnv() diff --git a/cmd/spinifex/cmd/templates/predastore-multinode.toml b/cmd/spinifex/cmd/templates/predastore-multinode.toml index a743a5fcf..24a81396f 100644 --- a/cmd/spinifex/cmd/templates/predastore-multinode.toml +++ b/cmd/spinifex/cmd/templates/predastore-multinode.toml @@ -1,10 +1,10 @@ # Predastore S3-compatible storage configuration file in TOML format. # Generated for multi-node cluster -version = "4" +version = "5" region = "{{.Region}}" -host = "0.0.0.0" -port = 8443 +# The gateway listen address comes from the daemon CLI options: a top-level +# host key would collide with the [[host]] topology tables below. debug = false @@ -16,26 +16,28 @@ parity = 1 data = 2 parity = 1 {{- end}} + +# One host per machine: a predastore process owning a socket and a data +# directory. bind_addr binds every interface; public_addr is what the other +# hosts dial. {{- range .Nodes}} -[[db]] +[[host]] id = {{.ID}} -host = "{{.Host}}" -port = 6660 -path = "distributed/db/node-{{.ID}}/" -access_key_id = "{{$.AccessKey}}" -secret_access_key = "{{$.SecretKey}}" -{{- if eq .ID 1}} -leader = true +bind_addr = "0.0.0.0:6660" +public_addr = "{{.Host}}:6660" +data_dir = "{{$.PredastoreDataDir}}" {{- end}} -{{- end}} -{{- range .Nodes}} -[[nodes]] +# Nodes are roles pinned to a host: each machine carries one shard-storage +# node for erasure-coded object shards and one state replica for the Raft +# quorum over global state. +{{- range .ClusterNodes}} + +[[node]] id = {{.ID}} -host = "{{.Host}}" -port = 9991 -path = "distributed/nodes/node-{{.ID}}/" +host_id = {{.HostID}} +role = "{{.Role}}" {{- end}} [[buckets]] diff --git a/cmd/spinifex/cmd/templates/predastore.toml b/cmd/spinifex/cmd/templates/predastore.toml index e5082499d..3adee8288 100644 --- a/cmd/spinifex/cmd/templates/predastore.toml +++ b/cmd/spinifex/cmd/templates/predastore.toml @@ -1,11 +1,10 @@ # Predastore S3-compatible storage configuration file in TOML format. # MULTI NODE used for testing -version = "4" +version = "5" region = "{{.Region}}" -# Optionally configure port and host, or specify via daemon CLI options -host = "0.0.0.0" -port = 8443 +# The gateway listen address comes from the daemon CLI options: a top-level +# host key would collide with the [[host]] topology table below. # Enable verbose debugging debug = false @@ -20,52 +19,50 @@ debug = false data = 2 parity = 1 -[[db]] +# A host is one predastore process: it owns a socket and a data directory, +# and runs the nodes pinned to it. A single-node deployment declares one host +# and colocates every node in it, so all intra-cluster traffic goes over the +# in-process pipe and no socket is bound. bind_addr/public_addr are unused +# while everything is colocated; they only matter once nodes are split across +# processes. +[[host]] id = 1 -host = "127.0.0.1" -port = 6660 -#raft_port = 7660 # Optional: Raft consensus port (default: port + 1000) -#advertise_host = "" # Optional: Address to advertise to other Raft nodes -path = "distributed/db/node-1/" -access_key_id = "{{.AccessKey}}" -secret_access_key = "{{.SecretKey}}" -leader = true # Bootstrap leader hint +bind_addr = "127.0.0.1:6660" +public_addr = "127.0.0.1:6660" +data_dir = "{{.PredastoreDataDir}}" -[[db]] -id = 2 -host = "127.0.0.1" -port = 6661 -path = "distributed/db/node-2/" -access_key_id = "{{.AccessKey}}" -secret_access_key = "{{.SecretKey}}" - -[[db]] -id = 3 -host = "127.0.0.1" -port = 6662 -path = "distributed/db/node-3/" -access_key_id = "{{.AccessKey}}" -secret_access_key = "{{.SecretKey}}" - -# Shard nodes -# Using distributed hash ring for object > shard placement. -[[nodes]] +# Shard nodes: the distributed hash ring places object shards across these. +# RS(2,1) needs 3. +[[node]] id = 1 -host = "127.0.0.1" -port = 9991 -path = "distributed/nodes/node-1/" +host_id = 1 +role = "shard-storage" -[[nodes]] +[[node]] id = 2 -host = "127.0.0.1" -port = 9992 -path = "distributed/nodes/node-2/" +host_id = 1 +role = "shard-storage" -[[nodes]] +[[node]] id = 3 -host = "127.0.0.1" -port = 9993 -path = "distributed/nodes/node-3/" +host_id = 1 +role = "shard-storage" + +# State replicas: the Raft quorum over global state (buckets, object index). +[[node]] +id = 4 +host_id = 1 +role = "state-replica" + +[[node]] +id = 5 +host_id = 1 +role = "state-replica" + +[[node]] +id = 6 +host_id = 1 +role = "state-replica" [[buckets]] name = "predastore" diff --git a/cmd/spinifex/cmd/templates/spinifex.toml b/cmd/spinifex/cmd/templates/spinifex.toml index 0031b7a83..a308638d7 100644 --- a/cmd/spinifex/cmd/templates/spinifex.toml +++ b/cmd/spinifex/cmd/templates/spinifex.toml @@ -1,5 +1,5 @@ # Domain configuration file in TOML format. -version = "3" +version = "4" epoch = 1 node = "{{.Node}}" {{- if .OperatorEmail}} @@ -140,8 +140,8 @@ region = "{{.Region}}" bucket = "predastore" accesskey = "{{.AccessKey}}" secretkey = "{{.SecretKey}}" -{{- if gt .PredastoreNodeID 0}} -node_id = {{.PredastoreNodeID}} +{{- if gt .PredastoreHostID 0}} +host_id = {{.PredastoreHostID}} {{- end}} {{- if .NorthstarConfigPath}} diff --git a/docs/COMMANDS.md b/docs/COMMANDS.md index 197d10c6c..567cf9187 100644 --- a/docs/COMMANDS.md +++ b/docs/COMMANDS.md @@ -10,7 +10,7 @@ Service lifecycle commands for starting, stopping, and checking status of all Sp | Command | Flags | Description | |---------|-------|-------------| -| `spx service predastore start` | `--port` (default: 8443), `--host` (default: 0.0.0.0), `--base-path`, `--config-path`, `--debug`, `--tls-cert`, `--tls-key`, `--backend` (distributed/filesystem, default: distributed), `--node-id` (default: -1), `--pprof`, `--pprof-output` | Creates predastore service instance with S3-compatible storage backend → starts service. When `--config`/`SPINIFEX_CONFIG_PATH` points at a cluster spinifex.toml, `--host`/`--port`/`--node-id` default to that node's `[nodes..predastore]` section instead of the flag defaults above; an explicit flag or `SPINIFEX_PREDASTORE_*` env var still overrides it. `node_id` absent from spinifex.toml resolves to -1 (co-located mode), never to 0. | +| `spx service predastore start` | `--port` (default: 8443), `--host` (default: 0.0.0.0), `--base-path`, `--config-path`, `--debug`, `--tls-cert`, `--tls-key`, `--encryption-key-file`, `--host-id` (default: 0 = run the whole topology in this process), `--pprof`, `--pprof-output` | Creates predastore service instance with S3-compatible storage backend → starts service. When `--config`/`SPINIFEX_CONFIG_PATH` points at a cluster spinifex.toml, `--host`/`--port`/`--host-id` default to that node's `[nodes..predastore]` section instead of the flag defaults above; an explicit flag or `SPINIFEX_PREDASTORE_*` env var still overrides it. `host_id` absent from spinifex.toml resolves to 0, running the whole topology in this process. | | `spx service predastore stop` | — | Stops the predastore service | | `spx service predastore status` | — | Reports predastore service status | | `spx service viperblock start` | `--s3-host` (default: 0.0.0.0:8443), `--s3-bucket` (default: predastore), `--s3-region` (default: ap-southeast-2), `--plugin-path` (auto-detected via `nbdkit --dump-config plugindir`; typically `/usr/lib/x86_64-linux-gnu/nbdkit/plugins/nbdkit-viperblock-plugin.so` on amd64, overridable via `SPINIFEX_VIPERBLOCK_PLUGIN_PATH` in `/etc/spinifex/systemd.env`) | Loads cluster config → connects to NATS and Predastore → starts viperblock block storage service with NBD plugin | diff --git a/docs/security/network-connections/README.md b/docs/security/network-connections/README.md index 9d6f38868..9eb0428fb 100644 --- a/docs/security/network-connections/README.md +++ b/docs/security/network-connections/README.md @@ -163,7 +163,7 @@ Every listener and outbound destination is controlled by one of these files. Cha |------|------|----------| | `/etc/spinifex/spinifex.toml` | `nodes..{awsgw,nats,predastore,daemon}.host`, `nodes..vpcd.ovn_{nb,sb}_addr`, `nodes..daemon.dev_networking` | Per-service bind addresses/ports; dev-mode QEMU port forwarding. | | `/etc/spinifex/nats.conf` | `listen`, `cluster.listen`, `cluster.routes`, `http`, `tls`, `cluster.authorization` | NATS client/cluster/monitoring listeners, peer routes, TLS, cluster token. | -| `/etc/spinifex/predastore.toml` | `host`, `port`, `[[db]].port`, `[[nodes]].port`, `tls.*` | Predastore S3 listener, Raft ports, shard ports, TLS certs. | +| `/etc/spinifex/predastore.toml` | `[[host]].bind_addr`, `[[host]].public_addr`, `[[node]].role`, `tls.*` | Predastore intra-cluster listener (one socket per host, carrying every node pinned to it), node placement, TLS certs. The S3 listener itself is set by the service's `--host`/`--port`. | | OVN packages (`ovn-central`, `ovn-host`) | `ovn-nb-db`, `ovn-sb-db` (via `ovs-vsctl set open_vswitch …`) | OVN DB bind addresses. | | Spinifex UI service | Built-in defaults: `host = "0.0.0.0"`, `port = 3000`. No `spinifex.toml` block today. | UI listener. | | `spx admin init` / `spx admin join` | `--port`, `--token-ttl`, `--no-telemetry` (or `SPX_NO_TELEMETRY=1`) | Formation port, token TTL, telemetry opt-out. | diff --git a/spinifex/admin/admin.go b/spinifex/admin/admin.go index 5f0322541..78a49f157 100644 --- a/spinifex/admin/admin.go +++ b/spinifex/admin/admin.go @@ -51,6 +51,9 @@ type ConfigSettings struct { DataDir string LogDir string ConfigDir string + // PredastoreDataDir is the [[host]] data_dir rendered into predastore.toml. + // Derived from DataDir via PredastoreDataDir when left unset. + PredastoreDataDir string Node string Az string @@ -65,8 +68,9 @@ type ConfigSettings struct { ClusterRoutes []string ClusterName string - // Predastore multi-node - PredastoreNodeID int + // Predastore multi-node: which [[host]] of the predastore topology this + // node is. Zero means single-node, where one process runs every node. + PredastoreHostID int // CompactionIntervalSeconds gates the predastore [compaction] block. Zero // means unset: no block is emitted and predastore keeps its built-in default. @@ -174,6 +178,10 @@ func GenerateConfigFiles(configs []ConfigFile, configSettings ConfigSettings) er // GenerateConfigFile creates a configuration file from a template. func GenerateConfigFile(configPath string, configTemplate string, configSettings ConfigSettings) error { + if configSettings.PredastoreDataDir == "" { + configSettings.PredastoreDataDir = PredastoreDataDir(configSettings.DataDir) + } + tmpl, err := template.New("config").Parse(configTemplate) if err != nil { return fmt.Errorf("failed to parse template: %w", err) @@ -303,6 +311,15 @@ func GenerateServerCertOnly(configDir string, bindIP, awsRegion, internalSuffix return GenerateSignedCert(serverCertPath, serverKeyPath, caCertPath, caKeyPath, extraIPs, extraDNS) } +// PredastoreDataDir is the root a Predastore host keeps its per-node state +// under. Each node gets a node- subdirectory beneath it, created by +// predastore itself. The path is absolute because the service runs under +// systemd, where the working directory the config would otherwise resolve +// against is not ours to depend on. +func PredastoreDataDir(spxRoot string) string { + return filepath.Join(spxRoot, "predastore", "cluster") +} + func CreateServiceDirectories(spxRoot string) { dirs := []string{ filepath.Join(spxRoot, "images"), @@ -934,34 +951,73 @@ func SetupAWSCredentials(accessKey, secretKey, region, certPath, bindIP string) return nil } +// PredastoreClusterNode is one [[node]] entry of the Predastore topology: a +// role pinned to a host. +type PredastoreClusterNode struct { + ID int + HostID int + Role string +} + +// Predastore node roles: shard storage holds erasure-coded object shards, +// state replicas form the Raft quorum over global state. +const ( + predastoreRoleShardStorage = "shard-storage" + predastoreRoleStateReplica = "state-replica" +) + +// PredastoreTopology derives the cluster nodes for a set of machines: each +// machine hosts one shard-storage node and one state replica. Node IDs are +// unique across roles, so storage takes 1..n and the replicas n+1..2n. +func PredastoreTopology(nodes []PredastoreNodeConfig) []PredastoreClusterNode { + out := make([]PredastoreClusterNode, 0, len(nodes)*2) + for _, n := range nodes { + out = append(out, PredastoreClusterNode{ID: n.ID, HostID: n.ID, Role: predastoreRoleShardStorage}) + } + for _, n := range nodes { + out = append(out, PredastoreClusterNode{ID: n.ID + len(nodes), HostID: n.ID, Role: predastoreRoleStateReplica}) + } + return out +} + // GenerateMultiNodePredastoreConfig produces a complete predastore.toml for a -// multi-node Predastore cluster. Each node gets its own DB entry (port 6660) -// and shard entry (port 9991) on a distinct IP. Node ID 1 is the bootstrap leader. +// multi-node Predastore cluster. Each machine becomes one [[host]] — a +// predastore process owning a socket on port 6660 and a data directory — +// carrying the shard-storage and state-replica nodes pinned to it. +// +// dataDir is the Spinifex data root; the hosts' data directories are absolute +// beneath it, since the service runs under systemd with no dependable working +// directory. // // A populated northstar credential provisions the zone bucket, grants the // system key write access to it, and adds the read-only entry the resolver // authenticates with. A zero value omits all three, yielding a config no // northstar service can use. -func GenerateMultiNodePredastoreConfig(templateStr string, nodes []PredastoreNodeConfig, accessKey, secretKey, region, natsToken, configDir, bindIP string, compactionIntervalSeconds int, northstar NorthstarCredentials) (string, error) { +func GenerateMultiNodePredastoreConfig(templateStr string, nodes []PredastoreNodeConfig, accessKey, secretKey, region, natsToken, configDir, dataDir, bindIP string, compactionIntervalSeconds int, northstar NorthstarCredentials) (string, error) { if len(nodes) < 2 { return "", fmt.Errorf("multi-node predastore requires at least 2 nodes, got %d", len(nodes)) } data := struct { Nodes []PredastoreNodeConfig + ClusterNodes []PredastoreClusterNode AccessKey string SecretKey string Region string NatsToken string ConfigDir string + PredastoreDataDir string BindIP string CompactionIntervalSeconds int NorthstarAccessKey string NorthstarSecretKey string NorthstarBucket string }{ - Nodes: nodes, AccessKey: accessKey, SecretKey: secretKey, Region: region, - NatsToken: natsToken, ConfigDir: configDir, BindIP: bindIP, + Nodes: nodes, ClusterNodes: PredastoreTopology(nodes), + AccessKey: accessKey, SecretKey: secretKey, Region: region, + NatsToken: natsToken, ConfigDir: configDir, + PredastoreDataDir: PredastoreDataDir(dataDir), + BindIP: bindIP, CompactionIntervalSeconds: compactionIntervalSeconds, NorthstarAccessKey: northstar.AccessKey, NorthstarSecretKey: northstar.SecretKey, @@ -992,17 +1048,32 @@ func FindNodeIDByIP(nodes []PredastoreNodeConfig, ip string) int { return 0 } -// ParsePredastoreNodeIDFromConfig parses a predastore.toml string and returns -// the node ID whose host matches the given IP, or 0 if not found. -func ParsePredastoreNodeIDFromConfig(tomlContent string, ip string) int { +// ParsePredastoreHostIDFromConfig parses a predastore.toml string and returns +// the ID of the [[host]] whose public address matches the given IP, or 0 if +// not found. The host is what an operator places on a machine; the nodes +// pinned to it follow from the topology. +func ParsePredastoreHostIDFromConfig(tomlContent string, ip string) int { var cfg struct { - DB []PredastoreNodeConfig `toml:"db"` + Hosts []struct { + ID int `toml:"id"` + PublicAddr string `toml:"public_addr"` + } `toml:"host"` } if err := toml.Unmarshal([]byte(tomlContent), &cfg); err != nil { slog.Warn("Failed to parse predastore.toml content", "error", err) return 0 } - return FindNodeIDByIP(cfg.DB, ip) + for _, h := range cfg.Hosts { + host, _, err := net.SplitHostPort(h.PublicAddr) + if err != nil { + // A bare address without a port is still a usable match. + host = h.PublicAddr + } + if host == ip { + return h.ID + } + } + return 0 } // SetMIGProfile idempotently writes mig_profile = "" for the given node diff --git a/spinifex/admin/admin_test.go b/spinifex/admin/admin_test.go index c6b79ccfb..2533904d2 100644 --- a/spinifex/admin/admin_test.go +++ b/spinifex/admin/admin_test.go @@ -800,9 +800,10 @@ func TestCreateServiceDirectories_Idempotent(t *testing.T) { // --- Predastore multi-node config --- func TestGenerateMultiNodePredastoreConfig_Success(t *testing.T) { - tmpl := `{{range .Nodes}}[[db]] + tmpl := `{{range .Nodes}}[[host]] id = {{.ID}} -host = "{{.Host}}" +public_addr = "{{.Host}}:6660" +data_dir = "{{$.PredastoreDataDir}}" {{end}}` nodes := []PredastoreNodeConfig{ {ID: 1, Host: "10.0.0.1"}, @@ -810,10 +811,51 @@ host = "{{.Host}}" {ID: 3, Host: "10.0.0.3"}, } - result, err := GenerateMultiNodePredastoreConfig(tmpl, nodes, "AK", "SK", "us-east-1", "nats-token", "/config", "10.0.0.1", 0, NorthstarCredentials{}) + result, err := GenerateMultiNodePredastoreConfig(tmpl, nodes, "AK", "SK", "us-east-1", "nats-token", "/config", "/var/lib/spinifex", "10.0.0.1", 0, NorthstarCredentials{}) require.NoError(t, err) - assert.Contains(t, result, `host = "10.0.0.1"`) - assert.Contains(t, result, `host = "10.0.0.3"`) + assert.Contains(t, result, `public_addr = "10.0.0.1:6660"`) + assert.Contains(t, result, `public_addr = "10.0.0.3:6660"`) + assert.Contains(t, result, `data_dir = "/var/lib/spinifex/predastore/cluster"`) +} + +// Each machine hosts one shard-storage node and one state replica, with node +// IDs unique across both roles so the topology validates. +func TestGenerateMultiNodePredastoreConfig_Topology(t *testing.T) { + tmpl := `{{range .ClusterNodes}}[[node]] +id = {{.ID}} +host_id = {{.HostID}} +role = "{{.Role}}" +{{end}}` + nodes := []PredastoreNodeConfig{ + {ID: 1, Host: "10.0.0.1"}, + {ID: 2, Host: "10.0.0.2"}, + {ID: 3, Host: "10.0.0.3"}, + } + + result, err := GenerateMultiNodePredastoreConfig(tmpl, nodes, "AK", "SK", "us-east-1", "nats-token", "/config", "/var/lib/spinifex", "10.0.0.1", 0, NorthstarCredentials{}) + require.NoError(t, err) + + assert.Contains(t, result, "id = 1\nhost_id = 1\nrole = \"shard-storage\"") + assert.Contains(t, result, "id = 3\nhost_id = 3\nrole = \"shard-storage\"") + assert.Contains(t, result, "id = 4\nhost_id = 1\nrole = \"state-replica\"") + assert.Contains(t, result, "id = 6\nhost_id = 3\nrole = \"state-replica\"") +} + +func TestPredastoreTopology_UniqueIDsAcrossRoles(t *testing.T) { + topology := PredastoreTopology([]PredastoreNodeConfig{ + {ID: 1, Host: "10.0.0.1"}, + {ID: 2, Host: "10.0.0.2"}, + }) + + require.Len(t, topology, 4) + seen := map[int]bool{} + for _, n := range topology { + assert.False(t, seen[n.ID], "duplicate node id %d", n.ID) + seen[n.ID] = true + assert.Contains(t, []int{1, 2}, n.HostID) + } + assert.Equal(t, "shard-storage", topology[0].Role) + assert.Equal(t, "state-replica", topology[2].Role) } // The northstar template fields were declared but never assigned, so every @@ -826,7 +868,7 @@ func TestGenerateMultiNodePredastoreConfig_NorthstarCredentialsReachTemplate(t * {ID: 2, Host: "10.0.0.2"}, } - result, err := GenerateMultiNodePredastoreConfig(tmpl, nodes, "AK", "SK", "us-east-1", "nats-token", "/config", "10.0.0.1", 0, + result, err := GenerateMultiNodePredastoreConfig(tmpl, nodes, "AK", "SK", "us-east-1", "nats-token", "/config", "/var/lib/spinifex", "10.0.0.1", 0, NorthstarCredentials{AccessKey: "NSAK", SecretKey: "NSSK", Bucket: "northstar"}) require.NoError(t, err) assert.Equal(t, `access = "NSAK" secret = "NSSK" bucket = "northstar"`, result) @@ -841,7 +883,7 @@ func TestGenerateMultiNodePredastoreConfig_NoNorthstarCredentials(t *testing.T) {ID: 2, Host: "10.0.0.2"}, } - result, err := GenerateMultiNodePredastoreConfig(tmpl, nodes, "AK", "SK", "us-east-1", "nats-token", "/config", "10.0.0.1", 0, + result, err := GenerateMultiNodePredastoreConfig(tmpl, nodes, "AK", "SK", "us-east-1", "nats-token", "/config", "/var/lib/spinifex", "10.0.0.1", 0, NorthstarCredentials{}) require.NoError(t, err) assert.Equal(t, `access = "" secret = "" bucket = ""`, result) @@ -852,7 +894,7 @@ func TestGenerateMultiNodePredastoreConfig_MinimumNodes(t *testing.T) { _, err := GenerateMultiNodePredastoreConfig(tmpl, []PredastoreNodeConfig{ {ID: 1, Host: "10.0.0.1"}, - }, "AK", "SK", "us-east-1", "nats-token", "/config", "10.0.0.1", 0, NorthstarCredentials{}) + }, "AK", "SK", "us-east-1", "nats-token", "/config", "/var/lib/spinifex", "10.0.0.1", 0, NorthstarCredentials{}) assert.Error(t, err) assert.Contains(t, err.Error(), "at least 2 nodes") } @@ -860,7 +902,7 @@ func TestGenerateMultiNodePredastoreConfig_MinimumNodes(t *testing.T) { func TestGenerateMultiNodePredastoreConfig_InvalidTemplate(t *testing.T) { _, err := GenerateMultiNodePredastoreConfig("{{.Unclosed", []PredastoreNodeConfig{ {ID: 1, Host: "a"}, {ID: 2, Host: "b"}, {ID: 3, Host: "c"}, - }, "AK", "SK", "us-east-1", "nats-token", "/config", "10.0.0.1", 0, NorthstarCredentials{}) + }, "AK", "SK", "us-east-1", "nats-token", "/config", "/var/lib/spinifex", "10.0.0.1", 0, NorthstarCredentials{}) assert.Error(t, err) assert.Contains(t, err.Error(), "failed to parse") } @@ -879,26 +921,40 @@ func TestFindNodeIDByIP(t *testing.T) { assert.Equal(t, 0, FindNodeIDByIP(nil, "10.0.0.1")) } -// --- ParsePredastoreNodeIDFromConfig --- +// --- ParsePredastoreHostIDFromConfig --- -func TestParsePredastoreNodeIDFromConfig(t *testing.T) { +func TestParsePredastoreHostIDFromConfig(t *testing.T) { tomlContent := ` -[[db]] +[[host]] id = 1 -host = "10.0.0.1" +bind_addr = "0.0.0.0:6660" +public_addr = "10.0.0.1:6660" -[[db]] +[[host]] id = 2 -host = "10.0.0.2" +bind_addr = "0.0.0.0:6660" +public_addr = "10.0.0.2:6660" -[[db]] +[[host]] id = 3 -host = "10.0.0.3" +bind_addr = "0.0.0.0:6660" +public_addr = "10.0.0.3:6660" ` - assert.Equal(t, 2, ParsePredastoreNodeIDFromConfig(tomlContent, "10.0.0.2")) - assert.Equal(t, 0, ParsePredastoreNodeIDFromConfig(tomlContent, "10.0.0.99")) - assert.Equal(t, 0, ParsePredastoreNodeIDFromConfig("invalid toml {{{", "10.0.0.1")) - assert.Equal(t, 0, ParsePredastoreNodeIDFromConfig("", "10.0.0.1")) + assert.Equal(t, 2, ParsePredastoreHostIDFromConfig(tomlContent, "10.0.0.2")) + assert.Equal(t, 0, ParsePredastoreHostIDFromConfig(tomlContent, "10.0.0.99")) + assert.Equal(t, 0, ParsePredastoreHostIDFromConfig("invalid toml {{{", "10.0.0.1")) + assert.Equal(t, 0, ParsePredastoreHostIDFromConfig("", "10.0.0.1")) +} + +// A public_addr without a port must still match, so a hand-edited config that +// omits it resolves to a host rather than silently to zero. +func TestParsePredastoreHostIDFromConfig_AddressWithoutPort(t *testing.T) { + tomlContent := ` +[[host]] +id = 7 +public_addr = "10.0.0.7" +` + assert.Equal(t, 7, ParsePredastoreHostIDFromConfig(tomlContent, "10.0.0.7")) } // --- Integration: Full config generation flow --- diff --git a/spinifex/config/config.go b/spinifex/config/config.go index 157fa9b18..fb60eb464 100644 --- a/spinifex/config/config.go +++ b/spinifex/config/config.go @@ -262,7 +262,9 @@ type PredastoreConfig struct { AccessKey string `json:"AccessKey" mapstructure:"accesskey"` SecretKey string `json:"SecretKey" mapstructure:"secretkey"` BaseDir string `json:"BaseDir" mapstructure:"base_dir"` - NodeID int `json:"NodeID" mapstructure:"node_id"` + // HostID is which [[host]] of the predastore topology this node runs; + // unset means the node runs the whole topology in one process. + HostID int `json:"HostID" mapstructure:"host_id"` } // GPUModelOverride maps a PCI vendor/device ID to a GPU instance family for diff --git a/spinifex/config/config_test.go b/spinifex/config/config_test.go index 6c9cae5df..ab9a97219 100644 --- a/spinifex/config/config_test.go +++ b/spinifex/config/config_test.go @@ -219,7 +219,7 @@ region = "ap-southeast-2" accesskey = "AK" secretkey = "SK" base_dir = "/data" -node_id = 1 +host_id = 1 [nodes.n1.awsgw] host = "0.0.0.0:9999" @@ -242,7 +242,7 @@ expected_nodes = 3 assert.Equal(t, "mybucket", n.Predastore.Bucket) assert.Equal(t, "AK", n.Predastore.AccessKey) assert.Equal(t, "/data", n.Predastore.BaseDir) - assert.Equal(t, 1, n.Predastore.NodeID) + assert.Equal(t, 1, n.Predastore.HostID) assert.Equal(t, "0.0.0.0:9999", n.AWSGW.Host) assert.True(t, n.AWSGW.Debug) assert.Equal(t, 3, n.AWSGW.ExpectedNodes) diff --git a/spinifex/daemon/daemon_handlers_storage.go b/spinifex/daemon/daemon_handlers_storage.go index 84c47518a..6695d9471 100644 --- a/spinifex/daemon/daemon_handlers_storage.go +++ b/spinifex/daemon/daemon_handlers_storage.go @@ -2,8 +2,10 @@ package daemon import ( "log/slog" + "net" "os" "path/filepath" + "strconv" "github.com/mulgadc/spinifex/spinifex/types" "github.com/nats-io/nats.go" @@ -14,8 +16,8 @@ import ( // containing only the fields needed for storage metrics (no credentials). type predastoreTOML struct { RS predastoreRS `toml:"rs"` - DB []predastoreDBNode `toml:"db"` - Nodes []predastoreNode `toml:"nodes"` + Hosts []predastoreHost `toml:"host"` + Nodes []predastoreNode `toml:"node"` Buckets []predastoreBucket `toml:"buckets"` } @@ -24,18 +26,27 @@ type predastoreRS struct { Parity int `toml:"parity"` } -type predastoreDBNode struct { - ID int `toml:"id"` - Host string `toml:"host"` - Port int `toml:"port"` +// predastoreHost is one predastore process; the nodes pinned to it are all +// reachable at its public address. +type predastoreHost struct { + ID int `toml:"id"` + PublicAddr string `toml:"public_addr"` } +// predastoreNode is a role pinned to a host. Its address is the host's, so +// reporting a node means resolving its host_id. type predastoreNode struct { - ID int `toml:"id"` - Host string `toml:"host"` - Port int `toml:"port"` + ID int `toml:"id"` + HostID int `toml:"host_id"` + Role string `toml:"role"` } +// Predastore node roles, as written in the [[node]] blocks. +const ( + predastoreRoleShardStorage = "shard-storage" + predastoreRoleStateReplica = "state-replica" +) + type predastoreBucket struct { Name string `toml:"name"` Type string `toml:"type"` @@ -69,23 +80,26 @@ func (d *Daemon) handleStorageConfig(msg *nats.Msg) { }, } - for _, db := range cfg.DB { - resp.DBNodes = append(resp.DBNodes, types.StorageDBNode{ - ID: db.ID, - Host: db.Host, - Port: db.Port, - }) - } - if resp.DBNodes == nil { - resp.DBNodes = []types.StorageDBNode{} + // A node's address is its host's: every node pinned to a host is served + // by that one process, keyed apart within it. + hostAddrs := make(map[int]predastoreHost, len(cfg.Hosts)) + for _, h := range cfg.Hosts { + hostAddrs[h.ID] = h } for _, n := range cfg.Nodes { - resp.ShardNodes = append(resp.ShardNodes, types.StorageShardNode{ - ID: n.ID, - Host: n.Host, - Port: n.Port, - }) + host, port := splitPredastoreAddr(hostAddrs[n.HostID].PublicAddr) + switch n.Role { + case predastoreRoleStateReplica: + resp.DBNodes = append(resp.DBNodes, types.StorageDBNode{ID: n.ID, Host: host, Port: port}) + case predastoreRoleShardStorage: + resp.ShardNodes = append(resp.ShardNodes, types.StorageShardNode{ID: n.ID, Host: host, Port: port}) + default: + slog.Warn("handleStorageConfig: unknown predastore node role", "node", n.ID, "role", n.Role) + } + } + if resp.DBNodes == nil { + resp.DBNodes = []types.StorageDBNode{} } if resp.ShardNodes == nil { resp.ShardNodes = []types.StorageShardNode{} @@ -104,3 +118,18 @@ func (d *Daemon) handleStorageConfig(msg *nats.Msg) { respondWithJSON(msg, resp) } + +// splitPredastoreAddr splits a host's "ip:port" into its parts. An address +// missing or malformed in the config yields the raw value and a zero port +// rather than dropping the node from the report entirely. +func splitPredastoreAddr(addr string) (string, int) { + host, portStr, err := net.SplitHostPort(addr) + if err != nil { + return addr, 0 + } + port, err := strconv.Atoi(portStr) + if err != nil { + return host, 0 + } + return host, port +} diff --git a/spinifex/daemon/daemon_handlers_storage_test.go b/spinifex/daemon/daemon_handlers_storage_test.go new file mode 100644 index 000000000..138721747 --- /dev/null +++ b/spinifex/daemon/daemon_handlers_storage_test.go @@ -0,0 +1,82 @@ +package daemon + +import ( + "testing" + + toml "github.com/pelletier/go-toml/v2" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// The storage status report is built by parsing predastore.toml directly, so +// it degrades silently when the schema moves under it: a config that no longer +// matches yields an empty topology rather than an error. This pins the parse +// to the [[host]]/[[node]] dialect the templates actually emit. +func TestPredastoreTOML_ParsesClusterTopology(t *testing.T) { + content := ` +[rs] +data = 2 +parity = 1 + +[[host]] +id = 1 +bind_addr = "0.0.0.0:6660" +public_addr = "10.0.0.1:6660" +data_dir = "/var/lib/spinifex/predastore/cluster" + +[[host]] +id = 2 +bind_addr = "0.0.0.0:6660" +public_addr = "10.0.0.2:6660" +data_dir = "/var/lib/spinifex/predastore/cluster" + +[[node]] +id = 1 +host_id = 1 +role = "shard-storage" + +[[node]] +id = 2 +host_id = 2 +role = "shard-storage" + +[[node]] +id = 3 +host_id = 1 +role = "state-replica" + +[[buckets]] +name = "predastore" +type = "distributed" +region = "ap-southeast-2" +` + + var cfg predastoreTOML + require.NoError(t, toml.Unmarshal([]byte(content), &cfg)) + + assert.Equal(t, 2, cfg.RS.Data) + assert.Equal(t, 1, cfg.RS.Parity) + require.Len(t, cfg.Hosts, 2) + require.Len(t, cfg.Nodes, 3) + + assert.Equal(t, "10.0.0.2:6660", cfg.Hosts[1].PublicAddr) + assert.Equal(t, predastoreRoleShardStorage, cfg.Nodes[0].Role) + assert.Equal(t, 1, cfg.Nodes[2].HostID) + assert.Equal(t, predastoreRoleStateReplica, cfg.Nodes[2].Role) +} + +func TestSplitPredastoreAddr(t *testing.T) { + host, port := splitPredastoreAddr("10.0.0.1:6660") + assert.Equal(t, "10.0.0.1", host) + assert.Equal(t, 6660, port) + + // A hand-edited address without a port must still name its node rather + // than dropping it from the report. + host, port = splitPredastoreAddr("10.0.0.1") + assert.Equal(t, "10.0.0.1", host) + assert.Equal(t, 0, port) + + host, port = splitPredastoreAddr("") + assert.Empty(t, host) + assert.Equal(t, 0, port) +} diff --git a/spinifex/migrate/005_spinifex_predastore_host_id.go b/spinifex/migrate/005_spinifex_predastore_host_id.go new file mode 100644 index 000000000..37df2df54 --- /dev/null +++ b/spinifex/migrate/005_spinifex_predastore_host_id.go @@ -0,0 +1,91 @@ +package migrate + +import ( + "fmt" + "os" + "path/filepath" + "regexp" + "strings" +) + +// The spinifex config target, relative to the Spinifex config directory. +const ( + spinifexTarget = "spinifex.toml" + spinifexRelPath = "spinifex.toml" + // spinifexHostIDVersion is the schema where the predastore section names a + // cluster host rather than a node. + spinifexHostIDVersion = 4 +) + +// spinifexPredastoreTableRe matches the per-node predastore section header, +// [nodes..predastore], which is the only place node_id ever appeared. +var spinifexPredastoreTableRe = regexp.MustCompile(`^\[nodes\.[^\[\]]+\.predastore\]$`) + +// spinifexNodeIDKeyRe matches the key to rename, preserving its indentation +// and spacing so the rest of the line is untouched. +var spinifexNodeIDKeyRe = regexp.MustCompile(`^(\s*)node_id(\s*=)`) + +// spinifexTableHeaderRe captures the name of a table or table-array header. +var spinifexTableHeaderRe = regexp.MustCompile(`^\[\[?([^\[\]]+)\]?\]$`) + +func init() { + DefaultRegistry.RegisterConfigTarget(spinifexTarget, spinifexRelPath, &TOMLVersionReader{}) + DefaultRegistry.RegisterConfig(spinifexTarget, ConfigMigration{ + FromVersion: 3, + ToVersion: spinifexHostIDVersion, + Description: "rename [nodes.*.predastore] node_id → host_id", + Run: migrateSpinifexPredastoreHostID, + }) +} + +// migrateSpinifexPredastoreHostID renames the predastore node selector. +// +// A machine is one predastore host now, and derivePredastoreBind reads this +// key to pick the process's host. Left as node_id, a multi-node member falls +// back to host 0 and tries to run every node in the cluster in its own process. +// +// Single-node installs never had the key — the template only emits it for +// multi-node — so for them this is a version bump and nothing else. +func migrateSpinifexPredastoreHostID(ctx ConfigContext) error { + path := filepath.Join(ctx.ConfigDir, spinifexRelPath) + + data, err := os.ReadFile(path) + if err != nil { + return fmt.Errorf("read %s: %w", path, err) + } + + lines := strings.Split(string(data), "\n") + inPredastore := false + renamed := 0 + + for i, line := range lines { + trimmed := strings.TrimSpace(line) + + if spinifexTableHeaderRe.MatchString(trimmed) { + inPredastore = spinifexPredastoreTableRe.MatchString(trimmed) + continue + } + if !inPredastore { + continue + } + if spinifexNodeIDKeyRe.MatchString(line) { + lines[i] = spinifexNodeIDKeyRe.ReplaceAllString(line, "${1}host_id${2}") + renamed++ + } + } + + if renamed == 0 { + return nil + } + + info, err := os.Stat(path) + if err != nil { + return fmt.Errorf("stat %s: %w", path, err) + } + if err := os.WriteFile(path, []byte(strings.Join(lines, "\n")), info.Mode()); err != nil { + return fmt.Errorf("write %s: %w", path, err) + } + + ctx.Logger.Info("Renamed predastore node_id to host_id", "path", path, "keys", renamed) + return nil +} diff --git a/spinifex/migrate/005_spinifex_predastore_host_id_test.go b/spinifex/migrate/005_spinifex_predastore_host_id_test.go new file mode 100644 index 000000000..ee0e61c02 --- /dev/null +++ b/spinifex/migrate/005_spinifex_predastore_host_id_test.go @@ -0,0 +1,99 @@ +package migrate + +import ( + "os" + "path/filepath" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// spinifexTomlFixture writes a spinifex.toml and returns its config directory. +func spinifexTomlFixture(t *testing.T, body string) string { + t.Helper() + configDir := t.TempDir() + require.NoError(t, os.WriteFile(filepath.Join(configDir, spinifexRelPath), []byte(body), 0640)) + return configDir +} + +func readSpinifexToml(t *testing.T, configDir string) string { + t.Helper() + data, err := os.ReadFile(filepath.Join(configDir, spinifexRelPath)) + require.NoError(t, err) + return string(data) +} + +// TestSpinifexPredastoreHostIDRename covers a multi-node member, where the key +// selects which predastore host this machine runs. +func TestSpinifexPredastoreHostIDRename(t *testing.T) { + configDir := spinifexTomlFixture(t, `version = "3" + +[nodes.node1] +node = "node1" +host = "10.0.0.1" + +[nodes.node1.predastore] +host = "10.0.0.1:8443" +region = "ap-southeast-2" +bucket = "predastore" +node_id = 2 + +[nodes.node2] +node = "node2" +`) + + require.NoError(t, DefaultRegistry.RunConfig(spinifexTarget, configDir, t.TempDir())) + + got := readSpinifexToml(t, configDir) + assert.Contains(t, got, "host_id = 2") + assert.NotContains(t, got, "node_id") + assert.Contains(t, got, `version = "4"`) + + // The surrounding tables are untouched. + assert.Contains(t, got, `host = "10.0.0.1:8443"`) + assert.Contains(t, got, "[nodes.node2]") +} + +// TestSpinifexPredastoreHostIDLeavesOtherTables guards against renaming a +// node_id that belongs to something else — only the predastore section moved +// from naming a node to naming a host. +func TestSpinifexPredastoreHostIDLeavesOtherTables(t *testing.T) { + configDir := spinifexTomlFixture(t, `version = "3" + +[cluster] +node_id = 7 + +[nodes.node1.viperblock] +node_id = 9 + +[nodes.node1.predastore] +node_id = 2 +`) + + require.NoError(t, DefaultRegistry.RunConfig(spinifexTarget, configDir, t.TempDir())) + + got := readSpinifexToml(t, configDir) + assert.Contains(t, got, "[cluster]\nnode_id = 7") + assert.Contains(t, got, "[nodes.node1.viperblock]\nnode_id = 9") + assert.Contains(t, got, "[nodes.node1.predastore]\nhost_id = 2") +} + +// TestSpinifexPredastoreHostIDSingleNode is the common install: the key is +// only emitted for multi-node, so there is nothing to rename. +func TestSpinifexPredastoreHostIDSingleNode(t *testing.T) { + body := `version = "3" + +[nodes.node1.predastore] +host = "0.0.0.0:8443" +bucket = "predastore" +` + configDir := spinifexTomlFixture(t, body) + + require.NoError(t, DefaultRegistry.RunConfig(spinifexTarget, configDir, t.TempDir())) + + got := readSpinifexToml(t, configDir) + assert.Contains(t, got, `version = "4"`) + assert.Contains(t, got, `host = "0.0.0.0:8443"`) + assert.NotContains(t, got, "host_id") +} diff --git a/spinifex/migrate/VERSIONS.md b/spinifex/migrate/VERSIONS.md index 24405153f..a28457c61 100644 --- a/spinifex/migrate/VERSIONS.md +++ b/spinifex/migrate/VERSIONS.md @@ -6,22 +6,31 @@ Single source of truth for the current schema version of every config file Spini |---|---|---| | `nats.conf` | `3` | `cmd/spinifex/cmd/templates/nats.conf` | | `awsgw.toml` | `2` | `cmd/spinifex/cmd/templates/awsgw.toml` | -| `spinifex.toml` | `2` | `cmd/spinifex/cmd/templates/spinifex.toml` | -| `predastore.toml` | `4` | `cmd/spinifex/cmd/templates/predastore.toml` | -| `predastore-multinode.toml` | `4` | `cmd/spinifex/cmd/templates/predastore-multinode.toml` | +| `spinifex.toml` | `4` | `cmd/spinifex/cmd/templates/spinifex.toml` | +| `predastore.toml` | `5` | `cmd/spinifex/cmd/templates/predastore.toml` | +| `predastore-multinode.toml` | `5` | `cmd/spinifex/cmd/templates/predastore-multinode.toml` | ## How versions are stamped 1. **Fresh install (`spx admin init`)** — the version string is baked into the embedded template file and written verbatim to disk. There is no Go constant; the template is the source of truth. -2. **Upgrade (`spx admin upgrade`)** — the migration framework (`Registry.RunConfig`) calls `ConfigVersionReader.WriteVersion` after each registered migration step. With no migrations registered, the upgrade command reports "No pending config migrations" and exits cleanly. +2. **Upgrade (`spx admin upgrade`)** — the migration framework (`Registry.RunConfig`) calls `ConfigVersionReader.WriteVersion` after each registered migration step. A target with no registered migration is left alone, and the upgrade command reports nothing pending for it. -## Current state +## Registered config migrations -**No config migrations are currently registered.** Schema bumps ship via force-reinstall. To restore in-place upgrade for a given target, register a `ConfigMigration` against `DefaultRegistry` in a new file under `spinifex/migrate/` and bump the version in both the template and this document. +| Target | Migration | File | +|---|---|---| +| `predastore.toml` | `4` → `5`: `[[db]]`/`[[nodes]]` → `[[host]]`/`[[node]]`, relocate node data, recover raft | `spinifex/migrate/predastoretopology/004_predastore_topology.go` | +| `spinifex.toml` | `3` → `4`: rename `[nodes.*.predastore]` `node_id` → `host_id` | `spinifex/migrate/005_spinifex_predastore_host_id.go` | + +## Where migrations live + +Register a `ConfigMigration` against `DefaultRegistry` in a new numbered file under `spinifex/migrate/`, and bump the version in both the template and the table above. + +A migration that needs more than the framework and the standard library goes in its own package beside `migrate` instead, blank-imported from `cmd/spinifex/cmd/admin_upgrade.go`. `migrate` is imported by every handler that stamps a KV bucket version, so a heavy dependency there reaches the whole tree — `predastoretopology` needs Predastore's cluster runtime, which is why it sits apart. ## Framework -The framework (`migrate.go`, `version_readers.go`) is intentionally preserved: +The framework (`migrate.go`, `version_readers.go`) covers both kinds of target: -- `RunKV` is still called from service-startup paths to stamp NATS KV bucket versions. -- `RunConfig` / `RunAllConfig` / `PendingConfig` remain available so a future migration can register cleanly without re-introducing the framework. +- `RunKV` is called from service-startup paths to stamp NATS KV bucket versions. +- `RunConfig` / `RunAllConfig` / `PendingConfig` back `spx admin upgrade`, which `scripts/setup.sh` runs with `--yes` while the services are stopped. diff --git a/spinifex/migrate/predastoretopology/004_predastore_topology.go b/spinifex/migrate/predastoretopology/004_predastore_topology.go new file mode 100644 index 000000000..ef4844441 --- /dev/null +++ b/spinifex/migrate/predastoretopology/004_predastore_topology.go @@ -0,0 +1,447 @@ +// Package predastoretopology migrates a Predastore install onto the v5 cluster +// topology: hosts owning a data directory, nodes pinned to them by role. +// +// It sits outside the migrate package because it needs Predastore's cluster +// runtime to rewrite raft state, and migrate is imported by every handler that +// stamps a KV bucket version. Only the spx binary that runs `admin upgrade` +// links this. +package predastoretopology + +import ( + "fmt" + "os" + "path/filepath" + "regexp" + "sort" + "strings" + + toml "github.com/pelletier/go-toml/v2" + + "github.com/mulgadc/predastore/clusterrun" + "github.com/mulgadc/spinifex/spinifex/migrate" +) + +// The predastore config target, relative to the Spinifex config directory. +const ( + predastoreTarget = "predastore.toml" + predastoreRelPath = "predastore/predastore.toml" + // predastoreClusterVersion is the schema the cluster topology lands on; + // the install templates stamp it directly. + predastoreClusterVersion = 5 +) + +// predastoreHostPort is the port a v5 host binds. Every v4 [[db]] entry used +// it too, so a migrated topology keeps the same firewall surface. +const predastoreHostPort = 6660 + +// Predastore v5 node roles, mirroring internal/cluster.Role. +const ( + predastoreRoleShard = "shard-storage" + predastoreRoleReplica = "state-replica" +) + +// predastoreV4Node is the v4 shape of a [[db]] or [[nodes]] entry, reduced to +// the fields the migration needs. A frozen copy keeps the migration +// independent of predastore's config surface, which no longer parses these. +type predastoreV4Node struct { + ID int `toml:"id"` + Host string `toml:"host"` + Path string `toml:"path"` +} + +// predastoreV4Config is the v4 topology. Only the two tables the migration +// rewrites are declared; every other key rides through the rewrite as text. +type predastoreV4Config struct { + DB []predastoreV4Node `toml:"db"` + Nodes []predastoreV4Node `toml:"nodes"` +} + +// predastoreHost is one v5 [[host]]: a predastore process owning a socket and +// a data directory. +type predastoreHost struct { + ID int + BindAddr string + PublicAddr string + DataDir string +} + +// predastoreNode is one v5 [[node]]: a role pinned to a host. +type predastoreNode struct { + ID int + HostID int + Role string +} + +// predastoreMove relocates one node's state from its v4 directory to the v5 +// one derived from the host data directory and the node id. +type predastoreMove struct { + NodeID int + Role string + From string + To string +} + +// predastoreTopology is everything derived from a v4 config: the tables to +// write, the directories to move, and the replica set any relocated state +// replica must be recovered onto. +type predastoreTopology struct { + Hosts []predastoreHost + Nodes []predastoreNode + Moves []predastoreMove + ReplicaIDs []int +} + +// predastoreHostTableRe detects a config that already carries the v5 topology, +// so a part-applied migration can be re-run. +var predastoreHostTableRe = regexp.MustCompile(`(?m)^\s*\[\[host\]\]\s*$`) + +// predastoreTopLevelAddrRe matches the v4 gateway address keys, which must go: +// a scalar host cannot coexist with the [[host]] tables. +var predastoreTopLevelAddrRe = regexp.MustCompile(`^(host|port)\s*=`) + +// predastoreTableHeaderRe captures the name of a table or table-array header. +var predastoreTableHeaderRe = regexp.MustCompile(`^\[\[?([^\[\]]+)\]?\]$`) + +func init() { + migrate.DefaultRegistry.RegisterConfigTarget(predastoreTarget, predastoreRelPath, &migrate.TOMLVersionReader{}) + migrate.DefaultRegistry.RegisterConfig(predastoreTarget, migrate.ConfigMigration{ + FromVersion: 4, + ToVersion: predastoreClusterVersion, + Description: "predastore [[db]]/[[nodes]] → [[host]]/[[node]]; relocate node data and recover raft", + Run: migratePredastoreTopology, + }) +} + +// migratePredastoreTopology carries a v4 install onto the v5 cluster topology: +// the data moves first, since a failure there leaves a config still describing +// where the data actually is. +func migratePredastoreTopology(ctx migrate.ConfigContext) error { + path := filepath.Join(ctx.ConfigDir, predastoreRelPath) + + data, err := os.ReadFile(path) + if err != nil { + return fmt.Errorf("read %s: %w", path, err) + } + + if predastoreHostTableRe.Match(data) { + ctx.Logger.Info("Predastore config already carries [[host]] tables, leaving it alone", "path", path) + return nil + } + + var old predastoreV4Config + if err := toml.Unmarshal(data, &old); err != nil { + return fmt.Errorf("parse %s: %w", path, err) + } + + topo, err := predastoreTopologyFromV4(old, ctx.DataDir) + if err != nil { + return err + } + + if err := relocatePredastoreData(ctx, topo); err != nil { + return err + } + + out, err := rewritePredastoreConfig(data, topo) + if err != nil { + return err + } + + info, err := os.Stat(path) + if err != nil { + return fmt.Errorf("stat %s: %w", path, err) + } + if err := os.WriteFile(path, out, info.Mode()); err != nil { + return fmt.Errorf("write %s: %w", path, err) + } + return nil +} + +// predastoreTopologyFromV4 derives the v5 topology from a v4 config. +// +// A machine is one host. Shard nodes keep their ids, because object metadata +// records which node holds each shard; the state replicas are the ones that +// move, taking ids above the shard range. +func predastoreTopologyFromV4(old predastoreV4Config, spxRoot string) (*predastoreTopology, error) { + if len(old.DB) == 0 || len(old.Nodes) == 0 { + return nil, fmt.Errorf("predastore config has %d [[db]] and %d [[nodes]] entries; not a v4 distributed topology", len(old.DB), len(old.Nodes)) + } + + db := append([]predastoreV4Node(nil), old.DB...) + shards := append([]predastoreV4Node(nil), old.Nodes...) + sort.Slice(db, func(i, j int) bool { return db[i].ID < db[j].ID }) + sort.Slice(shards, func(i, j int) bool { return shards[i].ID < shards[j].ID }) + + // A single-node install lists every db peer on one address; that is one + // machine, so one host carries the lot. + singleMachine := true + for _, n := range db[1:] { + if n.Host != db[0].Host { + singleMachine = false + break + } + } + + dataDir := predastoreDataDir(spxRoot) + basePath := predastoreBasePath(spxRoot) + + topo := &predastoreTopology{} + hostByAddr := make(map[string]int, len(db)) + + if singleMachine { + hostByAddr[db[0].Host] = 1 + topo.Hosts = append(topo.Hosts, predastoreHost{ + ID: 1, + BindAddr: fmt.Sprintf("%s:%d", db[0].Host, predastoreHostPort), + PublicAddr: fmt.Sprintf("%s:%d", db[0].Host, predastoreHostPort), + DataDir: dataDir, + }) + } else { + for _, n := range db { + if _, dup := hostByAddr[n.Host]; dup { + return nil, fmt.Errorf("predastore config lists two [[db]] entries on %s; cannot derive one host per machine", n.Host) + } + hostByAddr[n.Host] = n.ID + topo.Hosts = append(topo.Hosts, predastoreHost{ + ID: n.ID, + // The v4 multi-node template bound every interface; keep that + // rather than narrowing an operator's reachability on upgrade. + BindAddr: fmt.Sprintf("0.0.0.0:%d", predastoreHostPort), + PublicAddr: fmt.Sprintf("%s:%d", n.Host, predastoreHostPort), + DataDir: dataDir, + }) + } + } + + for _, n := range shards { + hostID, ok := hostByAddr[n.Host] + if !ok { + return nil, fmt.Errorf("predastore shard node %d is on %s, which has no [[db]] entry to place it", n.ID, n.Host) + } + topo.Nodes = append(topo.Nodes, predastoreNode{ID: n.ID, HostID: hostID, Role: predastoreRoleShard}) + topo.Moves = append(topo.Moves, predastoreMove{ + NodeID: n.ID, + Role: predastoreRoleShard, + From: predastoreV4Path(basePath, n.Path), + To: filepath.Join(dataDir, fmt.Sprintf("node-%d", n.ID)), + }) + } + + // State replica ids start above the shard range so the two never collide. + offset := shards[len(shards)-1].ID + for _, n := range db { + id := n.ID + offset + hostID := hostByAddr[n.Host] + topo.Nodes = append(topo.Nodes, predastoreNode{ID: id, HostID: hostID, Role: predastoreRoleReplica}) + topo.ReplicaIDs = append(topo.ReplicaIDs, id) + topo.Moves = append(topo.Moves, predastoreMove{ + NodeID: id, + Role: predastoreRoleReplica, + From: predastoreV4Path(basePath, n.Path), + To: filepath.Join(dataDir, fmt.Sprintf("node-%d", id)), + }) + } + + return topo, nil +} + +// relocatePredastoreData moves each node's state into the v5 layout and +// rewrites the raft configuration of every state replica it moves. +// +// Only this machine's directories exist, but the config describes the whole +// cluster, so a missing source is the normal multi-node case rather than an +// error. A destination that already exists is a re-run. +func relocatePredastoreData(ctx migrate.ConfigContext, topo *predastoreTopology) error { + for _, m := range topo.Moves { + moved, err := movePredastoreNodeDir(ctx, m) + if err != nil { + return err + } + if !moved || m.Role != predastoreRoleReplica { + continue + } + + // The replica set is renumbered, so the configuration persisted here + // names servers that no longer exist. Raft will not bootstrap over + // existing state to correct that; it has to be rewritten. + recovered, err := clusterrun.RecoverStateReplica(m.To, m.NodeID, topo.ReplicaIDs) + if err != nil { + return fmt.Errorf("recover raft configuration for state replica %d in %s: %w", m.NodeID, m.To, err) + } + ctx.Logger.Info("Migrated predastore state replica", "nodeID", m.NodeID, "dataDir", m.To, "raftRecovered", recovered) + } + return nil +} + +// movePredastoreNodeDir moves one node directory, reporting whether the +// destination now holds state that came from a v4 layout. +func movePredastoreNodeDir(ctx migrate.ConfigContext, m predastoreMove) (bool, error) { + if _, err := os.Stat(m.From); err != nil { + if !os.IsNotExist(err) { + return false, fmt.Errorf("stat %s: %w", m.From, err) + } + // Already moved by an earlier run; the replica still needs recovering + // if that run failed before it got there. + if _, err := os.Stat(m.To); err == nil { + return true, nil + } + return false, nil + } + + if _, err := os.Stat(m.To); err == nil { + return false, fmt.Errorf("cannot move %s to %s: destination already exists", m.From, m.To) + } + + if err := os.MkdirAll(filepath.Dir(m.To), 0750); err != nil { + return false, fmt.Errorf("create %s: %w", filepath.Dir(m.To), err) + } + if err := os.Rename(m.From, m.To); err != nil { + return false, fmt.Errorf("move %s to %s: %w", m.From, m.To, err) + } + ctx.Logger.Info("Moved predastore node data", "nodeID", m.NodeID, "role", m.Role, "from", m.From, "to", m.To) + return true, nil +} + +// rewritePredastoreConfig replaces the v4 topology tables and the gateway +// address keys with the v5 tables, leaving every other table as the operator +// left it. Re-rendering from the template would instead discard their buckets, +// credentials and rate limits. +func rewritePredastoreConfig(data []byte, topo *predastoreTopology) ([]byte, error) { + lines := strings.Split(string(data), "\n") + + var out, pending []string + var table string + dropping := false + wroteTopology := false + wroteGatewayNote := false + + // flush emits the buffered comment/blank block ahead of a line that keeps it. + flush := func() { + out = append(out, pending...) + pending = nil + } + + for _, line := range lines { + trimmed := strings.TrimSpace(line) + + if m := predastoreTableHeaderRe.FindStringSubmatch(trimmed); m != nil { + table = strings.TrimSpace(m[1]) + // The comment block introducing a dropped table goes with it. + if table == "db" || table == "nodes" { + pending = nil + dropping = true + if !wroteTopology { + out = append(out, predastoreTopologyBlock(topo)...) + wroteTopology = true + } + continue + } + dropping = false + flush() + out = append(out, line) + continue + } + + blankOrComment := trimmed == "" || strings.HasPrefix(trimmed, "#") + + if dropping { + // Buffer separators and trailing comments; a key line means the + // comments above it described the dropped table, so let them go. + if blankOrComment { + pending = append(pending, line) + } else { + pending = nil + } + continue + } + + if blankOrComment { + pending = append(pending, line) + continue + } + + if table == "" && predastoreTopLevelAddrRe.MatchString(trimmed) { + pending = nil + if !wroteGatewayNote { + out = append(out, predastoreGatewayNote...) + wroteGatewayNote = true + } + continue + } + + flush() + out = append(out, line) + } + flush() + + if !wroteTopology { + return nil, fmt.Errorf("predastore config has no [[db]] or [[nodes]] tables to replace") + } + return []byte(strings.Join(out, "\n")), nil +} + +// predastoreGatewayNote replaces the top-level host/port keys, which the v5 +// schema cannot accept. +var predastoreGatewayNote = []string{ + "", + "# The gateway listen address comes from the daemon CLI options: a top-level", + "# host key would collide with the [[host]] topology tables below.", +} + +// predastoreTopologyBlock renders the v5 tables, matching the layout the +// install templates generate so a migrated config reads like a fresh one. +func predastoreTopologyBlock(topo *predastoreTopology) []string { + out := []string{ + "", + "# One host per machine: a predastore process owning a socket and a data", + "# directory. bind_addr binds every interface; public_addr is what the other", + "# hosts dial.", + } + for _, h := range topo.Hosts { + out = append(out, + "", + "[[host]]", + fmt.Sprintf("id = %d", h.ID), + fmt.Sprintf("bind_addr = %q", h.BindAddr), + fmt.Sprintf("public_addr = %q", h.PublicAddr), + fmt.Sprintf("data_dir = %q", h.DataDir), + ) + } + + out = append(out, + "", + "# Nodes are roles pinned to a host: shard-storage nodes hold erasure-coded", + "# object shards, state replicas form the Raft quorum over global state.", + ) + for _, n := range topo.Nodes { + out = append(out, + "", + "[[node]]", + fmt.Sprintf("id = %d", n.ID), + fmt.Sprintf("host_id = %d", n.HostID), + fmt.Sprintf("role = %q", n.Role), + ) + } + return out +} + +// predastoreV4Path resolves a v4 node path, which is relative to the +// predastore base path unless the operator made it absolute. +func predastoreV4Path(basePath, path string) string { + if filepath.IsAbs(path) { + return filepath.Clean(path) + } + return filepath.Join(basePath, path) +} + +// predastoreBasePath mirrors SPINIFEX_PREDASTORE_BASE_PATH in +// build/systemd/spinifex-predastore.service; v4 node paths resolve against it. +func predastoreBasePath(spxRoot string) string { + return filepath.Join(spxRoot, "predastore") +} + +// predastoreDataDir mirrors admin.PredastoreDataDir. Duplicated to keep the +// migrate package free of the admin dependency; if either moves, both must. +func predastoreDataDir(spxRoot string) string { + return filepath.Join(spxRoot, "predastore", "cluster") +} diff --git a/spinifex/migrate/predastoretopology/004_predastore_topology_test.go b/spinifex/migrate/predastoretopology/004_predastore_topology_test.go new file mode 100644 index 000000000..2081269c8 --- /dev/null +++ b/spinifex/migrate/predastoretopology/004_predastore_topology_test.go @@ -0,0 +1,265 @@ +package predastoretopology + +import ( + "flag" + "fmt" + "log/slog" + "os" + "path/filepath" + "strings" + "testing" + + "github.com/mulgadc/predastore/s3" + "github.com/mulgadc/spinifex/spinifex/migrate" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +var updateGolden = flag.Bool("update-golden", false, "rewrite the predastore migration golden files") + +// predastoreFixture lays out a v4 install under a temporary root: the config +// in place, and a data directory for each node this machine owns. +func predastoreFixture(t *testing.T, name string, nodeDirs []string) (configDir, dataDir string) { + t.Helper() + + root := t.TempDir() + configDir = filepath.Join(root, "etc") + dataDir = filepath.Join(root, "var") + + require.NoError(t, os.MkdirAll(filepath.Join(configDir, "predastore"), 0750)) + + src, err := os.ReadFile(filepath.Join("testdata", name)) + require.NoError(t, err) + require.NoError(t, os.WriteFile(filepath.Join(configDir, predastoreRelPath), src, 0640)) + + for _, dir := range nodeDirs { + full := filepath.Join(dataDir, "predastore", dir) + require.NoError(t, os.MkdirAll(full, 0750)) + // A marker per directory proves the contents travelled, not just the path. + require.NoError(t, os.WriteFile(filepath.Join(full, "marker"), []byte(dir), 0640)) + } + return configDir, dataDir +} + +// runPredastoreMigration drives the real registry path, so registration, the +// backup and the version stamp are covered alongside the rewrite itself. +func runPredastoreMigration(t *testing.T, configDir, dataDir string) { + t.Helper() + require.NoError(t, migrate.DefaultRegistry.RunConfig(predastoreTarget, configDir, dataDir)) +} + +// dataDirPlaceholder stands in for the test's temporary root, which is +// rendered into data_dir and cannot be checked in. +const dataDirPlaceholder = "{{DATA_DIR}}" + +// assertGolden compares the migrated config against its golden file, which is +// checked in so a review sees the exact config an operator ends up with. +func assertGolden(t *testing.T, configDir, dataDir, golden string) { + t.Helper() + + raw, err := os.ReadFile(filepath.Join(configDir, predastoreRelPath)) + require.NoError(t, err) + got := strings.ReplaceAll(string(raw), dataDir, dataDirPlaceholder) + + path := filepath.Join("testdata", golden) + if *updateGolden { + require.NoError(t, os.WriteFile(path, []byte(got), 0640)) + return + } + want, err := os.ReadFile(path) + require.NoError(t, err) + assert.Equal(t, string(want), got) +} + +// readMigratedConfig parses the migrated config the way predastore does, so +// the test fails on anything predastore would reject rather than on text. +func readMigratedConfig(t *testing.T, configDir, dataDir string) *s3.Config { + t.Helper() + + cfg := &s3.Config{ + ConfigPath: filepath.Join(configDir, predastoreRelPath), + BasePath: filepath.Join(dataDir, "predastore"), + } + require.NoError(t, cfg.ReadConfig()) + return cfg +} + +// TestPredastoreTopologySingleNode covers the common install: three db peers +// and three shard nodes on one machine collapse to one host running all six. +func TestPredastoreTopologySingleNode(t *testing.T) { + configDir, dataDir := predastoreFixture(t, "predastore_v4_singlenode.toml", []string{ + "distributed/db/node-1", "distributed/db/node-2", "distributed/db/node-3", + "distributed/nodes/node-1", "distributed/nodes/node-2", "distributed/nodes/node-3", + }) + + runPredastoreMigration(t, configDir, dataDir) + assertGolden(t, configDir, dataDir, "predastore_v5_singlenode.golden") + + cfg := readMigratedConfig(t, configDir, dataDir) + + require.Len(t, cfg.Hosts, 1) + assert.Equal(t, 1, cfg.Hosts[0].ID) + assert.Equal(t, "127.0.0.1:6660", cfg.Hosts[0].BindAddr) + assert.Equal(t, "127.0.0.1:6660", cfg.Hosts[0].PublicAddr) + assert.Equal(t, filepath.Join(dataDir, "predastore", "cluster"), cfg.Hosts[0].DataDir) + + require.Len(t, cfg.ClusterNodes, 6) + for i, n := range cfg.ClusterNodes { + assert.Equal(t, i+1, n.ID) + assert.Equal(t, 1, n.HostID) + if i < 3 { + assert.Equal(t, "shard-storage", string(n.Role)) + } else { + assert.Equal(t, "state-replica", string(n.Role)) + } + } + + // Shard nodes keep their ids because object metadata records which node + // holds each shard; the state replicas are the ones that move. + cluster := filepath.Join(dataDir, "predastore", "cluster") + assertNodeDirContents(t, cluster, map[int]string{ + 1: "distributed/nodes/node-1", + 2: "distributed/nodes/node-2", + 3: "distributed/nodes/node-3", + 4: "distributed/db/node-1", + 5: "distributed/db/node-2", + 6: "distributed/db/node-3", + }) + + assert.NoDirExists(t, filepath.Join(dataDir, "predastore", "distributed", "db", "node-1")) + assert.NoDirExists(t, filepath.Join(dataDir, "predastore", "distributed", "nodes", "node-1")) + + // Everything outside the topology is the operator's, and must survive. + raw, err := os.ReadFile(filepath.Join(configDir, predastoreRelPath)) + require.NoError(t, err) + for _, keep := range []string{ + `[ratelimit.action."s3:PutObject"]`, + "access_keys_bucket = \"spinifex-iam-access-keys\"", + "interval_seconds = 300", + `{ bucket = "predastore", actions = ["s3:ListBucket", "s3:GetObject",`, + } { + assert.Contains(t, string(raw), keep) + } + assert.NotContains(t, string(raw), "\nhost = \"0.0.0.0\"") + assert.NotContains(t, string(raw), "[[db]]") + assert.NotContains(t, string(raw), "[[nodes]]") +} + +// TestPredastoreTopologyMultiNode covers a three-machine cluster from the +// perspective of one member, which only holds its own node directories. +func TestPredastoreTopologyMultiNode(t *testing.T) { + configDir, dataDir := predastoreFixture(t, "predastore_v4_multinode.toml", []string{ + "distributed/db/node-2", "distributed/nodes/node-2", + }) + + runPredastoreMigration(t, configDir, dataDir) + assertGolden(t, configDir, dataDir, "predastore_v5_multinode.golden") + + cfg := readMigratedConfig(t, configDir, dataDir) + + require.Len(t, cfg.Hosts, 3) + for i, h := range cfg.Hosts { + assert.Equal(t, i+1, h.ID) + // Binding every interface preserves the v4 template's reachability; + // only the advertised address is per machine. + assert.Equal(t, "0.0.0.0:6660", h.BindAddr) + assert.Equal(t, fmt.Sprintf("10.0.0.%d:6660", h.ID), h.PublicAddr) + assert.Equal(t, filepath.Join(dataDir, "predastore", "cluster"), h.DataDir) + } + + require.Len(t, cfg.ClusterNodes, 6) + byID := map[int]struct { + host int + role string + }{} + for _, n := range cfg.ClusterNodes { + byID[n.ID] = struct { + host int + role string + }{n.HostID, string(n.Role)} + } + for id, want := range map[int]struct { + host int + role string + }{ + 1: {1, "shard-storage"}, 2: {2, "shard-storage"}, 3: {3, "shard-storage"}, + 4: {1, "state-replica"}, 5: {2, "state-replica"}, 6: {3, "state-replica"}, + } { + assert.Equal(t, want, byID[id], "node %d", id) + } + + // Only this machine's directories exist; the rest belong to its peers. + cluster := filepath.Join(dataDir, "predastore", "cluster") + assertNodeDirContents(t, cluster, map[int]string{ + 2: "distributed/nodes/node-2", + 5: "distributed/db/node-2", + }) + assert.NoDirExists(t, filepath.Join(cluster, "node-1")) + assert.NoDirExists(t, filepath.Join(cluster, "node-4")) +} + +// TestPredastoreTopologyRerunIsNoop covers a run that failed after writing the +// config but before the version was stamped: the migration is invoked directly, +// so the version gate cannot mask a missing guard on the already-migrated file. +func TestPredastoreTopologyRerunIsNoop(t *testing.T) { + configDir, dataDir := predastoreFixture(t, "predastore_v4_singlenode.toml", []string{ + "distributed/db/node-1", "distributed/nodes/node-1", + }) + ctx := migrate.ConfigContext{ConfigDir: configDir, DataDir: dataDir, Logger: slog.Default()} + + require.NoError(t, migratePredastoreTopology(ctx)) + first, err := os.ReadFile(filepath.Join(configDir, predastoreRelPath)) + require.NoError(t, err) + + require.NoError(t, migratePredastoreTopology(ctx)) + second, err := os.ReadFile(filepath.Join(configDir, predastoreRelPath)) + require.NoError(t, err) + + assert.Equal(t, string(first), string(second)) + assertNodeDirContents(t, filepath.Join(dataDir, "predastore", "cluster"), map[int]string{ + 1: "distributed/nodes/node-1", + 4: "distributed/db/node-1", + }) +} + +// TestPredastoreTopologyRejectsNonV4Config fails loudly rather than writing a +// config with no topology at all. +func TestPredastoreTopologyRejectsNonV4Config(t *testing.T) { + configDir := filepath.Join(t.TempDir(), "etc") + require.NoError(t, os.MkdirAll(filepath.Join(configDir, "predastore"), 0750)) + require.NoError(t, os.WriteFile(filepath.Join(configDir, predastoreRelPath), + []byte("version = \"4\"\nregion = \"ap-southeast-2\"\n"), 0640)) + + err := migratePredastoreTopology(migrate.ConfigContext{ + ConfigDir: configDir, + DataDir: t.TempDir(), + Logger: slog.Default(), + }) + require.Error(t, err) + assert.Contains(t, err.Error(), "not a v4 distributed topology") +} + +// TestPredastoreVersionReaderReadsFixtures ties the migration's FromVersion to +// what the reader actually finds in a shipped config. +func TestPredastoreVersionReaderReadsFixtures(t *testing.T) { + r := &migrate.TOMLVersionReader{} + for _, name := range []string{"predastore_v4_singlenode.toml", "predastore_v4_multinode.toml"} { + v, err := r.ReadVersion(filepath.Join("testdata", name)) + require.NoError(t, err) + assert.Equal(t, 4, v, name) + } +} + +// assertNodeDirContents checks that each node directory holds the marker +// written into the v4 directory it came from. +func assertNodeDirContents(t *testing.T, clusterDir string, want map[int]string) { + t.Helper() + for id, origin := range want { + marker := filepath.Join(clusterDir, fmt.Sprintf("node-%d", id), "marker") + got, err := os.ReadFile(marker) + if !assert.NoError(t, err, "node-%d", id) { + continue + } + assert.Equal(t, origin, string(got), "node-%d", id) + } +} diff --git a/spinifex/migrate/predastoretopology/testdata/predastore_v4_multinode.toml b/spinifex/migrate/predastoretopology/testdata/predastore_v4_multinode.toml new file mode 100644 index 000000000..476ea5c6b --- /dev/null +++ b/spinifex/migrate/predastoretopology/testdata/predastore_v4_multinode.toml @@ -0,0 +1,97 @@ +# Predastore S3-compatible storage configuration file in TOML format. +# Generated for multi-node cluster +version = "4" +region = "ap-southeast-2" + +host = "0.0.0.0" +port = 8443 + +debug = false + +[rs] +data = 2 +parity = 1 + +[[db]] +id = 1 +host = "10.0.0.1" +port = 6660 +path = "distributed/db/node-1/" +access_key_id = "AKIAIOSFODNN7EXAMPLE" +secret_access_key = "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY" +leader = true + +[[db]] +id = 2 +host = "10.0.0.2" +port = 6660 +path = "distributed/db/node-2/" +access_key_id = "AKIAIOSFODNN7EXAMPLE" +secret_access_key = "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY" + +[[db]] +id = 3 +host = "10.0.0.3" +port = 6660 +path = "distributed/db/node-3/" +access_key_id = "AKIAIOSFODNN7EXAMPLE" +secret_access_key = "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY" + +[[nodes]] +id = 1 +host = "10.0.0.1" +port = 9991 +path = "distributed/nodes/node-1/" + +[[nodes]] +id = 2 +host = "10.0.0.2" +port = 9991 +path = "distributed/nodes/node-2/" + +[[nodes]] +id = 3 +host = "10.0.0.3" +port = 9991 +path = "distributed/nodes/node-3/" + +[[buckets]] +name = "predastore" +region = "ap-southeast-2" +type = "distributed" +public = false +encryption = "" +account_id = "000000000000" + +# Northstar DNS zone storage. Read-only for the northstar service; the control +# plane writes zone files with the system credentials. +[[buckets]] +name = "northstar-zones" +region = "ap-southeast-2" +type = "distributed" +public = false +encryption = "" +account_id = "000000000000" + +[[auth]] +access_key_id = "AKIAIOSFODNN7EXAMPLE" +secret_access_key = "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY" +account_id = "000000000000" +policy = [ + { bucket = "predastore", actions = ["s3:ListBucket", "s3:GetObject", "s3:PutObject", "s3:DeleteObject", "s3:ListAllMyBuckets"] }, + { bucket = "northstar-zones", actions = ["s3:ListBucket", "s3:GetObject", "s3:PutObject", "s3:DeleteObject"] }, +] + +# API request throttling (per-account, per-action token bucket) +[ratelimit] +enabled = false +rate = 100 # Default sustained requests per second per account per action +burst = 500 # Default maximum burst capacity + +# IAM authentication via NATS KV (enables multi-account S3 access) +[iam] +nats_url = "nats://10.0.0.2:4222" +nats_token = "s3cr3t" +nats_ca_cert = "/etc/spinifex/ca.pem" +master_key_path = "/etc/spinifex/master.key" +access_keys_bucket = "spinifex-iam-access-keys" diff --git a/spinifex/migrate/predastoretopology/testdata/predastore_v4_singlenode.toml b/spinifex/migrate/predastoretopology/testdata/predastore_v4_singlenode.toml new file mode 100644 index 000000000..d9643bb3f --- /dev/null +++ b/spinifex/migrate/predastoretopology/testdata/predastore_v4_singlenode.toml @@ -0,0 +1,109 @@ +# Predastore S3-compatible storage configuration file in TOML format. +# MULTI NODE used for testing +version = "4" +region = "ap-southeast-2" + +# Optionally configure port and host, or specify via daemon CLI options +host = "0.0.0.0" +port = 8443 + +# Enable verbose debugging +debug = false + +# Optionally configure base-path if full directory path not defined for "fs" buckets +# base_path = "/home/predastore/data/" + +# Specify reed solomon encoding - +# RS(2,1), 2 data, 1 parity. Minimum +# RS(3,2), 3 data, 2 parity. Recommended for production +[rs] +data = 2 +parity = 1 + +[[db]] +id = 1 +host = "127.0.0.1" +port = 6660 +#raft_port = 7660 # Optional: Raft consensus port (default: port + 1000) +#advertise_host = "" # Optional: Address to advertise to other Raft nodes +path = "distributed/db/node-1/" +access_key_id = "AKIAIOSFODNN7EXAMPLE" +secret_access_key = "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY" +leader = true # Bootstrap leader hint + +[[db]] +id = 2 +host = "127.0.0.1" +port = 6661 +path = "distributed/db/node-2/" +access_key_id = "AKIAIOSFODNN7EXAMPLE" +secret_access_key = "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY" + +[[db]] +id = 3 +host = "127.0.0.1" +port = 6662 +path = "distributed/db/node-3/" +access_key_id = "AKIAIOSFODNN7EXAMPLE" +secret_access_key = "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY" + +# Shard nodes +# Using distributed hash ring for object > shard placement. +[[nodes]] +id = 1 +host = "127.0.0.1" +port = 9991 +path = "distributed/nodes/node-1/" + +[[nodes]] +id = 2 +host = "127.0.0.1" +port = 9992 +path = "distributed/nodes/node-2/" + +[[nodes]] +id = 3 +host = "127.0.0.1" +port = 9993 +path = "distributed/nodes/node-3/" + +[[buckets]] +name = "predastore" +region = "ap-southeast-2" +type = "distributed" +public = false +encryption = "" +account_id = "000000000000" + +[[auth]] +access_key_id = "AKIAIOSFODNN7EXAMPLE" +secret_access_key = "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY" +account_id = "000000000000" +policy = [ + { bucket = "predastore", actions = ["s3:ListBucket", "s3:GetObject", "s3:PutObject", "s3:DeleteObject", "s3:ListAllMyBuckets"] }, +] + +# API request throttling (per-account, per-action token bucket) +[ratelimit] +enabled = false +rate = 100 # Default sustained requests per second per account per action +burst = 500 # Default maximum burst capacity + +[ratelimit.action."s3:PutObject"] +rate = 50 +burst = 200 + +[ratelimit.action."s3:DeleteObject"] +rate = 50 +burst = 200 + +# IAM authentication via NATS KV (enables multi-account S3 access) +[iam] +nats_url = "nats://localhost:4222" +nats_token = "s3cr3t" +nats_ca_cert = "/etc/spinifex/ca.pem" +master_key_path = "/etc/spinifex/master.key" +access_keys_bucket = "spinifex-iam-access-keys" + +[compaction] +interval_seconds = 300 diff --git a/spinifex/migrate/predastoretopology/testdata/predastore_v5_multinode.golden b/spinifex/migrate/predastoretopology/testdata/predastore_v5_multinode.golden new file mode 100644 index 000000000..4701b3949 --- /dev/null +++ b/spinifex/migrate/predastoretopology/testdata/predastore_v5_multinode.golden @@ -0,0 +1,109 @@ +# Predastore S3-compatible storage configuration file in TOML format. +# Generated for multi-node cluster +version = "5" +region = "ap-southeast-2" + +# The gateway listen address comes from the daemon CLI options: a top-level +# host key would collide with the [[host]] topology tables below. + +debug = false + +[rs] +data = 2 +parity = 1 + +# One host per machine: a predastore process owning a socket and a data +# directory. bind_addr binds every interface; public_addr is what the other +# hosts dial. + +[[host]] +id = 1 +bind_addr = "0.0.0.0:6660" +public_addr = "10.0.0.1:6660" +data_dir = "{{DATA_DIR}}/predastore/cluster" + +[[host]] +id = 2 +bind_addr = "0.0.0.0:6660" +public_addr = "10.0.0.2:6660" +data_dir = "{{DATA_DIR}}/predastore/cluster" + +[[host]] +id = 3 +bind_addr = "0.0.0.0:6660" +public_addr = "10.0.0.3:6660" +data_dir = "{{DATA_DIR}}/predastore/cluster" + +# Nodes are roles pinned to a host: shard-storage nodes hold erasure-coded +# object shards, state replicas form the Raft quorum over global state. + +[[node]] +id = 1 +host_id = 1 +role = "shard-storage" + +[[node]] +id = 2 +host_id = 2 +role = "shard-storage" + +[[node]] +id = 3 +host_id = 3 +role = "shard-storage" + +[[node]] +id = 4 +host_id = 1 +role = "state-replica" + +[[node]] +id = 5 +host_id = 2 +role = "state-replica" + +[[node]] +id = 6 +host_id = 3 +role = "state-replica" + +[[buckets]] +name = "predastore" +region = "ap-southeast-2" +type = "distributed" +public = false +encryption = "" +account_id = "000000000000" + +# Northstar DNS zone storage. Read-only for the northstar service; the control +# plane writes zone files with the system credentials. +[[buckets]] +name = "northstar-zones" +region = "ap-southeast-2" +type = "distributed" +public = false +encryption = "" +account_id = "000000000000" + +[[auth]] +access_key_id = "AKIAIOSFODNN7EXAMPLE" +secret_access_key = "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY" +account_id = "000000000000" +policy = [ + { bucket = "predastore", actions = ["s3:ListBucket", "s3:GetObject", "s3:PutObject", "s3:DeleteObject", "s3:ListAllMyBuckets"] }, + { bucket = "northstar-zones", actions = ["s3:ListBucket", "s3:GetObject", "s3:PutObject", "s3:DeleteObject"] }, +] + +# API request throttling (per-account, per-action token bucket) +[ratelimit] +enabled = false +rate = 100 # Default sustained requests per second per account per action +burst = 500 # Default maximum burst capacity + +# IAM authentication via NATS KV (enables multi-account S3 access) +[iam] +nats_url = "nats://10.0.0.2:4222" +nats_token = "s3cr3t" +nats_ca_cert = "/etc/spinifex/ca.pem" +master_key_path = "/etc/spinifex/master.key" +access_keys_bucket = "spinifex-iam-access-keys" diff --git a/spinifex/migrate/predastoretopology/testdata/predastore_v5_singlenode.golden b/spinifex/migrate/predastoretopology/testdata/predastore_v5_singlenode.golden new file mode 100644 index 000000000..db58a80ce --- /dev/null +++ b/spinifex/migrate/predastoretopology/testdata/predastore_v5_singlenode.golden @@ -0,0 +1,104 @@ +# Predastore S3-compatible storage configuration file in TOML format. +# MULTI NODE used for testing +version = "5" +region = "ap-southeast-2" + +# The gateway listen address comes from the daemon CLI options: a top-level +# host key would collide with the [[host]] topology tables below. + +# Enable verbose debugging +debug = false + +# Optionally configure base-path if full directory path not defined for "fs" buckets +# base_path = "/home/predastore/data/" + +# Specify reed solomon encoding - +# RS(2,1), 2 data, 1 parity. Minimum +# RS(3,2), 3 data, 2 parity. Recommended for production +[rs] +data = 2 +parity = 1 + +# One host per machine: a predastore process owning a socket and a data +# directory. bind_addr binds every interface; public_addr is what the other +# hosts dial. + +[[host]] +id = 1 +bind_addr = "127.0.0.1:6660" +public_addr = "127.0.0.1:6660" +data_dir = "{{DATA_DIR}}/predastore/cluster" + +# Nodes are roles pinned to a host: shard-storage nodes hold erasure-coded +# object shards, state replicas form the Raft quorum over global state. + +[[node]] +id = 1 +host_id = 1 +role = "shard-storage" + +[[node]] +id = 2 +host_id = 1 +role = "shard-storage" + +[[node]] +id = 3 +host_id = 1 +role = "shard-storage" + +[[node]] +id = 4 +host_id = 1 +role = "state-replica" + +[[node]] +id = 5 +host_id = 1 +role = "state-replica" + +[[node]] +id = 6 +host_id = 1 +role = "state-replica" + +[[buckets]] +name = "predastore" +region = "ap-southeast-2" +type = "distributed" +public = false +encryption = "" +account_id = "000000000000" + +[[auth]] +access_key_id = "AKIAIOSFODNN7EXAMPLE" +secret_access_key = "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY" +account_id = "000000000000" +policy = [ + { bucket = "predastore", actions = ["s3:ListBucket", "s3:GetObject", "s3:PutObject", "s3:DeleteObject", "s3:ListAllMyBuckets"] }, +] + +# API request throttling (per-account, per-action token bucket) +[ratelimit] +enabled = false +rate = 100 # Default sustained requests per second per account per action +burst = 500 # Default maximum burst capacity + +[ratelimit.action."s3:PutObject"] +rate = 50 +burst = 200 + +[ratelimit.action."s3:DeleteObject"] +rate = 50 +burst = 200 + +# IAM authentication via NATS KV (enables multi-account S3 access) +[iam] +nats_url = "nats://localhost:4222" +nats_token = "s3cr3t" +nats_ca_cert = "/etc/spinifex/ca.pem" +master_key_path = "/etc/spinifex/master.key" +access_keys_bucket = "spinifex-iam-access-keys" + +[compaction] +interval_seconds = 300 diff --git a/spinifex/services/northstar/northstar.go b/spinifex/services/northstar/northstar.go index ff5b6d92e..6785f825f 100644 --- a/spinifex/services/northstar/northstar.go +++ b/spinifex/services/northstar/northstar.go @@ -26,7 +26,6 @@ type Config struct { ConfigPath string // BasePath is the node base dir where the PID file is written. BasePath string - NodeID int NatsHost string NatsToken string NatsCACert string diff --git a/spinifex/services/predastore/predastore.go b/spinifex/services/predastore/predastore.go index e350ac932..dd9d605d7 100644 --- a/spinifex/services/predastore/predastore.go +++ b/spinifex/services/predastore/predastore.go @@ -6,11 +6,15 @@ import ( "log/slog" "os" "os/signal" + "sync" "syscall" "time" + "github.com/mulgadc/predastore/clusterrun" + "github.com/mulgadc/predastore/pkg/masterkey" "github.com/mulgadc/predastore/s3" "github.com/mulgadc/spinifex/spinifex/utils" + "golang.org/x/sync/errgroup" ) var serviceName = "predastore" @@ -30,17 +34,27 @@ type Config struct { // fragments are only ever opened on the node that sealed them. EncryptionKeyFile string - NodeID int + // HostID selects which host of the predastore topology this process is. + // A host owns one socket and one data directory, and runs the nodes + // pinned to it. Zero runs every node in the topology in this process, + // which is the single-node deployment and needs no intra-cluster network. + HostID int // Profiling PprofEnabled bool PprofOutputPath string } -// Service wraps the predastore S3 server. +// Service wraps the predastore S3 gateway and the cluster runtime beneath it. type Service struct { Config *Config server *s3.Server + + // stop cancels the process context Start runs under, so Shutdown can + // drain the gateway and the nodes together. Guarded because Shutdown is + // called from a different goroutine than Start. + mu sync.Mutex + stop context.CancelFunc } // New creates a new predastore service. @@ -65,42 +79,109 @@ func (svc *Service) Start() (int, error) { return 0, fmt.Errorf("write pid file: %w", err) } + rt, err := svc.buildRuntime() + if err != nil { + return 0, err + } + server, err := s3.NewServer( s3.WithConfigPath(svc.Config.ConfigPath), s3.WithAddress(svc.Config.Host, svc.Config.Port), s3.WithTLS(svc.Config.TlsCert, svc.Config.TlsKey), s3.WithBasePath(svc.Config.BasePath), s3.WithDebug(svc.Config.Debug), - s3.WithNodeID(svc.Config.NodeID), s3.WithPprof(svc.Config.PprofEnabled, svc.Config.PprofOutputPath), s3.WithEncryptionKeyFile(svc.Config.EncryptionKeyFile), + s3.WithPreparedBackend(rt.Backend), ) if err != nil { + rt.Close() slog.Error("Failed to create predastore server", "error", err) return 0, err } - svc.server = server + // One context for the whole service: a signal, a fatal gateway error or + // Shutdown all stop the cluster nodes and the gateway together. + ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM) + defer stop() - if err := server.ListenAndServeAsync(); err != nil { - slog.Error("Failed to start predastore server", "error", err) + svc.mu.Lock() + svc.server = server + svc.stop = stop + svc.mu.Unlock() + + g, gctx := errgroup.WithContext(ctx) + g.Go(func() error { return rt.Run(gctx) }) + g.Go(func() error { + // Serving before consensus settles would fail writes that would have + // succeeded a moment later; a timeout degrades rather than aborts, + // since the state client retries. + if err := rt.WaitReady(30 * time.Second); err != nil { + slog.Warn("No predastore leader elected within timeout, serving anyway", "error", err) + } + return svc.serve(gctx, server) + }) + + if err := g.Wait(); err != nil { + slog.Error("Predastore service exited", "error", err) return 0, err } - // Wait for shutdown signal - sigChan := make(chan os.Signal, 1) - signal.Notify(sigChan, syscall.SIGINT, syscall.SIGTERM) - <-sigChan + return os.Getpid(), nil +} - slog.Info("Shutting down predastore service") - ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) - defer cancel() - - if err := server.Shutdown(ctx); err != nil { - slog.Error("Error during shutdown", "error", err) +// buildRuntime reads the predastore topology and assembles the cluster nodes +// this process is responsible for, returning the backend the S3 gateway runs +// on top of. +func (svc *Service) buildRuntime() (*clusterrun.Runtime, error) { + cfg := &s3.Config{ConfigPath: svc.Config.ConfigPath, BasePath: svc.Config.BasePath} + if err := cfg.ReadConfig(); err != nil { + return nil, fmt.Errorf("read predastore config %s: %w", svc.Config.ConfigPath, err) } - return os.Getpid(), nil + // An unset host means this process is the whole cluster, so it runs every + // node over the in-process pipe and opens no intra-cluster socket. + nodeIDs := clusterrun.AllNodeIDs(cfg) + if svc.Config.HostID > 0 { + nodeIDs = clusterrun.NodeIDsForHost(cfg, svc.Config.HostID) + if len(nodeIDs) == 0 { + return nil, fmt.Errorf("predastore host %d has no nodes in %s", svc.Config.HostID, svc.Config.ConfigPath) + } + } + + key, err := masterkey.Load(svc.Config.EncryptionKeyFile) + if err != nil { + return nil, fmt.Errorf("load predastore master key: %w", err) + } + + rt, err := clusterrun.Build(cfg, nodeIDs, svc.Config.TlsCert, svc.Config.TlsKey, key) + if err != nil { + return nil, fmt.Errorf("build predastore cluster runtime: %w", err) + } + return rt, nil +} + +// serve runs the S3 gateway until the context is cancelled, then drains it +// within a bounded grace period. A gateway that cannot bind is fatal: leaving +// the cluster running headless would look healthy while serving nothing. +func (svc *Service) serve(ctx context.Context, server *s3.Server) error { + if err := server.ListenAndServeAsync(); err != nil { + return fmt.Errorf("start predastore s3 gateway: %w", err) + } + + select { + case <-ctx.Done(): + case err := <-server.ServeError(): + return fmt.Errorf("predastore s3 gateway: %w", err) + } + + slog.Info("Shutting down predastore service") + shutdownCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + if err := server.Shutdown(shutdownCtx); err != nil { + return fmt.Errorf("shut down predastore s3 gateway: %w", err) + } + return nil } // Stop stops the predastore service. @@ -113,12 +194,17 @@ func (svc *Service) Status() (string, error) { return utils.ServiceStatus(svc.Config.BasePath, serviceName) } -// Shutdown gracefully shuts down the predastore service. +// Shutdown gracefully shuts down the predastore service. When this process is +// the one running the cluster it cancels the service context, which drains the +// gateway and every local node; otherwise it signals the running process. func (svc *Service) Shutdown() error { - if svc.server != nil { - ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) - defer cancel() - return svc.server.Shutdown(ctx) + svc.mu.Lock() + stop := svc.stop + svc.mu.Unlock() + + if stop != nil { + stop() + return nil } return svc.Stop() } diff --git a/spinifex/services/predastore/predastore_runtime_test.go b/spinifex/services/predastore/predastore_runtime_test.go new file mode 100644 index 000000000..c377a467c --- /dev/null +++ b/spinifex/services/predastore/predastore_runtime_test.go @@ -0,0 +1,504 @@ +package predastore + +import ( + "context" + "crypto/ecdsa" + "crypto/elliptic" + "crypto/rand" + "crypto/x509" + "crypto/x509/pkix" + "encoding/pem" + "fmt" + "math/big" + "net" + "os" + "path/filepath" + "strconv" + "strings" + "sync/atomic" + "testing" + "time" + + "github.com/mulgadc/predastore/clusterrun" + "github.com/mulgadc/predastore/s3" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +const ( + roleShardStorage = "shard-storage" + roleStateReplica = "state-replica" +) + +// nodeIDSeq hands out node ids that are unique within this test binary. The +// pipe transport's listener registry is process-wide, and this package's +// integration tests hold a fixture daemon on the low ids for the life of the +// binary, so topologies here must not reuse them. +var nodeIDSeq atomic.Int64 + +func nextNodeIDs(n int) []int { + ids := make([]int, n) + for i := range ids { + ids[i] = 100 + int(nodeIDSeq.Add(1)) + } + return ids +} + +// hostSpec and nodeSpec are one entry each of a topology's [[host]] and +// [[node]] tables. +type hostSpec struct { + id int + dataDir string +} + +type nodeSpec struct { + id int + hostID int + role string +} + +// writeTopology writes a predastore config for hosts and nodes, returning its +// path. Hosts bind an ephemeral port so several topologies can coexist in one +// test binary. +func writeTopology(t *testing.T, dir string, hosts []hostSpec, nodes []nodeSpec) string { + t.Helper() + + var b strings.Builder + fmt.Fprintf(&b, "version = \"1.0\"\nregion = \"us-east-1\"\nbase_path = %q\n\n[rs]\ndata = 1\nparity = 1\n", dir) + for _, h := range hosts { + fmt.Fprintf(&b, "\n[[host]]\nid = %d\nbind_addr = \"127.0.0.1:0\"\npublic_addr = \"127.0.0.1:0\"\ndata_dir = %q\n", h.id, h.dataDir) + } + for _, n := range nodes { + fmt.Fprintf(&b, "\n[[node]]\nid = %d\nhost_id = %d\nrole = %q\n", n.id, n.hostID, n.role) + } + + path := filepath.Join(dir, "predastore.toml") + require.NoError(t, os.WriteFile(path, []byte(b.String()), 0600)) + return path +} + +// writeMasterKey writes a 32-byte master key at mode. The mode matters: +// masterkey.Load is fail-closed on anything group- or other-readable. +func writeMasterKey(t *testing.T, dir string, mode os.FileMode) string { + t.Helper() + + key := make([]byte, 32) + _, err := rand.Read(key) + require.NoError(t, err) + + path := filepath.Join(dir, "master.key") + require.NoError(t, os.WriteFile(path, key, mode)) + return path +} + +// writeCertificate writes a self-signed P-256 certificate serving both the +// intra-cluster QUIC socket and the S3 gateway. Nothing verifies it: the peers +// are this process and the tests' own dialers. +func writeCertificate(t *testing.T, dir string) (certPath, keyPath string) { + t.Helper() + + priv, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader) + require.NoError(t, err) + + tmpl := x509.Certificate{ + SerialNumber: big.NewInt(1), + Subject: pkix.Name{CommonName: "localhost"}, + NotBefore: time.Now().Add(-time.Hour), + NotAfter: time.Now().Add(24 * time.Hour), + KeyUsage: x509.KeyUsageDigitalSignature | x509.KeyUsageKeyEncipherment, + ExtKeyUsage: []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth}, + IPAddresses: []net.IP{net.ParseIP("127.0.0.1")}, + DNSNames: []string{"localhost"}, + BasicConstraintsValid: true, + } + der, err := x509.CreateCertificate(rand.Reader, &tmpl, &tmpl, &priv.PublicKey, priv) + require.NoError(t, err) + + certPath = filepath.Join(dir, "tls.crt") + require.NoError(t, os.WriteFile(certPath, pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: der}), 0600)) + + keyDER, err := x509.MarshalECPrivateKey(priv) + require.NoError(t, err) + keyPath = filepath.Join(dir, "tls.key") + require.NoError(t, os.WriteFile(keyPath, pem.EncodeToMemory(&pem.Block{Type: "EC PRIVATE KEY", Bytes: keyDER}), 0600)) + + return certPath, keyPath +} + +// freePort reserves and releases a loopback port, so a gateway under test +// binds somewhere the fixture daemon and other packages are not. +func freePort(t *testing.T) int { + t.Helper() + + ln, err := net.Listen("tcp", "127.0.0.1:0") + require.NoError(t, err) + defer ln.Close() + return ln.Addr().(*net.TCPAddr).Port +} + +// newTwoHostConfig returns the config for the first host of a two-host +// topology: the shard stores run here, the state replica belongs to the other +// host's process. Returns the local node ids and the remote one. +func newTwoHostConfig(t *testing.T) (cfg *Config, local []int, remote int) { + t.Helper() + + dir := t.TempDir() + certPath, keyPath := writeCertificate(t, dir) + ids := nextNodeIDs(3) + + hosts := []hostSpec{ + {id: 1, dataDir: filepath.Join(dir, "host-1")}, + {id: 2, dataDir: filepath.Join(dir, "host-2")}, + } + nodes := []nodeSpec{ + {id: ids[0], hostID: 1, role: roleShardStorage}, + {id: ids[1], hostID: 1, role: roleShardStorage}, + {id: ids[2], hostID: 2, role: roleStateReplica}, + } + + cfg = &Config{ + ConfigPath: writeTopology(t, dir, hosts, nodes), + Host: "127.0.0.1", + Port: freePort(t), + BasePath: dir, + TlsCert: certPath, + TlsKey: keyPath, + EncryptionKeyFile: writeMasterKey(t, dir, 0600), + HostID: 1, + } + return cfg, ids[:2], ids[2] +} + +// newSingleHostConfig returns the single-node deployment's config: one host +// running every node in this process, over the in-process pipe. +func newSingleHostConfig(t *testing.T) *Config { + t.Helper() + + dir := t.TempDir() + certPath, keyPath := writeCertificate(t, dir) + ids := nextNodeIDs(3) + + hosts := []hostSpec{{id: 1, dataDir: filepath.Join(dir, "host-1")}} + nodes := []nodeSpec{ + {id: ids[0], hostID: 1, role: roleShardStorage}, + {id: ids[1], hostID: 1, role: roleShardStorage}, + {id: ids[2], hostID: 1, role: roleStateReplica}, + } + + // HostID is left zero: this process is the whole cluster. + return &Config{ + ConfigPath: writeTopology(t, dir, hosts, nodes), + Host: "127.0.0.1", + Port: freePort(t), + BasePath: dir, + TlsCert: certPath, + TlsKey: keyPath, + EncryptionKeyFile: writeMasterKey(t, dir, 0600), + } +} + +// newGateway builds the S3 server serve runs, on top of an already-built +// cluster runtime. +func newGateway(t *testing.T, cfg *Config, rt *clusterrun.Runtime) *s3.Server { + t.Helper() + + server, err := s3.NewServer( + s3.WithConfigPath(cfg.ConfigPath), + s3.WithAddress(cfg.Host, cfg.Port), + s3.WithTLS(cfg.TlsCert, cfg.TlsKey), + s3.WithBasePath(cfg.BasePath), + s3.WithEncryptionKeyFile(cfg.EncryptionKeyFile), + s3.WithPreparedBackend(rt.Backend), + ) + require.NoError(t, err) + return server +} + +// requireGatewayAccepts waits for the S3 gateway to accept connections, which +// is also the point at which Start has published the cancel func Shutdown +// needs. +func requireGatewayAccepts(t *testing.T, cfg *Config) { + t.Helper() + + addr := net.JoinHostPort(cfg.Host, strconv.Itoa(cfg.Port)) + require.Eventually(t, func() bool { + conn, err := net.DialTimeout("tcp", addr, time.Second) + if err != nil { + return false + } + conn.Close() + return true + }, 30*time.Second, 100*time.Millisecond, "gateway never accepted connections on %s", addr) +} + +// TestBuildRuntimeErrors covers every way assembling the cluster runtime can +// fail before a single node is running. +func TestBuildRuntimeErrors(t *testing.T) { + testCases := []struct { + name string + config func(t *testing.T, dir string) *Config + wantErr string + }{ + { + name: "Unreadable Config", + config: func(t *testing.T, dir string) *Config { + return &Config{ + ConfigPath: filepath.Join(dir, "absent.toml"), + BasePath: dir, + EncryptionKeyFile: writeMasterKey(t, dir, 0600), + } + }, + wantErr: "read predastore config", + }, + { + name: "Host Owns No Nodes", + config: func(t *testing.T, dir string) *Config { + ids := nextNodeIDs(1) + return &Config{ + ConfigPath: writeTopology(t, dir, + []hostSpec{{id: 1, dataDir: filepath.Join(dir, "host-1")}}, + []nodeSpec{{id: ids[0], hostID: 1, role: roleStateReplica}}), + BasePath: dir, + EncryptionKeyFile: writeMasterKey(t, dir, 0600), + HostID: 9, + } + }, + wantErr: "predastore host 9 has no nodes in", + }, + { + name: "Missing Master Key", + config: func(t *testing.T, dir string) *Config { + ids := nextNodeIDs(1) + return &Config{ + ConfigPath: writeTopology(t, dir, + []hostSpec{{id: 1, dataDir: filepath.Join(dir, "host-1")}}, + []nodeSpec{{id: ids[0], hostID: 1, role: roleStateReplica}}), + BasePath: dir, + EncryptionKeyFile: filepath.Join(dir, "absent.key"), + } + }, + wantErr: "load predastore master key", + }, + { + name: "World Readable Master Key", + config: func(t *testing.T, dir string) *Config { + ids := nextNodeIDs(1) + return &Config{ + ConfigPath: writeTopology(t, dir, + []hostSpec{{id: 1, dataDir: filepath.Join(dir, "host-1")}}, + []nodeSpec{{id: ids[0], hostID: 1, role: roleStateReplica}}), + BasePath: dir, + EncryptionKeyFile: writeMasterKey(t, dir, 0644), + } + }, + wantErr: "load predastore master key", + }, + { + name: "Remote Nodes Without TLS Material", + config: func(t *testing.T, dir string) *Config { + ids := nextNodeIDs(2) + return &Config{ + ConfigPath: writeTopology(t, dir, + []hostSpec{ + {id: 1, dataDir: filepath.Join(dir, "host-1")}, + {id: 2, dataDir: filepath.Join(dir, "host-2")}, + }, + []nodeSpec{ + {id: ids[0], hostID: 1, role: roleShardStorage}, + {id: ids[1], hostID: 2, role: roleStateReplica}, + }), + BasePath: dir, + EncryptionKeyFile: writeMasterKey(t, dir, 0600), + HostID: 1, + // TlsCert and TlsKey deliberately left empty: reaching the + // other host needs the intra-cluster socket. + } + }, + wantErr: "build predastore cluster runtime", + }, + } + + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + svc, err := New(tc.config(t, t.TempDir())) + require.NoError(t, err) + + rt, err := svc.buildRuntime() + + require.Error(t, err) + assert.Nil(t, rt) + assert.Contains(t, err.Error(), tc.wantErr) + }) + } +} + +// TestBuildRuntimeRunsOnlyThisHostsNodes verifies host selection: a host runs +// the nodes pinned to it and leaves the rest to the processes that own them. +func TestBuildRuntimeRunsOnlyThisHostsNodes(t *testing.T) { + cfg, local, remote := newTwoHostConfig(t) + + svc, err := New(cfg) + require.NoError(t, err) + + rt, err := svc.buildRuntime() + require.NoError(t, err) + t.Cleanup(rt.Close) + + assert.NotNil(t, rt.Backend) + for _, id := range local { + assert.DirExists(t, filepath.Join(cfg.BasePath, "host-1", fmt.Sprintf("node-%d", id))) + } + assert.NoDirExists(t, filepath.Join(cfg.BasePath, "host-2", fmt.Sprintf("node-%d", remote))) +} + +// TestServeReportsGatewayFailures covers the two ways the S3 gateway can fail +// the service: refusing to start, and dying once started. +func TestServeReportsGatewayFailures(t *testing.T) { + testCases := []struct { + name string + tls func(cfg *Config) + wantErr string + }{ + { + name: "Missing TLS Material", + tls: func(cfg *Config) { cfg.TlsCert, cfg.TlsKey = "", "" }, + wantErr: "start predastore s3 gateway", + }, + { + name: "Unloadable Certificate", + tls: func(cfg *Config) { + cfg.TlsCert = filepath.Join(cfg.BasePath, "absent.crt") + cfg.TlsKey = filepath.Join(cfg.BasePath, "absent.key") + }, + wantErr: "predastore s3 gateway", + }, + } + + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + cfg, _, _ := newTwoHostConfig(t) + svc, err := New(cfg) + require.NoError(t, err) + + rt, err := svc.buildRuntime() + require.NoError(t, err) + t.Cleanup(rt.Close) + + // The runtime is built first: reaching the other host's node needs + // the certificate this case is about to take away. + tc.tls(cfg) + gateway := newGateway(t, cfg, rt) + + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + + err = svc.serve(ctx, gateway) + + require.Error(t, err) + assert.Contains(t, err.Error(), tc.wantErr) + }) + } +} + +// TestServeDrainsOnContextCancel verifies the gateway stops serving when the +// service context is cancelled, rather than outliving the cluster beneath it. +func TestServeDrainsOnContextCancel(t *testing.T) { + cfg, _, _ := newTwoHostConfig(t) + svc, err := New(cfg) + require.NoError(t, err) + + rt, err := svc.buildRuntime() + require.NoError(t, err) + t.Cleanup(rt.Close) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + gateway := newGateway(t, cfg, rt) + served := make(chan error, 1) + go func() { served <- svc.serve(ctx, gateway) }() + + requireGatewayAccepts(t, cfg) + cancel() + + select { + case err := <-served: + assert.NoError(t, err) + case <-time.After(30 * time.Second): + t.Fatal("serve did not return after the context was cancelled") + } +} + +// TestStartFailsWhenGatewayCannotServe verifies a fatal gateway error takes +// the whole service down: a cluster left running headless would look healthy +// while serving nothing. +func TestStartFailsWhenGatewayCannotServe(t *testing.T) { + cfg := newSingleHostConfig(t) + cfg.TlsCert = filepath.Join(cfg.BasePath, "absent.crt") + cfg.TlsKey = filepath.Join(cfg.BasePath, "absent.key") + + svc, err := New(cfg) + require.NoError(t, err) + + type result struct { + pid int + err error + } + done := make(chan result, 1) + go func() { + pid, err := svc.Start() + done <- result{pid: pid, err: err} + }() + + select { + case res := <-done: + require.Error(t, res.err) + assert.Zero(t, res.pid) + assert.Contains(t, res.err.Error(), "predastore s3 gateway") + case <-time.After(60 * time.Second): + t.Fatal("Start did not return after the gateway failed") + } +} + +// TestStartServesUntilShutdown covers the service lifecycle end to end: Start +// brings the cluster nodes and the gateway up together, and Shutdown drains +// both by cancelling the context they share. +func TestStartServesUntilShutdown(t *testing.T) { + cfg := newSingleHostConfig(t) + svc, err := New(cfg) + require.NoError(t, err) + + type result struct { + pid int + err error + } + done := make(chan result, 1) + go func() { + pid, err := svc.Start() + done <- result{pid: pid, err: err} + }() + + // Shutdown must not run before Start publishes its cancel func, or it + // falls through to the pid file and signals this test binary. + requireGatewayAccepts(t, cfg) + require.NoError(t, svc.Shutdown()) + + select { + case res := <-done: + require.NoError(t, res.err) + assert.Equal(t, os.Getpid(), res.pid) + case <-time.After(60 * time.Second): + t.Fatal("Start did not return after Shutdown") + } +} + +// TestShutdownWithoutRunningCluster verifies the fallback path: with no +// cluster in this process there is no context to cancel, so Shutdown goes +// through the pid file instead. +func TestShutdownWithoutRunningCluster(t *testing.T) { + svc, err := New(&Config{BasePath: t.TempDir()}) + require.NoError(t, err) + + // No pid file was ever written, so this reports rather than signals. + assert.Error(t, svc.Shutdown()) +} diff --git a/spinifex/testutil/vbscan/vbscan.go b/spinifex/testutil/vbscan/vbscan.go index fe404045d..7c778d37f 100644 --- a/spinifex/testutil/vbscan/vbscan.go +++ b/spinifex/testutil/vbscan/vbscan.go @@ -13,9 +13,9 @@ // mulga, the module that contains it, without a repo-level cycle. It is // deliberately its own leaf package, not part of the shared spinifex/testutil // package that ~80 test files import — see tests/fixtures/predastore's package -// comment for why that matters (goleak poisoning via a transitively-linked -// init-time goroutine). vbscan does not import predastore/quic/quicclient, -// directly or transitively, and must stay that way. +// comment for why that matters (goleak poisoning via transitively-linked +// long-lived goroutines). vbscan does not pull in a predastore cluster +// runtime, directly or transitively, and must stay that way. // // # Scope: no chunk-reference-integrity scan // diff --git a/tests/fixtures/predastore/predastore.go b/tests/fixtures/predastore/predastore.go index 314841b89..2a31d9457 100644 --- a/tests/fixtures/predastore/predastore.go +++ b/tests/fixtures/predastore/predastore.go @@ -2,16 +2,17 @@ // exercise an actual S3-compatible backend rather than a mock. It is // deliberately its own leaf package (not folded into spinifex/testutil, // which is imported by most of the module's test files for lightweight NATS -// helpers): this package transitively imports predastore/quic/quicclient, -// whose package-level DefaultPool starts a permanent cleanup goroutine at -// init time. Folding this file into the shared testutil package would pull -// that goroutine into every test binary that imports testutil, and trip up -// any unrelated goroutine-leak check (go.uber.org/goleak) running in that -// same binary. Only callers that actually need a real predastore should pay -// for this import. +// helpers): starting a daemon here builds a whole predastore cluster runtime +// — shard stores, Raft replicas and their transports — whose goroutines run +// for the life of the test binary. Folding this file into the shared testutil +// package would pull all of that into every test binary that imports +// testutil, and trip up any unrelated goroutine-leak check (go.uber.org/goleak) +// running in that same binary. Only callers that actually need a real +// predastore should pay for it. package predastore import ( + "context" "crypto/ecdsa" "crypto/elliptic" "crypto/rand" @@ -19,6 +20,7 @@ import ( "crypto/x509" "crypto/x509/pkix" "encoding/pem" + "log/slog" "math/big" "net" "net/http" @@ -33,14 +35,15 @@ import ( "github.com/aws/aws-sdk-go/aws/credentials" "github.com/aws/aws-sdk-go/aws/session" "github.com/aws/aws-sdk-go/service/s3" - "github.com/mulgadc/predastore/quic/quicclient" + "github.com/mulgadc/predastore/clusterrun" + "github.com/mulgadc/predastore/pkg/masterkey" predastoreserver "github.com/mulgadc/predastore/s3" "github.com/mulgadc/spinifex/tests/fixtures/scratch" ) // Fixed connection details for the shared predastore fixture daemon started -// by Start. NodeID -1 (dev mode) runs every QUIC shard node in-process, so -// these values never need to vary per caller or per test run. +// by Start. Every cluster node runs in-process, so these values never need to +// vary per caller or per test run. const ( Host = "127.0.0.1:18443" Region = "us-east-1" @@ -65,8 +68,8 @@ const ( ) // Fixture describes a running predastore daemon ready for real -// S3/viperblock clients: a reachable endpoint, TLS material already trusted -// process-wide, and its default buckets created. +// S3/viperblock clients: a reachable endpoint and its default buckets +// created. Its TLS cert is self-signed, so clients skip verification. type Fixture struct { Host string Region string @@ -128,21 +131,22 @@ func Start(t *testing.T) *Fixture { certPath := filepath.Join(certDir, "test.crt") keyPath := filepath.Join(certDir, "test.key") - caPool, err := generateCertificate(certPath, keyPath) - if err != nil { + // The cert serves the S3 HTTPS frontend only. Intra-cluster traffic never + // leaves the process here, so it needs no TLS material of its own. + if err := generateCertificate(certPath, keyPath); err != nil { t.Fatalf("predastore fixture: generate certificate: %v", err) } - // Inject the ephemeral cert for the QUIC client (s3d -> shard nodes). - quicclient.SetDefaultRootCAs(caPool) - - // SSL_CERT_FILE injects the cert for the s3db REST client's OS trust - // store (sync.Once-cached there, so it must be set before that client's - // first dial). Scoped to this call via t.Setenv, but since the daemon and - // its first client dial both happen inside this same call, the cache is - // warm before this function returns — later tests in the process don't - // need the env var set again. + // SSL_CERT_FILE injects the cert into the OS trust store for clients that + // verify it rather than skipping verification (objectstore's S3 client, + // for one). t.Setenv reverts when this test ends, while the daemon lives + // on for the whole binary, so the pool is loaded here and now: crypto/x509 + // caches it behind a sync.Once, and this call is what fixes our cert in it + // for every later test in the process. t.Setenv("SSL_CERT_FILE", certPath) + if _, err := x509.SystemCertPool(); err != nil { + t.Fatalf("predastore fixture: load system cert pool: %v", err) + } // Predastore mandates a 32-byte master key at mode 0600 (rejected // otherwise by internal/keyfile.Load). @@ -155,12 +159,12 @@ func Start(t *testing.T) *Fixture { t.Fatalf("predastore fixture: write encryption key: %v", err) } - // Five nodes trigger dev-mode: all QUIC shards start as local goroutines. + // One host carrying every node, so the whole cluster runs in this process + // over the in-process pipe: no sockets to bind, no intra-cluster certs. + // RS(3,2) needs 5 shard-storage nodes; 3 state replicas form the quorum. configPath := filepath.Join(testDir, "predastore_test.toml") configContent := `version = "1.0" region = "us-east-1" -host = "127.0.0.1" -port = 18443 debug = false disable_logging = false base_path = "` + testDir + `/" @@ -169,35 +173,51 @@ base_path = "` + testDir + `/" data = 3 parity = 2 -[[nodes]] +[[host]] id = 1 -host = "127.0.0.1" -port = 19991 -path = "store/node-1/" +bind_addr = "127.0.0.1:16660" +public_addr = "127.0.0.1:16660" +data_dir = "` + filepath.Join(testDir, "cluster") + `" -[[nodes]] +[[node]] +id = 1 +host_id = 1 +role = "shard-storage" + +[[node]] id = 2 -host = "127.0.0.1" -port = 19992 -path = "store/node-2/" +host_id = 1 +role = "shard-storage" -[[nodes]] +[[node]] id = 3 -host = "127.0.0.1" -port = 19993 -path = "store/node-3/" +host_id = 1 +role = "shard-storage" -[[nodes]] +[[node]] id = 4 -host = "127.0.0.1" -port = 19994 -path = "store/node-4/" +host_id = 1 +role = "shard-storage" -[[nodes]] +[[node]] id = 5 -host = "127.0.0.1" -port = 19995 -path = "store/node-5/" +host_id = 1 +role = "shard-storage" + +[[node]] +id = 6 +host_id = 1 +role = "state-replica" + +[[node]] +id = 7 +host_id = 1 +role = "state-replica" + +[[node]] +id = 8 +host_id = 1 +role = "state-replica" [[auth]] access_key_id = "` + AccessKey + `" @@ -209,6 +229,22 @@ account_id = "123456789012" t.Fatalf("predastore fixture: write config: %v", err) } + // The S3 gateway runs on top of a backend the caller wires: the cluster + // runtime owns the shard stores and the Raft replicas, and this process + // runs all of them. + cfg := &predastoreserver.Config{ConfigPath: configPath, BasePath: testDir} + if err := cfg.ReadConfig(); err != nil { + t.Fatalf("predastore fixture: read config: %v", err) + } + key, err := masterkey.Load(encryptionKeyPath) + if err != nil { + t.Fatalf("predastore fixture: load master key: %v", err) + } + rt, err := clusterrun.Build(cfg, clusterrun.AllNodeIDs(cfg), certPath, keyPath, key) + if err != nil { + t.Fatalf("predastore fixture: build cluster runtime: %v", err) + } + // Built directly against predastore/s3 rather than spinifex's own // services/predastore wrapper: that wrapper pulls in spinifex/utils, // and spinifex/utils' own tests import spinifex/testutil, which would be @@ -222,13 +258,29 @@ account_id = "123456789012" predastoreserver.WithTLS(certPath, keyPath), predastoreserver.WithBasePath(testDir), predastoreserver.WithDebug(false), - predastoreserver.WithNodeID(-1), predastoreserver.WithPprof(false, ""), predastoreserver.WithEncryptionKeyFile(encryptionKeyPath), + predastoreserver.WithPreparedBackend(rt.Backend), ) if err != nil { t.Fatalf("predastore fixture: create server: %v", err) } + + // The runtime is as permanent as the daemon it backs: an uncancellable + // context, left running until the test process exits. A t.Fatalf below + // takes the binary down with it, so nothing is left orphaned. + go func() { + if err := rt.Run(context.Background()); err != nil { + slog.Error("predastore fixture: cluster runtime exited", "error", err) + } + }() + + // Writes need a committed leader; serving before one exists would fail + // the bucket creation below for no reason other than timing. + if err := rt.WaitReady(30 * time.Second); err != nil { + t.Fatalf("predastore fixture: no leader elected: %v", err) + } + if err := server.ListenAndServeAsync(); err != nil { t.Fatalf("predastore fixture: start server: %v", err) } @@ -259,14 +311,13 @@ account_id = "123456789012" return fixture } -// generateCertificate generates a self-signed TLS certificate for the -// fixture daemon and returns a CertPool containing it, for injection into -// whichever client trust store a caller needs (see quicclient.SetDefaultRootCAs -// and the SSL_CERT_FILE handling in Start). -func generateCertificate(certPath, keyPath string) (*x509.CertPool, error) { +// generateCertificate writes a self-signed TLS certificate and key for the +// fixture daemon's S3 HTTPS frontend. Clients reach it with +// InsecureSkipVerify, so nothing needs the cert in a trust store. +func generateCertificate(certPath, keyPath string) error { priv, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader) if err != nil { - return nil, err + return err } notBefore := time.Now() @@ -274,7 +325,7 @@ func generateCertificate(certPath, keyPath string) (*x509.CertPool, error) { serialNumber, err := rand.Int(rand.Reader, new(big.Int).Lsh(big.NewInt(1), 128)) if err != nil { - return nil, err + return err } template := x509.Certificate{ @@ -294,40 +345,34 @@ func generateCertificate(certPath, keyPath string) (*x509.CertPool, error) { derBytes, err := x509.CreateCertificate(rand.Reader, &template, &template, &priv.PublicKey, priv) if err != nil { - return nil, err + return err } certOut, err := os.Create(certPath) if err != nil { - return nil, err + return err } defer certOut.Close() if err := pem.Encode(certOut, &pem.Block{Type: "CERTIFICATE", Bytes: derBytes}); err != nil { - return nil, err + return err } keyOut, err := os.Create(keyPath) if err != nil { - return nil, err + return err } defer keyOut.Close() privBytes, err := x509.MarshalECPrivateKey(priv) if err != nil { - return nil, err + return err } if err := pem.Encode(keyOut, &pem.Block{Type: "EC PRIVATE KEY", Bytes: privBytes}); err != nil { - return nil, err + return err } - parsed, err := x509.ParseCertificate(derBytes) - if err != nil { - return nil, err - } - pool := x509.NewCertPool() - pool.AddCert(parsed) - return pool, nil + return nil } // waitForReady polls the fixture daemon's HTTPS endpoint until it accepts diff --git a/tests/integration/predastore_test.go b/tests/integration/predastore_test.go index 74e4a07c7..14353df74 100644 --- a/tests/integration/predastore_test.go +++ b/tests/integration/predastore_test.go @@ -83,10 +83,9 @@ func TestPredastoreObjectLifecycle(t *testing.T) { } // predastoreS3Client builds an aws-sdk-go v1 S3 client against the shared -// predastore fixture daemon. TLS verification is skipped: the fixture's -// self-signed certificate is trusted process-wide for quicclient/the s3db -// REST client (see testpredastore.Start), but a plain s3.S3 built here has -// its own independent HTTP transport. +// predastore fixture daemon. TLS verification is skipped: the fixture serves +// the S3 API under a self-signed certificate it generates for itself (see +// testpredastore.Start), which no trust store knows about. func predastoreS3Client(t *testing.T, fixture *testpredastore.Fixture) *s3.S3 { t.Helper() httpClient := &http.Client{