Repository: kubernetes-sigs/aws-ebs-csi-driver (go)

Issue:
Volume partitions are not supported
/kind bug

**What happened?**
For CSI migration we need all in-tree features supported in the CSI driver. In-tree volumes support mounting volume partitions in  `pv.spec.awsElasticBlockStore.partition` flag. Such capability should be present in the CSI driver too.

**How to reproduce it (as minimally and precisely as possible)?**
Enable CSI migration and use following Pod:

```
apiVersion: v1
kind: Pod
metadata:
  name: inline
spec:
  containers:
  - name: inline
    image: centos:7
    args:
    - sleep
    - '99999'
    volumeMounts:
    - mountPath: "/mnt/storage"
      name: inline
  volumes:
  - name: inline
    awsElasticBlockStore:
      fsType: xfs
      volumeID: vol-072b2b73e088c43b5
      partition: 1
```

**Anything else we need to know?**:
The CSI translation library makes the partition number available in `volume_attribute: partition`: https://github.com/kubernetes/kubernetes/blob/bbb58fa085ccfc4635aa8544686d960115449d95/staging/src/k8s.io/csi-translation-lib/plugins/aws_ebs.go#L112


Pull request description:
support volume partition
**Is this a bug fix or adding new feature?**
Fixes #812 
**What is this PR about? / Why do we need it?**
Add volume partition support when stage&publish volumes to sync with in-tree driver
**What testing is done?** 
unit test
e2e test

Patch:
diff --git a/pkg/driver/constants.go b/pkg/driver/constants.go
index 67a65f44..bc7e0b26 100644
--- a/pkg/driver/constants.go
+++ b/pkg/driver/constants.go
@@ -23,6 +23,19 @@ const (
 	DevicePathKey = "devicePath"
 )
 
