Add extra fields in node expansion CSI call

pkg/volume/csi/csi_client.go

@@ -54,7 +54,7 @@ type csiClient interface {
 		fsType string,
 		mountOptions []string,
 	) error
-	NodeExpandVolume(ctx context.Context, volumeid, volumePath string, newSize resource.Quantity) (resource.Quantity, error)
+	NodeExpandVolume(ctx context.Context, rsOpts csiResizeOptions) (resource.Quantity, error)
 	NodeUnpublishVolume(
 		ctx context.Context,
 		volID string,
@@ -95,6 +95,16 @@ type csiDriverClient struct {
 	nodeV1ClientCreator nodeV1ClientCreator
 }
 
+type csiResizeOptions struct {
+	volumeid          string
+	volumePath        string
+	stagingTargetPath string
+	fsType            string
+	accessMode        api.PersistentVolumeAccessMode
+	newSize           resource.Quantity
+	mountOptions      []string
+}
+
 var _ csiClient = &csiDriverClient{}
 
 type nodeV1ClientCreator func(addr csiAddr) (
@@ -245,36 +255,55 @@ func (c *csiDriverClient) NodePublishVolume(
 	return err
 }
 
-func (c *csiDriverClient) NodeExpandVolume(ctx context.Context, volumeID, volumePath string, newSize resource.Quantity) (resource.Quantity, error) {
+func (c *csiDriverClient) NodeExpandVolume(ctx context.Context, opts csiResizeOptions) (resource.Quantity, error) {
 	if c.nodeV1ClientCreator == nil {
-		return newSize, fmt.Errorf("version of CSI driver does not support volume expansion")
+		return opts.newSize, fmt.Errorf("version of CSI driver does not support volume expansion")
 	}
 
-	if volumeID == "" {
+	if opts.volumeid == "" {
-		return newSize, errors.New("missing volume id")
+		return opts.newSize, errors.New("missing volume id")
 	}
-	if volumePath == "" {
+	if opts.volumeid == "" {
-		return newSize, errors.New("missing volume path")
+		return opts.newSize, errors.New("missing volume path")
 	}
 
-	if newSize.Value() < 0 {
+	if opts.newSize.Value() < 0 {
-		return newSize, errors.New("size can not be less than 0")
+		return opts.newSize, errors.New("size can not be less than 0")
 	}
 
 	nodeClient, closer, err := c.nodeV1ClientCreator(c.addr)
 	if err != nil {
-		return newSize, err
+		return opts.newSize, err
 	}
 	defer closer.Close()
 
 	req := &csipbv1.NodeExpandVolumeRequest{
-		VolumeId:      volumeID,
+		VolumeId:          opts.volumeid,
-		VolumePath:    volumePath,
+		VolumePath:        opts.volumePath,
-		CapacityRange: &csipbv1.CapacityRange{RequiredBytes: newSize.Value()},
+		StagingTargetPath: opts.stagingTargetPath,
+		CapacityRange:     &csipbv1.CapacityRange{RequiredBytes: opts.newSize.Value()},
+		VolumeCapability: &csipbv1.VolumeCapability{
+			AccessMode: &csipbv1.VolumeCapability_AccessMode{
+				Mode: asCSIAccessModeV1(opts.accessMode),
+			},
+		},
 	}
+	if opts.fsType == fsTypeBlockName {
+		req.VolumeCapability.AccessType = &csipbv1.VolumeCapability_Block{
+			Block: &csipbv1.VolumeCapability_BlockVolume{},
+		}
+	} else {
+		req.VolumeCapability.AccessType = &csipbv1.VolumeCapability_Mount{
+			Mount: &csipbv1.VolumeCapability_MountVolume{
+				FsType:     opts.fsType,
+				MountFlags: opts.mountOptions,
+			},
+		}
+	}
+
 	resp, err := nodeClient.NodeExpandVolume(ctx, req)
 	if err != nil {
-		return newSize, err
+		return opts.newSize, err
 	}
 	updatedQuantity := resource.NewQuantity(resp.CapacityBytes, resource.BinarySI)
 	return *updatedQuantity, nil

pkg/volume/csi/csi_client_test.go

@@ -284,16 +284,34 @@ func (c *fakeCsiDriverClient) NodeSupportsStageUnstage(ctx context.Context) (boo
 	return stageUnstageSet, nil
 }
 
-func (c *fakeCsiDriverClient) NodeExpandVolume(ctx context.Context, volumeid, volumePath string, newSize resource.Quantity) (resource.Quantity, error) {
+func (c *fakeCsiDriverClient) NodeExpandVolume(ctx context.Context, opts csiResizeOptions) (resource.Quantity, error) {
 	c.t.Log("calling fake.NodeExpandVolume")
 	req := &csipbv1.NodeExpandVolumeRequest{
-		VolumeId:      volumeid,
+		VolumeId:          opts.volumeid,
-		VolumePath:    volumePath,
+		VolumePath:        opts.volumePath,
-		CapacityRange: &csipbv1.CapacityRange{RequiredBytes: newSize.Value()},
+		StagingTargetPath: opts.stagingTargetPath,
+		CapacityRange:     &csipbv1.CapacityRange{RequiredBytes: opts.newSize.Value()},
+		VolumeCapability: &csipbv1.VolumeCapability{
+			AccessMode: &csipbv1.VolumeCapability_AccessMode{
+				Mode: asCSIAccessModeV1(opts.accessMode),
+			},
+		},
+	}
+	if opts.fsType == fsTypeBlockName {
+		req.VolumeCapability.AccessType = &csipbv1.VolumeCapability_Block{
+			Block: &csipbv1.VolumeCapability_BlockVolume{},
+		}
+	} else {
+		req.VolumeCapability.AccessType = &csipbv1.VolumeCapability_Mount{
+			Mount: &csipbv1.VolumeCapability_MountVolume{
+				FsType:     opts.fsType,
+				MountFlags: opts.mountOptions,
+			},
+		}
 	}
 	resp, err := c.nodeClient.NodeExpandVolume(ctx, req)
 	if err != nil {
-		return newSize, err
+		return opts.newSize, err
 	}
 	updatedQuantity := resource.NewQuantity(resp.CapacityBytes, resource.BinarySI)
 	return *updatedQuantity, nil
@@ -635,7 +653,8 @@ func TestNodeExpandVolume(t *testing.T) {
 				return nodeClient, fakeCloser, nil
 			},
 		}
-		_, err := client.NodeExpandVolume(context.Background(), tc.volID, tc.volumePath, tc.newSize)
+		opts := csiResizeOptions{volumeid: tc.volID, volumePath: tc.volumePath, newSize: tc.newSize}
+		_, err := client.NodeExpandVolume(context.Background(), opts)
 		checkErr(t, tc.mustFail, err)
 		if !tc.mustFail {
 			fakeCloser.Check()

pkg/volume/csi/expander.go

@@ -94,12 +94,30 @@ func (c *csiPlugin) nodeExpandWithClient(
 		return false, nil
 	}
 
-	volumeTargetPath := resizeOptions.DeviceMountPath
+	pv := resizeOptions.VolumeSpec.PersistentVolume
-	if !fsVolume {
+	if pv == nil {
-		volumeTargetPath = resizeOptions.DevicePath
+		return false, fmt.Errorf("Expander.NodeExpand failed to find associated PersistentVolume for plugin %s", c.GetPluginName())
 	}
 
-	_, err = csClient.NodeExpandVolume(ctx, csiSource.VolumeHandle, volumeTargetPath, resizeOptions.NewSize)
+	opts := csiResizeOptions{
+		volumePath:        resizeOptions.DeviceMountPath,
+		stagingTargetPath: resizeOptions.DeviceStagePath,
+		volumeid:          csiSource.VolumeHandle,
+		newSize:           resizeOptions.NewSize,
+		fsType:            csiSource.FSType,
+		accessMode:        api.ReadWriteOnce,
+		mountOptions:      pv.Spec.MountOptions,
+	}
+	if !fsVolume {
+		opts.volumePath = resizeOptions.DevicePath
+		opts.fsType = fsTypeBlockName
+	}
+
+	if pv.Spec.AccessModes != nil {
+		opts.accessMode = pv.Spec.AccessModes[0]
+	}
+
+	_, err = csClient.NodeExpandVolume(ctx, opts)
 	if err != nil {
 		return false, fmt.Errorf("Expander.NodeExpand failed to expand the volume : %v", err)
 	}

pkg/volume/csi/expander_test.go

@@ -26,24 +26,26 @@ import (
 
 func TestNodeExpand(t *testing.T) {
 	tests := []struct {
-		name          string
+		name            string
-		nodeExpansion bool
+		nodeExpansion   bool
-		nodeStageSet  bool
+		nodeStageSet    bool
-		volumePhase   volume.CSIVolumePhaseType
+		volumePhase     volume.CSIVolumePhaseType
-		success       bool
+		success         bool
-		fsVolume      bool
+		fsVolume        bool
+		deviceStagePath string
 	}{
 		{
 			name:    "when node expansion is not set",
 			success: false,
 		},
 		{
-			name:          "when nodeExpansion=on, nodeStage=on, volumePhase=staged",
+			name:            "when nodeExpansion=on, nodeStage=on, volumePhase=staged",
-			nodeExpansion: true,
+			nodeExpansion:   true,
-			nodeStageSet:  true,
+			nodeStageSet:    true,
-			volumePhase:   volume.CSIVolumeStaged,
+			volumePhase:     volume.CSIVolumeStaged,
-			success:       true,
+			success:         true,
-			fsVolume:      true,
+			fsVolume:        true,
+			deviceStagePath: "/foo/bar",
 		},
 		{
 			name:          "when nodeExpansion=on, nodeStage=off, volumePhase=staged",
@@ -88,21 +90,42 @@ func TestNodeExpand(t *testing.T) {
 				VolumeSpec:      spec,
 				NewSize:         newSize,
 				DeviceMountPath: "/foo/bar",
+				DeviceStagePath: "/foo/bar",
 				DevicePath:      "/mnt/foobar",
 				CSIVolumePhase:  tc.volumePhase,
 			}
 			csiSource, _ := getCSISourceFromSpec(resizeOptions.VolumeSpec)
-
 			csClient := setupClientWithExpansion(t, tc.nodeStageSet, tc.nodeExpansion)
 
+			fakeCSIClient, _ := csClient.(*fakeCsiDriverClient)
+			fakeNodeClient := fakeCSIClient.nodeClient
 			ok, err := plug.nodeExpandWithClient(resizeOptions, csiSource, csClient, tc.fsVolume)
+
+			// verify device staging targer path
+			stagingTargetPath := fakeNodeClient.FakeNodeExpansionRequest.GetStagingTargetPath()
+			if tc.deviceStagePath != "" && tc.deviceStagePath != stagingTargetPath {
+				t.Errorf("For %s: expected staging path %s got %s", tc.name, tc.deviceStagePath, stagingTargetPath)
+			}
+
 			if ok != tc.success {
 				if err != nil {
 					t.Errorf("For %s : expected %v got %v with %v", tc.name, tc.success, ok, err)
 				} else {
 					t.Errorf("For %s : expected %v got %v", tc.name, tc.success, ok)
 				}
-
+			}
+			// verify volume capability received by node expansion request
+			if tc.success {
+				capability := fakeNodeClient.FakeNodeExpansionRequest.GetVolumeCapability()
+				if tc.fsVolume {
+					if capability.GetMount() == nil {
+						t.Errorf("For %s: expected mount accesstype got: %v", tc.name, capability)
+					}
+				} else {
+					if capability.GetBlock() == nil {
+						t.Errorf("For %s: expected block accesstype got: %v", tc.name, capability)
+					}
+				}
 			}
 		})
 	}

pkg/volume/csi/fake/fake_client.go

@@ -79,14 +79,15 @@ type CSIVolume struct {
 
 // NodeClient returns CSI node client
 type NodeClient struct {
-	nodePublishedVolumes map[string]CSIVolume
+	nodePublishedVolumes     map[string]CSIVolume
-	nodeStagedVolumes    map[string]CSIVolume
+	nodeStagedVolumes        map[string]CSIVolume
-	stageUnstageSet      bool
+	stageUnstageSet          bool
-	expansionSet         bool
+	expansionSet             bool
-	volumeStatsSet       bool
+	volumeStatsSet           bool
-	nodeGetInfoResp      *csipb.NodeGetInfoResponse
+	nodeGetInfoResp          *csipb.NodeGetInfoResponse
-	nodeVolumeStatsResp  *csipb.NodeGetVolumeStatsResponse
+	nodeVolumeStatsResp      *csipb.NodeGetVolumeStatsResponse
-	nextErr              error
+	FakeNodeExpansionRequest *csipb.NodeExpandVolumeRequest
+	nextErr                  error
 }
 
 // NewNodeClient returns fake node client
@@ -296,6 +297,8 @@ func (f *NodeClient) NodeExpandVolume(ctx context.Context, req *csipb.NodeExpand
 		return nil, errors.New("required bytes should be greater than 0")
 	}
 
+	f.FakeNodeExpansionRequest = req
+
 	resp := &csipb.NodeExpandVolumeResponse{
 		CapacityBytes: req.GetCapacityRange().RequiredBytes,
 	}

pkg/volume/plugins.go

@@ -115,6 +115,9 @@ type NodeResizeOptions struct {
 	// it would be location where volume was mounted for the pod
 	DeviceMountPath string
 
+	// DeviceStagingPath stores location where the volume is staged
+	DeviceStagePath string
+
 	NewSize resource.Quantity
 	OldSize resource.Quantity
 

pkg/volume/util/operationexecutor/operation_generator.go

@@ -604,6 +604,7 @@ func (og *operationGenerator) GenerateMountVolumeFunc(
 			}
 
 			resizeOptions.DeviceMountPath = deviceMountPath
+			resizeOptions.DeviceStagePath = deviceMountPath
 			resizeOptions.CSIVolumePhase = volume.CSIVolumeStaged
 
 			// NodeExpandVolume will resize the file system if user has requested a resize of
@@ -1505,6 +1506,7 @@ func (og *operationGenerator) GenerateExpandInUseVolumeFunc(
 					return volumeToMount.GenerateError("NodeExpandVolume.GetDeviceMountPath failed", err)
 				}
 				resizeOptions.DeviceMountPath = dmp
+				resizeOptions.DeviceStagePath = dmp
 				resizeDone, simpleErr, detailedErr = og.doOnlineExpansion(volumeToMount, actualStateOfWorld, resizeOptions)
 				if simpleErr != nil || detailedErr != nil {
 					return simpleErr, detailedErr