fix(openstack): wait for volume attachment state

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,

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:
+		}
+	}
+}

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) {

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
 }