mirror of
https://github.com/mulgadc/spinifex.git
synced 2026-10-20 03:11:46 +00:00
feat: multi-node e2e test v2 (#8)
This commit is contained in:
Julian Sommer
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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 }}
|
||||
|
||||
|
||||
+56
-4
@@ -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)
|
||||
}
|
||||
|
||||
@@ -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_<bucket>"
|
||||
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
|
||||
}
|
||||
|
||||
@@ -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")
|
||||
}
|
||||
|
||||
+3
-1
@@ -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
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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 \
|
||||
|
||||
Executable
+327
@@ -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 <node_num> <data_dir>
|
||||
# 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 <node_num> <data_dir>
|
||||
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 <instance_id> <target_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 <node_num>
|
||||
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"
|
||||
}
|
||||
Executable
+332
@@ -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
|
||||
Reference in New Issue
Block a user