mirror of
https://github.com/mulgadc/spinifex.git
synced 2026-10-15 17:11:30 +00:00
revert daemon.go (#13)
This commit is contained in:
Julian Sommer
+111
-132
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user