Merge pull request kubernetes#86968 from gnufied/add-extra-csi-fields

Add extra fields in node expansion CSI call

Author: k8s-ci-robot

pkg/volume/csi/csi_block.go

@@ -107,9 +107,9 @@ func (m *csiBlockMapper) GetGlobalMapPath(spec *volume.Spec) (string, error) {
 	return dir, nil
 }
 
-// getStagingPath returns a staging path for a directory (on the node) that should be used on NodeStageVolume/NodeUnstageVolume
+// GetStagingPath returns a staging path for a directory (on the node) that should be used on NodeStageVolume/NodeUnstageVolume
 // Example: plugins/kubernetes.io/csi/volumeDevices/staging/{specName}
-func (m *csiBlockMapper) getStagingPath() string {
+func (m *csiBlockMapper) GetStagingPath() string {
 	return filepath.Join(m.plugin.host.GetVolumeDevicePluginDir(CSIPluginName), "staging", m.specName)
 }
 
@@ -143,7 +143,7 @@ func (m *csiBlockMapper) stageVolumeForBlock(
 ) (string, error) {
 	klog.V(4).Infof(log("blockMapper.stageVolumeForBlock called"))
 
-	stagingPath := m.getStagingPath()
+	stagingPath := m.GetStagingPath()
 	klog.V(4).Infof(log("blockMapper.stageVolumeForBlock stagingPath set [%s]", stagingPath))
 
 	// Check whether "STAGE_UNSTAGE_VOLUME" is set
@@ -237,7 +237,7 @@ func (m *csiBlockMapper) publishVolumeForBlock(
 		ctx,
 		m.volumeID,
 		m.readOnly,
-		m.getStagingPath(),
+		m.GetStagingPath(),
 		publishPath,
 		accessMode,
 		publishVolumeInfo,
@@ -255,26 +255,26 @@ func (m *csiBlockMapper) publishVolumeForBlock(
 }
 
 // SetUpDevice ensures the device is attached returns path where the device is located.
-func (m *csiBlockMapper) SetUpDevice() error {
+func (m *csiBlockMapper) SetUpDevice() (string, error) {
 	if !m.plugin.blockEnabled {
-		return errors.New("CSIBlockVolume feature not enabled")
+		return "", errors.New("CSIBlockVolume feature not enabled")
 	}
 	klog.V(4).Infof(log("blockMapper.SetUpDevice called"))
 
 	// Get csiSource from spec
 	if m.spec == nil {
-		return errors.New(log("blockMapper.SetUpDevice spec is nil"))
+		return "", errors.New(log("blockMapper.SetUpDevice spec is nil"))
 	}
 
 	csiSource, err := getCSISourceFromSpec(m.spec)
 	if err != nil {
-		return errors.New(log("blockMapper.SetUpDevice failed to get CSI persistent source: %v", err))
+		return "", errors.New(log("blockMapper.SetUpDevice failed to get CSI persistent source: %v", err))
 	}
 
 	driverName := csiSource.Driver
 	skip, err := m.plugin.skipAttach(driverName)
 	if err != nil {
-		return errors.New(log("blockMapper.SetupDevice failed to check CSIDriver for %s: %v", driverName, err))
+		return "", errors.New(log("blockMapper.SetupDevice failed to check CSIDriver for %s: %v", driverName, err))
 	}
 
 	var attachment *storage.VolumeAttachment
@@ -284,7 +284,7 @@ func (m *csiBlockMapper) SetUpDevice() error {
 		attachID := getAttachmentName(csiSource.VolumeHandle, csiSource.Driver, nodeName)
 		attachment, err = m.k8s.StorageV1().VolumeAttachments().Get(context.TODO(), attachID, meta.GetOptions{})
 		if err != nil {
-			return errors.New(log("blockMapper.SetupDevice failed to get volume attachment [id=%v]: %v", attachID, err))
+			return "", errors.New(log("blockMapper.SetupDevice failed to get volume attachment [id=%v]: %v", attachID, err))
 		}
 	}
 
@@ -299,11 +299,11 @@ func (m *csiBlockMapper) SetUpDevice() error {
 
 	csiClient, err := m.csiClientGetter.Get()
 	if err != nil {
-		return errors.New(log("blockMapper.SetUpDevice failed to get CSI client: %v", err))
+		return "", errors.New(log("blockMapper.SetUpDevice failed to get CSI client: %v", err))
 	}
 
 	// Call NodeStageVolume
-	_, err = m.stageVolumeForBlock(ctx, csiClient, accessMode, csiSource, attachment)
+	stagingPath, err := m.stageVolumeForBlock(ctx, csiClient, accessMode, csiSource, attachment)
 	if err != nil {
 		if volumetypes.IsOperationFinishedError(err) {
 			cleanupErr := m.cleanupOrphanDeviceFiles()
@@ -312,10 +312,10 @@ func (m *csiBlockMapper) SetUpDevice() error {
 				klog.V(4).Infof("Failed to clean up block volume directory %s", cleanupErr)
 			}
 		}
-		return err
+		return "", err
 	}
 
-	return nil
+	return stagingPath, nil
 }
 
 func (m *csiBlockMapper) MapPodDevice() (string, error) {
@@ -435,7 +435,7 @@ func (m *csiBlockMapper) TearDownDevice(globalMapPath, devicePath string) error
 	}
 
 	// Call NodeUnstageVolume
-	stagingPath := m.getStagingPath()
+	stagingPath := m.GetStagingPath()
 	if _, err := os.Stat(stagingPath); err != nil {
 		if os.IsNotExist(err) {
 			klog.V(4).Infof(log("blockMapper.TearDownDevice stagingPath(%s) has already been deleted, skip calling NodeUnstageVolume", stagingPath))
@@ -471,7 +471,7 @@ func (m *csiBlockMapper) cleanupOrphanDeviceFiles() error {
 
 	// Remove artifacts of NodeStage.
 	// stagingPath: xxx/plugins/kubernetes.io/csi/volumeDevices/staging/<volume name>
-	stagingPath := m.getStagingPath()
+	stagingPath := m.GetStagingPath()
 	if err := os.Remove(stagingPath); err != nil && !os.IsNotExist(err) {
 		return errors.New(log("failed to delete volume staging path [%s]: %v", stagingPath, err))
 	}

pkg/volume/csi/csi_block_test.go

@@ -123,7 +123,7 @@ func TestBlockMapperGetStagingPath(t *testing.T) {
 			t.Fatalf("Failed to make a new Mapper: %v", err)
 		}
 
-		path := csiMapper.getStagingPath()
+		path := csiMapper.GetStagingPath()
 
 		if tc.path != path {
 			t.Errorf("expecting path %s, got %s", tc.path, path)
@@ -234,13 +234,12 @@ func TestBlockMapperSetupDevice(t *testing.T) {
 	}
 	t.Log("created attachement ", attachID)
 
-	err = csiMapper.SetUpDevice()
+	stagingPath, err := csiMapper.SetUpDevice()
 	if err != nil {
 		t.Fatalf("mapper failed to SetupDevice: %v", err)
 	}
 
 	// Check if NodeStageVolume staged to the right path
-	stagingPath := csiMapper.getStagingPath()
 	svols := csiMapper.csiClient.(*fakeCsiDriverClient).nodeClient.GetNodeStagedVolumes()
 	svol, ok := svols[csiMapper.volumeID]
 	if !ok {
@@ -278,7 +277,7 @@ func TestBlockMapperSetupDeviceError(t *testing.T) {
 	}
 	t.Log("created attachement ", attachID)
 
-	err = csiMapper.SetUpDevice()
+	stagingPath, err := csiMapper.SetUpDevice()
 	if err == nil {
 		t.Fatal("mapper unexpectedly succeeded")
 	}
@@ -293,7 +292,7 @@ func TestBlockMapperSetupDeviceError(t *testing.T) {
 	if _, err := os.Stat(devDir); err == nil {
 		t.Errorf("volume publish device directory %s was not deleted", devDir)
 	}
-	stagingPath := csiMapper.getStagingPath()
+
 	if _, err := os.Stat(stagingPath); err == nil {
 		t.Errorf("volume staging path %s was not deleted", stagingPath)
 	}
@@ -475,12 +474,11 @@ func TestVolumeSetupTeardown(t *testing.T) {
 	}
 	t.Log("created attachement ", attachID)
 
-	err = csiMapper.SetUpDevice()
+	stagingPath, err := csiMapper.SetUpDevice()
 	if err != nil {
 		t.Fatalf("mapper failed to SetupDevice: %v", err)
 	}
 	// Check if NodeStageVolume staged to the right path
-	stagingPath := csiMapper.getStagingPath()
 	svols := csiMapper.csiClient.(*fakeCsiDriverClient).nodeClient.GetNodeStagedVolumes()
 	svol, ok := svols[csiMapper.volumeID]
 	if !ok {

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,20 @@ type csiDriverClient struct {
 	nodeV1ClientCreator nodeV1ClientCreator
 }
 
+type csiResizeOptions struct {
+	volumeID string
+	// volumePath is path where volume is available. It could be:
+	//   - path where node is staged if NodeExpandVolume is called after NodeStageVolume
+	//   - path where volume is published if NodeExpandVolume is called after NodePublishVolume
+	// DEPRECATION NOTICE: in future NodeExpandVolume will be always called after NodePublish
+	volumePath        string
+	stagingTargetPath string
+	fsType            string
+	accessMode        api.PersistentVolumeAccessMode
+	newSize           resource.Quantity
+	mountOptions      []string
+}
+
 var _ csiClient = &csiDriverClient{}
 
 type nodeV1ClientCreator func(addr csiAddr) (
@@ -245,36 +259,61 @@ 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 == "" {
-		return newSize, errors.New("missing volume id")
+	if opts.volumeID == "" {
+		return opts.newSize, errors.New("missing volume id")
 	}
-	if volumePath == "" {
-		return newSize, errors.New("missing volume path")
+	if opts.volumePath == "" {
+		return opts.newSize, errors.New("missing volume path")
 	}
 
-	if newSize.Value() < 0 {
-		return newSize, errors.New("size can not be less than 0")
+	if opts.newSize.Value() < 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,
-		VolumePath:    volumePath,
-		CapacityRange: &csipbv1.CapacityRange{RequiredBytes: newSize.Value()},
+		VolumeId:      opts.volumeID,
+		VolumePath:    opts.volumePath,
+		CapacityRange: &csipbv1.CapacityRange{RequiredBytes: opts.newSize.Value()},
+		VolumeCapability: &csipbv1.VolumeCapability{
+			AccessMode: &csipbv1.VolumeCapability_AccessMode{
+				Mode: asCSIAccessModeV1(opts.accessMode),
+			},
+		},
+	}
+
+	// not all CSI drivers support NodeStageUnstage and hence the StagingTargetPath
+	// should only be set when available
+	if opts.stagingTargetPath != "" {
+		req.StagingTargetPath = opts.stagingTargetPath
 	}
+
+	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,
-		VolumePath:    volumePath,
-		CapacityRange: &csipbv1.CapacityRange{RequiredBytes: newSize.Value()},
+		VolumeId:          opts.volumeID,
+		VolumePath:        opts.volumePath,
+		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,34 @@ func (c *csiPlugin) nodeExpandWithClient(
 		return false, nil
 	}
 
-	volumeTargetPath := resizeOptions.DeviceMountPath
+	pv := resizeOptions.VolumeSpec.PersistentVolume
+	if pv == nil {
+		return false, fmt.Errorf("Expander.NodeExpand failed to find associated PersistentVolume for plugin %s", c.GetPluginName())
+	}
+
+	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 {
-		volumeTargetPath = resizeOptions.DevicePath
+		// for block volumes the volumePath in CSI NodeExpandvolumeRequest is
+		// basically same as DevicePath because block devices are not mounted and hence
+		// DeviceMountPath does not get populated in resizeOptions.DeviceMountPath
+		opts.volumePath = resizeOptions.DevicePath
+		opts.fsType = fsTypeBlockName
+	}
+
+	if pv.Spec.AccessModes != nil {
+		opts.accessMode = pv.Spec.AccessModes[0]
 	}
 
-	_, err = csClient.NodeExpandVolume(ctx, csiSource.VolumeHandle, volumeTargetPath, resizeOptions.NewSize)
+	_, err = csClient.NodeExpandVolume(ctx, opts)
 	if err != nil {
 		return false, fmt.Errorf("Expander.NodeExpand failed to expand the volume : %v", err)
 	}