+// constants of keys in VolumeContext
+const (
+	// VolumeAttributePartition represents key for partition config in VolumeContext
+	// this represents the partition number on a device used to mount
+	VolumeAttributePartition = "partition"
+)
+
+// constants of disk partition suffix
+const (
+	diskPartitionSuffix     = ""
+	nvmeDiskPartitionSuffix = "p"
+)
+
 // constants of keys in volume parameters
 const (
 	// VolumeTypeKey represents key for volume type
diff --git a/pkg/driver/controller.go b/pkg/driver/controller.go
index 45d3b5c7..381187e3 100644
--- a/pkg/driver/controller.go
+++ b/pkg/driver/controller.go
@@ -450,6 +450,23 @@ func isValidVolumeCapabilities(volCaps []*csi.VolumeCapability) bool {
 	return foundAll
 }
 
+func isValidVolumeContext(volContext map[string]string) bool {
+	//There could be multiple volume attributes in the volumeContext map
+	//Validate here case by case
+	if partition, ok := volContext[VolumeAttributePartition]; ok {
+		partitionInt, err := strconv.ParseInt(partition, 10, 64)
+		if err != nil {
+			klog.Errorf("failed to parse partition %s as int", partition)
+			return false
+		}
+		if partitionInt < 0 {
+			klog.Errorf("invalid partition config, partition = %s", partition)
+			return false
+		}
+	}
+	return true
+}
+
 func (d *controllerService) CreateSnapshot(ctx context.Context, req *csi.CreateSnapshotRequest) (*csi.CreateSnapshotResponse, error) {
 	klog.V(4).Infof("CreateSnapshot: called with args %+v", req)
 	snapshotName := req.GetName()
diff --git a/pkg/driver/node.go b/pkg/driver/node.go
index 6ada6ee5..1ad3ad72 100644
--- a/pkg/driver/node.go
+++ b/pkg/driver/node.go
@@ -121,6 +121,10 @@ func (d *nodeService) NodeStageVolume(ctx context.Context, req *csi.NodeStageVol
 	if !isValidVolumeCapabilities([]*csi.VolumeCapability{volCap}) {
 		return nil, status.Error(codes.InvalidArgument, "Volume capability not supported")
 	}
+	volumeContext := req.GetVolumeContext()
+	if isValidVolumeContext := isValidVolumeContext(volumeContext); !isValidVolumeContext {
+		return nil, status.Error(codes.InvalidArgument, "Volume Attribute is not valid")
+	}
 
 	// If the access type is block, do nothing for stage
 	switch volCap.GetAccessType().(type) {
@@ -158,7 +162,15 @@ func (d *nodeService) NodeStageVolume(ctx context.Context, req *csi.NodeStageVol
 		return nil, status.Error(codes.InvalidArgument, "Device path not provided")
 	}
 
-	source, err := d.findDevicePath(devicePath, volumeID)
+	partition := ""
+	if part, ok := volumeContext[VolumeAttributePartition]; ok {
+		if part != "0" {
+			partition = part
+		} else {
+			klog.Warningf("NodeStageVolume: invalid partition config, will ignore. partition = %v", part)
+		}
+	}
+	source, err := d.findDevicePath(devicePath, volumeID, partition)
 	if err != nil {
 		return nil, status.Errorf(codes.Internal, "Failed to find device path %s. %v", devicePath, err)
 	}
@@ -516,12 +528,26 @@ func (d *nodeService) NodeGetInfo(ctx context.Context, req *csi.NodeGetInfoReque
 func (d *nodeService) nodePublishVolumeForBlock(req *csi.NodePublishVolumeRequest, mountOptions []string) error {
 	target := req.GetTargetPath()
 	volumeID := req.GetVolumeId()
+	volumeContext := req.GetVolumeContext()
 
 	devicePath, exists := req.PublishContext[DevicePathKey]
 	if !exists {
 		return status.Error(codes.InvalidArgument, "Device path not provided")
 	}
-	source, err := d.findDevicePath(devicePath, volumeID)
+	if isValidVolumeContext := isValidVolumeContext(volumeContext); !isValidVolumeContext {
+		return status.Error(codes.InvalidArgument, "Volume Attribute is invalid")
+	}
+
+	partition := ""
+	if part, ok := req.GetVolumeContext()[VolumeAttributePartition]; ok {
+		if part != "0" {
+			partition = part
+		} else {
+			klog.Warningf("NodePublishVolume: invalid partition config, will ignore. partition = %v", part)
+		}
+	}
+
+	source, err := d.findDevicePath(devicePath, volumeID, partition)
 	if err != nil {
 		return status.Errorf(codes.Internal, "Failed to find device path %s. %v", devicePath, err)
 	}
diff --git a/pkg/driver/node_linux.go b/pkg/driver/node_linux.go
index 5454db9b..71f2c53e 100644
--- a/pkg/driver/node_linux.go
+++ b/pkg/driver/node_linux.go
@@ -32,7 +32,7 @@ import (
 // findDevicePath finds path of device and verifies its existence
 // if the device is not nvme, return the path directly
 // if the device is nvme, finds and returns the nvme device path eg. /dev/nvme1n1
-func (d *nodeService) findDevicePath(devicePath, volumeID string) (string, error) {
+func (d *nodeService) findDevicePath(devicePath, volumeID string, partition string) (string, error) {
 	exists, err := d.mounter.PathExists(devicePath)
 	if err != nil {
 		return "", err
@@ -40,6 +40,9 @@ func (d *nodeService) findDevicePath(devicePath, volumeID string) (string, error
 
 	// If the path exists, assume it is not nvme device
 	if exists {
+		if partition != "" {
+			devicePath = devicePath + diskPartitionSuffix + partition
+		}
 		return devicePath, nil
 	}
 
@@ -49,7 +52,14 @@ func (d *nodeService) findDevicePath(devicePath, volumeID string) (string, error
 	// /dev/disk/by-id/nvme-Amazon_Elastic_Block_Store_vol0fab1d5e3f72a5e23
 	nvmeName := "nvme-Amazon_Elastic_Block_Store_" + strings.Replace(volumeID, "-", "", -1)
 
-	return findNvmeVolume(nvmeName)
+	nvmeDevicePath, err := findNvmeVolume(nvmeName)
+	if err != nil {
+		return "", err
+	}
+	if partition != "" {
+		nvmeDevicePath = nvmeDevicePath + nvmeDiskPartitionSuffix + partition
+	}
+	return nvmeDevicePath, nil
 }
 
 // findNvmeVolume looks for the nvme volume with the specified name


Test patch:
diff --git a/pkg/driver/node_test.go b/pkg/driver/node_test.go
index b2198786..1142eadc 100644
--- a/pkg/driver/node_test.go
+++ b/pkg/driver/node_test.go
@@ -48,6 +48,8 @@ func TestNodeStageVolume(t *testing.T) {
 				Mode: csi.VolumeCapability_AccessMode_SINGLE_NODE_WRITER,
 			},
 		}
+		stdVolContext           = map[string]string{VolumeAttributePartition: "1"}
+		devicePathWithPartition = devicePath + "1"
 	)
 	testCases := []struct {
 		name         string
@@ -241,6 +243,17 @@ func TestNodeStageVolume(t *testing.T) {
 			},
 			expectedCode: codes.InvalidArgument,
 		},
+		{
+			name: "fail invalid volumeContext",
+			request: &csi.NodeStageVolumeRequest{
+				PublishContext:    map[string]string{DevicePathKey: devicePath},
+				StagingTargetPath: targetPath,
+				VolumeCapability:  stdVolCap,
+				VolumeContext:     map[string]string{VolumeAttributePartition: "partition1"},
+				VolumeId:          "vol-test",
+			},
+			expectedCode: codes.InvalidArgument,
+		},
 		{
 			name: "success device already mounted at target",
 			request: &csi.NodeStageVolumeRequest{
@@ -260,6 +273,44 @@ func TestNodeStageVolume(t *testing.T) {
 				mockMounter.EXPECT().FormatAndMount(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Times(0)
 			},
 		},
+		{
+			name: "success with partition",
+			request: &csi.NodeStageVolumeRequest{
+				PublishContext:    map[string]string{DevicePathKey: devicePath},
+				StagingTargetPath: targetPath,
+				VolumeCapability:  stdVolCap,
+				VolumeContext:     stdVolContext,
+				VolumeId:          "vol-test",
+			},
+			expectMock: func(mockMounter mocks.MockMounter) {
+				gomock.InOrder(
+					mockMounter.EXPECT().PathExists(gomock.Eq(devicePath)).Return(true, nil),
+					mockMounter.EXPECT().PathExists(gomock.Eq(targetPath)).Return(false, nil),
+				)
+				mockMounter.EXPECT().MakeDir(targetPath).Return(nil)
+				mockMounter.EXPECT().GetDeviceNameFromMount(targetPath).Return("", 1, nil)
+				mockMounter.EXPECT().FormatAndMount(gomock.Eq(devicePathWithPartition), gomock.Eq(targetPath), gomock.Eq(defaultFsType), gomock.Any())
+			},
+		},
+		{
+			name: "success with invalid partition config, will ignore partition",
+			request: &csi.NodeStageVolumeRequest{
+				PublishContext:    map[string]string{DevicePathKey: devicePath},
+				StagingTargetPath: targetPath,
+				VolumeCapability:  stdVolCap,
+				VolumeContext:     map[string]string{VolumeAttributePartition: "0"},
+				VolumeId:          "vol-test",
+			},
+			expectMock: func(mockMounter mocks.MockMounter) {
+				gomock.InOrder(
+					mockMounter.EXPECT().PathExists(gomock.Eq(devicePath)).Return(true, nil),
+					mockMounter.EXPECT().PathExists(gomock.Eq(targetPath)).Return(false, nil),
+				)
+				mockMounter.EXPECT().MakeDir(targetPath).Return(nil)
+				mockMounter.EXPECT().GetDeviceNameFromMount(targetPath).Return("", 1, nil)
+				mockMounter.EXPECT().FormatAndMount(gomock.Eq(devicePath), gomock.Eq(targetPath), gomock.Eq(defaultFsType), gomock.Any())
+			},
+		},
 		{
 			name: "fail with in-flight request",
 			request: &csi.NodeStageVolumeRequest{
@@ -517,6 +568,8 @@ func TestNodePublishVolume(t *testing.T) {
 			Mode: csi.VolumeCapability_AccessMode_SINGLE_NODE_WRITER,
 		},
 	}
+	stdVolContext := map[string]string{"partition": "1"}
+	devicePathWithPartition := devicePath + "1"
 	testCases := []struct {
 		name     string
 		testFunc func(t *testing.T)
@@ -716,6 +769,131 @@ func TestNodePublishVolume(t *testing.T) {
 				}
 			},
 		},
+		{
+			name: "success normal with partition [raw block]",
+			testFunc: func(t *testing.T) {
+				mockCtl := gomock.NewController(t)
+				defer mockCtl.Finish()
+
+				mockMetadata := mocks.NewMockMetadataService(mockCtl)
+				mockMounter := mocks.NewMockMounter(mockCtl)
+
+				awsDriver := &nodeService{
+					metadata: mockMetadata,
+					mounter:  mockMounter,
+					inFlight: internal.NewInFlight(),
+				}
+
+				gomock.InOrder(
+					mockMounter.EXPECT().PathExists(gomock.Eq(devicePath)).Return(true, nil),
+					mockMounter.EXPECT().PathExists(gomock.Eq("/test")).Return(false, nil),
+				)
+				mockMounter.EXPECT().MakeDir(gomock.Eq("/test")).Return(nil)
+				mockMounter.EXPECT().MakeFile(targetPath).Return(nil)
+				mockMounter.EXPECT().Mount(gomock.Eq(devicePathWithPartition), gomock.Eq(targetPath), gomock.Eq(""), gomock.Eq([]string{"bind"})).Return(nil)
+
+				req := &csi.NodePublishVolumeRequest{
+					PublishContext:    map[string]string{DevicePathKey: "/dev/fake"},
+					StagingTargetPath: stagingTargetPath,
+					TargetPath:        targetPath,
+					VolumeCapability: &csi.VolumeCapability{
+						AccessType: &csi.VolumeCapability_Block{
+							Block: &csi.VolumeCapability_BlockVolume{},
+						},
+						AccessMode: &csi.VolumeCapability_AccessMode{
+							Mode: csi.VolumeCapability_AccessMode_SINGLE_NODE_WRITER,
+						},
+					},
+					VolumeContext: stdVolContext,
+					VolumeId:      "vol-test",
+				}
+
+				_, err := awsDriver.NodePublishVolume(context.TODO(), req)
+				if err != nil {
+					t.Fatalf("Expect no error but got: %v", err)
+				}
+			},
+		},
+		{
+			name: "success normal with invalid partition config, will ignore the config [raw block]",
+			testFunc: func(t *testing.T) {
+				mockCtl := gomock.NewController(t)
+				defer mockCtl.Finish()
+
+				mockMetadata := mocks.NewMockMetadataService(mockCtl)
+				mockMounter := mocks.NewMockMounter(mockCtl)
+
+				awsDriver := &nodeService{
+					metadata: mockMetadata,
+					mounter:  mockMounter,
+					inFlight: internal.NewInFlight(),
+				}
+
+				gomock.InOrder(
+					mockMounter.EXPECT().PathExists(gomock.Eq(devicePath)).Return(true, nil),
+					mockMounter.EXPECT().PathExists(gomock.Eq("/test")).Return(false, nil),
+				)
+				mockMounter.EXPECT().MakeDir(gomock.Eq("/test")).Return(nil)
+				mockMounter.EXPECT().MakeFile(targetPath).Return(nil)
+				mockMounter.EXPECT().Mount(gomock.Eq(devicePath), gomock.Eq(targetPath), gomock.Eq(""), gomock.Eq([]string{"bind"})).Return(nil)
+
+				req := &csi.NodePublishVolumeRequest{
+					PublishContext:    map[string]string{DevicePathKey: "/dev/fake"},
+					StagingTargetPath: stagingTargetPath,
+					TargetPath:        targetPath,
+					VolumeCapability: &csi.VolumeCapability{
+						AccessType: &csi.VolumeCapability_Block{
+							Block: &csi.VolumeCapability_BlockVolume{},
+						},
+						AccessMode: &csi.VolumeCapability_AccessMode{
+							Mode: csi.VolumeCapability_AccessMode_SINGLE_NODE_WRITER,
+						},
+					},
+					VolumeContext: map[string]string{VolumeAttributePartition: "0"},
+					VolumeId:      "vol-test",
+				}
+
+				_, err := awsDriver.NodePublishVolume(context.TODO(), req)
+				if err != nil {
+					t.Fatalf("Expect no error but got: %v", err)
+				}
+			},
+		},
+		{
+			name: "Fail invalid volumeContext config [raw block]",
+			testFunc: func(t *testing.T) {
+				mockCtl := gomock.NewController(t)
+				defer mockCtl.Finish()
+
+				mockMetadata := mocks.NewMockMetadataService(mockCtl)
+				mockMounter := mocks.NewMockMounter(mockCtl)
+
+				awsDriver := &nodeService{
+					metadata: mockMetadata,
+					mounter:  mockMounter,
+					inFlight: internal.NewInFlight(),
+				}
+
+				req := &csi.NodePublishVolumeRequest{
+					PublishContext:    map[string]string{DevicePathKey: "/dev/fake"},
+					StagingTargetPath: stagingTargetPath,
+					TargetPath:        targetPath,
+					VolumeCapability: &csi.VolumeCapability{
+						AccessType: &csi.VolumeCapability_Block{
+							Block: &csi.VolumeCapability_BlockVolume{},
+						},
+						AccessMode: &csi.VolumeCapability_AccessMode{
+							Mode: csi.VolumeCapability_AccessMode_SINGLE_NODE_WRITER,
+						},
+					},
+					VolumeContext: map[string]string{VolumeAttributePartition: "partition1"},
+					VolumeId:      "vol-test",
+				}
+
+				_, err := awsDriver.NodePublishVolume(context.TODO(), req)
+				expectErr(t, err, codes.InvalidArgument)
+			},
+		},
 		{
 			name: "fail no device path [raw block]",
 			testFunc: func(t *testing.T) {