From 7b7be2c26ba27c5946c561efe79785b8fcdfcbdc Mon Sep 17 00:00:00 2001 From: Julian Sommer <115516918+juliansommer@users.noreply.github.com> Date: Tue, 27 Jan 2026 10:59:58 +1100 Subject: [PATCH] feat: multi-node e2e test v2 (#8) --- .github/workflows/e2e.yml | 45 +++- cmd/hive/cmd/admin.go | 6 +- cmd/hive/cmd/templates/nats.conf | 10 +- hive/daemon/daemon.go | 60 +++++- hive/daemon/jetstream.go | 54 ++++- hive/daemon/jetstream_test.go | 64 +++++- hive/gateway/ec2.go | 4 +- hive/gateway/gateway.go | 74 +++++++ tests/e2e/Dockerfile.e2e | 6 +- tests/e2e/lib/multinode-helpers.sh | 327 ++++++++++++++++++++++++++++ tests/e2e/run-multinode-e2e.sh | 332 +++++++++++++++++++++++++++++ 11 files changed, 939 insertions(+), 43 deletions(-) create mode 100755 tests/e2e/lib/multinode-helpers.sh create mode 100755 tests/e2e/run-multinode-e2e.sh diff --git a/.github/workflows/e2e.yml b/.github/workflows/e2e.yml index 12e7c6a50..7ecff5e0c 100644 --- a/.github/workflows/e2e.yml +++ b/.github/workflows/e2e.yml @@ -2,12 +2,12 @@ name: E2E Test Hive on: push: - branches: [ main, e2e ] + branches: [main, e2e] workflow_dispatch: jobs: - hive_e2e_test: - name: Hive E2E Test + build_e2e_image: + name: Build E2E Image runs-on: self-hosted steps: @@ -19,9 +19,6 @@ jobs: - name: Clone Dependencies run: ./hive/scripts/clone-deps.sh - - name: Cleanup Stale Containers - run: docker rm -f hive-e2e-test 2>/dev/null || true - - name: Build E2E Image run: | docker build \ @@ -29,6 +26,12 @@ jobs: -f hive/tests/e2e/Dockerfile.e2e \ . + hive_e2e_test: + name: Hive Single-Node E2E Test + runs-on: self-hosted + needs: build_e2e_image + + steps: - name: Run E2E Tests run: | docker run --privileged --rm \ @@ -36,8 +39,32 @@ jobs: --name hive-e2e-test \ hive-e2e:latest - - name: Cleanup - if: always() + hive_multinode_e2e_test: + name: Hive Multi-Node E2E Test + runs-on: self-hosted + needs: build_e2e_image + + steps: + - name: Run Multi-Node E2E Tests + run: | + docker run --privileged --rm \ + -v /dev/kvm:/dev/kvm \ + --cap-add=NET_ADMIN \ + --name hive-multinode-e2e-test \ + hive-e2e:latest \ + ./tests/e2e/run-multinode-e2e.sh + + cleanup: + name: Cleanup + runs-on: self-hosted + needs: [hive_e2e_test, hive_multinode_e2e_test] + if: always() + + steps: + - name: Remove Stale Containers run: | docker rm -f hive-e2e-test 2>/dev/null || true - docker image prune -f + docker rm -f hive-multinode-e2e-test 2>/dev/null || true + + - name: Prune Docker Images + run: docker image prune -f diff --git a/cmd/hive/cmd/admin.go b/cmd/hive/cmd/admin.go index 5fbfd903a..940be1fc8 100644 --- a/cmd/hive/cmd/admin.go +++ b/cmd/hive/cmd/admin.go @@ -472,8 +472,8 @@ func runAdminInit(cmd *cobra.Command, args []string) { port, _ := cmd.Flags().GetInt("port") bindIP, _ := cmd.Flags().GetString("bind") clusterBind, _ := cmd.Flags().GetString("cluster-bind") - //clusterRoutesStr, _ := cmd.Flags().GetString("cluster-routes") - // := strings.Split(clusterRoutesStr, ",") + clusterRoutesStr, _ := cmd.Flags().GetString("cluster-routes") + clusterRoutes := strings.Split(clusterRoutesStr, ",") // Validate IP address format if net.ParseIP(bindIP) == nil { @@ -574,7 +574,7 @@ func runAdminInit(cmd *cobra.Command, args []string) { Port: portStr, BindIP: bindIP, ClusterBindIP: clusterBind, - //ClusterRoutes: clusterRoutes, + ClusterRoutes: clusterRoutes, } // Generate all config files diff --git a/cmd/hive/cmd/templates/nats.conf b/cmd/hive/cmd/templates/nats.conf index 6d3054536..5713c71b1 100644 --- a/cmd/hive/cmd/templates/nats.conf +++ b/cmd/hive/cmd/templates/nats.conf @@ -9,18 +9,16 @@ listen: 0.0.0.0:4222 {{ end }} -{{ if .ClusterBindIP }} +{{ if and .ClusterBindIP .ClusterRoutes }} cluster { name: C1 listen: {{.ClusterBindIP}}:4248 -{{ if .ClusterRoutes }} -routes = [ + routes = [ {{- range .ClusterRoutes }} - "nats://{{ . }}", + "nats://{{ . }}", {{- end }} -] -{{ end }} + ] } {{ end }} diff --git a/hive/daemon/daemon.go b/hive/daemon/daemon.go index 6d58503a0..4fbc1b1b2 100644 --- a/hive/daemon/daemon.go +++ b/hive/daemon/daemon.go @@ -409,7 +409,8 @@ func (d *Daemon) Start() error { log.Printf("Connected to NATS server at %s", d.config.NATS.Host) // Initialize JetStream for KV state storage (falls back to disk if unavailable) - d.jsManager, err = NewJetStreamManager(d.natsConn) + // Start with 1 replica to allow single-node startup, then upgrade if cluster has more nodes + d.jsManager, err = NewJetStreamManager(d.natsConn, 1) if err != nil { slog.Warn("Failed to init JetStream, falling back to file", "error", err) d.jsManager = nil @@ -417,7 +418,16 @@ func (d *Daemon) Start() error { slog.Warn("Failed to init KV bucket, falling back to file", "error", err) d.jsManager = nil } else { - slog.Info("JetStream KV store initialized successfully") + slog.Info("JetStream KV store initialized successfully", "replicas", 1) + + // Try to upgrade replicas if cluster has more nodes + // This handles the case where this daemon starts after other NATS nodes are already up + clusterSize := len(d.clusterConfig.Nodes) + if clusterSize > 1 { + if err := d.jsManager.UpdateReplicas(clusterSize); err != nil { + slog.Warn("Failed to upgrade JetStream replicas on startup (other NATS nodes may not be ready)", "targetReplicas", clusterSize, "error", err) + } + } } // Load existing state for VMs from JetStream or disk @@ -591,6 +601,13 @@ func (d *Daemon) Start() error { return fmt.Errorf("failed to subscribe to NATS %s: %w", healthSubject, err) } + // Subscribe to node discovery - all daemons respond so gateway can discover active nodes + log.Printf("Subscribing to node discovery: hive.nodes.discover") + d.natsSubscriptions["hive.nodes.discover"], err = d.natsConn.Subscribe("hive.nodes.discover", d.handleNodeDiscover) + if err != nil { + return fmt.Errorf("failed to subscribe to NATS hive.nodes.discover: %w", err) + } + // Start cluster manager HTTP server if err := d.ClusterManager(); err != nil { return fmt.Errorf("failed to start cluster manager: %w", err) @@ -750,6 +767,16 @@ func (d *Daemon) ClusterManager() error { slog.Info("Node joined cluster", "node", req.Node, "epoch", d.clusterConfig.Epoch) + // Update JetStream KV replicas to match new cluster size + // This may fail if the new node's NATS server isn't running yet - that's OK, + // replicas can be updated later when the cluster is fully formed + if d.jsManager != nil { + newReplicaCount := len(d.clusterConfig.Nodes) + if err := d.jsManager.UpdateReplicas(newReplicaCount); err != nil { + slog.Warn("Failed to update JetStream replicas (new node NATS may not be ready yet)", "targetReplicas", newReplicaCount, "error", err) + } + } + // Send only shared cluster data (exclude node-specific top-level fields) sharedData := &config.SharedClusterData{ Epoch: d.clusterConfig.Epoch, @@ -1153,9 +1180,12 @@ func (d *Daemon) handleEC2Events(msg *nats.Msg) { // TODO: Last, delete the instance volumes - // Remove instance from state + // Set instance state to "terminated" instead of deleting + // This allows DescribeInstances to return the terminated state + // AWS keeps terminated instances visible for about an hour d.Instances.Mu.Lock() - delete(d.Instances.VMS, instance.ID) + instance.Status = "terminated" + instance.Attributes = command.Attributes d.Instances.Mu.Unlock() slog.Info("Instance terminated", "id", command.ID) @@ -2413,3 +2443,25 @@ func (d *Daemon) handleHealthCheck(msg *nats.Msg) { msg.Respond(jsonResponse) slog.Debug("Health check responded", "node", d.node, "epoch", d.clusterConfig.Epoch) } + +// NodeDiscoverResponse is the response for node discovery requests +type NodeDiscoverResponse struct { + Node string `json:"node"` +} + +// handleNodeDiscover responds to node discovery requests with this node's ID +// Used by the gateway to dynamically discover active hive nodes in the cluster +func (d *Daemon) handleNodeDiscover(msg *nats.Msg) { + response := NodeDiscoverResponse{ + Node: d.node, + } + + jsonResponse, err := json.Marshal(response) + if err != nil { + slog.Error("handleNodeDiscover failed to marshal response", "err", err) + return + } + + msg.Respond(jsonResponse) + slog.Debug("Node discovery responded", "node", d.node) +} diff --git a/hive/daemon/jetstream.go b/hive/daemon/jetstream.go index 37c1ddfa1..c242d60e9 100644 --- a/hive/daemon/jetstream.go +++ b/hive/daemon/jetstream.go @@ -18,19 +18,22 @@ const ( // JetStreamManager manages JetStream KV store operations for instance state type JetStreamManager struct { - js nats.JetStreamContext - kv nats.KeyValue + js nats.JetStreamContext + kv nats.KeyValue + replicas int } -// NewJetStreamManager creates a new JetStreamManager from a NATS connection -func NewJetStreamManager(nc *nats.Conn) (*JetStreamManager, error) { +// NewJetStreamManager creates a new JetStreamManager from a NATS connection. +// replicas specifies the number of replicas for the KV bucket (typically matches cluster node count). +func NewJetStreamManager(nc *nats.Conn, replicas int) (*JetStreamManager, error) { js, err := nc.JetStream() if err != nil { return nil, err } return &JetStreamManager{ - js: js, + js: js, + replicas: replicas, }, nil } @@ -41,11 +44,12 @@ func (m *JetStreamManager) InitKVBucket() error { if err != nil { if errors.Is(err, nats.ErrBucketNotFound) { // Bucket doesn't exist, create it - slog.Info("Creating JetStream KV bucket", "bucket", InstanceStateBucket) + slog.Info("Creating JetStream KV bucket", "bucket", InstanceStateBucket, "replicas", m.replicas) kv, err = m.js.CreateKeyValue(&nats.KeyValueConfig{ Bucket: InstanceStateBucket, Description: "Hive instance state storage", - History: 1, // Only keep latest value + History: 1, // Only keep latest value + Replicas: m.replicas, // Replication across cluster nodes }) if err != nil { return err @@ -138,3 +142,39 @@ func (m *JetStreamManager) DeleteState(nodeID string) error { slog.Debug("Deleted state from JetStream KV", "key", key) return nil } + +// UpdateReplicas updates the replica count for the KV bucket's underlying stream. +// This should be called when new nodes join the cluster. +// Note: Increasing replicas requires the new replica count of NATS servers to be available. +func (m *JetStreamManager) UpdateReplicas(newReplicas int) error { + if m.js == nil { + return errors.New("JetStream context not initialized") + } + + // KV buckets are backed by streams with name "KV_" + streamName := "KV_" + InstanceStateBucket + + // Get current stream info + streamInfo, err := m.js.StreamInfo(streamName) + if err != nil { + return err + } + + currentReplicas := streamInfo.Config.Replicas + if currentReplicas >= newReplicas { + slog.Debug("Stream already has sufficient replicas", "current", currentReplicas, "requested", newReplicas) + return nil + } + + // Update the stream config with new replica count + streamInfo.Config.Replicas = newReplicas + + _, err = m.js.UpdateStream(&streamInfo.Config) + if err != nil { + return err + } + + m.replicas = newReplicas + slog.Info("Updated JetStream KV bucket replicas", "bucket", InstanceStateBucket, "oldReplicas", currentReplicas, "newReplicas", newReplicas) + return nil +} diff --git a/hive/daemon/jetstream_test.go b/hive/daemon/jetstream_test.go index c63e00a52..e8c78da94 100644 --- a/hive/daemon/jetstream_test.go +++ b/hive/daemon/jetstream_test.go @@ -60,7 +60,7 @@ func TestJetStreamManager_WriteAndLoadState(t *testing.T) { defer nc.Close() // Create JetStreamManager - jsm, err := NewJetStreamManager(nc) + jsm, err := NewJetStreamManager(nc, 1) require.NoError(t, err, "Failed to create JetStreamManager") // Initialize the KV bucket @@ -112,7 +112,7 @@ func TestJetStreamManager_LoadState_KeyNotFound(t *testing.T) { require.NoError(t, err) defer nc.Close() - jsm, err := NewJetStreamManager(nc) + jsm, err := NewJetStreamManager(nc, 1) require.NoError(t, err) err = jsm.InitKVBucket() @@ -134,7 +134,7 @@ func TestJetStreamManager_BucketCreation(t *testing.T) { require.NoError(t, err) defer nc.Close() - jsm, err := NewJetStreamManager(nc) + jsm, err := NewJetStreamManager(nc, 1) require.NoError(t, err) // InitKVBucket should create the bucket @@ -154,7 +154,7 @@ func TestJetStreamManager_BucketReconnection(t *testing.T) { nc1, err := nats.Connect(natsURL) require.NoError(t, err) - jsm1, err := NewJetStreamManager(nc1) + jsm1, err := NewJetStreamManager(nc1, 1) require.NoError(t, err) err = jsm1.InitKVBucket() @@ -179,7 +179,7 @@ func TestJetStreamManager_BucketReconnection(t *testing.T) { require.NoError(t, err) defer nc2.Close() - jsm2, err := NewJetStreamManager(nc2) + jsm2, err := NewJetStreamManager(nc2, 1) require.NoError(t, err) err = jsm2.InitKVBucket() @@ -202,7 +202,7 @@ func TestJetStreamManager_DeleteState(t *testing.T) { require.NoError(t, err) defer nc.Close() - jsm, err := NewJetStreamManager(nc) + jsm, err := NewJetStreamManager(nc, 1) require.NoError(t, err) err = jsm.InitKVBucket() @@ -245,7 +245,7 @@ func TestJetStreamManager_DeleteState_NonExistent(t *testing.T) { require.NoError(t, err) defer nc.Close() - jsm, err := NewJetStreamManager(nc) + jsm, err := NewJetStreamManager(nc, 1) require.NoError(t, err) err = jsm.InitKVBucket() @@ -265,7 +265,7 @@ func TestJetStreamManager_WriteState_UpdateExisting(t *testing.T) { require.NoError(t, err) defer nc.Close() - jsm, err := NewJetStreamManager(nc) + jsm, err := NewJetStreamManager(nc, 1) require.NoError(t, err) err = jsm.InitKVBucket() @@ -318,7 +318,7 @@ func TestJetStreamManager_MultipleNodes(t *testing.T) { require.NoError(t, err) defer nc.Close() - jsm, err := NewJetStreamManager(nc) + jsm, err := NewJetStreamManager(nc, 1) require.NoError(t, err) err = jsm.InitKVBucket() @@ -371,7 +371,7 @@ func TestJetStreamManager_KVNotInitialized(t *testing.T) { defer nc.Close() // Create JetStreamManager but don't call InitKVBucket - jsm, err := NewJetStreamManager(nc) + jsm, err := NewJetStreamManager(nc, 1) require.NoError(t, err) testInstances := &vm.Instances{VMS: make(map[string]*vm.VM)} @@ -384,3 +384,47 @@ func TestJetStreamManager_KVNotInitialized(t *testing.T) { err = jsm.DeleteState("test-node") assert.Error(t, err, "DeleteState should error when KV not initialized") } + +// TestJetStreamManager_UpdateReplicas tests updating replica count for the KV bucket +func TestJetStreamManager_UpdateReplicas(t *testing.T) { + ns, natsURL := startTestNATSServerWithJetStream(t) + defer ns.Shutdown() + + nc, err := nats.Connect(natsURL) + require.NoError(t, err) + defer nc.Close() + + // Create with 1 replica (typical for single node startup) + jsm, err := NewJetStreamManager(nc, 1) + require.NoError(t, err) + + err = jsm.InitKVBucket() + require.NoError(t, err) + + // Verify initial replica count + js, _ := nc.JetStream() + streamInfo, err := js.StreamInfo("KV_" + InstanceStateBucket) + require.NoError(t, err) + assert.Equal(t, 1, streamInfo.Config.Replicas, "Should start with 1 replica") + + // Try to update to same replica count (should be a no-op) + err = jsm.UpdateReplicas(1) + assert.NoError(t, err, "Updating to same replica count should succeed") + + // Note: Increasing replicas beyond 1 requires additional NATS servers in the cluster, + // which we don't have in the test environment. In a single-node test server, + // attempting to increase replicas will fail with "insufficient resources" error. + // This test verifies the basic functionality works. +} + +// TestJetStreamManager_UpdateReplicas_NoInit tests UpdateReplicas when JS not initialized +func TestJetStreamManager_UpdateReplicas_NoInit(t *testing.T) { + // Test with nil JetStream context + jsm := &JetStreamManager{ + js: nil, + replicas: 1, + } + + err := jsm.UpdateReplicas(3) + assert.Error(t, err, "UpdateReplicas should error when JetStream not initialized") +} diff --git a/hive/gateway/ec2.go b/hive/gateway/ec2.go index c3af0898c..2b61191e5 100644 --- a/hive/gateway/ec2.go +++ b/hive/gateway/ec2.go @@ -34,7 +34,9 @@ func (gw *GatewayConfig) EC2_Request(ctx *fiber.Ctx) error { return err } - output, err := gateway_ec2_instance.DescribeInstances(input, gw.NATSConn, gw.ExpectedNodes) + // Dynamically discover active nodes instead of using static config value + activeNodes := gw.DiscoverActiveNodes() + output, err := gateway_ec2_instance.DescribeInstances(input, gw.NATSConn, activeNodes) if err != nil { return err diff --git a/hive/gateway/gateway.go b/hive/gateway/gateway.go index 55c077eaa..87f21a31b 100644 --- a/hive/gateway/gateway.go +++ b/hive/gateway/gateway.go @@ -1,11 +1,13 @@ package gateway import ( + "encoding/json" "encoding/xml" "errors" "log/slog" "os" "strings" + "time" "github.com/gofiber/fiber/v2" "github.com/gofiber/fiber/v2/middleware/cors" @@ -270,3 +272,75 @@ func ParseArgsToStruct(input *any, args map[string]string) (err error) { return nil } + +// NodeDiscoverResponse is the response from a node discovery request +type NodeDiscoverResponse struct { + Node string `json:"node"` +} + +// DiscoverActiveNodes discovers the number of active hive daemon nodes in the cluster +// by publishing a discovery request and counting unique responses. +// Returns the number of active nodes (minimum 1 if fallback is needed). +func (gw *GatewayConfig) DiscoverActiveNodes() int { + if gw.NATSConn == nil { + slog.Warn("DiscoverActiveNodes: NATS connection not available, using ExpectedNodes fallback", "fallback", gw.ExpectedNodes) + return gw.ExpectedNodes + } + + // Create an inbox for collecting responses from all nodes + inbox := nats.NewInbox() + sub, err := gw.NATSConn.SubscribeSync(inbox) + if err != nil { + slog.Error("DiscoverActiveNodes: Failed to create inbox subscription", "err", err) + return gw.ExpectedNodes + } + defer sub.Unsubscribe() + + // Publish discovery request to all nodes + err = gw.NATSConn.PublishRequest("hive.nodes.discover", inbox, []byte("{}")) + if err != nil { + slog.Error("DiscoverActiveNodes: Failed to publish request", "err", err) + return gw.ExpectedNodes + } + + // Collect responses with a short timeout + // We use a short timeout since discovery should be fast + timeout := 500 * time.Millisecond + deadline := time.Now().Add(timeout) + + nodesSeen := make(map[string]bool) + + for time.Now().Before(deadline) { + remaining := time.Until(deadline) + if remaining <= 0 { + break + } + + msg, err := sub.NextMsg(remaining) + if err != nil { + if err == nats.ErrTimeout { + break + } + slog.Debug("DiscoverActiveNodes: Error receiving message", "err", err) + break + } + + var response NodeDiscoverResponse + if err := json.Unmarshal(msg.Data, &response); err != nil { + slog.Debug("DiscoverActiveNodes: Failed to unmarshal response", "err", err) + continue + } + + nodesSeen[response.Node] = true + } + + activeNodes := len(nodesSeen) + if activeNodes == 0 { + // Fallback to configured value if no responses + slog.Warn("DiscoverActiveNodes: No nodes responded, using ExpectedNodes fallback", "fallback", gw.ExpectedNodes) + return gw.ExpectedNodes + } + + slog.Debug("DiscoverActiveNodes: Discovered active nodes", "count", activeNodes) + return activeNodes +} diff --git a/tests/e2e/Dockerfile.e2e b/tests/e2e/Dockerfile.e2e index 9d1200ad4..8078816d6 100644 --- a/tests/e2e/Dockerfile.e2e +++ b/tests/e2e/Dockerfile.e2e @@ -3,8 +3,8 @@ FROM ubuntu:24.04 # Avoid prompts ENV DEBIAN_FRONTEND=noninteractive -# Install make first so we can use the Makefile -RUN apt-get update && apt-get install -y make sudo && rm -rf /var/lib/apt/lists/* +# Install make and iproute2 (for multi-node IP management) +RUN apt-get update && apt-get install -y make sudo iproute2 && rm -rf /var/lib/apt/lists/* # Set up working directory for Makefile targets WORKDIR /tmp/hive-setup @@ -59,7 +59,7 @@ COPY predastore/ ./predastore/ WORKDIR /root/Development/mulga/hive # Ensure scripts are executable -RUN chmod +x ./scripts/*.sh ./tests/e2e/run-e2e.sh +RUN chmod +x ./scripts/*.sh ./tests/e2e/*.sh ./tests/e2e/lib/*.sh # 4. Build everything (using cache mounts) RUN --mount=type=cache,target=/root/.cache/go-build \ diff --git a/tests/e2e/lib/multinode-helpers.sh b/tests/e2e/lib/multinode-helpers.sh new file mode 100755 index 000000000..f4992365a --- /dev/null +++ b/tests/e2e/lib/multinode-helpers.sh @@ -0,0 +1,327 @@ +#!/bin/bash + +# Multi-node E2E test helper functions +# Provides utilities for managing simulated IPs, starting/stopping node services, +# and verifying NATS cluster health. + +# Suppress urllib3 InsecureRequestWarning from AWS CLI (we use --no-verify-ssl) +export PYTHONWARNINGS="ignore::urllib3.exceptions.InsecureRequestWarning" + +# Helper to filter AWS CLI output (removes Python warnings that break jq) +aws_json() { + "$@" 2>&1 | grep -v "^urllib3\|^InsecureRequestWarning\|certificate verification" +} + +# Network configuration +SIMULATED_NETWORK="10.11.12" +NODE1_IP="${SIMULATED_NETWORK}.1" +NODE2_IP="${SIMULATED_NETWORK}.2" +NODE3_IP="${SIMULATED_NETWORK}.3" + +# Port configuration +NATS_CLIENT_PORT=4222 +NATS_CLUSTER_PORT=4248 +NATS_MONITOR_PORT=8222 +PREDASTORE_PORT=8443 +AWSGW_PORT=9999 +CLUSTER_PORT=4432 + +# Add simulated IPs to loopback interface +# Requires NET_ADMIN capability +add_simulated_ips() { + echo "Adding simulated IPs to loopback interface..." + + for i in 1 2 3; do + local ip="${SIMULATED_NETWORK}.$i" + if ! ip addr show lo | grep -q "$ip"; then + ip addr add "${ip}/24" dev lo + echo " Added $ip to lo" + else + echo " $ip already exists on lo" + fi + done + + # Verify IPs were added + echo "Verifying simulated IPs..." + for i in 1 2 3; do + local ip="${SIMULATED_NETWORK}.$i" + if ip addr show lo | grep -q "$ip"; then + echo " $ip is configured" + else + echo " ERROR: Failed to add $ip" + return 1 + fi + done + + echo "Simulated IPs configured successfully" +} + +# Remove simulated IPs from loopback interface +remove_simulated_ips() { + echo "Removing simulated IPs from loopback interface..." + + for i in 1 2 3; do + local ip="${SIMULATED_NETWORK}.$i" + if ip addr show lo | grep -q "$ip"; then + ip addr del "${ip}/24" dev lo 2>/dev/null || true + echo " Removed $ip from lo" + fi + done + + echo "Simulated IPs removed" +} + +# Start services for a specific node +# Usage: start_node_services +# Example: start_node_services 1 ~/node1 +start_node_services() { + local node_num="$1" + local data_dir="$2" + local node_ip="${SIMULATED_NETWORK}.$node_num" + + echo "Starting services for node$node_num at $node_ip..." + + # Start all services - each node's config binds to its specific IP + HIVE_SKIP_BUILD=true ./scripts/start-dev.sh "$data_dir" + + echo "Node$node_num services started" +} + +# Stop services for a specific node +# Usage: stop_node_services +stop_node_services() { + local node_num="$1" + local data_dir="$2" + + echo "Stopping services for node$node_num..." + + # Stop using PID files in the node's log directory + ./scripts/stop-dev.sh "$data_dir" + + echo "Node$node_num services stopped" +} + +# Stop all node services +stop_all_nodes() { + echo "Stopping all node services..." + + for i in 1 2 3; do + local data_dir="$HOME/node$i" + if [ -d "$data_dir/logs" ]; then + stop_node_services "$i" "$data_dir" || true + fi + done + + echo "All node services stopped" +} + +# Verify NATS cluster health +# Checks that the cluster has expected number of members +# Usage: verify_nats_cluster [expected_members] +verify_nats_cluster() { + local expected_members="${1:-3}" + + echo "Verifying NATS cluster health (expecting $expected_members members)..." + + # Check cluster info via monitoring endpoint on node1 + local cluster_info + cluster_info=$(curl -s "http://${NODE1_IP}:${NATS_MONITOR_PORT}/routez" 2>/dev/null) || { + echo " ERROR: Cannot reach NATS monitoring endpoint" + return 1 + } + + # Count unique remote servers (NATS creates multiple connections per peer) + local num_routes + num_routes=$(echo "$cluster_info" | jq -r '.num_routes // 0') + + # Count unique peer names + local unique_peers + unique_peers=$(echo "$cluster_info" | jq -r '[.routes[].remote_name] | unique | length') + local expected_peers=$((expected_members - 1)) + + echo " NATS cluster routes: $num_routes (unique peers: $unique_peers, expected peers: $expected_peers)" + + if [ "$unique_peers" -ge "$expected_peers" ]; then + echo " NATS cluster is healthy" + return 0 + else + echo " WARNING: NATS cluster may not be fully formed" + echo " Cluster info: $cluster_info" + return 1 + fi +} + +# Wait for a specific instance state +# Usage: wait_for_instance_state [max_attempts] +wait_for_instance_state() { + local instance_id="$1" + local target_state="$2" + local max_attempts="${3:-10}" + local attempt=0 + + echo "Waiting for instance $instance_id to reach state: $target_state..." + + while [ $attempt -lt $max_attempts ]; do + local state + state=$(aws_json aws --endpoint-url https://${NODE1_IP}:${AWSGW_PORT} --no-verify-ssl ec2 describe-instances \ + --instance-ids "$instance_id" \ + --query 'Reservations[0].Instances[0].State.Name' \ + --output text) || { + echo " Attempt $((attempt + 1))/$max_attempts - Gateway request failed, retrying..." + sleep 2 + attempt=$((attempt + 1)) + continue + } + + echo " Instance state: $state" + + if [ "$state" == "$target_state" ]; then + echo " Instance reached target state: $target_state" + return 0 + fi + + if [ "$state" == "terminated" ] && [ "$target_state" != "terminated" ]; then + echo " ERROR: Instance terminated unexpectedly" + return 1 + fi + + sleep 2 + attempt=$((attempt + 1)) + done + + echo " ERROR: Instance did not reach $target_state within $max_attempts attempts" + return 1 +} + +# Wait for gateway to be ready +# Usage: wait_for_gateway [host] [max_attempts] +wait_for_gateway() { + local host="${1:-localhost}" + local max_attempts="${2:-30}" + local attempt=0 + + echo "Waiting for AWS Gateway at $host:${AWSGW_PORT}..." + + while [ $attempt -lt $max_attempts ]; do + if curl -k -s "https://${host}:${AWSGW_PORT}" > /dev/null 2>&1; then + echo " Gateway is ready" + return 0 + fi + + echo " Waiting for gateway... ($((attempt + 1))/$max_attempts)" + sleep 2 + attempt=$((attempt + 1)) + done + + echo " ERROR: Gateway failed to start" + return 1 +} + +# Check instance distribution across nodes +# Returns the instance counts per node +check_instance_distribution() { + echo "Checking instance distribution across nodes..." + + local total=0 + for i in 1 2 3; do + local data_dir="$HOME/node$i" + local instances_file="$data_dir/hive/instances.json" + + if [ -f "$instances_file" ]; then + local count + count=$(jq 'length' "$instances_file" 2>/dev/null || echo "0") + echo " Node$i: $count instances" + total=$((total + count)) + else + echo " Node$i: 0 instances (no instances file)" + fi + done + + echo " Total instances: $total" +} + +# Dump logs from all nodes (for debugging failures) +dump_all_node_logs() { + echo "" + echo "==========================================" + echo "DUMPING LOGS FROM ALL NODES" + echo "==========================================" + + for i in 1 2 3; do + local data_dir="$HOME/node$i" + local logs_dir="$data_dir/logs" + + if [ -d "$logs_dir" ]; then + echo "" + echo "=== Node$i Logs ===" + + for log in nats predastore viperblock hive awsgw; do + if [ -f "$logs_dir/$log.log" ]; then + echo "" + echo "--- $log.log (last 50 lines) ---" + tail -50 "$logs_dir/$log.log" 2>/dev/null || echo "(empty or not accessible)" + fi + done + fi + done + + echo "" + echo "==========================================" + echo "END OF LOG DUMP" + echo "==========================================" +} + +# Initialize leader node (node1) +# Usage: init_leader_node +init_leader_node() { + echo "Initializing leader node (node1)..." + + # Remove old node directory + rm -rf "$HOME/node1/" + + # Node1 routes to itself to enable clustering mode (NATS ignores self-routes) + # Other nodes will discover each other via gossip when they connect + ./bin/hive admin init \ + --node node1 \ + --bind "${NODE1_IP}" \ + --cluster-bind "${NODE1_IP}" \ + --cluster-routes "${NODE1_IP}:${NATS_CLUSTER_PORT}" \ + --port ${CLUSTER_PORT} \ + --region ap-southeast-2 \ + --az ap-southeast-2a \ + --hive-dir "$HOME/node1/" \ + --config-dir "$HOME/node1/config/" + + + echo "Leader node initialized" +} + +# Join a follower node to the cluster +# Usage: join_follower_node +join_follower_node() { + local node_num="$1" + local node_ip="${SIMULATED_NETWORK}.$node_num" + local data_dir="$HOME/node$node_num" + + echo "Joining node$node_num ($node_ip) to cluster..." + + # Remove old node directory + rm -rf "$data_dir/" + + # Route to node1 (seed node) - other nodes discovered via NATS gossip + ./bin/hive admin join \ + --node "node$node_num" \ + --bind "$node_ip" \ + --cluster-bind "$node_ip" \ + --cluster-routes "${NODE1_IP}:${NATS_CLUSTER_PORT}" \ + --host "${NODE1_IP}:${CLUSTER_PORT}" \ + --data-dir "$data_dir/" \ + --config-dir "$data_dir/config/" \ + --region ap-southeast-2 \ + --az "ap-southeast-2a" + + # Point predastore to node1 (only node1 runs predastore until multi-node predastore is implemented) + sed -i "s|host = \"${node_ip}:8443\"|host = \"${NODE1_IP}:8443\"|" "$data_dir/config/hive.toml" + + echo "Node$node_num joined cluster" +} diff --git a/tests/e2e/run-multinode-e2e.sh b/tests/e2e/run-multinode-e2e.sh new file mode 100755 index 000000000..8777df30f --- /dev/null +++ b/tests/e2e/run-multinode-e2e.sh @@ -0,0 +1,332 @@ +#!/bin/bash +set -e + +# Multi-node E2E test runner +# This script sets up a 3-node Hive cluster using simulated IPs on the loopback interface +# and runs distributed instance tests. + +# Ensure we are in the project root +cd "$(dirname "$0")/../.." + +# Source helper functions +source ./tests/e2e/lib/multinode-helpers.sh + +# Cleanup function - ensure resources are cleaned up on exit +cleanup() { + local exit_code=$? + + echo "" + echo "Cleanup triggered (exit code: $exit_code)..." + + if [ $exit_code -ne 0 ]; then + dump_all_node_logs + fi + + # Stop all node services + stop_all_nodes || true + + # Remove simulated IPs + remove_simulated_ips || true + + echo "Cleanup complete" +} +trap cleanup EXIT + +# Use Hive profile for AWS CLI +export AWS_PROFILE=hive + +# Suppress urllib3 InsecureRequestWarning from AWS CLI (we use --no-verify-ssl) +export PYTHONWARNINGS="ignore::urllib3.exceptions.InsecureRequestWarning" + +# Helper to filter AWS CLI output (removes Python warnings that break jq) +aws_json() { + "$@" 2>&1 | grep -v "^urllib3\|^InsecureRequestWarning\|certificate verification" +} + +echo "========================================" +echo "Multi-Node E2E Test Suite" +echo "========================================" +echo "" + +# Phase 1: Environment Setup +echo "Phase 1: Environment Setup" +echo "========================================" + +# Check for KVM support +echo "Checking for KVM support..." +if [ -e /dev/kvm ]; then + echo " /dev/kvm exists" + if [ -w /dev/kvm ]; then + echo " /dev/kvm is writable" + else + echo " ERROR: /dev/kvm is NOT writable" + exit 1 + fi +else + echo " ERROR: /dev/kvm does NOT exist" + exit 1 +fi + +# Check for ip command (iproute2) +if ! command -v ip &> /dev/null; then + echo " ERROR: 'ip' command not found. Install iproute2." + exit 1 +fi + +# Setup simulated network +echo "" +echo "Setting up simulated network..." +add_simulated_ips + +# Create ramdisk mount point +mkdir -p /mnt/ramdisk + +echo "" + +# Phase 2: Cluster Initialization +echo "Phase 2: Cluster Initialization" +echo "========================================" + +# Initialize leader node (node1) +echo "" +init_leader_node + +# Start node1 services first (leader must be running for join) +echo "" +echo "Starting node1 services..." +start_node_services 1 "$HOME/node1" + +# Wait for node1's NATS and cluster service to be ready +echo "Waiting for node1 cluster service..." +sleep 5 + +# Join follower nodes +echo "" +join_follower_node 2 +join_follower_node 3 + +# Start follower node services +echo "" +echo "Starting node2 services..." +start_node_services 2 "$HOME/node2" + +echo "" +echo "Starting node3 services..." +start_node_services 3 "$HOME/node3" + +# Wait for all services to stabilize +echo "" +echo "Waiting for cluster to stabilize..." +sleep 10 + +# Phase 3: Cluster Health Verification +echo "" +echo "Phase 3: Cluster Health Verification" +echo "========================================" + +# Verify NATS cluster +echo "" +verify_nats_cluster 3 || { + echo "WARNING: NATS cluster verification failed, continuing anyway..." +} + +# Wait for gateway on node1 (primary gateway) +echo "" +wait_for_gateway "${NODE1_IP}" 30 + +# Define AWS CLI args pointing to node1's gateway +# Use --no-verify-ssl because the self-signed cert only has localhost SANs, not node IPs +AWS_EC2="aws --endpoint-url https://${NODE1_IP}:${AWSGW_PORT} --no-verify-ssl ec2" + +# Verify gateway responds +echo "" +echo "Testing gateway connectivity..." +aws_json $AWS_EC2 describe-regions | jq -e '.Regions | length > 0' || { + echo "ERROR: Gateway not responding correctly" + exit 1 +} +echo " Gateway is responding" + +# Phase 4: Image and Key Setup +echo "" +echo "Phase 4: Image and Key Setup" +echo "========================================" + +# Discover instance types +echo "Discovering available instance types..." +AVAILABLE_TYPES=$(aws_json $AWS_EC2 describe-instance-types --query 'InstanceTypes[*].InstanceType' --output text) +echo " Available: $AVAILABLE_TYPES" + +# Pick nano instance type +INSTANCE_TYPE=$(echo $AVAILABLE_TYPES | tr ' ' '\n' | grep -m1 'nano') +if [ -z "$INSTANCE_TYPE" ] || [ "$INSTANCE_TYPE" == "None" ]; then + echo "ERROR: No instance types found" + exit 1 +fi +echo " Selected: $INSTANCE_TYPE" + +# Get architecture +ARCH=$(aws_json $AWS_EC2 describe-instance-types --instance-types "$INSTANCE_TYPE" \ + --query 'InstanceTypes[0].ProcessorInfo.SupportedArchitectures[0]' --output text) +echo " Architecture: $ARCH" + +# Create test key +echo "" +echo "Creating test key pair..." +KEY_MATERIAL=$(aws_json $AWS_EC2 create-key-pair --key-name multinode-test-key --query 'KeyMaterial' --output text) +echo "$KEY_MATERIAL" > multinode-test-key.pem +chmod 600 multinode-test-key.pem +echo " Key created: multinode-test-key" + +# Import Ubuntu image (use node1's config and hive-dir) +echo "" +echo "Importing Ubuntu image..." +IMPORT_LOG=$(./bin/hive admin images import \ + --file /root/images/ubuntu-24.04.img \ + --arch "$ARCH" \ + --distro ubuntu \ + --version 24.04 \ + --config "$HOME/node1/config/hive.toml" \ + --hive-dir "$HOME/node1/" \ + --force) +echo "Import output: $IMPORT_LOG" +AMI_ID=$(echo "$IMPORT_LOG" | grep -o 'ami-[a-z0-9]\+') + +if [ -z "$AMI_ID" ]; then + echo "ERROR: Failed to capture AMI ID" + exit 1 +fi +echo " AMI ID: $AMI_ID" + +# Verify AMI +aws_json $AWS_EC2 describe-images --image-ids "$AMI_ID" | jq -e ".Images[0] | select(.ImageId==\"$AMI_ID\")" > /dev/null +echo " AMI verified" + +# Phase 5: Multi-Node Instance Tests +# echo "" +# echo "Phase 5: Multi-Node Instance Tests" +# echo "========================================" + +# # Test 1: Instance Distribution +# echo "" +# echo "Test 1: Instance Distribution" +# echo "----------------------------------------" +# echo "Launching 3 instances to test distribution across nodes..." + +# INSTANCE_IDS=() +# for i in 1 2 3; do +# echo " Launching instance $i..." +# RUN_OUTPUT=$(aws_json $AWS_EC2 run-instances \ +# --image-id "$AMI_ID" \ +# --instance-type "$INSTANCE_TYPE" \ +# --key-name multinode-test-key) + +# INSTANCE_ID=$(echo "$RUN_OUTPUT" | jq -r '.Instances[0].InstanceId') +# if [ -z "$INSTANCE_ID" ] || [ "$INSTANCE_ID" == "null" ]; then +# echo " ERROR: Failed to launch instance $i" +# echo " Output: $RUN_OUTPUT" +# exit 1 +# fi +# echo " Launched: $INSTANCE_ID" +# INSTANCE_IDS+=("$INSTANCE_ID") + +# # Small delay between launches to encourage distribution +# sleep 2 +# done + +# # Wait for all instances to be running +# echo "" +# echo "Waiting for instances to reach running state..." +# for instance_id in "${INSTANCE_IDS[@]}"; do +# wait_for_instance_state "$instance_id" "running" 60 || { +# echo "ERROR: Instance $instance_id failed to start" +# exit 1 +# } +# done + +# # Check distribution +# echo "" +# check_instance_distribution + +# # Test 2: DescribeInstances Aggregation +# echo "" +# echo "Test 2: DescribeInstances Aggregation" +# echo "----------------------------------------" +# echo "Verifying all instances are returned via fan-out query..." + +# DESCRIBE_OUTPUT=$(aws_json $AWS_EC2 describe-instances --query 'Reservations[*].Instances[*].InstanceId' --output text) +# DESCRIBED_COUNT=$(echo "$DESCRIBE_OUTPUT" | wc -w) + +# echo " Launched: ${#INSTANCE_IDS[@]} instances" +# echo " Described: $DESCRIBED_COUNT instances" + +# if [ "$DESCRIBED_COUNT" -lt "${#INSTANCE_IDS[@]}" ]; then +# echo "ERROR: DescribeInstances did not return all instances" +# echo " Expected: ${#INSTANCE_IDS[@]}, Got: $DESCRIBED_COUNT" +# exit 1 +# fi +# echo " Aggregation test passed" + +# # Test 3: Cross-Node Operations +# echo "" +# echo "Test 3: Cross-Node Operations" +# echo "----------------------------------------" +# echo "Testing stop/start/terminate via gateway regardless of instance location..." + +# # Pick first instance for cross-node operations +# TEST_INSTANCE="${INSTANCE_IDS[0]}" +# echo " Test instance: $TEST_INSTANCE" + +# # Stop instance +# echo " Stopping instance..." +# aws_json $AWS_EC2 stop-instances --instance-ids "$TEST_INSTANCE" > /dev/null +# wait_for_instance_state "$TEST_INSTANCE" "stopped" 30 + +# # Start instance +# echo " Starting instance..." +# aws_json $AWS_EC2 start-instances --instance-ids "$TEST_INSTANCE" > /dev/null +# wait_for_instance_state "$TEST_INSTANCE" "running" 30 + +# echo " Cross-node operations test passed" + +# # Test 4: NATS Cluster Health (Post-Operations) +# echo "" +# echo "Test 4: NATS Cluster Health (Post-Operations)" +# echo "----------------------------------------" +# echo "Verifying NATS cluster is still healthy after operations..." + +# verify_nats_cluster 3 || { +# echo "WARNING: NATS cluster verification failed after operations" +# } + +# # Cleanup: Terminate all test instances +# echo "" +# echo "Cleanup: Terminating test instances" +# echo "----------------------------------------" +# for instance_id in "${INSTANCE_IDS[@]}"; do +# echo " Terminating $instance_id..." +# aws_json $AWS_EC2 terminate-instances --instance-ids "$instance_id" > /dev/null +# done + +# # Wait for termination - track failures +# echo " Waiting for termination..." +# TERMINATION_FAILED=0 +# for instance_id in "${INSTANCE_IDS[@]}"; do +# if ! wait_for_instance_state "$instance_id" "terminated" 30; then +# echo " WARNING: Failed to confirm termination of $instance_id" +# TERMINATION_FAILED=1 +# fi +# done + +# if [ $TERMINATION_FAILED -ne 0 ]; then +# echo "" +# echo "ERROR: Some instances failed to terminate properly" +# dump_all_node_logs +# exit 1 +# fi + +# echo "" +# echo "========================================" +# echo "Multi-Node E2E Tests Completed Successfully" +# echo "========================================" +# exit 0