[cinder-csi-plugin] support multi-node attachments (kubernetes#1073)

Author: ramineni

docs/using-cinder-csi-plugin.md

@@ -436,6 +436,25 @@ Following prerequisites needed for volume cloning to work :
 
 Sample yamls can be found [here](https://github.com/kubernetes/cloud-provider-openstack/tree/master/examples/cinder-csi-plugin/clone)
 
+### Multi-Attach Volumes
+
+To avail the multiattach feature of cinder, specify the ID/name of cinder volume type that includes an extra-spec capability setting of `multiattach=<is> True` in storage class parameters as shown below.
+
+> Note: This volume type must exist in cinder already (`openstack volume type list`)
+
+This should enable to attach a volume to multiple hosts/servers simultaneously.
+
+Example:
+
+```
+apiVersion: storage.k8s.io/v1
+kind: StorageClass
+metadata:
+  name: csi-sc-cinderplugin
+provisioner: cinder.csi.openstack.org
+parameters:
+  type: <multiattach-volume-type>
+
 ## Running Sanity Tests
 
 Sanity tests create a real instance of driver and fake cloud provider.

pkg/csi/cinder/controllerserver.go

@@ -41,16 +41,18 @@ type controllerServer struct {
 }
 
 func (cs *controllerServer) CreateVolume(ctx context.Context, req *csi.CreateVolumeRequest) (*csi.CreateVolumeResponse, error) {
+	klog.V(4).Infof("CreateVolume: called with args %+v", *req)
 
 	// Volume Name
 	volName := req.GetName()
+	volCapabilities := req.GetVolumeCapabilities()
 
 	if len(volName) == 0 {
-		return nil, status.Error(codes.InvalidArgument, "CreateVolume request missing Volume Name")
+		return nil, status.Error(codes.InvalidArgument, "[CreateVolume] missing Volume Name")
 	}
 
-	if req.VolumeCapabilities == nil {
-		return nil, status.Error(codes.InvalidArgument, "CreateVolume request missing Volume capability")
+	if volCapabilities == nil {
+		return nil, status.Error(codes.InvalidArgument, "[CreateVolume] missing Volume capability")
 	}
 
 	// Volume Size - Default is 1 GiB
@@ -123,18 +125,20 @@ func (cs *controllerServer) CreateVolume(ctx context.Context, req *csi.CreateVol
 	}
 
 	vol, err := cloud.CreateVolume(volName, volSizeGB, volType, volAvailability, snapshotID, sourcevolID, &properties)
+
 	if err != nil {
-		klog.V(3).Infof("Failed to CreateVolume: %v", err)
+		klog.Errorf("Failed to CreateVolume: %v", err)
 		return nil, status.Error(codes.Internal, fmt.Sprintf("CreateVolume failed with error %v", err))
 
 	}
 
-	klog.V(4).Infof("Create volume %s in Availability Zone: %s of size %d GiB", vol.ID, vol.AvailabilityZone, vol.Size)
+	klog.V(4).Infof("CreateVolume: Successfully created volume %s in Availability Zone: %s of size %d GiB", vol.ID, vol.AvailabilityZone, vol.Size)
 
 	return getCreateVolumeResponse(vol), nil
 }
 
 func (cs *controllerServer) DeleteVolume(ctx context.Context, req *csi.DeleteVolumeRequest) (*csi.DeleteVolumeResponse, error) {
+	klog.V(4).Infof("DeleteVolume: called with args %+v", *req)
 
 	// Volume Delete
 	volID := req.GetVolumeId()
@@ -147,11 +151,11 @@ func (cs *controllerServer) DeleteVolume(ctx context.Context, req *csi.DeleteVol
 			klog.V(3).Infof("Volume %s is already deleted.", volID)
 			return &csi.DeleteVolumeResponse{}, nil
 		}
-		klog.V(3).Infof("Failed to DeleteVolume: %v", err)
+		klog.Errorf("Failed to DeleteVolume: %v", err)
 		return nil, status.Error(codes.Internal, fmt.Sprintf("DeleteVolume failed with error %v", err))
 	}
 
-	klog.V(4).Infof("Delete volume %s", volID)
+	klog.V(4).Infof("DeleteVolume: Successfully deleted volume %s", volID)
 
 	return &csi.DeleteVolumeResponse{}, nil
 }

pkg/csi/cinder/controllerserver_test.go

@@ -57,11 +57,12 @@ func TestCreateVolume(t *testing.T) {
 		Name: FakeVolName,
 		VolumeCapabilities: []*csi.VolumeCapability{
 			{
-				AccessType: &csi.VolumeCapability_Mount{
-					Mount: &csi.VolumeCapability_MountVolume{},
+				AccessMode: &csi.VolumeCapability_AccessMode{
+					Mode: csi.VolumeCapability_AccessMode_SINGLE_NODE_WRITER,
 				},
 			},
 		},
+
 		AccessibilityRequirements: &csi.TopologyRequirement{
 			Requisite: []*csi.Topology{
 				{
@@ -109,8 +110,8 @@ func TestCreateVolumeFromSnapshot(t *testing.T) {
 		Name: FakeVolName,
 		VolumeCapabilities: []*csi.VolumeCapability{
 			{
-				AccessType: &csi.VolumeCapability_Mount{
-					Mount: &csi.VolumeCapability_MountVolume{},
+				AccessMode: &csi.VolumeCapability_AccessMode{
+					Mode: csi.VolumeCapability_AccessMode_SINGLE_NODE_WRITER,
 				},
 			},
 		},
@@ -157,8 +158,8 @@ func TestCreateVolumeFromSourceVolume(t *testing.T) {
 		Name: FakeVolName,
 		VolumeCapabilities: []*csi.VolumeCapability{
 			{
-				AccessType: &csi.VolumeCapability_Mount{
-					Mount: &csi.VolumeCapability_MountVolume{},
+				AccessMode: &csi.VolumeCapability_AccessMode{
+					Mode: csi.VolumeCapability_AccessMode_SINGLE_NODE_WRITER,
 				},
 			},
 		},
@@ -195,8 +196,8 @@ func TestCreateVolumeDuplicate(t *testing.T) {
 		Name: "fake-duplicate",
 		VolumeCapabilities: []*csi.VolumeCapability{
 			{
-				AccessType: &csi.VolumeCapability_Mount{
-					Mount: &csi.VolumeCapability_MountVolume{},
+				AccessMode: &csi.VolumeCapability_AccessMode{
+					Mode: csi.VolumeCapability_AccessMode_SINGLE_NODE_WRITER,
 				},
 			},
 		},

pkg/csi/cinder/openstack/openstack_volumes.go

@@ -59,6 +59,7 @@ func (os *OpenStack) CheckBlockStorageAPI() error {
 
 // CreateVolume creates a volume of given size
 func (os *OpenStack) CreateVolume(name string, size int, vtype, availability string, snapshotID string, sourcevolID string, tags *map[string]string) (*volumes.Volume, error) {
+
 	opts := &volumes.CreateOpts{
 		Name:             name,
 		Size:             size,
@@ -145,20 +146,31 @@ func (os *OpenStack) GetVolume(volumeID string) (*volumes.Volume, error) {
 
 // AttachVolume attaches given cinder volume to the compute
 func (os *OpenStack) AttachVolume(instanceID, volumeID string) (string, error) {
+	computeServiceClient := os.compute
+
 	volume, err := os.GetVolume(volumeID)
 	if err != nil {
 		return "", err
 	}
 
-	if len(volume.Attachments) > 0 {
-		if instanceID == volume.Attachments[0].ServerID {
+	for _, att := range volume.Attachments {
+		if instanceID == att.ServerID {
 			klog.V(4).Infof("Disk %s is already attached to instance %s", volumeID, instanceID)
 			return volume.ID, nil
 		}
-		return "", fmt.Errorf("disk %s is attached to a different instance (%s)", volumeID, volume.Attachments[0].ServerID)
 	}
 
-	_, err = volumeattach.Create(os.compute, instanceID, &volumeattach.CreateOpts{
+	if volume.Multiattach {
+		// For multiattach volumes, supported compute api version is 2.60
+		// Init a local thread safe copy of the compute ServiceClient
+		computeServiceClient, err = openstack.NewComputeV2(os.compute.ProviderClient, os.epOpts)
+		if err != nil {
+			return "", err
+		}
+		computeServiceClient.Microversion = "2.60"
+	}
+
+	_, err = volumeattach.Create(computeServiceClient, instanceID, &volumeattach.CreateOpts{
 		VolumeID: volume.ID,
 	}).Extract()
 
@@ -209,21 +221,19 @@ func (os *OpenStack) DetachVolume(instanceID, volumeID string) error {
 		return fmt.Errorf("can not detach volume %s, its status is %s", volume.Name, volume.Status)
 	}
 
-	if len(volume.Attachments) > 0 {
-		if volume.Attachments[0].ServerID != instanceID {
-			return fmt.Errorf("disk: %s is not attached to compute: %s", volume.Name, instanceID)
+	// Incase volume is of type multiattach, it could be attached to more than one instance
+	for _, att := range volume.Attachments {
+		if att.ServerID == instanceID {
+			err = volumeattach.Delete(os.compute, instanceID, volume.ID).ExtractErr()
+			if err != nil {
+				return fmt.Errorf("failed to detach volume %s from compute %s : %v", volume.ID, instanceID, err)
+			}
+			klog.V(2).Infof("Successfully detached volume: %s from compute: %s", volume.ID, instanceID)
+			return nil
 		}
-		err = volumeattach.Delete(os.compute, instanceID, volume.ID).ExtractErr()
-		if err != nil {
-			return fmt.Errorf("failed to delete volume %s from compute %s attached %v", volume.ID, instanceID, err)
-		}
-		klog.V(2).Infof("Successfully detached volume: %s from compute: %s", volume.ID, instanceID)
-
-	} else {
-		return fmt.Errorf("disk: %s has no attachments", volume.Name)
 	}
 
-	return nil
+	return fmt.Errorf("disk: %s has no attachments or not attached to compute %s", volume.ID, instanceID)
 }
 
 // WaitDiskDetached waits for detached
@@ -259,13 +269,15 @@ func (os *OpenStack) GetAttachmentDiskPath(instanceID, volumeID string) (string,
 		return "", fmt.Errorf("can not get device path of volume %s, its status is %s ", volume.Name, volume.Status)
 	}
 
-	if len(volume.Attachments) > 0 && volume.Attachments[0].ServerID != "" {
-		if instanceID == volume.Attachments[0].ServerID {
-			return volume.Attachments[0].Device, nil
+	if len(volume.Attachments) > 0 {
+		for _, att := range volume.Attachments {
+			if att.ServerID == instanceID {
+				return att.Device, nil
+			}
 		}
-		return "", fmt.Errorf("disk %q is attached to a different compute: %q, should be detached before proceeding", volumeID, volume.Attachments[0].ServerID)
+		return "", fmt.Errorf("disk %q is not attached to compute: %q", volumeID, instanceID)
 	}
-	return "", fmt.Errorf("volume %s has no ServerId", volumeID)
+	return "", fmt.Errorf("volume %s has no Attachments", volumeID)
 }
 
 // ExpandVolume expands the volume to new size
@@ -315,11 +327,11 @@ func (os *OpenStack) diskIsAttached(instanceID, volumeID string) (bool, error) {
 	if err != nil {
 		return false, err
 	}
-
-	if len(volume.Attachments) > 0 {
-		return instanceID == volume.Attachments[0].ServerID, nil
+	for _, att := range volume.Attachments {
+		if att.ServerID == instanceID {
+			return true, nil
+		}
 	}
-
 	return false, nil
 }