diff --git a/hive/daemon/daemon.go b/hive/daemon/daemon.go index 40d616ea8..694aa3957 100644 --- a/hive/daemon/daemon.go +++ b/hive/daemon/daemon.go @@ -445,9 +445,101 @@ func (d *Daemon) Start() error { return fmt.Errorf("failed to start cluster manager: %w", err) } - // 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() + // 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 + + } // Create instance service for handling EC2 instance operations d.instanceService = handlers_ec2_instance.NewInstanceServiceImpl(d.config, d.resourceMgr.instanceTypes, d.natsConn, &d.Instances) @@ -801,16 +893,10 @@ func (d *Daemon) ClusterManager() error { // WriteState writes the instance state to JetStream KV store (required) func (d *Daemon) WriteState() error { - 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 d.jsManager == nil { + return fmt.Errorf("JetStream manager not initialized - cannot write state") } - if err := jsm.WriteState(d.node, &d.Instances); err != nil { + if err := d.jsManager.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) } @@ -845,121 +931,21 @@ func (d *Daemon) InitaliseVMs() { // LoadState loads the instance state from JetStream KV store (required) func (d *Daemon) LoadState() error { - d.mu.Lock() - jsm := d.jsManager - d.mu.Unlock() - - if jsm == nil { + if d.jsManager == nil { return fmt.Errorf("JetStream manager not initialized - cannot load state") } - instances, err := jsm.LoadState(d.node) + instances, err := d.jsManager.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) { @@ -1091,7 +1077,6 @@ func (d *Daemon) SendQMPCommand(q *qmp.QMPClient, cmd qmp.QMPCommand, instanceId // handleEC2Events processes incoming EC2 events (start, stop, terminate) func (d *Daemon) handleEC2Events(msg *nats.Msg) { - slog.Info("handleEC2Events called", "subject", msg.Subject) var command qmp.Command var resp *qmp.QMPResponse @@ -1108,7 +1093,8 @@ func (d *Daemon) handleEC2Events(msg *nats.Msg) { return } - slog.Info("handleEC2Events received command", "subject", msg.Subject, "instanceId", command.ID, "start", command.Attributes.StartInstance, "stop", command.Attributes.StopInstance, "terminate", command.Attributes.TerminateInstance) + log.Printf("Received message on subject: %s", msg.Subject) + log.Printf("Message data: %s", string(msg.Data)) d.Instances.Mu.Lock() instance, ok := d.Instances.VMS[command.ID] @@ -1573,12 +1559,10 @@ func (d *Daemon) stopInstance(instances map[string]*vm.VM, deleteVolume bool) er // For stop operations, keep the subscription so we can receive start commands if deleteVolume { for _, instance := range instances { - subKey := fmt.Sprintf("ec2.cmd.%s", instance.ID) - if sub, ok := d.natsSubscriptions[subKey]; ok && sub != nil { - slog.Info("Unsubscribing from NATS subject", "instance", instance.ID, "subject", subKey) - sub.Unsubscribe() - delete(d.natsSubscriptions, subKey) - } + slog.Info("Unsubscribing from NATS subject", "instance", instance.ID) + d.natsSubscriptions[fmt.Sprintf("ec2.cmd.%s", instance.ID)].Unsubscribe() + // TODO: Remove redundant subscription if not used + //d.natsSubscriptions[fmt.Sprintf("ec2.describe.%s", instance.ID)].Unsubscribe() } } return nil @@ -1737,16 +1721,11 @@ func (d *Daemon) LaunchInstance(instance *vm.VM) (err error) { d.mu.Lock() defer d.mu.Unlock() - // Check if subscription already exists (e.g., when restarting a stopped instance) - subKey := fmt.Sprintf("ec2.cmd.%s", instance.ID) - if existingSub, ok := d.natsSubscriptions[subKey]; ok && existingSub != nil && existingSub.IsValid() { - slog.Info("NATS subscription already exists for instance", "instance", instance.ID) - } else { - d.natsSubscriptions[subKey], err = d.natsConn.QueueSubscribe(subKey, "hive-events", d.handleEC2Events) - if err != nil { - slog.Error("failed to subscribe to NATS", "err", err) - return err - } + d.natsSubscriptions[instance.ID], err = d.natsConn.QueueSubscribe(fmt.Sprintf("ec2.cmd.%s", instance.ID), "hive-events", d.handleEC2Events) + + if err != nil { + slog.Error("failed to subscribe to NATS", "err", err) + return err } // TODO: Replaced with describe-instances with Inbox subscription