diff --git a/internal/openstack/client.go b/internal/openstack/client.go index 03aaf7e..b9b6b6a 100644 --- a/internal/openstack/client.go +++ b/internal/openstack/client.go @@ -264,6 +264,12 @@ func (c *ClientSet) CreateResourcesForVirtualMachine(ctx context.Context, vm *ob return err } + waitOpts := VolumeWaitOptsFromContext(ctx) + volume, err = c.EnsureVolumeAvailable(ctx, volume.ID, waitOpts.DetachTimeout) + if err != nil { + return err + } + blockDevices = append(blockDevices, servers.BlockDevice{ BootIndex: diskIndex, SourceType: servers.SourceVolume, diff --git a/internal/openstack/volume_wait.go b/internal/openstack/volume_wait.go new file mode 100644 index 0000000..1e84000 --- /dev/null +++ b/internal/openstack/volume_wait.go @@ -0,0 +1,135 @@ +package openstack + +import ( + "context" + "fmt" + "strings" + "time" + + "github.com/gophercloud/gophercloud/v2/openstack/blockstorage/v3/volumes" + log "github.com/sirupsen/logrus" +) + +const ( + DefaultVolumeCreateTimeout = 10 * time.Minute + DefaultVolumeAttachTimeout = 10 * time.Minute + DefaultVolumeDetachTimeout = 10 * time.Minute +) + +type volumeWaitOptsContextKey struct{} + +type VolumeWaitOpts struct { + CreateTimeout time.Duration + AttachTimeout time.Duration + DetachTimeout time.Duration +} + +func DefaultVolumeWaitOpts() VolumeWaitOpts { + return VolumeWaitOpts{ + CreateTimeout: DefaultVolumeCreateTimeout, + AttachTimeout: DefaultVolumeAttachTimeout, + DetachTimeout: DefaultVolumeDetachTimeout, + } +} + +func WithVolumeWaitOpts(ctx context.Context, opts VolumeWaitOpts) context.Context { + return context.WithValue(ctx, volumeWaitOptsContextKey{}, opts) +} + +func VolumeWaitOptsFromContext(ctx context.Context) VolumeWaitOpts { + opts, ok := ctx.Value(volumeWaitOptsContextKey{}).(VolumeWaitOpts) + if !ok { + return DefaultVolumeWaitOpts() + } + + defaults := DefaultVolumeWaitOpts() + if opts.CreateTimeout == 0 { + opts.CreateTimeout = defaults.CreateTimeout + } + if opts.AttachTimeout == 0 { + opts.AttachTimeout = defaults.AttachTimeout + } + if opts.DetachTimeout == 0 { + opts.DetachTimeout = defaults.DetachTimeout + } + + return opts +} + +func volumeAttachedTo(volume *volumes.Volume, serverID string) bool { + for _, attachment := range volume.Attachments { + if strings.EqualFold(attachment.ServerID, serverID) { + return true + } + } + + return false +} + +func (c *ClientSet) WaitForVolumeAvailable(ctx context.Context, volumeID string, timeout time.Duration) (*volumes.Volume, error) { + return c.waitForVolume(ctx, volumeID, timeout, "available with no attachments", func(volume *volumes.Volume) bool { + return volume.Status == "available" && len(volume.Attachments) == 0 + }) +} + +func (c *ClientSet) EnsureVolumeAvailable(ctx context.Context, volumeID string, timeout time.Duration) (*volumes.Volume, error) { + volume, err := volumes.Get(ctx, c.BlockStorage, volumeID).Extract() + if err != nil { + return nil, err + } + + if (volume.Status == "reserved" || volume.Status == "attaching") && len(volume.Attachments) == 0 { + log.WithFields(log.Fields{ + "volume_id": volume.ID, + "status": volume.Status, + }).Warn("Volume has no attachments but is not available, unreserving") + + err = volumes.Unreserve(ctx, c.BlockStorage, volume.ID).ExtractErr() + if err != nil { + return nil, err + } + } + + return c.WaitForVolumeAvailable(ctx, volumeID, timeout) +} + +func (c *ClientSet) WaitForVolumeAttached(ctx context.Context, volumeID, serverID string, timeout time.Duration) (*volumes.Volume, error) { + return c.waitForVolume(ctx, volumeID, timeout, fmt.Sprintf("in-use and attached to %s", serverID), func(volume *volumes.Volume) bool { + return volume.Status == "in-use" && volumeAttachedTo(volume, serverID) + }) +} + +func (c *ClientSet) waitForVolume(ctx context.Context, volumeID string, timeout time.Duration, desiredState string, ready func(*volumes.Volume) bool) (*volumes.Volume, error) { + ctx, cancel := context.WithTimeout(ctx, timeout) + defer cancel() + + ticker := time.NewTicker(2 * time.Second) + defer ticker.Stop() + + var lastVolume *volumes.Volume + + for { + volume, err := volumes.Get(ctx, c.BlockStorage, volumeID).Extract() + if err != nil { + return nil, err + } + + lastVolume = volume + if ready(volume) { + return volume, nil + } + + log.WithFields(log.Fields{ + "volume_id": volume.ID, + "status": volume.Status, + "attachments": len(volume.Attachments), + "desired": desiredState, + }).Debug("Waiting for volume state") + + select { + case <-ctx.Done(): + return nil, fmt.Errorf("timed out waiting for volume %s to become %s; last status=%s attachments=%d: %w", volumeID, desiredState, lastVolume.Status, len(lastVolume.Attachments), ctx.Err()) + case <-ticker.C: + } + } +} diff --git a/internal/target/openstack.go b/internal/target/openstack.go index 20ce4d6..7e103c7 100644 --- a/internal/target/openstack.go +++ b/internal/target/openstack.go @@ -3,6 +3,7 @@ package target import ( "context" "errors" + "fmt" "math" "os" "path/filepath" @@ -69,6 +70,59 @@ func findDevice(volumeID string) (string, error) { return "", nil } +func hasVolumeAttachment(volume *volumes.Volume, serverID string) bool { + for _, attachment := range volume.Attachments { + if strings.EqualFold(attachment.ServerID, serverID) { + return true + } + } + + return false +} + +func describeVolumeAttachments(volume *volumes.Volume) []string { + attachments := make([]string, 0, len(volume.Attachments)) + for _, attachment := range volume.Attachments { + attachments = append(attachments, fmt.Sprintf("%s:%s", attachment.ServerID, attachment.ID)) + } + + return attachments +} + +func waitForDevice(ctx context.Context, volumeID string, timeout time.Duration) (string, error) { + ctx, cancel := context.WithTimeout(ctx, timeout) + defer cancel() + + ticker := time.NewTicker(1 * time.Second) + defer ticker.Stop() + + for { + devicePath, err := findDevice(volumeID) + if err != nil { + return "", err + } + + if devicePath != "" { + log.WithFields(log.Fields{ + "volume_id": volumeID, + "device": devicePath, + }).Info("Device found") + + return devicePath, nil + } + + log.WithFields(log.Fields{ + "volume_id": volumeID, + }).Debug("Device for volume not found, checking again...") + + select { + case <-ctx.Done(): + return "", errors.Join(errors.New("timed out waiting for volume device to appear"), ctx.Err()) + case <-ticker.C: + } + } +} + func (t *OpenStack) Connect(ctx context.Context) error { volume, err := t.ClientSet.GetVolumeForDisk(ctx, t.VirtualMachine, t.Disk) volumeMetadata := map[string]string{ @@ -180,13 +234,18 @@ func (t *OpenStack) Connect(ctx context.Context) error { "volume_id": volume.ID, }).Info("Attaching volume") - path, err := t.GetPath(ctx) + instanceUUID, err := openstack.GetCurrentInstanceUUID() if err != nil { return err } - if path == "" { - instanceUUID, err := openstack.GetCurrentInstanceUUID() + waitOpts := openstack.VolumeWaitOptsFromContext(ctx) + if !hasVolumeAttachment(volume, instanceUUID) { + if len(volume.Attachments) > 0 { + return fmt.Errorf("volume %s is attached to other server(s): %v", volume.ID, describeVolumeAttachments(volume)) + } + + volume, err = t.ClientSet.EnsureVolumeAvailable(ctx, volume.ID, waitOpts.AttachTimeout) if err != nil { return err } @@ -201,38 +260,20 @@ func (t *OpenStack) Connect(ctx context.Context) error { if err != nil { return err } + } else { + log.WithFields(log.Fields{ + "instance_uuid": instanceUUID, + "volume_id": volume.ID, + }).Info("Volume is already attached to this instance") + } - timeoutTimer := time.After(2 * time.Minute) - ticker := time.NewTicker(1 * time.Second) - defer ticker.Stop() - - for { - select { - case <-timeoutTimer: - return errors.New("timed out waiting for volume to attach") - case <-ticker.C: - devicePath, err := findDevice(volume.ID) - if err != nil { - return err - } - - if devicePath != "" { - log.WithFields(log.Fields{ - "volume_id": volume.ID, - "device": devicePath, - }).Info("Device found") - - return nil - } - - log.WithFields(log.Fields{ - "volume_id": volume.ID, - }).Info("Device for volume not found, checking again...") - } - } + _, err = t.ClientSet.WaitForVolumeAttached(ctx, volume.ID, instanceUUID, waitOpts.AttachTimeout) + if err != nil { + return err } - return nil + _, err = waitForDevice(ctx, volume.ID, waitOpts.AttachTimeout) + return err } func (t *OpenStack) createVolume(ctx context.Context, opts *VolumeCreateOpts, metadata map[string]string) (*volumes.Volume, error) { @@ -248,12 +289,10 @@ func (t *OpenStack) createVolume(ctx context.Context, opts *VolumeCreateOpts, me return nil, err } - ctx, cancel := context.WithTimeout(ctx, 60*time.Second) - defer cancel() - - err = volumes.WaitForStatus(ctx, t.ClientSet.BlockStorage, volume.ID, "available") + waitOpts := openstack.VolumeWaitOptsFromContext(ctx) + volume, err = t.ClientSet.WaitForVolumeAvailable(ctx, volume.ID, waitOpts.CreateTimeout) if err != nil { - return nil, errors.Join(errors.New("timed out waiting for volume to be available"), err) + return nil, err } return volume, nil @@ -281,32 +320,28 @@ func (t *OpenStack) Disconnect(ctx context.Context) error { return err } - devicePath, err := findDevice(volume.ID) + instanceUUID, err := openstack.GetCurrentInstanceUUID() if err != nil { return err } - if devicePath != "" { - instanceUUID, err := openstack.GetCurrentInstanceUUID() - if err != nil { - return err - } + if hasVolumeAttachment(volume, instanceUUID) { + log.WithFields(log.Fields{ + "volume_id": volume.ID, + "instance_uuid": instanceUUID, + }).Info("Detaching volume") err = volumeattach.Delete(ctx, t.ClientSet.Compute, instanceUUID, volume.ID).ExtractErr() if err != nil { return err } - - ctx, cancel := context.WithTimeout(ctx, 60*time.Second) - defer cancel() - - err = volumes.WaitForStatus(ctx, t.ClientSet.BlockStorage, volume.ID, "available") - if err != nil { - return errors.Join(errors.New("timed out waiting for volume to be available"), err) - } + } else if len(volume.Attachments) > 0 { + return fmt.Errorf("volume %s is attached to other server(s): %v", volume.ID, describeVolumeAttachments(volume)) } - return nil + waitOpts := openstack.VolumeWaitOptsFromContext(ctx) + _, err = t.ClientSet.WaitForVolumeAvailable(ctx, volume.ID, waitOpts.DetachTimeout) + return err } func (t *OpenStack) Exists(ctx context.Context) (bool, error) { diff --git a/internal/vmware_nbdkit/vmware_nbdkit.go b/internal/vmware_nbdkit/vmware_nbdkit.go index 78b5134..490582b 100644 --- a/internal/vmware_nbdkit/vmware_nbdkit.go +++ b/internal/vmware_nbdkit/vmware_nbdkit.go @@ -323,7 +323,15 @@ func (s *NbdkitServer) SyncToTarget(ctx context.Context, t target.Target, runV2V if err != nil { return err } - defer t.Disconnect(ctx) + connected := true + defer func() { + if connected { + err := t.Disconnect(ctx) + if err != nil { + log.WithError(err).Error("Failed to disconnect from target") + } + } + }() c := make(chan os.Signal, 1) signal.Notify(c, os.Interrupt, syscall.SIGTERM) @@ -387,5 +395,7 @@ func (s *NbdkitServer) SyncToTarget(ctx context.Context, t target.Target, runV2V } } - return nil + err = t.Disconnect(ctx) + connected = false + return err } diff --git a/main.go b/main.go index e0244ec..3135dbc 100644 --- a/main.go +++ b/main.go @@ -72,7 +72,10 @@ var ( busType BusTypeOpts vzUnsafeVolumeByName bool osType string - enableQemuGuestAgent bool + enableQemuGuestAgent bool + volumeCreateTimeout time.Duration + volumeAttachTimeout time.Duration + volumeDetachTimeout time.Duration ) var rootCmd = &cobra.Command{ @@ -199,6 +202,12 @@ var rootCmd = &cobra.Command{ ctx = context.WithValue(ctx, "enableQemuGuestAgent", enableQemuGuestAgent) + ctx = openstack.WithVolumeWaitOpts(ctx, openstack.VolumeWaitOpts{ + CreateTimeout: volumeCreateTimeout, + AttachTimeout: volumeAttachTimeout, + DetachTimeout: volumeDetachTimeout, + }) + cmd.SetContext(ctx) return nil @@ -350,9 +359,15 @@ func init() { rootCmd.PersistentFlags().BoolVar(&vzUnsafeVolumeByName, "vz-unsafe-volume-by-name", false, "Only use the name to find a volume - workaround for virtuozzu - dangerous option") - rootCmd.PersistentFlags().StringVar(&osType, "os-type", "", "Set os_type in the volume (image) metadata, (if set to \"auto\", it tries to detect the type from VMware GuestId)") + rootCmd.PersistentFlags().StringVar(&osType, "os-type", "", "Set os_type in the volume (image) metadata, (if set to \"auto\", it tries to detect the type from VMware GuestId)") + + rootCmd.PersistentFlags().DurationVar(&volumeCreateTimeout, "volume-create-timeout", openstack.DefaultVolumeCreateTimeout, "Maximum time to wait for a created OpenStack volume to become available") + + rootCmd.PersistentFlags().DurationVar(&volumeAttachTimeout, "volume-attach-timeout", openstack.DefaultVolumeAttachTimeout, "Maximum time to wait for an OpenStack volume attach and local device discovery") + + rootCmd.PersistentFlags().DurationVar(&volumeDetachTimeout, "volume-detach-timeout", openstack.DefaultVolumeDetachTimeout, "Maximum time to wait for an OpenStack volume detach") - rootCmd.PersistentFlags().BoolVar(&enableQemuGuestAgent, "enable-qemu-guest-agent", false, "Sets the hw_qemu_guest_agent metadata parameter to yes") + rootCmd.PersistentFlags().BoolVar(&enableQemuGuestAgent, "enable-qemu-guest-agent", false, "Sets the hw_qemu_guest_agent metadata parameter to yes") cutoverCmd.Flags().StringVar(&flavorId, "flavor", "", "OpenStack Flavor ID") cutoverCmd.MarkFlagRequired("flavor")