Ensure that StagingPath is supplied to blockVolume expansion

Author: gnufied

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,
@@ -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)

pkg/volume/csi/csi_client.go

@@ -282,16 +282,22 @@ func (c *csiDriverClient) NodeExpandVolume(ctx context.Context, opts csiResizeOp
 	defer closer.Close()
 
 	req := &csipbv1.NodeExpandVolumeRequest{
-		VolumeId:          opts.volumeID,
-		VolumePath:        opts.volumePath,
-		StagingTargetPath: opts.stagingTargetPath,
-		CapacityRange:     &csipbv1.CapacityRange{RequiredBytes: opts.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{},

pkg/volume/csi/expander.go

@@ -108,7 +108,11 @@ func (c *csiPlugin) nodeExpandWithClient(
 		accessMode:        api.ReadWriteOnce,
 		mountOptions:      pv.Spec.MountOptions,
 	}
+
 	if !fsVolume {
+		// 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
 	}

pkg/volume/local/local.go

@@ -621,6 +621,11 @@ func (m *localVolumeMapper) MapPodDevice() (string, error) {
 	return globalPath, nil
 }
 
+// GetStagingPath returns
+func (m *localVolumeMapper) GetStagingPath() string {
+	return ""
+}
+
 // localVolumeUnmapper implements the BlockVolumeUnmapper interface for local volumes.
 type localVolumeUnmapper struct {
 	*localVolume

pkg/volume/testing/testing.go

@@ -998,6 +998,10 @@ func (fv *FakeVolume) SetUpDevice() (string, error) {
 	return "", nil
 }
 
+func (fv *FakeVolume) GetStagingPath() string {
+	return filepath.Join(fv.Plugin.Host.GetVolumeDevicePluginDir(utilstrings.EscapeQualifiedName(fv.Plugin.PluginName)), "staging", fv.VolName)
+}
+
 // Block volume support
 func (fv *FakeVolume) GetSetUpDeviceCallCount() int {
 	fv.RLock()

pkg/volume/util/operationexecutor/operation_generator.go

@@ -1498,41 +1498,64 @@ func (og *operationGenerator) GenerateExpandInUseVolumeFunc(
 		var simpleErr, detailedErr error
 		resizeOptions := volume.NodeResizeOptions{
 			VolumeSpec: volumeToMount.VolumeSpec,
+			DevicePath: volumeToMount.DevicePath,
+		}
+		fsVolume, err := util.CheckVolumeModeFilesystem(volumeToMount.VolumeSpec)
+		if err != nil {
+			return volumeToMount.GenerateError("NodeExpandvolume.CheckVolumeModeFilesystem failed", err)
 		}
 
-		attachableVolumePlugin, _ :=
-			og.volumePluginMgr.FindAttachablePluginBySpec(volumeToMount.VolumeSpec)
+		if fsVolume {
+			volumeMounter, newMounterErr := volumePlugin.NewMounter(
+				volumeToMount.VolumeSpec,
+				volumeToMount.Pod,
+				volume.VolumeOptions{})
+			if newMounterErr != nil {
+				return volumeToMount.GenerateError("NodeExpandVolume.NewMounter initialization failed", newMounterErr)
+			}
 
-		if attachableVolumePlugin != nil {
-			volumeAttacher, _ := attachableVolumePlugin.NewAttacher()
-			if volumeAttacher != nil {
-				resizeOptions.CSIVolumePhase = volume.CSIVolumeStaged
-				resizeOptions.DevicePath = volumeToMount.DevicePath
-				dmp, err := volumeAttacher.GetDeviceMountPath(volumeToMount.VolumeSpec)
+			resizeOptions.DeviceMountPath = volumeMounter.GetPath()
+
+			deviceMountableVolumePlugin, _ := og.volumePluginMgr.FindDeviceMountablePluginBySpec(volumeToMount.VolumeSpec)
+			var volumeDeviceMounter volume.DeviceMounter
+			if deviceMountableVolumePlugin != nil {
+				volumeDeviceMounter, _ = deviceMountableVolumePlugin.NewDeviceMounter()
+			}
+
+			if volumeDeviceMounter != nil {
+				deviceStagePath, err := volumeDeviceMounter.GetDeviceMountPath(volumeToMount.VolumeSpec)
 				if err != nil {
 					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
-				}
-				if resizeDone {
-					return nil, nil
-				}
+				resizeOptions.DeviceStagePath = deviceStagePath
+			}
+		} else {
+			// Get block volume mapper plugin
+			blockVolumePlugin, err :=
+				og.volumePluginMgr.FindMapperPluginBySpec(volumeToMount.VolumeSpec)
+			if err != nil {
+				return volumeToMount.GenerateError("MapVolume.FindMapperPluginBySpec failed", err)
+			}
+
+			if blockVolumePlugin == nil {
+				return volumeToMount.GenerateError("MapVolume.FindMapperPluginBySpec failed to find BlockVolumeMapper plugin. Volume plugin is nil.", nil)
+			}
+
+			blockVolumeMapper, newMapperErr := blockVolumePlugin.NewBlockVolumeMapper(
+				volumeToMount.VolumeSpec,
+				volumeToMount.Pod,
+				volume.VolumeOptions{})
+			if newMapperErr != nil {
+				return volumeToMount.GenerateError("MapVolume.NewBlockVolumeMapper initialization failed", newMapperErr)
+			}
+
+			// if plugin supports custom mappers lets add DeviceStagePath
+			if customBlockVolumeMapper, ok := blockVolumeMapper.(volume.CustomBlockVolumeMapper); ok {
+				resizeOptions.DeviceStagePath = customBlockVolumeMapper.GetStagingPath()
 			}
-		}
-		// if we are here that means volume plugin does not support attach interface
-		volumeMounter, newMounterErr := volumePlugin.NewMounter(
-			volumeToMount.VolumeSpec,
-			volumeToMount.Pod,
-			volume.VolumeOptions{})
-		if newMounterErr != nil {
-			return volumeToMount.GenerateError("NodeExpandVolume.NewMounter initialization failed", newMounterErr)
 		}
 
-		resizeOptions.DeviceMountPath = volumeMounter.GetPath()
+		// if we are doing online expansion then volume is already published
 		resizeOptions.CSIVolumePhase = volume.CSIVolumePublished
 		resizeDone, simpleErr, detailedErr = og.doOnlineExpansion(volumeToMount, actualStateOfWorld, resizeOptions)
 		if simpleErr != nil || detailedErr != nil {

pkg/volume/volume.go

@@ -183,6 +183,10 @@ type CustomBlockVolumeMapper interface {
 	// If empty string is returned, the path retuned by attacher.Attach() and
 	// attacher.WaitForAttach() will be used.
 	MapPodDevice() (publishPath string, err error)
+
+	// GetStagingPath returns path that was used for staging the volume
+	// it is mainly used by CSI plugins
+	GetStagingPath() string
 }
 
 // BlockVolumeUnmapper interface is an unmapper interface for block volume.