diff --git a/hive/daemon/daemon.go b/hive/daemon/daemon.go index 694aa3957..c21ddbfcc 100644 --- a/hive/daemon/daemon.go +++ b/hive/daemon/daemon.go @@ -445,101 +445,9 @@ func (d *Daemon) Start() error { return fmt.Errorf("failed to start cluster manager: %w", err) } - // Initialize JetStream for KV state storage (required - no disk fallback) - // Start with 1 replica to allow single-node startup, then upgrade if cluster has more nodes - // Retry with backoff if JetStream is not ready yet - maxRetries := 10 - retryDelay := 500 * time.Millisecond - - for attempt := 1; attempt <= maxRetries; attempt++ { - d.jsManager, err = NewJetStreamManager(d.natsConn, 1) - if err != nil { - slog.Warn("Failed to init JetStream", "error", err, "attempt", attempt, "maxRetries", maxRetries) - if attempt < maxRetries { - time.Sleep(retryDelay) - retryDelay *= 2 // Exponential backoff - if retryDelay > 5*time.Second { - retryDelay = 5 * time.Second - } - continue - } - return fmt.Errorf("failed to initialize JetStream after %d attempts: %w", maxRetries, err) - } - - if err := d.jsManager.InitKVBucket(); err != nil { - slog.Warn("Failed to init KV bucket", "error", err, "attempt", attempt, "maxRetries", maxRetries) - if attempt < maxRetries { - time.Sleep(retryDelay) - retryDelay *= 2 - if retryDelay > 5*time.Second { - retryDelay = 5 * time.Second - } - continue - } - return fmt.Errorf("failed to initialize JetStream KV bucket after %d attempts: %w", maxRetries, err) - } - - // Success - slog.Info("JetStream KV store initialized successfully", "replicas", 1, "attempts", attempt) - break - } - - // 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 - err = d.LoadState() - if err != nil { - slog.Warn("Failed to load state from disk, continuing with empty state", "error", err) - } else { - - slog.Info("Loaded state from disk", "instance count", len(d.Instances.VMS)) - - //d.mu.Lock() - // Ensure mutex is usable after unmarshalling - d.Instances.Mu = sync.Mutex{} - - for i := range d.Instances.VMS { - instance := d.Instances.VMS[i] - instance.EBSRequests.Mu = sync.Mutex{} - instance.QMPClient = &qmp.QMPClient{} - d.Instances.VMS[i] = instance - //d.Instances.VMS[i].EBSRequests.Mu = sync.Mutex{} - // d.Instances.VMS[i].QMPClient.Mu = sync.Mutex{} - - if instance.Attributes.StopInstance { - slog.Info("Instance flagged as user initiated stop, skipping", "instance", instance.ID) - - } else if instance.Status != "terminated" { - instanceType, ok := d.resourceMgr.instanceTypes[instance.InstanceType] - if ok { - slog.Info("Re-allocating resources for instance", "instanceId", instance.ID, "type", instance.InstanceType) - if err := d.resourceMgr.allocate(instanceType); err != nil { - slog.Error("Failed to re-allocate resources for instance on startup", "instanceId", instance.ID, "err", err) - } - } - - slog.Info("Launching instance", "instance", instance.ID) - err = d.LaunchInstance(instance) - if err != nil { - slog.Error("Failed to launch instance:", "err", err) - } - } else { - slog.Info("Instance state is terminated, skipping", "instance", instance.ID) - } - - } - //d.mu.Unlock() - - // Launch running instances - - } + // Initialize JetStream in background - don't block daemon startup + // This allows the daemon to start accepting requests while waiting for NATS cluster to form + go d.initJetStreamBackground() // Create instance service for handling EC2 instance operations d.instanceService = handlers_ec2_instance.NewInstanceServiceImpl(d.config, d.resourceMgr.instanceTypes, d.natsConn, &d.Instances) @@ -893,10 +801,16 @@ func (d *Daemon) ClusterManager() error { // WriteState writes the instance state to JetStream KV store (required) func (d *Daemon) WriteState() error { - if d.jsManager == nil { - return fmt.Errorf("JetStream manager not initialized - cannot write state") + d.mu.Lock() + jsm := d.jsManager + d.mu.Unlock() + + if jsm == nil { + // JetStream not ready yet - this is expected during startup + slog.Debug("WriteState skipped - JetStream not initialized yet") + return nil } - if err := d.jsManager.WriteState(d.node, &d.Instances); err != nil { + if err := jsm.WriteState(d.node, &d.Instances); err != nil { slog.Error("JetStream write failed", "error", err) return fmt.Errorf("failed to write state to JetStream: %w", err) } @@ -931,21 +845,121 @@ func (d *Daemon) InitaliseVMs() { // LoadState loads the instance state from JetStream KV store (required) func (d *Daemon) LoadState() error { - if d.jsManager == nil { + d.mu.Lock() + jsm := d.jsManager + d.mu.Unlock() + + if jsm == nil { return fmt.Errorf("JetStream manager not initialized - cannot load state") } - instances, err := d.jsManager.LoadState(d.node) + instances, err := jsm.LoadState(d.node) if err != nil { slog.Error("JetStream load failed", "error", err) return fmt.Errorf("failed to load state from JetStream: %w", err) } // Copy only the VMS map, not the mutex + d.Instances.Mu.Lock() d.Instances.VMS = instances.VMS + d.Instances.Mu.Unlock() return nil } +// initJetStreamBackground initializes JetStream in a background goroutine. +// This allows the daemon to start and accept cluster join requests while waiting +// for the NATS cluster to form and JetStream to become available. +func (d *Daemon) initJetStreamBackground() { + retryDelay := 500 * time.Millisecond + const maxDelay = 10 * time.Second + + for attempt := 1; ; attempt++ { + // Check for shutdown + select { + case <-d.ctx.Done(): + slog.Info("JetStream init cancelled due to shutdown") + return + default: + } + + // Try to create JetStream manager and init KV bucket + jsm, err := NewJetStreamManager(d.natsConn, 1) + if err != nil { + slog.Warn("Failed to init JetStream (will retry)", "error", err, "attempt", attempt) + } else if err := jsm.InitKVBucket(); err != nil { + slog.Warn("Failed to init KV bucket (will retry)", "error", err, "attempt", attempt) + } else { + // Success - store the manager + d.mu.Lock() + d.jsManager = jsm + d.mu.Unlock() + slog.Info("JetStream KV store initialized successfully", "replicas", 1, "attempts", attempt) + break + } + + // Wait before retry (with shutdown check) + select { + case <-d.ctx.Done(): + slog.Info("JetStream init cancelled due to shutdown") + return + case <-time.After(retryDelay): + } + + // Exponential backoff + retryDelay *= 2 + if retryDelay > maxDelay { + retryDelay = maxDelay + } + } + + // Note: Replica upgrades are handled by the /join endpoint on the leader node. + // Non-leader nodes should not try to update replicas to avoid race conditions. + + // Load state and restore VMs + if err := d.LoadState(); err != nil { + slog.Warn("Failed to load state from JetStream, continuing with empty state", "error", err) + slog.Info("JetStream background initialization complete") + return + } + + slog.Info("Loaded state from JetStream", "instance_count", len(d.Instances.VMS)) + + // Restore VMs from loaded state + d.Instances.Mu.Lock() + defer d.Instances.Mu.Unlock() + + for id, instance := range d.Instances.VMS { + instance.EBSRequests.Mu = sync.Mutex{} + instance.QMPClient = &qmp.QMPClient{} + d.Instances.VMS[id] = instance + + if instance.Attributes.StopInstance { + slog.Info("Instance flagged as user initiated stop, skipping", "instance", instance.ID) + continue + } + + if instance.Status == "terminated" { + slog.Info("Instance state is terminated, skipping", "instance", instance.ID) + continue + } + + // Re-allocate resources + if instanceType, ok := d.resourceMgr.instanceTypes[instance.InstanceType]; ok { + slog.Info("Re-allocating resources for instance", "instanceId", instance.ID, "type", instance.InstanceType) + if err := d.resourceMgr.allocate(instanceType); err != nil { + slog.Error("Failed to re-allocate resources for instance on startup", "instanceId", instance.ID, "err", err) + } + } + + slog.Info("Launching instance", "instance", instance.ID) + if err := d.LaunchInstance(instance); err != nil { + slog.Error("Failed to launch instance", "err", err) + } + } + + slog.Info("JetStream background initialization complete") +} + // NATS events func (d *Daemon) handleEC2StartInstances(msg *nats.Msg) { diff --git a/tests/e2e/lib/multinode-helpers.sh b/tests/e2e/lib/multinode-helpers.sh index aaef66389..0b9e2240e 100755 --- a/tests/e2e/lib/multinode-helpers.sh +++ b/tests/e2e/lib/multinode-helpers.sh @@ -218,23 +218,18 @@ wait_for_gateway() { } # Check instance distribution across nodes -# Returns the instance counts per node +# Counts QEMU processes per node IP from hostfwd bindings 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 + local node_ip="${SIMULATED_NETWORK}.$i" + # Count QEMU processes with hostfwd bound to this node's IP + local count + count=$(ps auxw | grep qemu-system | grep -v grep | grep -c "hostfwd=tcp:${node_ip}:" 2>/dev/null || echo "0") + echo " Node$i ($node_ip): $count instances" + total=$((total + count)) done echo " Total instances: $total" @@ -272,6 +267,32 @@ get_ssh_port() { return 1 } +# Find the SSH host IP for an instance from the QEMU process hostfwd setting +# Usage: get_ssh_host +# Returns: the SSH host IP (e.g., 10.11.12.1, 127.0.0.1), or "127.0.0.1" if not found/empty +get_ssh_host() { + local instance_id="$1" + + local qemu_cmd + qemu_cmd=$(ps auxw | grep "$instance_id" | grep qemu-system | grep -v grep || true) + + if [ -n "$qemu_cmd" ]; then + # Extract host from hostfwd=tcp:HOST:PORT-:22 + # Pattern matches: hostfwd=tcp:IP:PORT-:22 where IP can be empty + local ssh_host + ssh_host=$(echo "$qemu_cmd" | sed -n 's/.*hostfwd=tcp:\([^:]*\):[0-9]*-:22.*/\1/p') + + if [ -n "$ssh_host" ]; then + echo "$ssh_host" + return 0 + fi + fi + + # Default to localhost if not found + echo "127.0.0.1" + return 0 +} + # Find which node an instance is running on (for multi-node setups) # Usage: get_instance_node # Returns: node number (1, 2, or 3), or empty if not found