This commit is contained in:
Julian Sommer
2026-02-11 15:10:55 +10:00
parent 998649cbab
commit 0ff03be0dd
3 changed files with 38 additions and 36 deletions
+18 -12
View File
@@ -643,20 +643,21 @@ func (d *Daemon) restoreInstances() {
continue
}
// QEMU is not running - handle transitional states from interrupted operations
// QEMU is not running -- resolve transitional states from interrupted operations
switch instance.Status {
case vm.StateStopping:
instance.Status = vm.StateStopped
slog.Info("Instance was stopping, QEMU exited, marking stopped", "instance", instance.ID)
if err := d.WriteState(); err != nil {
slog.Error("Failed to write state", "instance", instance.ID, "error", err)
case vm.StateStopping, vm.StateShuttingDown:
prevStatus := instance.Status
if instance.Status == vm.StateStopping {
instance.Status = vm.StateStopped
} else {
instance.Status = vm.StateTerminated
}
continue
case vm.StateShuttingDown:
instance.Status = vm.StateTerminated
slog.Info("Instance was shutting-down, QEMU exited, marking terminated", "instance", instance.ID)
slog.Info("QEMU exited during transition, finalizing state",
"instance", instance.ID, "from", prevStatus, "to", instance.Status)
if err := d.WriteState(); err != nil {
slog.Error("Failed to write state", "instance", instance.ID, "error", err)
slog.Error("Failed to persist state, will retry on next restart",
"instance", instance.ID, "error", err)
instance.Status = prevStatus // revert so next restart retries
}
continue
case vm.StateRunning:
@@ -697,6 +698,11 @@ func (d *Daemon) reconnectInstance(instance *vm.VM) error {
sub, err := d.natsConn.QueueSubscribe(fmt.Sprintf("ec2.cmd.%s", instance.ID), "hive-events", d.handleEC2Events)
if err != nil {
d.mu.Unlock()
// Clean up the QMP connection we just opened
if instance.QMPClient != nil && instance.QMPClient.Conn != nil {
instance.QMPClient.Conn.Close()
instance.QMPClient = nil
}
return fmt.Errorf("failed to subscribe to NATS: %w", err)
}
d.natsSubscriptions[instance.ID] = sub
@@ -705,7 +711,7 @@ func (d *Daemon) reconnectInstance(instance *vm.VM) error {
instance.Status = vm.StateRunning
if err := d.WriteState(); err != nil {
slog.Error("Failed to persist reconnected instance state", "instanceId", instance.ID, "err", err)
return fmt.Errorf("failed to persist reconnected instance state: %w", err)
}
slog.Info("Successfully reconnected to running instance", "instance", instance.ID)
+2 -2
View File
@@ -874,13 +874,13 @@ func (d *Daemon) handleEC2CreateImage(msg *nats.Msg) {
d.Instances.Mu.Unlock()
if !ok {
slog.Error("CreateImage: instance not found", "instanceId", instanceID)
slog.Warn("CreateImage: instance not found", "instanceId", instanceID)
respondWithError(awserrors.ErrorInvalidInstanceIDNotFound)
return
}
if status != vm.StateRunning && status != vm.StateStopped {
slog.Error("CreateImage: instance not in valid state", "instanceId", instanceID, "status", status)
slog.Warn("CreateImage: instance not in valid state", "instanceId", instanceID, "status", status)
respondWithError(awserrors.ErrorIncorrectInstanceState)
return
}
+18 -22
View File
@@ -253,15 +253,13 @@ func (s *ImageServiceImpl) CreateImageFromInstance(params CreateImageParams) (*e
"rootVolumeId", params.RootVolumeID, "amiId", amiID, "snapshotId", snapshotID,
"isRunning", params.IsRunning)
// Step 1: Snapshot the root volume
// Step 1: Snapshot root volume (live via NATS or offline from S3)
snapshotFn := s.snapshotStoppedVolume
if params.IsRunning {
if err := s.snapshotRunningVolume(params.RootVolumeID, snapshotID); err != nil {
return nil, err
}
} else {
if err := s.snapshotStoppedVolume(params.RootVolumeID, snapshotID); err != nil {
return nil, err
}
snapshotFn = s.snapshotRunningVolume
}
if err := snapshotFn(params.RootVolumeID, snapshotID); err != nil {
return nil, err
}
// Step 2: Read source volume config for size
@@ -273,7 +271,11 @@ func (s *ImageServiceImpl) CreateImageFromInstance(params CreateImageParams) (*e
volumeSizeGiB := volumeConfig.VolumeMetadata.SizeGiB
// Step 3: Read source AMI config for architecture, platform, etc.
var sourceAMI viperblock.AMIMetadata
sourceAMI := viperblock.AMIMetadata{
Architecture: "x86_64",
PlatformDetails: "Linux/UNIX",
Virtualization: "hvm",
}
if params.SourceImageID != "" {
srcCfg, err := s.getAMIConfig(params.SourceImageID)
if err != nil {
@@ -281,12 +283,6 @@ func (s *ImageServiceImpl) CreateImageFromInstance(params CreateImageParams) (*e
return nil, errors.New(awserrors.ErrorServerInternal)
}
sourceAMI = srcCfg
} else {
sourceAMI = viperblock.AMIMetadata{
Architecture: "x86_64",
PlatformDetails: "Linux/UNIX",
Virtualization: "hvm",
}
}
// Step 4: Store snapshot metadata
@@ -422,18 +418,18 @@ func (s *ImageServiceImpl) snapshotStoppedVolume(volumeID, snapshotID string) er
return errors.New(awserrors.ErrorServerInternal)
}
// Clean up local WAL files regardless of snapshot success or failure
defer func() {
if err := vb.RemoveLocalFiles(); err != nil {
slog.Error("snapshotStoppedVolume: failed to remove local files", "volumeId", volumeID, "err", err)
}
}()
if _, err := vb.CreateSnapshot(snapshotID); err != nil {
slog.Error("snapshotStoppedVolume: CreateSnapshot failed", "volumeId", volumeID, "snapshotId", snapshotID, "err", err)
if cleanupErr := vb.RemoveLocalFiles(); cleanupErr != nil {
slog.Error("snapshotStoppedVolume: failed to remove local files after snapshot failure", "volumeId", volumeID, "err", cleanupErr)
}
return errors.New(awserrors.ErrorServerInternal)
}
if err := vb.RemoveLocalFiles(); err != nil {
slog.Error("snapshotStoppedVolume: failed to remove local files", "volumeId", volumeID, "err", err)
}
slog.Info("snapshotStoppedVolume: snapshot created", "volumeId", volumeID, "snapshotId", snapshotID)
return nil
}