mirror of
https://github.com/mulgadc/spinifex.git
synced 2026-10-16 01:22:00 +00:00
Improved multi-server node testing and tweaks
This commit is contained in:
Ben Duncan
@@ -28,3 +28,8 @@ Thumbs.db
|
||||
|
||||
# Test reports
|
||||
*-report.txt
|
||||
|
||||
# Agent direction (internal use only)
|
||||
docs/INTERNAL.md
|
||||
docs/PRIVATE.md
|
||||
docs/DEVNOTES.md
|
||||
|
||||
+287
-89
@@ -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.
|
||||
|
||||
+341
-128
@@ -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 <name> --bind <ip>\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()
|
||||
}
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -11,7 +11,7 @@ listen: 0.0.0.0:4222
|
||||
|
||||
{{ if and .ClusterBindIP .ClusterRoutes }}
|
||||
cluster {
|
||||
name: C1
|
||||
name: {{ .ClusterName }}
|
||||
listen: {{.ClusterBindIP}}:4248
|
||||
|
||||
routes = [
|
||||
|
||||
@@ -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
|
||||
+10
-3
@@ -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 {
|
||||
|
||||
@@ -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) {
|
||||
|
||||
+10
-3
@@ -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
|
||||
|
||||
@@ -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)
|
||||
|
||||
+10
-10
@@ -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)
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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()
|
||||
|
||||
@@ -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/
|
||||
# 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/"
|
||||
|
||||
@@ -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"
|
||||
|
||||
+13
-12
@@ -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/"
|
||||
|
||||
Reference in New Issue
Block a user