From db0d46ca5b59d765d337f5a0b3f18fc4bab04df7 Mon Sep 17 00:00:00 2001 From: Ben Duncan <5178243+benduncan@users.noreply.github.com> Date: Fri, 13 Feb 2026 20:31:45 +1000 Subject: [PATCH] Improved multi-server node testing and tweaks --- .gitignore | 5 + INSTALL.md | 376 +++++++--- cmd/hive/cmd/admin.go | 469 ++++++++---- cmd/hive/cmd/service.go | 1 + cmd/hive/cmd/templates/nats.conf | 2 +- docs/development/improvements/multi-server.md | 675 ++++++++++++++++++ hive/admin/admin.go | 13 +- hive/admin/admin_test.go | 27 +- hive/daemon/daemon.go | 13 +- hive/daemon/daemon_handlers.go | 2 +- hive/daemon/daemon_test.go | 20 +- hive/formation/formation.go | 253 +++++++ hive/formation/formation_test.go | 313 ++++++++ hive/formation/helpers.go | 52 ++ hive/services/viperblockd/viperblockd.go | 41 +- .../viperblockd/viperblockd_handlers_test.go | 2 +- .../viperblockd_integration_test.go | 13 +- scripts/create-multi-node.sh | 142 ++-- scripts/openvswitch.sh | 56 ++ scripts/stop-multi-node.sh | 25 +- 20 files changed, 2185 insertions(+), 315 deletions(-) create mode 100644 docs/development/improvements/multi-server.md create mode 100644 hive/formation/formation.go create mode 100644 hive/formation/formation_test.go create mode 100644 hive/formation/helpers.go create mode 100644 scripts/openvswitch.sh diff --git a/.gitignore b/.gitignore index 261af9103..bf5d475e1 100644 --- a/.gitignore +++ b/.gitignore @@ -28,3 +28,8 @@ Thumbs.db # Test reports *-report.txt + +# Agent direction (internal use only) +docs/INTERNAL.md +docs/PRIVATE.md +docs/DEVNOTES.md diff --git a/INSTALL.md b/INSTALL.md index e1d31efe9..1310aaf22 100644 --- a/INSTALL.md +++ b/INSTALL.md @@ -9,6 +9,7 @@ The currently supported operating systems that are validated and tested include: - Ubuntu 22.04 - Ubuntu 24.04 - Ubuntu 25.10 +- Debian 12.13 For the development preview, please use one of the supported versions above. @@ -551,133 +552,330 @@ The Hive Platform also offers a web interface for managing hive services. Simply **Note:** If you are not viewing the website on the same machine that is running the hive server, you will need to go to [https://localhost:9999](https://localhost:9999) and [https://localhost:8443](https://localhost:8443) and accept the certificates. If you have added the CA to your local machine this is not needed. -# Multi-node Configuration +# Multi-Node Configuration -To create a simulated multi-node installation on a single server (e.g node1, node2, node3) review the script `./scripts/create-multi-node.sh`. +Hive is designed for distributed, multi-server deployments. A Hive cluster operates as a fully distributed infrastructure region — similar to an AWS Region — where multiple nodes work together to provide high availability, data durability, and fault tolerance across the platform. -Specifically this will create 3 IPs on a specified ethernet adapter. +When deploying a multi-node cluster, you define a **region** (e.g., `ap-southeast-2`, `us-east-1`) and **availability zones** to organize your infrastructure. Services are distributed across nodes: NATS provides a clustered message bus for request routing and JetStream replication, Predastore uses Raft consensus with Reed-Solomon erasure coding (RS 2+1) for durable S3-compatible object storage, and the Hive daemon on each node can independently serve AWS API requests. The result is a platform where compute, storage, and API services are replicated — an AMI stored on one node is available from any node, an EC2 instance launched via any gateway is visible cluster-wide, and the loss of a single node does not compromise data or availability. + +Cluster formation is automatic. When initializing with `--nodes 3`, the init node starts a formation server and waits for other nodes to join. Once all nodes have registered, each node receives the full cluster topology — credentials, CA certificates, NATS routes, Predastore peer lists — and generates its own configuration files. No manual configuration synchronization is required. + +## Simulated Multi-Node (Single Server) + +For development and testing, a 3-node cluster can be simulated on a single machine using loopback addresses. Linux handles the full `127.0.0.0/8` range natively, so `127.0.0.1`, `127.0.0.2`, and `127.0.0.3` all work without any network configuration. + +Each node uses a separate data directory (`~/node1/`, `~/node2/`, `~/node3/`) to isolate configuration, logs, and state. + +### 1. Build ```bash -ip addr add 10.11.12.1/24 dev eth0 -ip addr add 10.11.12.2/24 dev eth0 -ip addr add 10.11.12.3/24 dev eth0 +make build ``` -### Initialize the Leader Node +### 2. Form the Cluster -Next, `hive` will be installed and initiated in `$HOME/node1` for configuration files. +The formation process requires running init and join commands concurrently. The init node blocks until all expected nodes have joined. + +**Terminal 1 — Initialize the leader node:** ```bash ./bin/hive admin init \ ---node node1 \ ---bind 10.11.12.1 \ ---cluster-bind 10.11.12.1 \ ---cluster-routes 10.11.12.1:4248 \ ---predastore-nodes 10.11.12.1,10.11.12.2,10.11.12.3 \ ---port 4432 \ ---hive-dir ~/node1/ \ ---config-dir ~/node1/config/ \ ---region ap-southeast-2 \ ---az ap-southeast-2a \ - -./scripts/start-dev.sh ~/node1/ + --node node1 \ + --nodes 3 \ + --bind 127.0.0.1 \ + --cluster-bind 127.0.0.1 \ + --port 4432 \ + --hive-dir ~/node1/ \ + --config-dir ~/node1/config/ \ + --region ap-southeast-2 \ + --az ap-southeast-2a ``` -### Join Follower Nodes +This starts a formation server on `127.0.0.1:4432` and waits for 2 more nodes to join. -Two other nodes will be created, joining node1 to exchange configuration files and identify as a new node. +**Terminal 2 — Join node 2 (while init is running):** ```bash ./bin/hive admin join \ ---node node2 \ ---bind 10.11.12.2 \ ---cluster-bind 10.11.12.2 \ ---cluster-routes 10.11.12.1:4248 \ ---host 10.11.12.1:4432 \ ---data-dir ~/node2/ \ ---config-dir ~/node2/config/ \ ---region ap-southeast-2 \ ---az ap-southeast-2a \ - -./scripts/start-dev.sh ~/node2/ + --node node2 \ + --bind 127.0.0.2 \ + --cluster-bind 127.0.0.2 \ + --host 127.0.0.1:4432 \ + --data-dir ~/node2/ \ + --config-dir ~/node2/config/ \ + --region ap-southeast-2 \ + --az ap-southeast-2a ``` +**Terminal 3 — Join node 3 (while init is running):** + ```bash ./bin/hive admin join \ ---node node3 \ ---bind 10.11.12.3 \ ---cluster-bind 10.11.12.3 \ ---cluster-routes 10.11.12.1:4248 \ ---host 10.11.12.1:4432 \ ---data-dir ~/node3/ \ ---config-dir ~/node3/config/ \ ---region ap-southeast-2 \ ---az ap-southeast-2a \ + --node node3 \ + --bind 127.0.0.3 \ + --cluster-bind 127.0.0.3 \ + --host 127.0.0.1:4432 \ + --data-dir ~/node3/ \ + --config-dir ~/node3/config/ \ + --region ap-southeast-2 \ + --az ap-southeast-2a +``` -./scripts/start-dev.sh ~/node3/ +All three processes will exit with a cluster summary once formation is complete. + +### 3. Start Services + +Start services on each node (in separate terminals or background): + +```bash +HIVE_SKIP_BUILD=true UI=false ./scripts/start-dev.sh ~/node1/ +HIVE_SKIP_BUILD=true UI=false ./scripts/start-dev.sh ~/node2/ +HIVE_SKIP_BUILD=true UI=false ./scripts/start-dev.sh ~/node3/ +``` + +### 4. Verify the Cluster + +Check that all daemons are healthy: + +```bash +curl -s http://127.0.0.1:4432/health +curl -s http://127.0.0.2:4432/health +curl -s http://127.0.0.3:4432/health +``` + +Check NATS cluster routing (from any node): + +```bash +grep -i "route\|cluster" ~/node1/logs/nats.log +``` + +Check Predastore Raft consensus: + +```bash +grep -i "leader\|election" ~/node1/logs/predastore.log +``` + +### 5. Using the Cluster + +When using the AWS CLI, specify the endpoint URL pointing to any node's gateway: + +```bash +AWS_ENDPOINT_URL=https://127.0.0.1:9999 aws ec2 describe-instances +AWS_ENDPOINT_URL=https://127.0.0.2:9999 aws ec2 describe-instances +``` + +Each node has its own logs for debugging: + +```bash +tail -f ~/node1/logs/hive.log +tail -f ~/node2/logs/hive.log +tail -f ~/node3/logs/hive.log +``` + +### 6. Stop the Cluster + +Stop services on each node in reverse order: + +```bash +./scripts/stop-dev.sh ~/node3/ +./scripts/stop-dev.sh ~/node2/ +./scripts/stop-dev.sh ~/node1/ +``` + +### 7. Clean Up + +To fully remove the simulated cluster: + +```bash +rm -rf ~/node1/ ~/node2/ ~/node3/ +``` + +### Alternative: IP Aliases + +For a more production-like simulation using dedicated IP addresses on a network interface: + +```bash +sudo ip addr add 10.11.12.1/24 dev eth0 +sudo ip addr add 10.11.12.2/24 dev eth0 +sudo ip addr add 10.11.12.3/24 dev eth0 +``` + +Replace `eth0` with your network interface name (e.g., `enp0s3`, `ens33`). Then use these IPs instead of `127.0.0.x` in the formation commands above. + +To remove the aliases: + +```bash +sudo ip addr del 10.11.12.1/24 dev eth0 +sudo ip addr del 10.11.12.2/24 dev eth0 +sudo ip addr del 10.11.12.3/24 dev eth0 +``` + +## Real Multi-Node (Physical Servers) + +For production or production-like deployments across multiple physical servers or VMs. + +### Prerequisites + +On **each server**: + +1. Install Hive dependencies (see [Dependencies](#dependencies) above) +2. Clone the repository and build: + +```bash +mkdir -p ~/Development/mulga/ +cd ~/Development/mulga/ +git clone https://github.com/mulgadc/hive.git +cd hive +./scripts/clone-deps.sh +make build +``` + +3. Identify the bind IP for each server (e.g., the primary network interface IP): + +```bash +ip addr show | grep "inet " +``` + +### Example: 3-Node Cluster + +Assume three servers with these IPs: + +| Node | IP Address | Role | +|------|------------|------| +| node1 | 192.168.1.10 | Init node | +| node2 | 192.168.1.11 | Join node | +| node3 | 192.168.1.12 | Join node | + +**Server 1 — Initialize the cluster:** + +```bash +cd ~/Development/mulga/hive + +./bin/hive admin init \ + --node node1 \ + --nodes 3 \ + --bind 192.168.1.10 \ + --cluster-bind 192.168.1.10 \ + --port 4432 \ + --hive-dir ~/hive/ \ + --config-dir ~/hive/config/ \ + --region us-east-1 \ + --az us-east-1a +``` + +**Server 2 — Join the cluster (while init is running):** + +```bash +cd ~/Development/mulga/hive + +./bin/hive admin join \ + --node node2 \ + --bind 192.168.1.11 \ + --cluster-bind 192.168.1.11 \ + --host 192.168.1.10:4432 \ + --data-dir ~/hive/ \ + --config-dir ~/hive/config/ \ + --region us-east-1 \ + --az us-east-1a +``` + +**Server 3 — Join the cluster (while init is running):** + +```bash +cd ~/Development/mulga/hive + +./bin/hive admin join \ + --node node3 \ + --bind 192.168.1.12 \ + --cluster-bind 192.168.1.12 \ + --host 192.168.1.10:4432 \ + --data-dir ~/hive/ \ + --config-dir ~/hive/config/ \ + --region us-east-1 \ + --az us-east-1a +``` + +All three processes will exit with a cluster summary once formation is complete. + +### Service Co-location + +Every node that runs EC2 instances (compute) **must** also run the viperblock service. EBS volumes are mounted locally via Unix domain sockets (NBD) — the daemon publishes to a node-specific NATS topic (`ebs.{nodeName}.mount`) which is handled by the viperblock instance on the same server. Nodes without viperblock cannot run EC2 instances. + +In the default deployment, every node runs all services (NATS, Predastore, Viperblock, Daemon, Gateway), which satisfies this requirement. + +### Start Services on Each Server + +After formation completes, start services on **all servers**: + +```bash +./scripts/start-dev.sh ``` ### Verify the Cluster -Each node will have its own storage for config, data and state in `~/node[1,2,3]` to simulate a unique node and network. - -Once the multi-node cluster is configured, follow the original installation instructions above to: - -- Import an SSH key -- Import an AMI -- Launch instances - -Note - when using the AWS CLI tool you must specify the IP address of a gateway node (10.11.12.1, 10.11.12.2, or 10.11.12.3). +From any server, check daemon health: ```bash -aws --endpoint-url https://10.11.12.3:9999 ec2 describe-instances --instance-ids i-9f5f648adc57ea46d +curl -s http://192.168.1.10:4432/health +curl -s http://192.168.1.11:4432/health +curl -s http://192.168.1.12:4432/health ``` -```json -{ - "Reservations": [ - { - "ReservationId": "r-9d26866c3b0d53bf4", - "OwnerId": "123456789012", - "Instances": [ - { - "InstanceId": "i-9f5f648adc57ea46d", - "ImageId": "ami-d0de73dd18fa33ac9", - "State": { - "Code": 16, - "Name": "running" - }, - "KeyName": "hive-key", - "InstanceType": "t3.micro", - "LaunchTime": "2025-11-21T10:36:42.834000+00:00" - } - ] - } - ] -} -``` - -For debugging, reference the node unique configuration files, e.g +Check NATS cluster mesh: ```bash -tail -n 100 ~/node3/logs/hive.log +grep -i "route\|cluster" ~/hive/logs/nats.log ``` +Check Predastore Raft: + ```bash -... -2025/11/21 22:59:32 Received message on subject: ec2.DescribeInstances -2025/11/21 22:59:32 Message data: {"DryRun":null,"Filters":null,"InstanceIds":["i-9f5f648adc57ea46d"],"MaxResults":null,"NextToken":null} -2025/11/21 22:59:32 INFO Processing DescribeInstances request from this node -2025/11/21 22:59:32 INFO handleEC2DescribeInstances completed count=1 -... +grep -i "leader\|election" ~/hive/logs/predastore.log ``` -### Stop multi-node Instance +### Trust CA Certificate -To stop a simulated multi-node instance: +The CA certificate is generated by the init node and distributed to all joining nodes during formation. On **each server**, trust the CA: ```bash -./scripts/stop-multi-node.sh +sudo cp ~/hive/config/ca.pem /usr/local/share/ca-certificates/hive-ca.crt +sudo update-ca-certificates ``` -Note if multiple EC2 instances are running, it may take a few minutes to gracefully terminate the instance, unmount the attached EBS volume (via NBD) and push the write-ahead-log (WAL) to the S3 server (predastore). +### Using the Cluster + +Connect to any node's AWS Gateway: + +```bash +AWS_ENDPOINT_URL=https://192.168.1.10:9999 AWS_PROFILE=hive aws ec2 describe-instances +``` + +### Dual NIC Configuration + +If servers have separate management and cluster network interfaces, use `--cluster-bind` to specify the cluster network IP: + +```bash +# Server 1: management=192.168.1.10, cluster=10.0.0.10 +./bin/hive admin init \ + --node node1 \ + --nodes 3 \ + --bind 192.168.1.10 \ + --cluster-bind 10.0.0.10 \ + --port 4432 \ + --hive-dir ~/hive/ \ + --config-dir ~/hive/config/ \ + --region us-east-1 \ + --az us-east-1a +``` + +The `--bind` IP is used for the formation server, daemon, and AWS gateway. The `--cluster-bind` IP is used for NATS cluster routing and Predastore Raft consensus. If `--cluster-bind` is not specified, it defaults to the `--bind` IP. + +### Shutdown + +Stop services on each server: + +```bash +./scripts/stop-dev.sh +``` + +Note: if EC2 instances are running, the stop process will gracefully terminate them, unmount attached EBS volumes (via NBD), and flush the write-ahead-log (WAL) to the S3 server (Predastore). This may take a few minutes depending on instance volume sizes. diff --git a/cmd/hive/cmd/admin.go b/cmd/hive/cmd/admin.go index 36033b65a..a58aea24d 100644 --- a/cmd/hive/cmd/admin.go +++ b/cmd/hive/cmd/admin.go @@ -17,7 +17,7 @@ package cmd import ( "bytes" - "crypto/tls" + "context" _ "embed" "encoding/json" "fmt" @@ -33,14 +33,13 @@ import ( "github.com/mulgadc/hive/hive/admin" "github.com/mulgadc/hive/hive/config" + "github.com/mulgadc/hive/hive/formation" "github.com/mulgadc/hive/hive/utils" "github.com/mulgadc/viperblock/viperblock" "github.com/mulgadc/viperblock/viperblock/backends/s3" "github.com/mulgadc/viperblock/viperblock/v_utils" - "github.com/pelletier/go-toml/v2" "github.com/pterm/pterm" "github.com/spf13/cobra" - "golang.org/x/net/http2" ) //go:embed templates/hive.toml @@ -161,6 +160,8 @@ func init() { adminInitCmd.Flags().String("cluster-bind", "", "IP address to bind NATS cluster services to (e.g., 10.11.12.1 for multi-node)") adminInitCmd.Flags().String("cluster-routes", "", "NATS cluster hosts for routing specify multiple with comma (e.g., 10.11.12.1:4248,10.11.12.2:4248 for multi-node)") adminInitCmd.Flags().String("predastore-nodes", "", "Comma-separated IPs for multi-node Predastore cluster (e.g., 10.11.12.1,10.11.12.2,10.11.12.3). Requires >= 3 nodes.") + adminInitCmd.Flags().String("formation-timeout", "10m", "Timeout for cluster formation (e.g., 5m, 30s)") + adminInitCmd.Flags().String("cluster-name", "hive", "NATS cluster name") // Flags for admin join adminJoinCmd.Flags().String("region", "ap-southeast-2", "Region for this node") @@ -473,12 +474,15 @@ func runAdminInit(cmd *cobra.Command, args []string) { region, _ := cmd.Flags().GetString("region") az, _ := cmd.Flags().GetString("az") node, _ := cmd.Flags().GetString("node") + nodes, _ := cmd.Flags().GetInt("nodes") port, _ := cmd.Flags().GetInt("port") bindIP, _ := cmd.Flags().GetString("bind") clusterBind, _ := cmd.Flags().GetString("cluster-bind") clusterRoutesStr, _ := cmd.Flags().GetString("cluster-routes") clusterRoutes := strings.Split(clusterRoutesStr, ",") predastoreNodesStr, _ := cmd.Flags().GetString("predastore-nodes") + formationTimeoutStr, _ := cmd.Flags().GetString("formation-timeout") + clusterName, _ := cmd.Flags().GetString("cluster-name") // Validate IP address format if net.ParseIP(bindIP) == nil { @@ -492,6 +496,11 @@ func runAdminInit(cmd *cobra.Command, args []string) { os.Exit(1) } + // Default cluster-bind to bind IP if not specified + if clusterBind == "" { + clusterBind = bindIP + } + fmt.Printf("Initializing Hive with bind IP: %s, port: %d\n", bindIP, port) // Default config directory @@ -545,9 +554,20 @@ func runAdminInit(cmd *cobra.Command, args []string) { // Get home directory for data path homeDir, _ := os.UserHomeDir() hiveRoot = filepath.Join(homeDir, "hive") - } + // Determine if this is a multi-node formation + isMultiNode := nodes >= 2 && bindIP != "0.0.0.0" + + if isMultiNode { + runAdminInitMultiNode(cmd, accessKey, secretKey, accountID, natsToken, clusterName, + configDir, hiveRoot, certPath, region, az, node, bindIP, clusterBind, + port, nodes, formationTimeoutStr) + return + } + + // --- Single-node path (existing behavior) --- + // Create config files from embedded templates fmt.Println("\n📝 Creating configuration files...") @@ -566,7 +586,7 @@ func runAdminInit(cmd *cobra.Command, args []string) { portStr := fmt.Sprintf("%d", port) - // Parse multi-node predastore configuration + // Parse multi-node predastore configuration (legacy flag-based approach for single-node) var predastoreNodeID int if predastoreNodesStr != "" { ips := strings.Split(predastoreNodesStr, ",") @@ -623,6 +643,7 @@ func runAdminInit(cmd *cobra.Command, args []string) { BindIP: bindIP, ClusterBindIP: clusterBind, ClusterRoutes: clusterRoutes, + ClusterName: clusterName, PredastoreNodeID: predastoreNodeID, } @@ -648,7 +669,7 @@ func runAdminInit(cmd *cobra.Command, args []string) { // Update ~/.aws/credentials and ~/.aws/config fmt.Println("\n🔧 Configuring AWS credentials...") - if err := admin.SetupAWSCredentials(accessKey, secretKey, region, certPath); err != nil { + if err := admin.SetupAWSCredentials(accessKey, secretKey, region, certPath, bindIP); err != nil { fmt.Fprintf(os.Stderr, "Warning: Could not update AWS credentials: %v\n", err) } else { fmt.Println("✅ AWS credentials configured") @@ -672,6 +693,187 @@ func runAdminInit(cmd *cobra.Command, args []string) { fmt.Println() } +// runAdminInitMultiNode handles the multi-node formation path for admin init. +// It starts a formation server, registers this node, waits for all nodes to join, +// then generates configs with complete cluster topology. +func runAdminInitMultiNode(cmd *cobra.Command, accessKey, secretKey, accountID, natsToken, clusterName, + configDir, hiveRoot, certPath, region, az, node, bindIP, clusterBind string, + port, expectedNodes int, formationTimeoutStr string) { + + formationTimeout, err := time.ParseDuration(formationTimeoutStr) + if err != nil { + fmt.Fprintf(os.Stderr, "❌ Error: Invalid --formation-timeout: %v\n", err) + os.Exit(1) + } + + // Read CA cert/key for distribution to joining nodes + caCertData, err := os.ReadFile(filepath.Join(configDir, "ca.pem")) + if err != nil { + fmt.Fprintf(os.Stderr, "❌ Error reading CA cert: %v\n", err) + os.Exit(1) + } + caKeyData, err := os.ReadFile(filepath.Join(configDir, "ca.key")) + if err != nil { + fmt.Fprintf(os.Stderr, "❌ Error reading CA key: %v\n", err) + os.Exit(1) + } + + creds := &formation.SharedCredentials{ + AccessKey: accessKey, + SecretKey: secretKey, + AccountID: accountID, + NatsToken: natsToken, + ClusterName: clusterName, + Region: region, + } + + fs := formation.NewFormationServer(expectedNodes, creds, string(caCertData), string(caKeyData)) + + // Register self (init node) as the first node + selfNode := formation.NodeInfo{ + Name: node, + BindIP: bindIP, + ClusterIP: clusterBind, + Region: region, + AZ: az, + Port: port, + } + if err := fs.RegisterNode(selfNode); err != nil { + fmt.Fprintf(os.Stderr, "❌ Error registering self: %v\n", err) + os.Exit(1) + } + + // Start formation server + formationAddr := fmt.Sprintf("%s:%d", bindIP, port) + if err := fs.Start(formationAddr); err != nil { + fmt.Fprintf(os.Stderr, "❌ Error starting formation server: %v\n", err) + os.Exit(1) + } + + fmt.Printf("\n📡 Formation server started on %s\n", formationAddr) + fmt.Printf(" Waiting for %d more node(s) to join...\n", expectedNodes-1) + fmt.Printf(" Other nodes should run: hive admin join --host %s --node --bind \n\n", formationAddr) + + // Wait for all nodes to register + if err := fs.WaitForCompletion(formationTimeout); err != nil { + fmt.Fprintf(os.Stderr, "❌ %v\n", err) + fs.Shutdown(context.Background()) + os.Exit(1) + } + + fmt.Printf("✅ All %d nodes joined!\n", expectedNodes) + + // Build cluster topology from formation data + allNodes := fs.Nodes() + clusterRoutes := formation.BuildClusterRoutes(allNodes) + predastoreNodes := formation.BuildPredastoreNodes(allNodes) + + fmt.Println("\n📝 Creating configuration files...") + + // Create subdirectories + awsgwDir := filepath.Join(configDir, "awsgw") + predastoreDir := filepath.Join(configDir, "predastore") + natsDir := filepath.Join(configDir, "nats") + hiveDir := filepath.Join(configDir, "hive") + + for _, dir := range []string{awsgwDir, predastoreDir, natsDir, hiveDir} { + if err := os.MkdirAll(dir, 0700); err != nil { + fmt.Fprintf(os.Stderr, "Error creating directory %s: %v\n", dir, err) + os.Exit(1) + } + } + + portStr := fmt.Sprintf("%d", port) + + // Generate multi-node predastore config + var predastoreNodeID int + if len(predastoreNodes) >= 3 { + predastoreContent, err := admin.GenerateMultiNodePredastoreConfig(predastoreMultiNodeTemplate, predastoreNodes, accessKey, secretKey, region) + if err != nil { + fmt.Fprintf(os.Stderr, "Error generating multi-node predastore config: %v\n", err) + os.Exit(1) + } + + predastorePath := filepath.Join(predastoreDir, "predastore.toml") + if err := os.WriteFile(predastorePath, []byte(predastoreContent), 0600); err != nil { + fmt.Fprintf(os.Stderr, "Error writing predastore config: %v\n", err) + os.Exit(1) + } + + predastoreNodeID = admin.FindNodeIDByIP(predastoreNodes, bindIP) + fmt.Printf("✅ Created: multi-node predastore.toml (node ID: %d)\n", predastoreNodeID) + } + + hiveTomlPath := filepath.Join(configDir, "hive.toml") + + configSettings := admin.ConfigSettings{ + AccessKey: accessKey, + SecretKey: secretKey, + AccountID: accountID, + Region: region, + NatsToken: natsToken, + DataDir: hiveRoot, + + Node: node, + Az: az, + Port: portStr, + BindIP: bindIP, + ClusterBindIP: clusterBind, + ClusterRoutes: clusterRoutes, + ClusterName: clusterName, + + PredastoreNodeID: predastoreNodeID, + } + + // Generate config files + configs := []admin.ConfigFile{ + {Name: "hive.toml", Path: hiveTomlPath, Template: hiveTomlTemplate}, + {Name: filepath.Join(awsgwDir, "awsgw.toml"), Path: filepath.Join(awsgwDir, "awsgw.toml"), Template: awsgwTomlTemplate}, + {Name: filepath.Join(natsDir, "nats.conf"), Path: filepath.Join(natsDir, "nats.conf"), Template: natsConfTemplate}, + } + // Skip template-based predastore.toml if multi-node was generated + if len(predastoreNodes) < 3 { + configs = append(configs, admin.ConfigFile{ + Name: filepath.Join(predastoreDir, "predastore.toml"), Path: filepath.Join(predastoreDir, "predastore.toml"), Template: predastoreTomlTemplate, + }) + } + + if err := admin.GenerateConfigFiles(configs, configSettings); err != nil { + fmt.Fprintf(os.Stderr, "Error generating configuration files: %v\n", err) + os.Exit(1) + } + + // Update ~/.aws/credentials and ~/.aws/config + fmt.Println("\n🔧 Configuring AWS credentials...") + if err := admin.SetupAWSCredentials(accessKey, secretKey, region, certPath, bindIP); err != nil { + fmt.Fprintf(os.Stderr, "Warning: Could not update AWS credentials: %v\n", err) + } else { + fmt.Println("✅ AWS credentials configured") + } + + admin.CreateServiceDirectories(hiveRoot) + + // Keep formation server running briefly so joining nodes can fetch complete status + fmt.Println("\n⏳ Waiting for joining nodes to fetch cluster data...") + time.Sleep(15 * time.Second) + + // Shutdown formation server + fs.Shutdown(context.Background()) + + // Print cluster summary + fmt.Println("\n🎉 Cluster formation complete!") + fmt.Printf(" Cluster: %s (%d nodes)\n", clusterName, expectedNodes) + fmt.Printf(" Region: %s\n", region) + fmt.Println(" Nodes:") + for name, n := range allNodes { + fmt.Printf(" - %s (%s)\n", name, n.BindIP) + } + fmt.Println("\n📋 Next steps:") + fmt.Println(" 1. Start services on ALL nodes:") + fmt.Println(" ./scripts/start-dev.sh") + fmt.Println() +} + func runAdminJoin(cmd *cobra.Command, args []string) { node, _ := cmd.Flags().GetString("node") leaderHost, _ := cmd.Flags().GetString("host") @@ -682,12 +884,6 @@ func runAdminJoin(cmd *cobra.Command, args []string) { bindIP, _ := cmd.Flags().GetString("bind") configDir, _ := cmd.Flags().GetString("config-dir") clusterBind, _ := cmd.Flags().GetString("cluster-bind") - // Receive as string, split later - // TODO: Use GetStringArray - clusterRoutesStr, _ := cmd.Flags().GetString("cluster-routes") - - // Supply as array to template - clusterRoutes := strings.Split(clusterRoutesStr, ",") // Validate required parameters if node == "" { @@ -711,6 +907,11 @@ func runAdminJoin(cmd *cobra.Command, args []string) { os.Exit(1) } + // Default cluster-bind to bind IP if not specified + if clusterBind == "" { + clusterBind = bindIP + } + // Set default data directory if dataDir == "" { homeDir, err := os.UserHomeDir() @@ -721,9 +922,6 @@ func runAdminJoin(cmd *cobra.Command, args []string) { dataDir = filepath.Join(homeDir, "hive") } - // Set daemon host for this node - daemonHost := fmt.Sprintf("%s:%d", bindIP, port) - fmt.Println("🚀 Joining Hive cluster...") fmt.Printf("Node: %s\n", node) fmt.Printf("Leader: %s\n", leaderHost) @@ -732,13 +930,16 @@ func runAdminJoin(cmd *cobra.Command, args []string) { fmt.Printf("Bind IP: %s\n", bindIP) fmt.Printf("Port: %d\n\n", port) - // Create join request - joinReq := config.NodeJoinRequest{ - Node: node, - Region: region, - AZ: az, - DataDir: dataDir, - DaemonHost: daemonHost, + // POST join request to formation server + joinReq := formation.JoinRequest{ + NodeInfo: formation.NodeInfo{ + Name: node, + BindIP: bindIP, + ClusterIP: clusterBind, + Region: region, + AZ: az, + Port: port, + }, } reqBody, err := json.Marshal(joinReq) @@ -747,46 +948,24 @@ func runAdminJoin(cmd *cobra.Command, args []string) { os.Exit(1) } - // Send join request to leader with HTTP/2 support - tr := &http.Transport{ - TLSClientConfig: &tls.Config{ - InsecureSkipVerify: true, // Skip TLS verification for self-signed certs - NextProtos: []string{"h2", "http/1.1"}, - }, - ForceAttemptHTTP2: true, - } + client := &http.Client{Timeout: 30 * time.Second} - // Configure HTTP/2 support with custom TLS config - http2.ConfigureTransport(tr) - - client := &http.Client{ - Timeout: 30 * time.Second, - Transport: tr, - } - - joinURL := fmt.Sprintf("http://%s/join", leaderHost) + joinURL := fmt.Sprintf("http://%s/formation/join", leaderHost) resp, err := client.Post(joinURL, "application/json", bytes.NewBuffer(reqBody)) if err != nil { - fmt.Fprintf(os.Stderr, "❌ Error connecting to leader node: %v\n", err) - fmt.Fprintf(os.Stderr, "Make sure the leader node is running and accessible at %s\n", leaderHost) + fmt.Fprintf(os.Stderr, "❌ Error connecting to formation server: %v\n", err) + fmt.Fprintf(os.Stderr, "Make sure the leader node has run 'hive admin init' and is accessible at %s\n", leaderHost) os.Exit(1) } - defer resp.Body.Close() body, err := io.ReadAll(resp.Body) + resp.Body.Close() if err != nil { fmt.Fprintf(os.Stderr, "❌ Error reading response body: %v\n", err) os.Exit(1) } - if resp.StatusCode != 200 { - var errResp config.NodeJoinResponse - json.Unmarshal(body, &errResp) - fmt.Fprintf(os.Stderr, "❌ Failed to join cluster: %s\n", errResp.Message) - os.Exit(1) - } - - var joinResp config.NodeJoinResponse + var joinResp formation.JoinResponse if err := json.Unmarshal(body, &joinResp); err != nil { fmt.Fprintf(os.Stderr, "Error parsing join response: %v\n", err) os.Exit(1) @@ -797,12 +976,49 @@ func runAdminJoin(cmd *cobra.Command, args []string) { os.Exit(1) } - fmt.Println("✅ Successfully joined cluster!") - fmt.Printf("Epoch: %d\n", joinResp.SharedData.Epoch) - fmt.Printf("Config hash: %s\n\n", joinResp.ConfigHash) + fmt.Printf("✅ Registered with formation server (%d/%d nodes joined)\n", joinResp.Joined, joinResp.Expected) - // Save cluster config locally - // Confirm if default config directory required + // Poll status until formation is complete + statusURL := fmt.Sprintf("http://%s/formation/status", leaderHost) + var statusResp formation.StatusResponse + + for { + sResp, err := client.Get(statusURL) + if err != nil { + fmt.Fprintf(os.Stderr, "❌ Error polling formation status: %v\n", err) + os.Exit(1) + } + + sBody, err := io.ReadAll(sResp.Body) + sResp.Body.Close() + if err != nil { + fmt.Fprintf(os.Stderr, "❌ Error reading status response: %v\n", err) + os.Exit(1) + } + + if err := json.Unmarshal(sBody, &statusResp); err != nil { + fmt.Fprintf(os.Stderr, "Error parsing status response: %v\n", err) + os.Exit(1) + } + + if statusResp.Complete { + break + } + + fmt.Printf(" Waiting for cluster formation... (%d/%d nodes joined)\n", statusResp.Joined, statusResp.Expected) + time.Sleep(2 * time.Second) + } + + fmt.Printf("✅ Cluster formation complete! (%d nodes)\n\n", statusResp.Expected) + + // Extract credentials and CA from formation status + creds := statusResp.Credentials + if creds == nil { + fmt.Fprintf(os.Stderr, "❌ Error: formation server did not return credentials\n") + os.Exit(1) + } + + // Set up config directory if configDir == "" { homeDir, err := os.UserHomeDir() if err != nil { @@ -812,109 +1028,91 @@ func runAdminJoin(cmd *cobra.Command, args []string) { configDir = filepath.Join(homeDir, "hive", "config") } - hiveTomlPath := filepath.Join(configDir, "hive.toml") - - // Create config directory if needed if err := os.MkdirAll(configDir, 0700); err != nil { fmt.Fprintf(os.Stderr, "Error creating config directory: %v\n", err) os.Exit(1) } - // Construct local node config with THIS node's name at top level - localConfig := config.ClusterConfig{ - Epoch: joinResp.SharedData.Epoch, - Node: node, // THIS node's name, not the leader's - Version: joinResp.SharedData.Version, - Nodes: joinResp.SharedData.Nodes, - } + // Write CA cert and key + caCertPath := filepath.Join(configDir, "ca.pem") + caKeyPath := filepath.Join(configDir, "ca.key") - // Marshal config to TOML format - configTOML, err := toml.Marshal(&localConfig) - if err != nil { - fmt.Fprintf(os.Stderr, "Error marshaling config to TOML: %v\n", err) + if statusResp.CACert == "" || statusResp.CAKey == "" { + fmt.Fprintf(os.Stderr, "❌ Error: formation server did not return CA certificate\n") os.Exit(1) } - // Write config to file - if err := os.WriteFile(hiveTomlPath, configTOML, 0600); err != nil { - fmt.Fprintf(os.Stderr, "Error writing config file: %v\n", err) + if err := os.WriteFile(caCertPath, []byte(statusResp.CACert), 0600); err != nil { + fmt.Fprintf(os.Stderr, "Error writing CA cert: %v\n", err) os.Exit(1) } - - fmt.Printf("✅ Config saved to: %s\n\n", hiveTomlPath) - - // Handle CA and server certificate generation - // If leader provided CA files, use those; otherwise fall back to generating our own - if joinResp.CACert != "" && joinResp.CAKey != "" { - // Write CA files received from leader - caCertPath := filepath.Join(configDir, "ca.pem") - caKeyPath := filepath.Join(configDir, "ca.key") - - if err := os.WriteFile(caCertPath, []byte(joinResp.CACert), 0600); err != nil { - fmt.Fprintf(os.Stderr, "Error writing CA cert: %v\n", err) - os.Exit(1) - } - if err := os.WriteFile(caKeyPath, []byte(joinResp.CAKey), 0600); err != nil { - fmt.Fprintf(os.Stderr, "Error writing CA key: %v\n", err) - os.Exit(1) - } - fmt.Printf("✅ CA certificate received from leader: %s\n", caCertPath) - - // Generate server cert signed by CA with this node's bind IP - if err := admin.GenerateServerCertOnly(configDir, bindIP); err != nil { - fmt.Fprintf(os.Stderr, "Error generating server certificate: %v\n", err) - os.Exit(1) - } - fmt.Printf("✅ Server certificate generated with bind IP: %s\n\n", bindIP) - } else { - // Fallback: generate own CA if leader didn't provide one (backwards compatibility) - certPath := admin.GenerateCertificatesIfNeeded(configDir, false, bindIP) - fmt.Printf("✅ SSL certificates available at: %s\n\n", certPath) + if err := os.WriteFile(caKeyPath, []byte(statusResp.CAKey), 0600); err != nil { + fmt.Fprintf(os.Stderr, "Error writing CA key: %v\n", err) + os.Exit(1) } + fmt.Printf("✅ CA certificate received from leader: %s\n", caCertPath) - // Write individual node config files - portStr := fmt.Sprintf("%d", port) + // Generate server cert signed by CA with this node's bind IP + if err := admin.GenerateServerCertOnly(configDir, bindIP); err != nil { + fmt.Fprintf(os.Stderr, "Error generating server certificate: %v\n", err) + os.Exit(1) + } + fmt.Printf("✅ Server certificate generated with bind IP: %s\n\n", bindIP) + // Build cluster topology from formation data + clusterRoutes := formation.BuildClusterRoutes(statusResp.Nodes) + predastoreNodes := formation.BuildPredastoreNodes(statusResp.Nodes) + + fmt.Println("📝 Creating configuration files...") + + // Create subdirectories awsgwDir := filepath.Join(configDir, "awsgw") predastoreDir := filepath.Join(configDir, "predastore") natsDir := filepath.Join(configDir, "nats") + hiveDir := filepath.Join(configDir, "hive") - for _, dir := range []string{awsgwDir, predastoreDir, natsDir} { + for _, dir := range []string{awsgwDir, predastoreDir, natsDir, hiveDir} { if err := os.MkdirAll(dir, 0700); err != nil { fmt.Fprintf(os.Stderr, "Error creating directory %s: %v\n", dir, err) os.Exit(1) } } - // Handle multi-node predastore config from leader + portStr := fmt.Sprintf("%d", port) + + // Generate multi-node predastore config var predastoreNodeID int - hasLeaderPredastoreConfig := joinResp.PredastoreConfig != "" + hasPredastoreConfig := len(predastoreNodes) >= 3 + + if hasPredastoreConfig { + predastoreContent, err := admin.GenerateMultiNodePredastoreConfig(predastoreMultiNodeTemplate, predastoreNodes, creds.AccessKey, creds.SecretKey, creds.Region) + if err != nil { + fmt.Fprintf(os.Stderr, "Error generating multi-node predastore config: %v\n", err) + os.Exit(1) + } - if hasLeaderPredastoreConfig { - // Write the leader's predastore.toml predastorePath := filepath.Join(predastoreDir, "predastore.toml") - if err := os.WriteFile(predastorePath, []byte(joinResp.PredastoreConfig), 0600); err != nil { - fmt.Fprintf(os.Stderr, "Error writing predastore config from leader: %v\n", err) + if err := os.WriteFile(predastorePath, []byte(predastoreContent), 0600); err != nil { + fmt.Fprintf(os.Stderr, "Error writing predastore config: %v\n", err) os.Exit(1) } - fmt.Printf("✅ Predastore config received from leader: %s\n", predastorePath) - predastoreNodeID = admin.ParsePredastoreNodeIDFromConfig(joinResp.PredastoreConfig, bindIP) - - if predastoreNodeID > 0 { - fmt.Printf("✅ Detected Predastore node ID: %d (for bind IP %s)\n", predastoreNodeID, bindIP) - } else { - fmt.Fprintf(os.Stderr, "ERROR: Could not detect Predastore node ID for bind IP %s in leader config\n", bindIP) - fmt.Fprintf(os.Stderr, "This node's IP must be listed in the --predastore-nodes used during init\n") + predastoreNodeID = admin.FindNodeIDByIP(predastoreNodes, bindIP) + if predastoreNodeID == 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) } + hiveTomlPath := filepath.Join(configDir, "hive.toml") + configSettings := admin.ConfigSettings{ - AccessKey: localConfig.Nodes[node].AccessKey, - SecretKey: localConfig.Nodes[node].SecretKey, - Region: region, - NatsToken: localConfig.Nodes[node].NATS.ACL.Token, + AccessKey: creds.AccessKey, + SecretKey: creds.SecretKey, + AccountID: creds.AccountID, + Region: creds.Region, + NatsToken: creds.NatsToken, DataDir: dataDir, Node: node, @@ -923,6 +1121,7 @@ func runAdminJoin(cmd *cobra.Command, args []string) { BindIP: bindIP, ClusterBindIP: clusterBind, ClusterRoutes: clusterRoutes, + ClusterName: creds.ClusterName, PredastoreNodeID: predastoreNodeID, } @@ -933,8 +1132,8 @@ func runAdminJoin(cmd *cobra.Command, args []string) { {Name: filepath.Join(awsgwDir, "awsgw.toml"), Path: filepath.Join(awsgwDir, "awsgw.toml"), Template: awsgwTomlTemplate}, {Name: filepath.Join(natsDir, "nats.conf"), Path: filepath.Join(natsDir, "nats.conf"), Template: natsConfTemplate}, } - // Skip template-based predastore.toml if received from leader - if !hasLeaderPredastoreConfig { + // Skip template-based predastore.toml if multi-node was generated + if !hasPredastoreConfig { configs = append(configs, admin.ConfigFile{ Name: filepath.Join(predastoreDir, "predastore.toml"), Path: filepath.Join(predastoreDir, "predastore.toml"), Template: predastoreTomlTemplate, }) @@ -946,11 +1145,25 @@ func runAdminJoin(cmd *cobra.Command, args []string) { os.Exit(1) } + // Update ~/.aws/credentials and ~/.aws/config + fmt.Println("\n🔧 Configuring AWS credentials...") + if err := admin.SetupAWSCredentials(creds.AccessKey, creds.SecretKey, creds.Region, caCertPath, bindIP); err != nil { + fmt.Fprintf(os.Stderr, "Warning: Could not update AWS credentials: %v\n", err) + } else { + fmt.Println("✅ AWS credentials configured") + } + admin.CreateServiceDirectories(dataDir) - fmt.Println("🎉 Node successfully joined cluster!") + // Print cluster summary + fmt.Println("\n🎉 Node successfully joined cluster!") + fmt.Printf(" Cluster: %s (%d nodes)\n", creds.ClusterName, len(statusResp.Nodes)) + fmt.Println(" Nodes:") + for name, n := range statusResp.Nodes { + fmt.Printf(" - %s (%s)\n", name, n.BindIP) + } fmt.Println("\n📋 Next steps:") - fmt.Println(" 1. Start the hive service:") - fmt.Printf(" hive service hive start --config %s\n", hiveTomlPath) + fmt.Println(" 1. Start services:") + fmt.Println(" ./scripts/start-dev.sh") fmt.Println() } diff --git a/cmd/hive/cmd/service.go b/cmd/hive/cmd/service.go index a7307e4ac..55cc6c6c9 100644 --- a/cmd/hive/cmd/service.go +++ b/cmd/hive/cmd/service.go @@ -275,6 +275,7 @@ var viperblockStartCmd = &cobra.Command{ AccessKey: nodeConfig.Predastore.AccessKey, SecretKey: nodeConfig.Predastore.SecretKey, BaseDir: nodeConfig.Predastore.BaseDir, + NodeName: clusterConfig.Node, }) if err != nil { diff --git a/cmd/hive/cmd/templates/nats.conf b/cmd/hive/cmd/templates/nats.conf index 5713c71b1..139f00556 100644 --- a/cmd/hive/cmd/templates/nats.conf +++ b/cmd/hive/cmd/templates/nats.conf @@ -11,7 +11,7 @@ listen: 0.0.0.0:4222 {{ if and .ClusterBindIP .ClusterRoutes }} cluster { - name: C1 + name: {{ .ClusterName }} listen: {{.ClusterBindIP}}:4248 routes = [ diff --git a/docs/development/improvements/multi-server.md b/docs/development/improvements/multi-server.md new file mode 100644 index 000000000..d220d09a0 --- /dev/null +++ b/docs/development/improvements/multi-server.md @@ -0,0 +1,675 @@ +# Multi-Server Cluster Formation + +## Problem + +The current multi-server configuration has a fundamental design flaw: cluster formation and cluster operation are entangled. The `/join` endpoint lives inside the Hive daemon, which requires NATS to be running. This creates a chicken-and-egg problem — nodes cannot join until services are running, but services cannot be correctly configured until all nodes have joined. + +### Current bugs + +1. **NATS config missing all cluster routes** — each node's `nats.conf` only contains the leader's route, not routes to all other nodes. NATS cannot form a full mesh cluster. +2. **NATS cluster name hardcoded to `C1`** — should be configurable per-cluster. +3. **Predastore "Distributed mode requires -node flag"** — NODE_ID detection fails in multi-node setups. +4. **Predastore "no known peers" on joining nodes** — Raft starts with `servers=[]`, cannot form consensus. +5. **Sequence dependency** — node 1 must run `./scripts/start-dev.sh` before node 2 can `hive admin join`, because the `/join` endpoint is inside the daemon. +6. **Certs not installed to system CA store** — joining nodes receive the CA cert but don't install it, causing TLS failures. + +### Root cause + +All bugs trace back to one issue: configs are generated per-node during init/join with incomplete information (only the leader IP is known). The correct approach is to generate configs AFTER all nodes have registered, using complete cluster information. + +## Design + +Separate cluster **formation** (lightweight HTTP) from cluster **operation** (NATS, Predastore, etc). + +### Formation flow + +``` + ┌─────────────────────────────────┐ + │ hive admin init --nodes 3 │ + │ Starts formation HTTP server │ + │ Registers self as node 1 │ + │ Waits for 2 more nodes... │ + └──────────┬──────────────────────┘ + │ :4432 + ┌────────────────┼────────────────┐ + │ │ │ + ┌─────────▼──────┐ ┌─────▼────────┐ ┌───▼──────────┐ + │ hive admin join │ │ (waiting) │ │ (waiting) │ + │ POST /join │ │ │ │ │ + │ polls /status │ │ │ │ │ + └────────────────┘ └──────────────┘ └──────────────┘ + │ │ │ + └────────────────┼────────────────┘ + │ All 3 nodes joined + ┌──────────▼──────────────────────┐ + │ Formation complete │ + │ Generate configs with ALL IPs │ + │ Distribute CA, credentials │ + │ All processes exit │ + └──────────┬──────────────────────┘ + │ + ┌────────────────┼────────────────┐ + │ │ │ + ┌─────────▼──────┐ ┌─────▼────────┐ ┌───▼──────────┐ + │ start-dev.sh │ │ start-dev.sh │ │ start-dev.sh │ + │ Node 1 │ │ Node 2 │ │ Node 3 │ + └────────────────┘ └──────────────┘ └──────────────┘ +``` + +### Phase 1: Formation (init + join) + +**Node 1** starts a lightweight HTTP formation server (no NATS dependency): + +```sh +./bin/hive admin init \ + --node node1 \ + --nodes 3 \ + --bind 10.11.12.1 \ + --cluster-bind 10.11.12.1 \ + --port 4432 \ + --hive-dir ~/hive/ \ + --config-dir ~/hive/config/ \ + --region ap-southeast-2 \ + --az ap-southeast-2a +``` + +This generates credentials (AWS keys, NATS token, CA certs) and starts the formation HTTP server on `10.11.12.1:4432`. It registers itself as the first node and waits for 2 more. + +**Node 2 and Node 3** join while init is running: + +```sh +./bin/hive admin join \ + --node node2 \ + --bind 10.11.12.2 \ + --cluster-bind 10.11.12.2 \ + --host 10.11.12.1:4432 \ + --data-dir ~/hive/ \ + --config-dir ~/hive/config/ \ + --region ap-southeast-2 \ + --az ap-southeast-2a +``` + +Each joining node POSTs to the formation server, then polls for completion. + +Once all 3 nodes register, the formation server: +1. Computes NATS cluster routes from ALL node IPs (full mesh) +2. Builds predastore node configs from ALL node IPs (with correct node IDs) +3. Returns complete config package (credentials, CA, node list) to each polling node +4. Each node generates its own configs locally with complete cluster data +5. CA cert is installed to system trust store on each node +6. All processes (init + joins) display cluster summary and exit + +**Key improvements over current design:** +- `--predastore-nodes` is now **optional** — derived automatically from joined nodes +- `--cluster-routes` on join is now **optional** — derived from formation data +- No sequence dependency — init starts the formation server directly (no NATS needed) +- All configs have complete cluster data (all IPs, all routes) + +### Phase 2: Service startup + +After formation completes, run on each node: + +```sh +./scripts/start-dev.sh +``` + +Services start in order: NATS → Predastore → Viperblock → Daemon → AWS Gateway → UI. + +Since all configs already have the full cluster topology, NATS forms a mesh immediately and Predastore Raft finds all peers. + +## Network recommendations + +It is recommended that at least two network links are used by Hive: + +1. **Management link** — cluster sync, NATS, Predastore Raft, and daemon communication. This requires low latency (< 5ms) and dedicated bandwidth. +2. **Data/VM link** — EC2 instances, VPC traffic (Geneve/Open vSwitch), and user-facing services. Isolated from management to prevent contention. + +This follows the same model as Proxmox VE, which recommends a physically separated NIC for cluster (corosync) traffic: "The network should not be used heavily by other members, as while corosync does not use much bandwidth it is sensitive to latency jitters." + +## Proxmox Cluster Manager comparison + +Reviewed [Proxmox Cluster Manager](https://pve.proxmox.com/wiki/Cluster_Manager) for best practices. + +**What we adopt from Proxmox:** +- Formation-first workflow (`pvecm create` / `pvecm add` completes before services use the cluster) +- Shared CA model (joining node's cert is replaced with one signed by cluster CA) +- Dedicated cluster network recommendation (separate NIC for management traffic) +- Hostname/IP finalized before formation (cannot change post-creation) + +**Why native Go over Corosync:** + +Corosync is a C group communication library providing reliable messaging, quorum, and membership. Hive already has equivalent capabilities: + +| Capability | Corosync | Hive | +|---|---|---| +| Message bus | Totem protocol | NATS | +| State replication | pmxcfs (replicated filesystem) | NATS JetStream KV | +| Consensus/quorum | Corosync votequorum | Predastore Raft | +| Config distribution | pmxcfs | Formation server + hive.toml | + +Adding Corosync would be a redundant external C dependency providing overlapping functionality. The native Go implementation gives us: +- Single binary deployment (no `corosync` daemon to install/manage) +- Direct integration with NATS and Predastore +- Full control over the join protocol +- No CGo or external library dependency + +## Implementation + +### New package: `hive/formation/` + +Lightweight HTTP formation server using `net/http`. Zero dependency on NATS or any operational service. + +**`formation.go`** — core server: +- `FormationServer` struct: tracks expected nodes, joined nodes, shared credentials, CA cert/key +- `POST /formation/join` — register a node, return progress +- `GET /formation/status` — poll completion; returns full config data when all nodes joined +- `GET /formation/health` — liveness check +- `WaitForCompletion()` — blocks until all nodes join or timeout (default 10 minutes) + +**`helpers.go`** — config builders: +- `BuildClusterRoutes(nodes)` — returns all node cluster-bind IPs as `IP:4248` routes +- `BuildPredastoreNodes(nodes)` — returns `[]PredastoreNodeConfig` with 1-based IDs + +### Changes to `cmd/hive/cmd/admin.go` + +**`runAdminInit`:** +- `--nodes 1` (or no cluster flags): existing single-node behaviour, unchanged +- `--nodes >= 2`: start formation server, register self, wait for joins, generate configs with complete data +- `--predastore-nodes` becomes optional (derived from formation when not specified) +- New flag: `--formation-timeout` (default "10m") + +**`runAdminJoin`:** +- POST to `http://leader:port/formation/join` (not the daemon `/join`) +- Poll `GET /formation/status` every 2s until complete +- Extract full node list, CA, credentials from response +- Generate all configs locally with complete cluster data +- Install CA to system trust store + +### Changes to templates + +**`templates/nats.conf`:** Change `name: C1` to `name: {{ .ClusterName }}` + +**`hive/admin/admin.go`:** Add `ClusterName string` to `ConfigSettings` + +### Changes to `scripts/start-dev.sh` + +- Remove awk-based NODE_ID auto-detection (formation now writes correct node_id into hive.toml) +- Keep `HIVE_PREDASTORE_NODE_ID` env var override for manual use + +### Daemon `/join` endpoint (no changes) + +The existing `/join` endpoint in `hive/daemon/daemon.go` stays as-is for future "hot-add node to running cluster" scenarios. Formation handles initial bootstrap; daemon handles runtime expansion. + +### Graceful cluster shutdown + +The current `stop-dev.sh` sends SIGTERM to each service sequentially with no cluster coordination. In a multi-node cluster this causes cascading failures — if the Raft leader is stopped first, remaining predastore nodes enter an election storm trying to contact dead peers, goroutines block, and the cluster hangs instead of shutting down cleanly. + +#### Two shutdown modes + +**1. Cluster-wide shutdown** (`hive admin shutdown`): + +Coordinated shutdown of the entire cluster. The initiating node broadcasts a shutdown intent, all nodes agree, and services stop in a safe order across the cluster. + +```sh +# On any node — broadcasts to entire cluster +./bin/hive admin shutdown +``` + +**2. Single-node shutdown** (`stop-dev.sh` or `hive admin stop`): + +Stops services on one node only. The node announces its departure so peers can adjust (e.g., Raft leader transfer) rather than treating it as a crash. + +```sh +# On the node being stopped +./scripts/stop-dev.sh +``` + +#### Cluster-wide shutdown flow + +``` +Operator runs: hive admin shutdown (on any node) + │ + ▼ +┌──────────────────────────────────┐ +│ 1. Connect to local daemon │ +│ 2. POST /cluster/shutdown │ +│ 3. Daemon broadcasts │ +│ hive.cluster.shutdown via NATS │ +└──────────────┬───────────────────┘ + │ + ┌──────────┼──────────┐ + ▼ ▼ ▼ + Node 1 Node 2 Node 3 + │ │ │ + │ All nodes enter "shutting_down" state + │ Stop accepting new work (API returns 503) + │ │ │ + ▼ ▼ ▼ + Phase 1: Stop VMs (system_powerdown, wait, write state to JetStream) + │ │ │ + ▼ ▼ ▼ + Phase 2: Stop daemon + gateway (unsubscribe NATS, close connections) + │ │ │ + ▼ ▼ ▼ + Phase 3: Stop storage (predastore Raft shutdown, viperblock) + │ │ │ + ▼ ▼ ▼ + Phase 4: Stop NATS (last — needed for coordination until this point) +``` + +#### Shutdown protocol + +1. **Initiation** — any node's daemon receives `POST /cluster/shutdown` (from `hive admin shutdown` CLI) +2. **Broadcast** — daemon publishes `hive.cluster.shutdown` on NATS with initiator node ID and epoch +3. **Acknowledgement** — each daemon responds on `hive.cluster.shutdown.ack` with its node ID. Initiator waits up to 10s for all known nodes to ACK (proceeds anyway if some nodes are unreachable) +4. **Phase 1: VM drain** — all daemons stop accepting new `ec2.RunInstances` requests, send `system_powerdown` to running VMs, write instance state to JetStream, wait up to 60s for VMs to power off +5. **Phase 2: Daemon stop** — daemons unsubscribe from all NATS topics, shut down the cluster HTTP server and AWS Gateway +6. **Phase 3: Storage stop** — predastore Raft leader (if local) calls `raft.Shutdown()` first, then followers shut down Raft; viperblock stops. Because all nodes are shutting down simultaneously, there's no election storm — every node's Raft transport is closed before heartbeat timeouts fire +7. **Phase 4: NATS stop** — NATS server is the last to stop (it was needed for the shutdown broadcast and ACKs) +8. **Cleanup** — each node's `stop-dev.sh` waits for all service PIDs to exit, force-kills any stragglers after 120s + +#### Predastore Raft shutdown handling + +The core problem: when a Raft leader disappears without warning, followers detect missing heartbeats and start elections. With 2-of-3 nodes gone, elections can never succeed and goroutines spin indefinitely. + +**Solution for cluster-wide shutdown:** +- All predastore nodes receive the shutdown broadcast via NATS (Phase 3) +- Each node closes the Raft transport first (`transport.Close()`), which prevents any new RPCs +- Then calls `raft.Shutdown()` which drains in-flight operations +- Because all nodes close transport near-simultaneously, no node has time to start an election +- The 5-second Raft shutdown timeout in predastore (`s3db/raft.go`) is sufficient since no elections are triggered + +**Solution for single-node shutdown:** +- The departing node publishes `hive.cluster.node-leaving` with its node ID +- If the departing node is the Raft leader, it calls `raft.LeadershipTransfer()` before shutting down — this hands leadership to a follower cleanly +- If it's a follower, it calls `raft.Shutdown()` directly — the leader and remaining followers continue normally +- Remaining nodes log the departure and continue operating (Raft maintains quorum with N-1 nodes if majority survives) + +#### Single-node stop flow + +When `stop-dev.sh` runs on one node in a multi-node cluster: + +1. Daemon receives SIGTERM +2. Daemon publishes `hive.cluster.node-leaving` with its node ID on NATS +3. If this node is the predastore Raft leader: call `raft.LeadershipTransfer()`, wait up to 5s +4. Stop VMs: `system_powerdown` → wait → write state to JetStream +5. Unsubscribe from NATS topics, shut down cluster HTTP server +6. Stop predastore (Raft transport close, then `raft.Shutdown()`) +7. Stop viperblock +8. Close NATS connection +9. Stop NATS server + +The remaining cluster nodes see the `node-leaving` message and: +- Remove the node from active heartbeat tracking (no false alarms) +- Predastore Raft has already transferred leadership if needed +- AWS Gateway's next `hive.nodes.discover` broadcast gets fewer responses (expected) +- No config changes — the node is still in `hive.toml` but simply offline + +#### Changes to `scripts/stop-dev.sh` + +- Before stopping services, check if this is a multi-node cluster (read `hive.toml` node count) +- If multi-node: send `hive.cluster.node-leaving` via the daemon before stopping it +- If daemon is unreachable (already crashed): proceed with direct SIGTERM as today (best effort) +- Keep the existing service stop order but add a brief pause after daemon stop to allow the `node-leaving` broadcast to propagate + +#### New CLI command: `hive admin shutdown` + +```sh +./bin/hive admin shutdown [--force] [--timeout 120s] +``` + +- Connects to local daemon's cluster HTTP endpoint +- `POST /cluster/shutdown` triggers the coordinated shutdown flow +- `--force` skips VM drain and ACK wait (immediate SIGTERM to all services) +- `--timeout` controls max wait for VM powerdown (default 120s) +- Prints progress: "Waiting for ACKs... 3/3", "Draining VMs... 2 remaining", "Stopping storage...", "Shutdown complete" + +## Files + +| File | Action | +|---|---| +| `hive/formation/formation.go` | New — formation HTTP server | +| `hive/formation/helpers.go` | New — route and node config builders | +| `hive/formation/formation_test.go` | New — tests | +| `hive/admin/admin.go` | Modify — add ClusterName to ConfigSettings | +| `cmd/hive/cmd/admin.go` | Modify — rewrite init multi-node path + join, add `shutdown` subcommand | +| `cmd/hive/cmd/templates/nats.conf` | Modify — C1 → {{ .ClusterName }} | +| `scripts/start-dev.sh` | Modify — simplify NODE_ID detection | +| `scripts/stop-dev.sh` | Modify — check cluster mode, send `node-leaving` before stopping services | +| `hive/daemon/daemon.go` | Modify — add shutdown broadcast handler, `node-leaving` publish on SIGTERM, Raft leader transfer coordination | + +## Testing + +### Unit tests + +- Formation server: all nodes join, duplicate rejection, timeout, status polling +- Helper functions: route building, predastore node building +- NATS template: renders cluster name and all routes correctly +- Shutdown broadcast: all nodes ACK, timeout when node unreachable, phase ordering +- Node-leaving: Raft leader transfer triggered when departing node is leader + +### Integration test + +Note: `--config-dir ~/$NODE/config/` needs to be specified for each node, so not to conflict when testing locally to simulate 3 nodes. + +1. `hive admin init --nodes 3 --bind 127.0.0.1 --node node1` in terminal 1 +2. `hive admin join --node node2 --host 127.0.0.1:4432 --bind 127.0.0.2` in terminal 2 +3. `hive admin join --node node3 --host 127.0.0.1:4432 --bind 127.0.0.3` in terminal 3 +4. Verify: all 3 exit with cluster summary +5. Verify: each `nats.conf` has routes to ALL other nodes +6. Verify: each `predastore.toml` has all 3 nodes +7. Verify: each `hive.toml` has all 3 nodes in `[nodes.*]` +8. Run `./scripts/start-dev.sh` on all 3, verify NATS mesh and Predastore Raft form correctly + +### Shutdown integration test + +1. Start 3-node cluster (from formation integration test above) +2. Run `hive admin shutdown` on node1 +3. Verify: all 3 nodes receive shutdown broadcast (check logs for ACK) +4. Verify: all predastore Raft nodes shut down cleanly — no election storms in logs +5. Verify: all NATS servers stop last +6. Verify: all service PIDs are gone on all 3 nodes + +### Single-node stop integration test + +1. Start 3-node cluster +2. Run `./scripts/stop-dev.sh` on node3 (a Raft follower) +3. Verify: node3 publishes `node-leaving`, services stop cleanly +4. Verify: nodes 1 and 2 continue operating, Predastore Raft maintains quorum (2/3) +5. Run `./scripts/stop-dev.sh` on node1 (if node1 is Raft leader) +6. Verify: Raft leadership transfers to node2 before node1's predastore stops +7. Verify: node2 continues operating as single-node (degraded but functional) + +--- + +## Stage 2: Runtime Node Expansion + +### Overview + +After the initial cluster is formed (Stage 1) and services are running, operators need the ability to add new nodes without stopping the cluster. This covers "day 2" operations — e.g., a 3-node cluster formed on Monday, with a 4th node added on Tuesday. + +The daemon takes over formation duties once the cluster is running. The existing `/join` endpoint in `hive/daemon/daemon.go` is replaced by the formation server operating in "Stage 2 mode" — same formation package, different context. + +### Key difference from Stage 1 + +Stage 1 formation happens BEFORE any services start — it's a clean-room config generation step where all processes exit after writing configs. Stage 2 happens WHILE services are running, so each service must be notified of topology changes and reconfigure itself live. + +### Expansion flow + +``` +Existing running cluster (3 nodes) New node +┌──────────┐ ┌──────────┐ ┌──────────┐ +│ Node 1 │ │ Node 2 │ │ Node 3 │ +│ daemon │ │ daemon │ │ daemon │ +│ :4432 │ │ :4432 │ │ :4432 │ +└─────┬─────┘ └─────┬─────┘ └─────┬─────┘ + │ NATS mesh │ │ + └───────┬───────┘ │ + │ │ ┌─────────────────┐ + │ │ │ hive admin join │ + │ │ │ --host node1 │ + │ │ │ --node node4 │ + │ │ │ --capability │ + │ │ │ nats,viperblock│ + │ │ │ ,hive,awsgw │ + │ │ └────────┬────────┘ + │ │ │ + │ │ POST /formation/join + │ │ │ + ┌─────▼──────┐ │ │ + │ Node 1 │◄─────────────┼──────────────────┘ + │ validates │ │ + │ updates toml │ │ + │ returns creds│─────────────┘ + │ broadcasts │──► hive.cluster.config-update (NATS) + └─────┬───────┘ + │ + ▼ + All existing nodes receive broadcast: + - Regenerate nats.conf → SIGHUP nats-server + - Update predastore.toml if needed → restart predastore + - Update hive.toml with new node entry + │ + ▼ + New node generates configs, starts services: + ./scripts/start-dev.sh (on node4 only) +``` + +### Node capabilities + +Not all nodes need to run all services. The `--capability` flag controls which services a node provides: + +```sh +./bin/hive admin join \ + --node node4 \ + --bind 10.11.12.4 \ + --cluster-bind 10.11.12.4 \ + --host 10.11.12.1:4432 \ + --data-dir ~/hive/ \ + --config-dir ~/hive/config/ \ + --region ap-southeast-2 \ + --az ap-southeast-2a \ + --capability nats,viperblock,hive,awsgw +``` + +Available capabilities: + +| Capability | Service | Notes | +|---|---|---| +| `nats` | NATS server | Cluster messaging — required for all nodes | +| `predastore` | Predastore storage node | Object shard storage via QUIC | +| `predastoredb` | Predastore DB node | Raft-replicated metadata database | +| `viperblock` | Viperblock | Local NVMe WAL for block storage — required if running VMs | +| `hive` | Hive daemon | VM orchestration and scheduling | +| `awsgw` | AWS Gateway | TLS API endpoint for AWS SDK compatibility | +| `hive-ui` | Hive UI | Web management interface | + +Default: all capabilities enabled. + +**Example: compute-only node** (runs VMs, no object storage): +```sh +--capability nats,viperblock,hive,awsgw +``` + +**Example: storage-only node** (object storage, no VMs): +```sh +--capability nats,predastore +``` + +**Predastore DB scaling note:** The `predastoredb` capability controls whether a node participates in the Predastore Raft metadata cluster. For small clusters (3-5 nodes), every predastore node should also run predastoredb. For larger clusters (10+ nodes), Raft only needs 3-5 voters for consensus — additional predastore nodes store object shards via the hash ring but don't replicate the metadata database. This avoids replicating the KV store 25 times in a 25-node cluster. + +The `--capability` flag also applies to Stage 1 `hive admin join` for initial formation. Stage 1 `hive admin init` always has all capabilities (it's the bootstrap node). + +### Join protocol (Stage 2) + +The `hive admin join` command detects whether it's joining a formation (Stage 1) or a running cluster (Stage 2) based on the response from the leader. + +1. New node runs `hive admin join --host leader:4432 --node node4 --capability ...` +2. New node POSTs to `http://leader:4432/formation/join` with node info + capabilities +3. Leader daemon validates: + - Node name not already in cluster (409 Conflict if duplicate) + - Bind IP not already in use by another node + - Capabilities are valid strings +4. Leader updates local `hive.toml` with new node entry, increments epoch +5. Leader returns: CA cert/key, credentials (AWS access/secret key, NATS token), full node list including the new node +6. New node writes CA cert/key, generates server cert signed by cluster CA +7. New node generates local configs (nats.conf, hive.toml, predastore.toml, etc.) using complete cluster data, filtered by its capabilities +8. New node installs CA cert to system trust store +9. Leader broadcasts config update to all existing nodes via NATS (`hive.cluster.config-update`) +10. Existing nodes receive update and reconfigure affected services (see below) +11. `hive admin join` exits with cluster summary — operator then runs `start-dev.sh` on the new node + +### Service reconfiguration on existing nodes + +When a new node joins a running cluster, the leader broadcasts a `hive.cluster.config-update` message via NATS containing the updated node list, capabilities, and epoch. Each existing daemon handles the update: + +**NATS:** +1. Regenerate `nats.conf` with new route added (only if new node has `nats` capability) +2. Send `SIGHUP` to the `nats-server` process — NATS supports config reload via signal, including adding new cluster routes +3. NATS discovers and connects to the new node once it starts +4. If SIGHUP reload fails, log error and flag for manual restart + +**Predastore storage (`predastore` capability):** +- If the new node has `predastore` capability: regenerate `predastore.toml` with new `[[nodes]]` entry and restart the predastore process (predastore does not support hot config reload) +- The consistent hash ring updates on restart — new PUTs distribute across all nodes including the new one +- Existing objects are unaffected (see "Hash ring rebalancing" below) + +**Predastore DB (`predastoredb` capability):** +- If the new node has `predastoredb` capability: the leader calls `raft.AddVoter()` to add the new node to the Raft cluster +- No restart needed on existing DB nodes — Raft handles membership changes natively via joint consensus +- If the cluster already has sufficient DB voters (e.g., 5), the new node is not added as a voter even if it has the `predastoredb` capability — log a warning + +**Viperblock:** +- No reconfiguration needed on existing nodes — viperblock is local to each node +- New node starts its own viperblock instance independently + +**JetStream:** +- Call `UpdateReplicas()` to match new cluster size (already implemented in `jetstream.go`) +- Capped at the NATS cluster size (JetStream replicas cannot exceed the number of NATS nodes) + +**AWS Gateway:** +- No reconfiguration needed — uses dynamic NATS node discovery (`hive.nodes.discover` fan-out with 500ms timeout) +- New node's daemon responds to discovery automatically once it's connected to NATS + +**Hive UI:** +- No reconfiguration needed — UI queries the AWS Gateway, which discovers nodes dynamically + +### Predastore hash ring rebalancing + +When a new predastore storage node joins, the consistent hash ring (xxhash + `buraksezer/consistent`) changes. Shard placement for new writes shifts, but existing data is safe. + +**Immediate behaviour (no rebalancing — Stage 2 scope):** +- Existing objects stay on their original nodes. Shard locations are recorded per-object in `ObjectToShardNodes` in the Raft-replicated s3db — GETs look up this metadata, not the ring +- New objects (PUTs) use the updated ring, distributing across all nodes including the new one +- DELETEs use stored metadata to find and remove shards from the correct nodes +- The cluster is fully functional without rebalancing — the only downside is uneven storage distribution until a rebalance runs + +**Why this is safe:** The hash ring is only used at write time to decide where to place shards. Once placed, the mapping `object_hash → [node_ids]` is persisted in s3db. All subsequent reads and deletes use this stored mapping, not the ring. Adding a node to the ring does not invalidate any existing mappings. + +**Background rebalancing (future — not in Stage 2 scope):** +A future `hive admin rebalance` command would: +1. Scan all objects in s3db +2. For each object, compute new ring placement with current topology +3. If placement changed, migrate shards from old nodes to new node via QUIC +4. Update s3db metadata with new shard locations (atomic Raft write) +5. Delete old shard copies after successful migration + +This is a significant feature (data migration, progress tracking, failure recovery, bandwidth throttling) and is deferred to a dedicated implementation. + +### Config sync protocol + +All nodes must converge on the same cluster state. The protocol uses epoch-based versioning: + +1. **Epoch counter** — every config change (node join, capability change) increments the epoch on the leader +2. **Broadcast on change** — leader publishes `hive.cluster.config-update` via NATS with full node list + epoch +3. **Periodic heartbeat** — each daemon periodically publishes its epoch on `hive.cluster.heartbeat`; if a node detects it has a stale epoch (e.g., it was temporarily disconnected when the broadcast happened), it requests the latest config from the leader via `hive.cluster.config-request` +4. **Config hash** — SHA256 of shared cluster data (already implemented via `computeConfigHash()`) included in `/health` endpoint response; monitoring can detect inconsistencies +5. **Idempotent updates** — if a node receives a config update with the same or older epoch, it ignores it + +### New node startup sequence + +After `hive admin join` completes on the new node: + +```sh +./scripts/start-dev.sh # On the new node only +``` + +`start-dev.sh` reads the node's capabilities from `hive.toml` and starts only matching services: + +1. **NATS** (if `nats` capability) — joins existing mesh via routes in nats.conf +2. **Predastore** (if `predastore` capability) — starts storage node; existing nodes have already been restarted with updated config +3. **Predastore DB** (if `predastoredb` capability) — starts DB node; leader has already called `AddVoter()` so Raft accepts it +4. **Viperblock** (if `viperblock` capability) — starts local instance +5. **Daemon** (if `hive` capability) — connects to NATS, subscribes to topics, announces via heartbeat +6. **AWS Gateway** (if `awsgw` capability) — starts and discovers nodes via NATS fan-out +7. **UI** (if `hive-ui` capability) — starts web interface + +### Stage 2 implementation + +**Changes to `hive/formation/formation.go`:** +- Add `Stage2Join()` handler — validates node, updates config, returns credentials + full cluster data +- Add `BroadcastConfigUpdate(natsConn, clusterConfig)` — publishes update to `hive.cluster.config-update` NATS topic +- `FormationServer` gains `mode` field: `ModeFormation` (Stage 1) vs `ModeRunning` (Stage 2) +- In Stage 2 mode, the server has access to the NATS connection for broadcasting + +**Changes to `hive/daemon/daemon.go`:** +- On startup, start formation server in Stage 2 mode (replaces the existing `/join` handler) +- Remove old `/join` endpoint handler (lines 814-931) — replaced by formation server +- Add `handleConfigUpdate()` — NATS subscription on `hive.cluster.config-update`, applies received config +- Add `reloadNATS()` — regenerates nats.conf from current cluster config, sends SIGHUP to nats-server PID +- Add `reloadPredastore()` — regenerates predastore.toml, restarts predastore process +- Add `handleClusterHeartbeat()` — periodic epoch broadcast + stale detection + +**Changes to `hive/config/config.go`:** +- Add `Capabilities []string` to per-node `Config` struct +- Add `Capabilities []string` to `NodeJoinRequest` +- Add `HasCapability(cap string) bool` method on `Config` + +**Changes to `cmd/hive/cmd/admin.go`:** +- `runAdminJoin`: new `--capability` flag (comma-separated string, default all capabilities) +- Join detects Stage 1 vs Stage 2: if formation server responds with `mode: "formation"`, poll for completion (Stage 1); if `mode: "running"`, configs are returned immediately (Stage 2) + +**Changes to `hive/formation/helpers.go`:** +- `BuildClusterRoutes(nodes)` → `BuildClusterRoutes(nodes, capabilityFilter)` — only include nodes with `nats` capability +- `BuildPredastoreNodes(nodes)` → filter by `predastore` capability +- New: `BuildPredastoreDBNodes(nodes)` — filter by `predastoredb` capability +- New: `FilterNodesByCapability(nodes map[string]Config, cap string) []Config` + +**Changes to config templates:** +- `nats.conf` — routes only include nodes with `nats` capability +- `predastore-multinode.toml` — `[[nodes]]` only include `predastore` capable nodes; `[[db]]` only include `predastoredb` capable nodes + +**Changes to `scripts/start-dev.sh`:** +- Read capabilities from `hive.toml` for the local node +- Only start services matching the node's capabilities +- Skip predastore/predastoredb/viperblock/awsgw/hive-ui if not in capability list + +### Stage 2 files + +| File | Action | +|---|---| +| `hive/formation/formation.go` | Modify — add Stage 2 mode, Stage2Join handler, broadcast | +| `hive/formation/helpers.go` | Modify — capability-aware filtering for all config builders | +| `hive/formation/formation_test.go` | Modify — add Stage 2 tests | +| `hive/daemon/daemon.go` | Modify — replace `/join` with formation Stage 2, add config update + service reload handlers | +| `hive/config/config.go` | Modify — add Capabilities field to Config and NodeJoinRequest | +| `cmd/hive/cmd/admin.go` | Modify — add `--capability` flag, Stage 1 vs Stage 2 detection logic | +| `cmd/hive/cmd/templates/nats.conf` | Modify — capability-filtered routes | +| `cmd/hive/cmd/templates/predastore-multinode.toml` | Modify — capability-filtered `[[nodes]]` and `[[db]]` sections | +| `scripts/start-dev.sh` | Modify — capability-aware service startup | +| `hive/daemon/jetstream.go` | No changes (UpdateReplicas already handles dynamic scaling) | + +### Stage 2 testing + +**Unit tests:** +- Stage 2 join: validates capabilities, rejects duplicate node names and IPs, increments epoch +- Capability filtering: correct nodes appear in NATS routes and predastore config +- Config broadcast: serialization/deserialization of `hive.cluster.config-update` messages +- Service reload: NATS config regeneration produces valid config, predastore config includes correct node subsets +- Epoch-based sync: stale epoch detection triggers config request + +**Integration test:** +1. Form 3-node cluster via Stage 1 (all capabilities) +2. Start all 3 nodes, verify NATS mesh + Predastore Raft healthy +3. Run `hive admin join --node node4 --host 127.0.0.1:4432 --bind 127.0.0.4 --capability nats,viperblock,hive,awsgw` +4. Verify: node4 added to all nodes' `hive.toml` with correct capabilities +5. Verify: all `nats.conf` files updated with node4 route +6. Verify: `predastore.toml` NOT updated (node4 has no `predastore` capability) +7. Start node4 with `start-dev.sh`, verify it joins NATS mesh +8. Verify: AWS Gateway discovers node4 via `hive.nodes.discover` +9. Verify: predastore cluster unchanged (still 3 nodes) +10. Run `hive admin join --node node5 --host 127.0.0.1:4432 --bind 127.0.0.5 --capability nats,predastore` +11. Verify: `predastore.toml` updated on all nodes with node5 as storage node +12. Verify: predastore `[[db]]` section unchanged (node5 has no `predastoredb` capability) +13. Start node5, write new object via S3 API, verify shards distributed across 4 predastore nodes + +## Remote testing + +Integration tests above use localhost IPs for local simulation. For real multi-server testing, a 3-node cluster environment is available — see `docs/INTERNAL.md` for SSH access, deployment workflow (git diff → scp → git apply → make build), and cluster init/join commands with real IPs. + +The remote test flow: +1. Deploy code patches to all 3 nodes (INTERNAL.md: "Deploying code changes") +2. Fresh install or patch-only depending on scope (INTERNAL.md: "Testing workflows") +3. Run `hive admin init` on node 1, `hive admin join` on nodes 2 and 3 (INTERNAL.md: "Cluster init and join") +4. Start services on all nodes, verify NATS mesh, Predastore Raft, and daemon health +5. Run AWS CLI smoke test against the cluster endpoint diff --git a/hive/admin/admin.go b/hive/admin/admin.go index a20badcc3..fa07a6ff3 100644 --- a/hive/admin/admin.go +++ b/hive/admin/admin.go @@ -38,6 +38,7 @@ type ConfigSettings struct { // Cluster settings ClusterBindIP string ClusterRoutes []string + ClusterName string // Predastore multi-node PredastoreNodeID int @@ -516,8 +517,9 @@ func GenerateSelfSignedCert(certPath, keyPath string) error { return nil } -// setupAWSCredentials updates ~/.aws/credentials and ~/.aws/config -func SetupAWSCredentials(accessKey, secretKey, region, certPath string) error { +// SetupAWSCredentials updates ~/.aws/credentials and ~/.aws/config. +// bindIP is the IP the AWS gateway listens on. If empty or "0.0.0.0", defaults to "localhost". +func SetupAWSCredentials(accessKey, secretKey, region, certPath, bindIP string) error { homeDir, err := os.UserHomeDir() if err != nil { return err @@ -559,9 +561,14 @@ func SetupAWSCredentials(accessKey, secretKey, region, certPath string) error { configSection = "profile " + profileName } + endpointHost := bindIP + if endpointHost == "" || endpointHost == "0.0.0.0" { + endpointHost = "localhost" + } + if err := UpdateAWSINIFile(configPath, configSection, map[string]string{ "region": region, - "endpoint_url": "https://localhost:9999", + "endpoint_url": fmt.Sprintf("https://%s:9999", endpointHost), "ca_bundle": certPath, "output": "json", }); err != nil { diff --git a/hive/admin/admin_test.go b/hive/admin/admin_test.go index 70a40b214..e7f638bfb 100644 --- a/hive/admin/admin_test.go +++ b/hive/admin/admin_test.go @@ -209,7 +209,7 @@ func TestSetupAWSCredentials_CreatesFiles(t *testing.T) { dir := t.TempDir() t.Setenv("HOME", dir) - err := SetupAWSCredentials("AKIATEST123", "secret123", "us-east-1", "/path/to/ca.pem") + err := SetupAWSCredentials("AKIATEST123", "secret123", "us-east-1", "/path/to/ca.pem", "") require.NoError(t, err) credData, _ := os.ReadFile(filepath.Join(dir, ".aws", "credentials")) @@ -232,7 +232,7 @@ func TestSetupAWSCredentials_PreservesExistingProfiles(t *testing.T) { "aws_access_key_id": "EXISTING_KEY", })) - err := SetupAWSCredentials("NEWAKEY", "NEWSECRET", "us-west-2", "/ca.pem") + err := SetupAWSCredentials("NEWAKEY", "NEWSECRET", "us-west-2", "/ca.pem", "") require.NoError(t, err) data, _ := os.ReadFile(filepath.Join(awsDir, "credentials")) @@ -241,6 +241,29 @@ func TestSetupAWSCredentials_PreservesExistingProfiles(t *testing.T) { assert.Contains(t, content, "NEWAKEY") } +func TestSetupAWSCredentials_UsesBindIP(t *testing.T) { + dir := t.TempDir() + t.Setenv("HOME", dir) + + err := SetupAWSCredentials("AKIATEST123", "secret123", "us-east-1", "/ca.pem", "10.11.12.1") + require.NoError(t, err) + + configData, _ := os.ReadFile(filepath.Join(dir, ".aws", "config")) + assert.Contains(t, string(configData), "https://10.11.12.1:9999") + assert.NotContains(t, string(configData), "localhost") +} + +func TestSetupAWSCredentials_FallsBackToLocalhostForWildcard(t *testing.T) { + dir := t.TempDir() + t.Setenv("HOME", dir) + + err := SetupAWSCredentials("AKIATEST123", "secret123", "us-east-1", "/ca.pem", "0.0.0.0") + require.NoError(t, err) + + configData, _ := os.ReadFile(filepath.Join(dir, ".aws", "config")) + assert.Contains(t, string(configData), "https://localhost:9999") +} + // --- Certificate generation --- func TestGenerateCACert_CreatesValidCA(t *testing.T) { diff --git a/hive/daemon/daemon.go b/hive/daemon/daemon.go index 45a9bd056..00020149c 100644 --- a/hive/daemon/daemon.go +++ b/hive/daemon/daemon.go @@ -1071,7 +1071,7 @@ func (d *Daemon) stopInstance(instances map[string]*vm.VM, deleteVolume bool) er continue } - msg, err := d.natsConn.Request("ebs.unmount", ebsUnMountRequest, 30*time.Second) + msg, err := d.natsConn.Request(d.ebsTopic("unmount"), ebsUnMountRequest, 30*time.Second) if err != nil { slog.Error("Failed to unmount volume", "name", ebsRequest.Name, "id", instance.ID, "err", err) } else { @@ -1629,6 +1629,13 @@ func (d *Daemon) StartInstance(instance *vm.VM) error { return nil } +// ebsTopic returns a node-specific EBS NATS topic, e.g. "ebs.node1.mount". +// This ensures mount/unmount requests are routed to the viperblock instance +// running on the same node as the daemon (NBD sockets are local). +func (d *Daemon) ebsTopic(action string) string { + return fmt.Sprintf("ebs.%s.%s", d.node, action) +} + // MountVolumes mounts the volumes for an instance func (d *Daemon) MountVolumes(instance *vm.VM) error { @@ -1643,7 +1650,7 @@ func (d *Daemon) MountVolumes(instance *vm.VM) error { return err } - reply, err := d.natsConn.Request("ebs.mount", ebsMountRequest, 30*time.Second) + reply, err := d.natsConn.Request(d.ebsTopic("mount"), ebsMountRequest, 30*time.Second) slog.Info("Mounting volume", "Vol", v.Name, "NBDURI", v.NBDURI) @@ -1690,7 +1697,7 @@ func (d *Daemon) rollbackEBSMount(req config.EBSRequest) { slog.Error("rollbackEBSMount: failed to marshal unmount request", "volume", req.Name, "err", err) return } - msg, err := d.natsConn.Request("ebs.unmount", data, 10*time.Second) + msg, err := d.natsConn.Request(d.ebsTopic("unmount"), data, 10*time.Second) if err != nil { slog.Error("rollbackEBSMount: ebs.unmount NATS request failed", "volume", req.Name, "err", err) return diff --git a/hive/daemon/daemon_handlers.go b/hive/daemon/daemon_handlers.go index 7196d4720..1275603c4 100644 --- a/hive/daemon/daemon_handlers.go +++ b/hive/daemon/daemon_handlers.go @@ -157,7 +157,7 @@ func (d *Daemon) handleAttachVolume(msg *nats.Msg, command qmp.Command, instance return } - mountReply, err := d.natsConn.Request("ebs.mount", ebsMountData, 30*time.Second) + mountReply, err := d.natsConn.Request(d.ebsTopic("mount"), ebsMountData, 30*time.Second) if err != nil { slog.Error("AttachVolume: ebs.mount failed", "volumeId", volumeID, "err", err) respondWithError(awserrors.ErrorServerInternal) diff --git a/hive/daemon/daemon_test.go b/hive/daemon/daemon_test.go index 43f62e204..9475e511b 100644 --- a/hive/daemon/daemon_test.go +++ b/hive/daemon/daemon_test.go @@ -1577,7 +1577,7 @@ func TestStopInstance_DeleteOnTermination_VolumeDeletion(t *testing.T) { ebsDeletedVolumes := make(map[string]bool) // Mock ebs.unmount subscriber - unmountSub, err := daemon.natsConn.Subscribe("ebs.unmount", func(msg *nats.Msg) { + unmountSub, err := daemon.natsConn.Subscribe("ebs.node-1.unmount", func(msg *nats.Msg) { var req config.EBSRequest json.Unmarshal(msg.Data, &req) mu.Lock() @@ -1674,7 +1674,7 @@ func TestStopInstance_DeleteOnTermination_False_SkipsVolumeDeletion(t *testing.T ebsDeletedVolumes := make(map[string]bool) // Mock ebs.unmount subscriber - unmountSub, err := daemon.natsConn.Subscribe("ebs.unmount", func(msg *nats.Msg) { + unmountSub, err := daemon.natsConn.Subscribe("ebs.node-1.unmount", func(msg *nats.Msg) { var req config.EBSRequest json.Unmarshal(msg.Data, &req) mu.Lock() @@ -1771,7 +1771,7 @@ func TestStopInstance_NoDelete_OnStop(t *testing.T) { ebsDeletedVolumes := make(map[string]bool) // Mock ebs.unmount subscriber - unmountSub, err := daemon.natsConn.Subscribe("ebs.unmount", func(msg *nats.Msg) { + unmountSub, err := daemon.natsConn.Subscribe("ebs.node-1.unmount", func(msg *nats.Msg) { resp := config.EBSUnMountResponse{Mounted: false} data, _ := json.Marshal(resp) msg.Respond(data) @@ -2432,7 +2432,7 @@ func TestDetachVolume_SuccessPath(t *testing.T) { // Subscribe a mock ebs.unmount handler ebsUnmountCalled := make(chan string, 1) - ebsSub, err := daemon.natsConn.Subscribe("ebs.unmount", func(msg *nats.Msg) { + ebsSub, err := daemon.natsConn.Subscribe("ebs.node-1.unmount", func(msg *nats.Msg) { var req config.EBSRequest json.Unmarshal(msg.Data, &req) ebsUnmountCalled <- req.Name @@ -2573,7 +2573,7 @@ func TestDetachVolume_ForceFlag(t *testing.T) { daemon.Instances.VMS[instanceID] = instance // Mock ebs.unmount - ebsSub, err := daemon.natsConn.Subscribe("ebs.unmount", func(msg *nats.Msg) { + ebsSub, err := daemon.natsConn.Subscribe("ebs.node-1.unmount", func(msg *nats.Msg) { resp := config.EBSUnMountResponse{Mounted: false} data, _ := json.Marshal(resp) msg.Respond(data) @@ -2776,7 +2776,7 @@ func TestDetachVolume_SuccessWithDeviceMatch(t *testing.T) { } daemon.Instances.VMS[instanceID] = instance - ebsSub, err := daemon.natsConn.Subscribe("ebs.unmount", func(msg *nats.Msg) { + ebsSub, err := daemon.natsConn.Subscribe("ebs.node-1.unmount", func(msg *nats.Msg) { resp := config.EBSUnMountResponse{Mounted: false} data, _ := json.Marshal(resp) msg.Respond(data) @@ -2852,7 +2852,7 @@ func TestAttachVolume_ReplacesStaleEBSRequest(t *testing.T) { daemon.Instances.VMS[instanceID] = instance // Mock ebs.mount to return success with a new NBDURI - ebsSub, err := daemon.natsConn.Subscribe("ebs.mount", func(msg *nats.Msg) { + ebsSub, err := daemon.natsConn.Subscribe("ebs.node-1.mount", func(msg *nats.Msg) { resp := config.EBSMountResponse{URI: "nbd://new:2222"} data, _ := json.Marshal(resp) msg.Respond(data) @@ -3333,7 +3333,7 @@ func TestRollbackEBSMount_Success(t *testing.T) { unmountCalled := make(chan string, 1) // Mock ebs.unmount subscriber that returns success - sub, err := daemon.natsConn.Subscribe("ebs.unmount", func(msg *nats.Msg) { + sub, err := daemon.natsConn.Subscribe("ebs.node-1.unmount", func(msg *nats.Msg) { var req config.EBSRequest json.Unmarshal(msg.Data, &req) unmountCalled <- req.Name @@ -3367,7 +3367,7 @@ func TestRollbackEBSMount_UnmountError(t *testing.T) { daemon := createTestDaemon(t, natsURL) // Mock ebs.unmount subscriber that returns an error - sub, err := daemon.natsConn.Subscribe("ebs.unmount", func(msg *nats.Msg) { + sub, err := daemon.natsConn.Subscribe("ebs.node-1.unmount", func(msg *nats.Msg) { resp := config.EBSUnMountResponse{Error: "unmount failed: device busy"} data, _ := json.Marshal(resp) msg.Respond(data) @@ -3388,7 +3388,7 @@ func TestRollbackEBSMount_StillMounted(t *testing.T) { daemon := createTestDaemon(t, natsURL) - sub, err := daemon.natsConn.Subscribe("ebs.unmount", func(msg *nats.Msg) { + sub, err := daemon.natsConn.Subscribe("ebs.node-1.unmount", func(msg *nats.Msg) { resp := config.EBSUnMountResponse{Mounted: true} // still mounted data, _ := json.Marshal(resp) msg.Respond(data) diff --git a/hive/formation/formation.go b/hive/formation/formation.go new file mode 100644 index 000000000..2f5cb04f3 --- /dev/null +++ b/hive/formation/formation.go @@ -0,0 +1,253 @@ +package formation + +import ( + "context" + "encoding/json" + "fmt" + "log/slog" + "net" + "net/http" + "sync" + "time" +) + +// NodeInfo describes a node participating in cluster formation. +type NodeInfo struct { + Name string `json:"name"` + BindIP string `json:"bind_ip"` + ClusterIP string `json:"cluster_ip"` + Region string `json:"region"` + AZ string `json:"az"` + Port int `json:"port"` +} + +// JoinRequest is the payload POSTed by joining nodes. +type JoinRequest struct { + NodeInfo +} + +// JoinResponse is returned by the formation server on join. +type JoinResponse struct { + Success bool `json:"success"` + Message string `json:"message"` + Joined int `json:"joined"` + Expected int `json:"expected"` +} + +// StatusResponse is returned by the formation server status endpoint. +type StatusResponse struct { + Complete bool `json:"complete"` + Joined int `json:"joined"` + Expected int `json:"expected"` + Nodes map[string]NodeInfo `json:"nodes,omitempty"` + Credentials *SharedCredentials `json:"credentials,omitempty"` + CACert string `json:"ca_cert,omitempty"` + CAKey string `json:"ca_key,omitempty"` +} + +// SharedCredentials contains the cluster-wide credentials distributed during formation. +type SharedCredentials struct { + AccessKey string `json:"access_key"` + SecretKey string `json:"secret_key"` + AccountID string `json:"account_id"` + NatsToken string `json:"nats_token"` + ClusterName string `json:"cluster_name"` + Region string `json:"region"` +} + +// FormationServer is a lightweight HTTP server that coordinates cluster formation. +// Nodes register themselves via POST /formation/join. Once the expected number of +// nodes have joined, the done channel is closed and full cluster data (credentials, +// CA, node list) becomes available via GET /formation/status. +type FormationServer struct { + mu sync.RWMutex + expected int + nodes map[string]NodeInfo + credentials *SharedCredentials + caCert string + caKey string + done chan struct{} + server *http.Server +} + +// NewFormationServer creates a new formation server expecting the given number of nodes. +func NewFormationServer(expected int, creds *SharedCredentials, caCert, caKey string) *FormationServer { + return &FormationServer{ + expected: expected, + nodes: make(map[string]NodeInfo), + credentials: creds, + caCert: caCert, + caKey: caKey, + done: make(chan struct{}), + } +} + +// RegisterNode validates and adds a node. Returns an error for duplicates. +func (fs *FormationServer) RegisterNode(info NodeInfo) error { + fs.mu.Lock() + defer fs.mu.Unlock() + + if info.Name == "" { + return fmt.Errorf("node name is required") + } + if info.BindIP == "" { + return fmt.Errorf("bind_ip is required") + } + + // Check for duplicate name + if _, exists := fs.nodes[info.Name]; exists { + return fmt.Errorf("node %q already registered", info.Name) + } + + // Check for duplicate bind IP + for _, n := range fs.nodes { + if n.BindIP == info.BindIP { + return fmt.Errorf("bind IP %s already registered by node %q", info.BindIP, n.Name) + } + } + + fs.nodes[info.Name] = info + slog.Info("Node registered", "name", info.Name, "bind_ip", info.BindIP, "joined", len(fs.nodes), "expected", fs.expected) + + if fs.isComplete() { + close(fs.done) + } + + return nil +} + +// isComplete returns true when we have enough nodes. Must be called with lock held. +func (fs *FormationServer) isComplete() bool { + return len(fs.nodes) >= fs.expected +} + +// IsComplete returns true when the expected number of nodes have registered. +func (fs *FormationServer) IsComplete() bool { + fs.mu.RLock() + defer fs.mu.RUnlock() + return fs.isComplete() +} + +// WaitForCompletion blocks until all nodes have joined or the timeout fires. +func (fs *FormationServer) WaitForCompletion(timeout time.Duration) error { + select { + case <-fs.done: + return nil + case <-time.After(timeout): + fs.mu.RLock() + joined := len(fs.nodes) + fs.mu.RUnlock() + return fmt.Errorf("formation timed out after %s: %d/%d nodes joined", timeout, joined, fs.expected) + } +} + +// Nodes returns a copy of the registered nodes. +func (fs *FormationServer) Nodes() map[string]NodeInfo { + fs.mu.RLock() + defer fs.mu.RUnlock() + out := make(map[string]NodeInfo, len(fs.nodes)) + for k, v := range fs.nodes { + out[k] = v + } + return out +} + +// Start launches the HTTP server on the given address (e.g. "10.0.0.1:4432"). +func (fs *FormationServer) Start(bindAddr string) error { + mux := http.NewServeMux() + mux.HandleFunc("POST /formation/join", fs.handleJoin) + mux.HandleFunc("GET /formation/status", fs.handleStatus) + mux.HandleFunc("GET /formation/health", fs.handleHealth) + + fs.server = &http.Server{ + Addr: bindAddr, + Handler: mux, + } + + ln, err := net.Listen("tcp", bindAddr) + if err != nil { + return fmt.Errorf("formation server listen: %w", err) + } + + go func() { + if err := fs.server.Serve(ln); err != nil && err != http.ErrServerClosed { + slog.Error("Formation server error", "error", err) + } + }() + + slog.Info("Formation server started", "addr", bindAddr) + return nil +} + +// Shutdown gracefully stops the formation server. +func (fs *FormationServer) Shutdown(ctx context.Context) error { + if fs.server == nil { + return nil + } + return fs.server.Shutdown(ctx) +} + +func (fs *FormationServer) handleJoin(w http.ResponseWriter, r *http.Request) { + var req JoinRequest + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + writeJSON(w, http.StatusBadRequest, JoinResponse{ + Success: false, + Message: fmt.Sprintf("invalid request body: %v", err), + }) + return + } + + if err := fs.RegisterNode(req.NodeInfo); err != nil { + writeJSON(w, http.StatusConflict, JoinResponse{ + Success: false, + Message: err.Error(), + }) + return + } + + fs.mu.RLock() + joined := len(fs.nodes) + fs.mu.RUnlock() + + writeJSON(w, http.StatusOK, JoinResponse{ + Success: true, + Message: fmt.Sprintf("node %q registered", req.Name), + Joined: joined, + Expected: fs.expected, + }) +} + +func (fs *FormationServer) handleStatus(w http.ResponseWriter, r *http.Request) { + fs.mu.RLock() + defer fs.mu.RUnlock() + + resp := StatusResponse{ + Complete: fs.isComplete(), + Joined: len(fs.nodes), + Expected: fs.expected, + } + + // Only expose full data when formation is complete + if fs.isComplete() { + resp.Nodes = make(map[string]NodeInfo, len(fs.nodes)) + for k, v := range fs.nodes { + resp.Nodes[k] = v + } + resp.Credentials = fs.credentials + resp.CACert = fs.caCert + resp.CAKey = fs.caKey + } + + writeJSON(w, http.StatusOK, resp) +} + +func (fs *FormationServer) handleHealth(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusOK) + w.Write([]byte("ok")) +} + +func writeJSON(w http.ResponseWriter, status int, v any) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(status) + json.NewEncoder(w).Encode(v) +} diff --git a/hive/formation/formation_test.go b/hive/formation/formation_test.go new file mode 100644 index 000000000..78a1247b9 --- /dev/null +++ b/hive/formation/formation_test.go @@ -0,0 +1,313 @@ +package formation + +import ( + "bytes" + "encoding/json" + "net/http" + "net/http/httptest" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func testCreds() *SharedCredentials { + return &SharedCredentials{ + AccessKey: "AKIATEST1234567890AB", + SecretKey: "testSecretKey1234567890", + AccountID: "123456789012", + NatsToken: "nats_testtoken123456", + ClusterName: "hive", + Region: "us-west-1", + } +} + +func testNode(name, ip string) NodeInfo { + return NodeInfo{ + Name: name, + BindIP: ip, + Region: "us-west-1", + AZ: "us-west-1a", + Port: 4432, + } +} + +func TestNewFormationServer(t *testing.T) { + t.Parallel() + fs := NewFormationServer(3, testCreds(), "ca-cert-pem", "ca-key-pem") + + assert.Equal(t, 3, fs.expected) + assert.Empty(t, fs.nodes) + assert.False(t, fs.IsComplete()) +} + +func TestRegisterNode(t *testing.T) { + t.Parallel() + fs := NewFormationServer(3, testCreds(), "", "") + + err := fs.RegisterNode(testNode("node1", "10.0.0.1")) + require.NoError(t, err) + + nodes := fs.Nodes() + assert.Len(t, nodes, 1) + assert.Equal(t, "10.0.0.1", nodes["node1"].BindIP) +} + +func TestRegisterDuplicateName(t *testing.T) { + t.Parallel() + fs := NewFormationServer(3, testCreds(), "", "") + + require.NoError(t, fs.RegisterNode(testNode("node1", "10.0.0.1"))) + err := fs.RegisterNode(testNode("node1", "10.0.0.2")) + assert.Error(t, err) + assert.Contains(t, err.Error(), "already registered") +} + +func TestRegisterDuplicateIP(t *testing.T) { + t.Parallel() + fs := NewFormationServer(3, testCreds(), "", "") + + require.NoError(t, fs.RegisterNode(testNode("node1", "10.0.0.1"))) + err := fs.RegisterNode(testNode("node2", "10.0.0.1")) + assert.Error(t, err) + assert.Contains(t, err.Error(), "bind IP") +} + +func TestIsComplete(t *testing.T) { + t.Parallel() + fs := NewFormationServer(2, testCreds(), "", "") + + assert.False(t, fs.IsComplete()) + require.NoError(t, fs.RegisterNode(testNode("node1", "10.0.0.1"))) + assert.False(t, fs.IsComplete()) + require.NoError(t, fs.RegisterNode(testNode("node2", "10.0.0.2"))) + assert.True(t, fs.IsComplete()) +} + +func TestWaitForCompletion(t *testing.T) { + t.Parallel() + fs := NewFormationServer(2, testCreds(), "", "") + + go func() { + time.Sleep(50 * time.Millisecond) + fs.RegisterNode(testNode("node1", "10.0.0.1")) + fs.RegisterNode(testNode("node2", "10.0.0.2")) + }() + + err := fs.WaitForCompletion(5 * time.Second) + require.NoError(t, err) + assert.True(t, fs.IsComplete()) +} + +func TestWaitForCompletionTimeout(t *testing.T) { + t.Parallel() + fs := NewFormationServer(3, testCreds(), "", "") + + require.NoError(t, fs.RegisterNode(testNode("node1", "10.0.0.1"))) + + err := fs.WaitForCompletion(100 * time.Millisecond) + assert.Error(t, err) + assert.Contains(t, err.Error(), "timed out") +} + +func testServer(t *testing.T, fs *FormationServer) *httptest.Server { + t.Helper() + mux := http.NewServeMux() + mux.HandleFunc("POST /formation/join", fs.handleJoin) + mux.HandleFunc("GET /formation/status", fs.handleStatus) + mux.HandleFunc("GET /formation/health", fs.handleHealth) + return httptest.NewServer(mux) +} + +func TestJoinEndpoint(t *testing.T) { + t.Parallel() + fs := NewFormationServer(3, testCreds(), "", "") + ts := testServer(t, fs) + defer ts.Close() + + body, _ := json.Marshal(JoinRequest{NodeInfo: testNode("node1", "10.0.0.1")}) + resp, err := http.Post(ts.URL+"/formation/join", "application/json", bytes.NewReader(body)) + require.NoError(t, err) + defer resp.Body.Close() + + assert.Equal(t, http.StatusOK, resp.StatusCode) + + var jr JoinResponse + require.NoError(t, json.NewDecoder(resp.Body).Decode(&jr)) + assert.True(t, jr.Success) + assert.Equal(t, 1, jr.Joined) + assert.Equal(t, 3, jr.Expected) +} + +func TestJoinEndpointDuplicate(t *testing.T) { + t.Parallel() + fs := NewFormationServer(3, testCreds(), "", "") + ts := testServer(t, fs) + defer ts.Close() + + body, _ := json.Marshal(JoinRequest{NodeInfo: testNode("node1", "10.0.0.1")}) + + resp1, err := http.Post(ts.URL+"/formation/join", "application/json", bytes.NewReader(body)) + require.NoError(t, err) + resp1.Body.Close() + assert.Equal(t, http.StatusOK, resp1.StatusCode) + + resp2, err := http.Post(ts.URL+"/formation/join", "application/json", bytes.NewReader(body)) + require.NoError(t, err) + defer resp2.Body.Close() + assert.Equal(t, http.StatusConflict, resp2.StatusCode) +} + +func TestStatusEndpointIncomplete(t *testing.T) { + t.Parallel() + fs := NewFormationServer(3, testCreds(), "ca-cert", "ca-key") + ts := testServer(t, fs) + defer ts.Close() + + require.NoError(t, fs.RegisterNode(testNode("node1", "10.0.0.1"))) + + resp, err := http.Get(ts.URL + "/formation/status") + require.NoError(t, err) + defer resp.Body.Close() + + var sr StatusResponse + require.NoError(t, json.NewDecoder(resp.Body).Decode(&sr)) + assert.False(t, sr.Complete) + assert.Equal(t, 1, sr.Joined) + assert.Equal(t, 3, sr.Expected) + assert.Nil(t, sr.Credentials) + assert.Empty(t, sr.CACert) + assert.Nil(t, sr.Nodes) +} + +func TestStatusEndpointComplete(t *testing.T) { + t.Parallel() + creds := testCreds() + fs := NewFormationServer(2, creds, "ca-cert-data", "ca-key-data") + ts := testServer(t, fs) + defer ts.Close() + + require.NoError(t, fs.RegisterNode(testNode("node1", "10.0.0.1"))) + require.NoError(t, fs.RegisterNode(testNode("node2", "10.0.0.2"))) + + resp, err := http.Get(ts.URL + "/formation/status") + require.NoError(t, err) + defer resp.Body.Close() + + var sr StatusResponse + require.NoError(t, json.NewDecoder(resp.Body).Decode(&sr)) + assert.True(t, sr.Complete) + assert.Equal(t, 2, sr.Joined) + assert.Equal(t, 2, sr.Expected) + assert.Len(t, sr.Nodes, 2) + require.NotNil(t, sr.Credentials) + assert.Equal(t, creds.AccessKey, sr.Credentials.AccessKey) + assert.Equal(t, "ca-cert-data", sr.CACert) + assert.Equal(t, "ca-key-data", sr.CAKey) +} + +func TestHealthEndpoint(t *testing.T) { + t.Parallel() + fs := NewFormationServer(1, testCreds(), "", "") + ts := testServer(t, fs) + defer ts.Close() + + resp, err := http.Get(ts.URL + "/formation/health") + require.NoError(t, err) + defer resp.Body.Close() + + assert.Equal(t, http.StatusOK, resp.StatusCode) +} + +func TestBuildClusterRoutes(t *testing.T) { + t.Parallel() + nodes := map[string]NodeInfo{ + "node3": {Name: "node3", BindIP: "10.0.0.3"}, + "node1": {Name: "node1", BindIP: "10.0.0.1"}, + "node2": {Name: "node2", BindIP: "10.0.0.2"}, + } + + routes := BuildClusterRoutes(nodes) + assert.Equal(t, []string{"10.0.0.1:4248", "10.0.0.2:4248", "10.0.0.3:4248"}, routes) +} + +func TestBuildClusterRoutesUsesClusterIP(t *testing.T) { + t.Parallel() + nodes := map[string]NodeInfo{ + "node1": {Name: "node1", BindIP: "10.0.0.1", ClusterIP: "192.168.1.1"}, + "node2": {Name: "node2", BindIP: "10.0.0.2"}, // no cluster IP, fallback to bind + } + + routes := BuildClusterRoutes(nodes) + assert.Equal(t, []string{"192.168.1.1:4248", "10.0.0.2:4248"}, routes) +} + +func TestBuildPredastoreNodes(t *testing.T) { + t.Parallel() + nodes := map[string]NodeInfo{ + "node3": {Name: "node3", BindIP: "10.0.0.3"}, + "node1": {Name: "node1", BindIP: "10.0.0.1"}, + "node2": {Name: "node2", BindIP: "10.0.0.2"}, + } + + pnodes := BuildPredastoreNodes(nodes) + require.Len(t, pnodes, 3) + assert.Equal(t, 1, pnodes[0].ID) + assert.Equal(t, "10.0.0.1", pnodes[0].Host) + assert.Equal(t, 2, pnodes[1].ID) + assert.Equal(t, "10.0.0.2", pnodes[1].Host) + assert.Equal(t, 3, pnodes[2].ID) + assert.Equal(t, "10.0.0.3", pnodes[2].Host) +} + +func TestFullFormationFlow(t *testing.T) { + t.Parallel() + creds := testCreds() + fs := NewFormationServer(3, creds, "-----BEGIN CERTIFICATE-----\ntest\n-----END CERTIFICATE-----", "-----BEGIN PRIVATE KEY-----\ntest\n-----END PRIVATE KEY-----") + ts := testServer(t, fs) + defer ts.Close() + + // 3 nodes join sequentially + for i, n := range []struct{ name, ip string }{ + {"node1", "10.0.0.1"}, + {"node2", "10.0.0.2"}, + {"node3", "10.0.0.3"}, + } { + body, _ := json.Marshal(JoinRequest{NodeInfo: testNode(n.name, n.ip)}) + resp, err := http.Post(ts.URL+"/formation/join", "application/json", bytes.NewReader(body)) + require.NoError(t, err) + + var jr JoinResponse + require.NoError(t, json.NewDecoder(resp.Body).Decode(&jr)) + resp.Body.Close() + + assert.True(t, jr.Success) + assert.Equal(t, i+1, jr.Joined) + } + + // All nodes poll status — should be complete + resp, err := http.Get(ts.URL + "/formation/status") + require.NoError(t, err) + defer resp.Body.Close() + + var sr StatusResponse + require.NoError(t, json.NewDecoder(resp.Body).Decode(&sr)) + assert.True(t, sr.Complete) + assert.Equal(t, 3, sr.Joined) + assert.Len(t, sr.Nodes, 3) + require.NotNil(t, sr.Credentials) + assert.Equal(t, creds.AccessKey, sr.Credentials.AccessKey) + assert.Equal(t, creds.SecretKey, sr.Credentials.SecretKey) + assert.Equal(t, creds.NatsToken, sr.Credentials.NatsToken) + assert.NotEmpty(t, sr.CACert) + assert.NotEmpty(t, sr.CAKey) + + // Verify helper outputs + routes := BuildClusterRoutes(sr.Nodes) + assert.Len(t, routes, 3) + + pnodes := BuildPredastoreNodes(sr.Nodes) + assert.Len(t, pnodes, 3) + assert.Equal(t, 1, pnodes[0].ID) +} diff --git a/hive/formation/helpers.go b/hive/formation/helpers.go new file mode 100644 index 000000000..9a0fc050c --- /dev/null +++ b/hive/formation/helpers.go @@ -0,0 +1,52 @@ +package formation + +import ( + "sort" + + "github.com/mulgadc/hive/hive/admin" +) + +// BuildClusterRoutes returns a sorted list of "IP:4248" NATS cluster routes +// for each node. If ClusterIP is set it is used, otherwise BindIP. +// Nodes are sorted by name for deterministic ordering. +func BuildClusterRoutes(nodes map[string]NodeInfo) []string { + sorted := sortedNodes(nodes) + routes := make([]string, len(sorted)) + for i, n := range sorted { + ip := n.ClusterIP + if ip == "" { + ip = n.BindIP + } + routes[i] = ip + ":4248" + } + return routes +} + +// BuildPredastoreNodes returns a sorted list of PredastoreNodeConfig with +// 1-based IDs. Nodes are sorted by name for deterministic ordering. +func BuildPredastoreNodes(nodes map[string]NodeInfo) []admin.PredastoreNodeConfig { + sorted := sortedNodes(nodes) + out := make([]admin.PredastoreNodeConfig, len(sorted)) + for i, n := range sorted { + out[i] = admin.PredastoreNodeConfig{ + ID: i + 1, + Host: n.BindIP, + } + } + return out +} + +// sortedNodes returns nodes sorted by name. +func sortedNodes(nodes map[string]NodeInfo) []NodeInfo { + names := make([]string, 0, len(nodes)) + for name := range nodes { + names = append(names, name) + } + sort.Strings(names) + + sorted := make([]NodeInfo, len(names)) + for i, name := range names { + sorted[i] = nodes[name] + } + return sorted +} diff --git a/hive/services/viperblockd/viperblockd.go b/hive/services/viperblockd/viperblockd.go index 58cbd3bf7..09fb175ad 100644 --- a/hive/services/viperblockd/viperblockd.go +++ b/hive/services/viperblockd/viperblockd.go @@ -47,6 +47,11 @@ type Config struct { SecretKey string BaseDir string + // NodeName identifies this node in the cluster (e.g. "node1"). + // Used for node-specific NATS topics: ebs.{NodeName}.mount / ebs.{NodeName}.unmount. + // If empty, falls back to generic ebs.mount / ebs.unmount with queue group (single-node compat). + NodeName string + // NBDTransport controls the transport type: "socket" (default) or "tcp" // Socket is faster for local connections, TCP required for remote/DPU scenarios NBDTransport config.NBDTransport @@ -150,7 +155,11 @@ func launchService(cfg *Config) (err error) { return err } - slog.Info("Waiting for EBS events") + if cfg.NodeName != "" { + slog.Info("Waiting for EBS events", "node", cfg.NodeName) + } else { + slog.Info("Waiting for EBS events (single-node mode)") + } if _, err := nc.QueueSubscribe("ebs.delete", "hive-workers", func(msg *nats.Msg) { slog.Info("Received ebs.delete message", "data", string(msg.Data)) @@ -227,7 +236,18 @@ func launchService(cfg *Config) (err error) { slog.Error("Failed to subscribe to ebs.delete", "err", err) } - if _, err := nc.QueueSubscribe("ebs.unmount", "hive-workers", func(msg *nats.Msg) { + // Subscribe to node-specific unmount topic if NodeName is set, otherwise fall back to generic queue group + unmountTopic := "ebs.unmount" + if cfg.NodeName != "" { + unmountTopic = fmt.Sprintf("ebs.%s.unmount", cfg.NodeName) + } + unmountSubscribe := func(topic string, handler nats.MsgHandler) (*nats.Subscription, error) { + if cfg.NodeName != "" { + return nc.Subscribe(topic, handler) + } + return nc.QueueSubscribe(topic, "hive-workers", handler) + } + if _, err := unmountSubscribe(unmountTopic, func(msg *nats.Msg) { slog.Info("Received message", "data", string(msg.Data)) // Parse the message @@ -308,7 +328,7 @@ func launchService(cfg *Config) (err error) { slog.Error("Failed to publish ebs.unmount.response", "err", err) } }); err != nil { - slog.Error("Failed to subscribe to ebs.unmount", "err", err) + slog.Error("Failed to subscribe to unmount topic", "topic", unmountTopic, "err", err) } if _, err := nc.QueueSubscribe("ebs.sync", "hive-workers", func(msg *nats.Msg) { @@ -369,7 +389,18 @@ func launchService(cfg *Config) (err error) { // subscribed at mount time and unsubscribed at unmount time. This ensures // snapshot requests are routed to the node that owns the volume. - if _, err := nc.QueueSubscribe("ebs.mount", "hive-workers", func(msg *nats.Msg) { + // Subscribe to node-specific mount topic if NodeName is set, otherwise fall back to generic queue group + mountTopic := "ebs.mount" + if cfg.NodeName != "" { + mountTopic = fmt.Sprintf("ebs.%s.mount", cfg.NodeName) + } + mountSubscribe := func(topic string, handler nats.MsgHandler) (*nats.Subscription, error) { + if cfg.NodeName != "" { + return nc.Subscribe(topic, handler) + } + return nc.QueueSubscribe(topic, "hive-workers", handler) + } + if _, err := mountSubscribe(mountTopic, func(msg *nats.Msg) { slog.Info("Received message:", "data", string(msg.Data)) // Parse the message @@ -671,7 +702,7 @@ func launchService(cfg *Config) (err error) { slog.Debug("Sent ebs.mount response", "response", string(response)) }); err != nil { - slog.Error("Failed to subscribe to ebs.mount", "err", err) + slog.Error("Failed to subscribe to mount topic", "topic", mountTopic, "err", err) } // Create a channel to receive shutdown signals diff --git a/hive/services/viperblockd/viperblockd_handlers_test.go b/hive/services/viperblockd/viperblockd_handlers_test.go index ef6d9f8f2..ccacb3269 100644 --- a/hive/services/viperblockd/viperblockd_handlers_test.go +++ b/hive/services/viperblockd/viperblockd_handlers_test.go @@ -251,7 +251,7 @@ func TestIntegration_EBSUnmountRemovesSocket(t *testing.T) { defer nc.Close() reqData, _ := json.Marshal(config.EBSRequest{Name: "vol-unmount-socket"}) - msg, err := nc.Request("ebs.unmount", reqData, 3*time.Second) + msg, err := nc.Request("ebs.test-node.unmount", reqData, 3*time.Second) require.NoError(t, err) var resp config.EBSUnMountResponse diff --git a/hive/services/viperblockd/viperblockd_integration_test.go b/hive/services/viperblockd/viperblockd_integration_test.go index ec7c8ebe6..4de0d353f 100644 --- a/hive/services/viperblockd/viperblockd_integration_test.go +++ b/hive/services/viperblockd/viperblockd_integration_test.go @@ -56,6 +56,7 @@ func setupTestConfig(t *testing.T, natsURL string) *Config { PluginPath: filepath.Join(testDir, "plugins"), // Use temp dir to avoid actual nbdkit execution Debug: true, MountedVolumes: []MountedVolume{}, + NodeName: "test-node", } return cfg @@ -160,7 +161,7 @@ func TestIntegration_EBSMountRequest(t *testing.T) { assert.NoError(t, err) // Use Request instead of Publish to get direct response - msg, err := nc.Request("ebs.mount", requestData, 5*time.Second) + msg, err := nc.Request("ebs.test-node.mount", requestData, 5*time.Second) // We expect to get a response (even if it contains an error) if err != nil { @@ -230,7 +231,7 @@ func TestIntegration_EBSUnmountRequest(t *testing.T) { assert.NoError(t, err) // Send unmount request - msg, err := nc.Request("ebs.unmount", requestData, 3*time.Second) + msg, err := nc.Request("ebs.test-node.unmount", requestData, 3*time.Second) assert.NoError(t, err) assert.NotNil(t, msg) @@ -287,7 +288,7 @@ func TestIntegration_EBSUnmountNonExistentVolume(t *testing.T) { assert.NoError(t, err) // Send unmount request - msg, err := nc.Request("ebs.unmount", requestData, 3*time.Second) + msg, err := nc.Request("ebs.test-node.unmount", requestData, 3*time.Second) assert.NoError(t, err) assert.NotNil(t, msg) @@ -347,7 +348,7 @@ func TestIntegration_ConcurrentMountRequests(t *testing.T) { assert.NoError(t, err) // Use Request to get synchronous response - msg, err := nc.Request("ebs.mount", requestData, 5*time.Second) + msg, err := nc.Request("ebs.test-node.mount", requestData, 5*time.Second) if err != nil { // Timeout or no response - acceptable for this test @@ -400,7 +401,7 @@ func TestIntegration_MessageSubscriptions(t *testing.T) { // Test unmount (most reliable since it doesn't depend on external services) testMsg := []byte(`{"Name": "test-volume-sub"}`) - msg, err := nc.Request("ebs.unmount", testMsg, 3*time.Second) + msg, err := nc.Request("ebs.test-node.unmount", testMsg, 3*time.Second) // Should get a response assert.NoError(t, err) @@ -466,7 +467,7 @@ func TestIntegration_ServiceGracefulShutdown(t *testing.T) { request := config.EBSRequest{Name: "test"} requestData, _ := json.Marshal(request) - _, err = nc.Request("ebs.unmount", requestData, 2*time.Second) + _, err = nc.Request("ebs.test-node.unmount", requestData, 2*time.Second) assert.NoError(t, err) nc.Close() diff --git a/scripts/create-multi-node.sh b/scripts/create-multi-node.sh index decdb90e8..2e9ee756f 100755 --- a/scripts/create-multi-node.sh +++ b/scripts/create-multi-node.sh @@ -1,67 +1,101 @@ -# simulate node1 +#!/bin/bash + +# Simulated multi-node cluster on a single machine using loopback addresses. +# Uses ~/node{1,2,3}/ for separate data directories. +# Usage: ./scripts/create-multi-node.sh + +set -e + +SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +PROJECT_ROOT="$(cd "$SCRIPT_DIR/.." && pwd)" +cd "$PROJECT_ROOT" + echo "Multi-node simulation environment" -make +echo "=================================" -echo "Adding simulated IPs: 10.11.12.1, 10.11.12.2, 10.11.12.3" - -sudo ip addr add 10.11.12.1/24 dev eth0 - -# simulate node2 -sudo ip addr add 10.11.12.2/24 dev eth0 - -# simulate node3 -sudo ip addr add 10.11.12.3/24 dev eth0 +# Build first +make build +echo "" echo "Removing old node directories..." +rm -rf ~/node1/ ~/node2/ ~/node3/ -rm -rf ~/node1/ -rm -rf ~/node2/ -rm -rf ~/node3/ +echo "" +echo "Step 1: Cluster Formation" +echo "=========================" +echo "Starting init node (127.0.0.1) and joining nodes (127.0.0.2, 127.0.0.3)..." -echo "Node1 Setup:" - -# Initialize node1 +# Init blocks until all nodes join, so run in background ./bin/hive admin init \ ---region ap-southeast-2 \ ---az ap-southeast-2a \ ---node node1 \ ---bind 10.11.12.1 \ ---cluster-bind 10.11.12.1 \ ---port 4432 \ ---hive-dir ~/node1/ \ ---config-dir ~/node1/config/ + --node node1 \ + --nodes 3 \ + --bind 127.0.0.1 \ + --cluster-bind 127.0.0.1 \ + --port 4432 \ + --hive-dir ~/node1/ \ + --config-dir ~/node1/config/ \ + --region ap-southeast-2 \ + --az ap-southeast-2a & +INIT_PID=$! -# Start node1 -./scripts/start-dev.sh ~/node1/ +# Wait for formation server to start +sleep 2 -echo "Node2 Setup:" - -# Join cluster +# Join nodes concurrently ./bin/hive admin join \ ---region ap-southeast-2 \ ---az ap-southeast-2a \ ---node node2 \ ---bind 10.11.12.2 \ ---cluster-bind 10.11.12.2 \ ---cluster-routes 10.11.12.1:4248 \ ---host 10.11.12.1:4432 \ ---data-dir ~/node2/ \ ---config-dir ~/node2/config/ \ - -# Start node2 -./scripts/start-dev.sh ~/node2/ - -echo "Node3 Setup:" + --node node2 \ + --bind 127.0.0.2 \ + --cluster-bind 127.0.0.2 \ + --host 127.0.0.1:4432 \ + --data-dir ~/node2/ \ + --config-dir ~/node2/config/ \ + --region ap-southeast-2 \ + --az ap-southeast-2a & +JOIN2_PID=$! ./bin/hive admin join \ ---region ap-southeast-2 \ ---az ap-southeast-2a \ ---node node3 \ ---bind 10.11.12.3 \ ---cluster-bind 10.11.12.3 \ ---cluster-routes 10.11.12.1:4248 \ ---host 10.11.12.1:4432 \ ---data-dir ~/node3/ \ ---config-dir ~/node3/config/ + --node node3 \ + --bind 127.0.0.3 \ + --cluster-bind 127.0.0.3 \ + --host 127.0.0.1:4432 \ + --data-dir ~/node3/ \ + --config-dir ~/node3/config/ \ + --region ap-southeast-2 \ + --az ap-southeast-2a & +JOIN3_PID=$! -./scripts/start-dev.sh ~/node3/ \ No newline at end of file +# Wait for all formation processes to complete +wait $INIT_PID $JOIN2_PID $JOIN3_PID +echo "" +echo "Cluster formation complete!" + +echo "" +echo "Step 2: Starting Services" +echo "=========================" + +# Start node 1 first (Predastore needs quorum) +HIVE_SKIP_BUILD=true UI=false ./scripts/start-dev.sh ~/node1/ & +sleep 5 + +# Start nodes 2 and 3 +HIVE_SKIP_BUILD=true UI=false ./scripts/start-dev.sh ~/node2/ & +sleep 2 +HIVE_SKIP_BUILD=true UI=false ./scripts/start-dev.sh ~/node3/ & + +# Wait for all start scripts to finish +wait + +echo "" +echo "Step 3: Verification" +echo "====================" +sleep 5 + +echo "Daemon health:" +for IP in 127.0.0.1 127.0.0.2 127.0.0.3; do + echo " $IP: $(curl -s http://$IP:4432/health | head -c 100)" +done + +echo "" +echo "Multi-node cluster is running!" +echo " Stop with: ./scripts/stop-multi-node.sh" +echo " Logs: ~/node{1,2,3}/logs/" diff --git a/scripts/openvswitch.sh b/scripts/openvswitch.sh new file mode 100644 index 000000000..0eb662703 --- /dev/null +++ b/scripts/openvswitch.sh @@ -0,0 +1,56 @@ +#!/bin/sh + +### Setup openvswitch +echo "Install and enable OVS:" + +sudo apt update +sudo apt install -y openvswitch-switch +sudo systemctl enable --now openvswitch-switch + +echo "Sysctls (overlay-friendly, avoids rp_filter pain):" + +sudo tee /etc/sysctl.d/99-mulga-vpc.conf >/dev/null <<'EOF' +net.ipv4.ip_forward=1 +net.ipv4.conf.all.rp_filter=0 +net.ipv4.conf.default.rp_filter=0 +EOF +sudo sysctl --system + +echo "1) Create br-int on BOTH hosts" + +sudo ovs-vsctl --may-exist add-br br-int +sudo ip link set br-int up + +echo "Creating TAP for tap-vm1" + +# Consider, adduser hive, for all services used. +sudo ip tuntap add dev tap-vm1 mode tap user $USER +sudo ip link set tap-vm1 mtu 1450 +sudo ip link set tap-vm1 up +ip link show tap-vm1 + +echo "Creating flows on host1" + +sudo ovs-ofctl del-flows br-int + +sudo ovs-vsctl --may-exist add-port br-int tap-vm1 + +# VM -> Geneve (set VNI) +sudo ovs-ofctl add-flow br-int "in_port=tap-vm1,actions=set_tunnel:10302,output:geneve-h2" + +# Geneve -> VM (only that VNI) +sudo ovs-ofctl add-flow br-int "in_port=geneve-h2,tun_id=10302,actions=output:tap-vm1" + +# Creating +#echo "Creating flows on host2" + +#sudo ovs-vsctl --may-exist add-port br-int tap-vm1 + +#sudo ovs-ofctl del-flows br-int + +# VM -> Geneve (set VNI) +#sudo ovs-ofctl add-flow br-int "in_port=tap-vm1,actions=set_tunnel:10302,output:geneve-h1" + +# Geneve -> VM (only that VNI) +#sudo ovs-ofctl add-flow br-int "in_port=geneve-h1,tun_id=10302,actions=output:tap-vm1" + diff --git a/scripts/stop-multi-node.sh b/scripts/stop-multi-node.sh index 958dccca3..dee7ecc1a 100755 --- a/scripts/stop-multi-node.sh +++ b/scripts/stop-multi-node.sh @@ -1,17 +1,18 @@ +#!/bin/bash -#sudo ip addr del 10.11.12.1/24 dev eth0 -#sudo ip addr del 10.11.12.2/24 dev eth0 -#sudo ip addr del 10.11.12.3/24 dev eth0 +# Stop simulated multi-node cluster. +# Stops services in reverse order (node3 → node2 → node1). +# Usage: ./scripts/stop-multi-node.sh -# TODO: Reverse order, predastore running on node1 -# BUG: Need to wait for each QEMU to flush/commit WAL to disk > S3, before shutdown, else corruption may occur -# BUG: nbdkit: ramdisk.3: debug: sending error reply: Cannot send after transport endpoint shutdown -# TODO: Fallback, if WAL on local disk not flushed, on boot, viperblock should commit and verify. Requires UT and test-cases +SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" -./scripts/stop-dev.sh ~/node3/ - -./scripts/stop-dev.sh ~/node2/ - -./scripts/stop-dev.sh ~/node1/ +echo "Stopping multi-node cluster..." +# Stop in reverse order — node1 last since it hosts the Predastore leader +"$SCRIPT_DIR/stop-dev.sh" ~/node3/ +"$SCRIPT_DIR/stop-dev.sh" ~/node2/ +"$SCRIPT_DIR/stop-dev.sh" ~/node1/ +echo "" +echo "Multi-node cluster stopped." +echo "To remove all data: rm -rf ~/node1/ ~/node2/ ~/node3/"