[cinder-csi-plugin] Wait for volume availability before attach

docs/cinder-csi-plugin/using-cinder-csi-plugin.md

@@ -158,6 +158,8 @@ These configuration options pertain to block storage and should appear in the `[
   Optional. Set to `true` if your set of Block Storage (Cinder) AZs does not match your set of Compute (Nova) AZs and you are manually setting the `topology` parameter on your Storage Class(es). For more information, refer to [When trying to use the topology feature, pods are not able to schedule](./troubleshooting.md#when-trying-to-use-the-topology-feature-pods-are-not-able-to-schedule). Defaults to `false`.
 * `ignore-volume-microversion`
   Optional. Set to `true` only when your cinder microversion is older than 3.34. This might cause some features to not work as expected, but aims to allow basic operations like creating a volume. Defaults to `false`
+* `volume-status-poll-delay`
+  Optional. The interval in seconds between volume status polls when waiting for a volume to become available before attaching. Value must be a positive integer representing whole seconds (e.g. `3`, `5`, `10`). Defaults to `3`.
 
 ### Metadata
 These configuration options pertain to metadata and should appear in the `[Metadata]` section of the `$CLOUD_CONFIG` file.

pkg/csi/cinder/controllerserver.go

@@ -34,6 +34,7 @@ import (
 	"google.golang.org/grpc/status"
 	"google.golang.org/protobuf/types/known/timestamppb"
 
+	"k8s.io/apimachinery/pkg/util/wait"
 	sharedcsi "k8s.io/cloud-provider-openstack/pkg/csi"
 	"k8s.io/cloud-provider-openstack/pkg/csi/cinder/openstack"
 	"k8s.io/cloud-provider-openstack/pkg/util"
@@ -313,12 +314,42 @@ func (cs *controllerServer) ControllerPublishVolume(ctx context.Context, req *cs
 		return nil, status.Error(codes.InvalidArgument, "[ControllerPublishVolume] Volume capability must be provided")
 	}
 
-	_, err := cloud.GetVolume(ctx, volumeID)
+	// Wait for the volume to reach a state valid for attachment.
+	// Cinder rejects attachment requests for volumes not in "available"
+	// (or "in-use" for multiattach) state. Without this wait, the driver
+	// would immediately hit a 409 Conflict if CreateVolume has just been
+	// called and the volume is still being provisioned on the backend.
+	vol, err := cs.waitForVolumeAttachable(ctx, cloud, volumeID)
 	if err != nil {
 		if cpoerrors.IsNotFound(err) {
 			return nil, status.Errorf(codes.NotFound, "[ControllerPublishVolume] Volume %s not found", volumeID)
 		}
-		return nil, status.Errorf(codes.Internal, "[ControllerPublishVolume] get volume failed with error %v", err)
+		klog.Errorf("Failed waiting for volume %s to become attachable: %v", volumeID, err)
+		return nil, status.Errorf(codes.FailedPrecondition, "[ControllerPublishVolume] Volume %s is not available for attach: %v", volumeID, err)
+	}
+
+	// Check if the volume is already attached to this instance (idempotent).
+	// CSI spec requires that a repeated ControllerPublishVolume for the same
+	// volume+node returns 0 OK (see CSI v1.12 spec §ControllerPublishVolume).
+	// This uses the volume object returned by waitForVolumeAttachable to
+	// avoid a redundant volumes.Get API call.
+	for _, att := range vol.Attachments {
+		if instanceID == att.ServerID {
+			klog.V(4).Infof("Volume %s is already attached to instance %s", volumeID, instanceID)
+			devicePath, err := cloud.GetAttachmentDiskPath(ctx, instanceID, volumeID)
+			if err != nil {
+				klog.Errorf("Failed to GetAttachmentDiskPath: %v", err)
+				return nil, status.Errorf(codes.Internal, "[ControllerPublishVolume] failed to get device path of attached volume: %v", err)
+			}
+			pvInfo := map[string]string{"DevicePath": devicePath}
+			return &csi.ControllerPublishVolumeResponse{PublishContext: pvInfo}, nil
+		}
+	}
+
+	// If the volume is in-use and not multiattach-capable, it is attached to
+	// a different node (same-node was handled above). Fail immediately.
+	if vol.Status == openstack.VolumeInUseStatus && !vol.Multiattach {
+		return nil, status.Errorf(codes.FailedPrecondition, "[ControllerPublishVolume] Volume %s is already attached to a different node and does not support multiattach", volumeID)
 	}
 
 	_, err = cloud.GetInstanceByID(ctx, instanceID)
@@ -359,6 +390,45 @@ func (cs *controllerServer) ControllerPublishVolume(ctx context.Context, req *cs
 	}, nil
 }
 
+// waitForVolumeAttachable polls the volume status until it leaves transitional
+// states (e.g. "creating", "downloading") and reaches "available" or "in-use".
+// For "in-use" volumes, the caller is responsible for checking whether the
+// attachment is an idempotent same-node retry (CSI spec requires 0 OK) or a
+// conflict with a different node.
+// The wait is bounded by the context deadline (typically the gRPC RPC timeout).
+func (cs *controllerServer) waitForVolumeAttachable(ctx context.Context, cloud openstack.IOpenStack, volumeID string) (*volumes.Volume, error) {
+	bsOpts := cloud.GetBlockStorageOpts()
+
+	var lastVol *volumes.Volume
+	err := wait.PollUntilContextCancel(ctx, bsOpts.VolumeStatusPollInterval(), true, func(ctx context.Context) (bool, error) {
+		vol, err := cloud.GetVolume(ctx, volumeID)
+		if err != nil {
+			return false, err
+		}
+		lastVol = vol
+
+		if vol.Status == openstack.VolumeAvailableStatus {
+			return true, nil
+		}
+		if vol.Status == openstack.VolumeInUseStatus {
+			// Return the volume as-is. The caller will inspect
+			// attachments for idempotency before rejecting.
+			return true, nil
+		}
+
+		if slices.Contains(openstack.VolumeErrorStates[:], vol.Status) {
+			return false, fmt.Errorf("volume %s is in error state: %s", volumeID, vol.Status)
+		}
+		return false, nil
+	})
+
+	if err != nil && ctx.Err() != nil {
+		return lastVol, fmt.Errorf("timeout waiting for volume %s to become available: %w", volumeID, err)
+	}
+
+	return lastVol, err
+}
+
 func (cs *controllerServer) ControllerUnpublishVolume(ctx context.Context, req *csi.ControllerUnpublishVolumeRequest) (*csi.ControllerUnpublishVolumeResponse, error) {
 	klog.V(4).Infof("ControllerUnpublishVolume: called with args %+v", protosanitizer.StripSecrets(req))
 

pkg/csi/cinder/controllerserver_test.go

@@ -424,6 +424,7 @@ func TestCreateVolumeFromSourceVolume(t *testing.T) {
 	osmock.On("CreateVolume", FakeVolName, mock.AnythingOfType("int"), FakeVolType, "", "", FakeVolID, "", properties).Return(&FakeVolFromSourceVolume, nil)
 	osmock.On("GetVolumesByName", FakeVolName).Return(FakeVolListEmpty, nil)
 	osmock.On("GetBlockStorageOpts").Return(openstack.BlockStorageOpts{})
+	osmock.On("GetVolume", FakeVolID).Return(&FakeVol, nil)
 
 	assert := assert.New(t)
 
@@ -529,6 +530,13 @@ func TestDeleteVolume(t *testing.T) {
 func TestControllerPublishVolume(t *testing.T) {
 	fakeCs, osmock := fakeControllerServer()
 
+	fakeAvailableVol := &volumes.Volume{
+		ID:     FakeVolID,
+		Name:   FakeVolName,
+		Status: "available",
+	}
+	osmock.On("GetBlockStorageOpts").Return(openstack.BlockStorageOpts{})
+	osmock.On("GetVolume", FakeVolID).Return(fakeAvailableVol, nil)
 	osmock.On("AttachVolume", FakeNodeID, FakeVolID).Return(FakeVolID, nil)
 	osmock.On("WaitDiskAttached", FakeNodeID, FakeVolID).Return(nil)
 	osmock.On("GetAttachmentDiskPath", FakeNodeID, FakeVolID).Return(FakeDevicePath, nil)
@@ -564,6 +572,84 @@ func TestControllerPublishVolume(t *testing.T) {
 	assert.Equal(expectedRes, actualRes)
 }
 
+// TestControllerPublishVolumeIdempotentSameNode verifies that repeating a
+// ControllerPublishVolume request for the same volume+node after the volume is
+// already in-use returns success (CSI spec idempotency requirement).
+func TestControllerPublishVolumeIdempotentSameNode(t *testing.T) {
+	fakeCs, osmock := fakeControllerServer()
+
+	// Volume is already in-use with an attachment to the same node.
+	fakeInUseVol := &volumes.Volume{
+		ID:     FakeVolID,
+		Name:   FakeVolName,
+		Status: "in-use",
+		Attachments: []volumes.Attachment{
+			{ServerID: FakeNodeID},
+		},
+	}
+	osmock.On("GetBlockStorageOpts").Return(openstack.BlockStorageOpts{})
+	osmock.On("GetVolume", FakeVolID).Return(fakeInUseVol, nil)
+	osmock.On("GetAttachmentDiskPath", FakeNodeID, FakeVolID).Return(FakeDevicePath, nil)
+
+	assert := assert.New(t)
+
+	fakeReq := &csi.ControllerPublishVolumeRequest{
+		VolumeId: FakeVolID,
+		NodeId:   FakeNodeID,
+		VolumeCapability: &csi.VolumeCapability{
+			AccessType: &csi.VolumeCapability_Mount{
+				Mount: &csi.VolumeCapability_MountVolume{},
+			},
+		},
+		Readonly: false,
+	}
+
+	expectedRes := &csi.ControllerPublishVolumeResponse{
+		PublishContext: map[string]string{
+			"DevicePath": FakeDevicePath,
+		},
+	}
+
+	actualRes, err := fakeCs.ControllerPublishVolume(FakeCtx, fakeReq)
+	assert.Nil(err)
+	assert.Equal(expectedRes, actualRes)
+}
+
+// TestControllerPublishVolumeInUseDifferentNode verifies that attempting to
+// attach a non-multiattach volume already in-use on a different node returns
+// FailedPrecondition.
+func TestControllerPublishVolumeInUseDifferentNode(t *testing.T) {
+	fakeCs, osmock := fakeControllerServer()
+
+	// Volume is in-use on a different node, not multiattach.
+	fakeInUseVol := &volumes.Volume{
+		ID:     FakeVolID,
+		Name:   FakeVolName,
+		Status: "in-use",
+		Attachments: []volumes.Attachment{
+			{ServerID: "different-node-id"},
+		},
+	}
+	osmock.On("GetBlockStorageOpts").Return(openstack.BlockStorageOpts{})
+	osmock.On("GetVolume", FakeVolID).Return(fakeInUseVol, nil)
+
+	fakeReq := &csi.ControllerPublishVolumeRequest{
+		VolumeId: FakeVolID,
+		NodeId:   FakeNodeID,
+		VolumeCapability: &csi.VolumeCapability{
+			AccessType: &csi.VolumeCapability_Mount{
+				Mount: &csi.VolumeCapability_MountVolume{},
+			},
+		},
+		Readonly: false,
+	}
+
+	_, err := fakeCs.ControllerPublishVolume(FakeCtx, fakeReq)
+	assert := assert.New(t)
+	assert.NotNil(err)
+	assert.Contains(err.Error(), "already attached to a different node")
+}
+
 // Test ControllerUnpublishVolume
 func TestControllerUnpublishVolume(t *testing.T) {
 	fakeCs, osmock := fakeControllerServer()
@@ -1101,6 +1187,12 @@ func TestControllerExpandVolume(t *testing.T) {
 	tState := []string{"available", "in-use"}
 	osmock.On("ExpandVolume", FakeVolID, openstack.VolumeAvailableStatus, 5).Return(nil)
 	osmock.On("WaitVolumeTargetStatus", FakeVolID, tState).Return(nil)
+	osmock.On("GetVolume", FakeVolID).Return(&volumes.Volume{
+		ID:     FakeVolID,
+		Name:   FakeVolName,
+		Size:   FakeCapacityGiB,
+		Status: "available",
+	}, nil)
 
 	assert := assert.New(t)
 
@@ -1131,7 +1223,7 @@ func TestControllerExpandVolume(t *testing.T) {
 func TestValidateVolumeCapabilities(t *testing.T) {
 	fakeCs, osmock := fakeControllerServer()
 
-	osmock.On("GetVolume", FakeVolID).Return(FakeVol1)
+	osmock.On("GetVolume", FakeVolID).Return(&FakeVol1, nil)
 
 	assert := assert.New(t)
 

pkg/csi/cinder/nodeserver_test.go

@@ -92,7 +92,7 @@ func TestNodePublishVolume(t *testing.T) {
 
 	mmock.On("ScanForAttach", FakeDevicePath).Return(nil)
 	mmock.On("IsLikelyNotMountPointAttach", FakeTargetPath).Return(true, nil)
-	omock.On("GetVolume", FakeVolID).Return(FakeVol, nil)
+	omock.On("GetVolume", FakeVolID).Return(&FakeVol, nil)
 
 	assert := assert.New(t)
 
@@ -167,7 +167,7 @@ func TestNodeStageVolume(t *testing.T) {
 
 	mmock.On("GetDevicePath", FakeVolID).Return(FakeDevicePath, nil)
 	mmock.On("IsLikelyNotMountPointAttach", FakeStagingTargetPath).Return(true, nil)
-	omock.On("GetVolume", FakeVolID).Return(FakeVol, nil)
+	omock.On("GetVolume", FakeVolID).Return(&FakeVol, nil)
 
 	assert := assert.New(t)
 
@@ -241,7 +241,7 @@ func TestNodeUnpublishVolume(t *testing.T) {
 	fakeNs, omock, mmock, _ := fakeNodeServer()
 
 	mmock.On("UnmountPath", FakeTargetPath).Return(nil)
-	omock.On("GetVolume", FakeVolID).Return(FakeVol, nil)
+	omock.On("GetVolume", FakeVolID).Return(&FakeVol, nil)
 
 	assert := assert.New(t)
 
@@ -269,7 +269,7 @@ func TestNodeUnstageVolume(t *testing.T) {
 	fakeNs, omock, mmock, _ := fakeNodeServer()
 
 	mmock.On("UnmountPath", FakeStagingTargetPath).Return(nil)
-	omock.On("GetVolume", FakeVolID).Return(FakeVol, nil)
+	omock.On("GetVolume", FakeVolID).Return(&FakeVol, nil)
 
 	assert := assert.New(t)
 

pkg/csi/cinder/openstack/openstack.go

@@ -22,6 +22,7 @@ import (
 	"net/http"
 	"os"
 	"sync"
+	"time"
 
 	"github.com/gophercloud/gophercloud/v2"
 	"github.com/gophercloud/gophercloud/v2/openstack"
@@ -91,6 +92,22 @@ type BlockStorageOpts struct {
 	RescanOnResize           bool  `gcfg:"rescan-on-resize"`
 	IgnoreVolumeAZ           bool  `gcfg:"ignore-volume-az"`
 	IgnoreVolumeMicroversion bool  `gcfg:"ignore-volume-microversion"`
+	// VolumeStatusPollDelay is the interval in seconds between volume status
+	// polls when waiting for a volume to become available before attaching.
+	// Must be a positive integer representing whole seconds (e.g. 3, 5, 10).
+	// Defaults to 3 seconds if not set or zero.
+	VolumeStatusPollDelay int `gcfg:"volume-status-poll-delay"`
+}
+
+const defaultVolumeStatusPollInterval = 3 * time.Second
+
+// VolumeStatusPollInterval returns the configured poll interval for volume
+// status checks, or the default (3s) if not configured.
+func (opts BlockStorageOpts) VolumeStatusPollInterval() time.Duration {
+	if opts.VolumeStatusPollDelay > 0 {
+		return time.Duration(opts.VolumeStatusPollDelay) * time.Second
+	}
+	return defaultVolumeStatusPollInterval
 }
 
 type Config struct {

pkg/csi/cinder/openstack/openstack_mock.go

@@ -29,14 +29,6 @@ import (
 	"k8s.io/cloud-provider-openstack/pkg/util/metadata"
 )
 
-var fakeVol1 = volumes.Volume{
-	ID:               "261a8b81-3660-43e5-bab8-6470b65ee4e9",
-	Name:             "fake-duplicate",
-	Status:           "available",
-	AvailabilityZone: "nova",
-	Size:             1,
-}
-
 var fakeSnapshot = snapshots.Snapshot{
 	ID:       "261a8b81-3660-43e5-bab8-6470b65ee4e8",
 	Name:     "fake-snapshot",
@@ -134,7 +126,11 @@ func (_m *OpenStackMock) DeleteVolume(ctx context.Context, volumeID string) erro
 
 // GetVolume provides a mock function with given fields: volumeID
 func (_m *OpenStackMock) GetVolume(ctx context.Context, volumeID string) (*volumes.Volume, error) {
-	return &fakeVol1, nil
+	args := _m.Called(volumeID)
+	if args.Get(0) == nil {
+		return nil, args.Error(1)
+	}
+	return args.Get(0).(*volumes.Volume), args.Error(1)
 }
 
 // DetachVolume provides a mock function with given fields: instanceID, volumeID

pkg/csi/cinder/openstack/openstack_volumes.go

@@ -22,6 +22,7 @@ import (
 	"fmt"
 	"net/http"
 	"net/url"
+	"slices"
 	"strings"
 	"time"
 
@@ -56,7 +57,7 @@ const (
 	volumeDescription        = "Created by OpenStack Cinder CSI driver"
 )
 
-var volumeErrorStates = [...]string{"error", "error_extending", "error_deleting"}
+var VolumeErrorStates = [...]string{"error", "error_extending", "error_deleting"}
 
 // CreateVolume creates a volume of given size
 func (os *OpenStack) CreateVolume(ctx context.Context, opts *volumes.CreateOpts, schedulerHints volumes.SchedulerHintOptsBuilder) (*volumes.Volume, error) {
@@ -278,15 +279,11 @@ func (os *OpenStack) WaitVolumeTargetStatus(ctx context.Context, volumeID string
 		if err != nil {
 			return false, err
 		}
-		for _, t := range tStatus {
-			if vol.Status == t {
-				return true, nil
-			}
+		if slices.Contains(tStatus, vol.Status) {
+			return true, nil
 		}
-		for _, eState := range volumeErrorStates {
-			if vol.Status == eState {
-				return false, fmt.Errorf("volume is in Error State : %s", vol.Status)
-			}
+		if slices.Contains(VolumeErrorStates[:], vol.Status) {
+			return false, fmt.Errorf("volume is in Error State : %s", vol.Status)
 		}
 		return false, nil
 	})