Move resize function parameters to a new ResizeOptions type

pkg/volume/awsebs/aws_ebs.go

@@ -317,8 +317,8 @@ func (plugin *awsElasticBlockStorePlugin) ExpandVolumeDevice(
 	return awsVolume.ResizeDisk(volumeID, oldSize, newSize)
 }
 
-func (plugin *awsElasticBlockStorePlugin) NodeExpand(spec *volume.Spec, devicePath, deviceMountPath string, _, _ resource.Quantity) (bool, error) {
-	_, err := util.GenericResizeFS(plugin.host, plugin.GetPluginName(), devicePath, deviceMountPath)
+func (plugin *awsElasticBlockStorePlugin) NodeExpand(resizeOptions volume.NodeResizeOptions) (bool, error) {
+	_, err := util.GenericResizeFS(plugin.host, plugin.GetPluginName(), resizeOptions.DevicePath, resizeOptions.DeviceMountPath)
 	if err != nil {
 		return false, err
 	}

pkg/volume/azure_dd/azure_dd.go

@@ -312,8 +312,8 @@ func (plugin *azureDataDiskPlugin) ExpandVolumeDevice(
 	return diskController.ResizeDisk(spec.PersistentVolume.Spec.AzureDisk.DataDiskURI, oldSize, newSize)
 }
 
-func (plugin *azureDataDiskPlugin) NodeExpand(spec *volume.Spec, devicePath, deviceMountPath string, _, _ resource.Quantity) (bool, error) {
-	_, err := util.GenericResizeFS(plugin.host, plugin.GetPluginName(), devicePath, deviceMountPath)
+func (plugin *azureDataDiskPlugin) NodeExpand(resizeOptions volume.NodeResizeOptions) (bool, error) {
+	_, err := util.GenericResizeFS(plugin.host, plugin.GetPluginName(), resizeOptions.DevicePath, resizeOptions.DeviceMountPath)
 	if err != nil {
 		return false, err
 	}

pkg/volume/cinder/cinder.go

@@ -304,8 +304,8 @@ func (plugin *cinderPlugin) ExpandVolumeDevice(spec *volume.Spec, newSize resour
 	return expandedSize, nil
 }
 
-func (plugin *cinderPlugin) NodeExpand(spec *volume.Spec, devicePath, deviceMountPath string, _, _ resource.Quantity) (bool, error) {
-	_, err := util.GenericResizeFS(plugin.host, plugin.GetPluginName(), devicePath, deviceMountPath)
+func (plugin *cinderPlugin) NodeExpand(resizeOptions volume.NodeResizeOptions) (bool, error) {
+	_, err := util.GenericResizeFS(plugin.host, plugin.GetPluginName(), resizeOptions.DevicePath, resizeOptions.DeviceMountPath)
 	if err != nil {
 		return false, err
 	}

pkg/volume/csi/expander.go

@@ -21,21 +21,28 @@ import (
 	"errors"
 	"fmt"
 
-	"k8s.io/apimachinery/pkg/api/resource"
+	utilfeature "k8s.io/apiserver/pkg/util/feature"
 	"k8s.io/klog"
+	"k8s.io/kubernetes/pkg/features"
 	"k8s.io/kubernetes/pkg/volume"
 )
 
+var _ volume.NodeExpandableVolumePlugin = &csiPlugin{}
+
 func (c *csiPlugin) RequiresFSResize() bool {
 	// We could check plugin's node capability but we instead are going to rely on
 	// NodeExpand to do the right thing and return early if plugin does not have
 	// node expansion capability.
+	if !utilfeature.DefaultFeatureGate.Enabled(features.ExpandCSIVolumes) {
+		klog.V(4).Infof("Resizing is not enabled for this CSI volume")
+		return false
+	}
 	return true
 }
 
-func (c *csiPlugin) NodeExpand(spec *volume.Spec, devicePath, deviceMountPath string, newSize, oldSize resource.Quantity) (bool, error) {
-	klog.V(4).Infof(log("Expander.NodeExpand(%s)", deviceMountPath))
-	pvSource, err := getCSISourceFromSpec(spec)
+func (c *csiPlugin) NodeExpand(resizeOptions volume.NodeResizeOptions) (bool, error) {
+	klog.V(4).Infof(log("Expander.NodeExpand(%s)", resizeOptions.DeviceMountPath))
+	pvSource, err := getCSISourceFromSpec(resizeOptions.VolumeSpec)
 	if err != nil {
 		return false, err
 	}
@@ -50,7 +57,7 @@ func (c *csiPlugin) NodeExpand(spec *volume.Spec, devicePath, deviceMountPath st
 		return false, err
 	}
 
-	csiSource, err := getCSISourceFromSpec(spec)
+	csiSource, err := getCSISourceFromSpec(resizeOptions.VolumeSpec)
 	if err != nil {
 		klog.Error(log("Expander.NodeExpand failed to get CSI persistent source: %v", err))
 		return false, err
@@ -62,5 +69,28 @@ func (c *csiPlugin) NodeExpand(spec *volume.Spec, devicePath, deviceMountPath st
 	if err != nil {
 		return false, fmt.Errorf("Expander.NodeExpand failed to check if node supports expansion : %v", err)
 	}
-	return false, nil
+
+	if !nodeExpandSet {
+		return false, fmt.Errorf("Expander.NodeExpand found CSI plugin %s to not support node expansion", c.GetPluginName())
+	}
+
+	// Check whether "STAGE_UNSTAGE_VOLUME" is set
+	stageUnstageSet, err := csiClient.NodeSupportsStageUnstage(ctx)
+	if err != nil {
+		return false, fmt.Errorf("Expander.NodeExpand failed to check if plugins supports stage_unstage %v", err)
+	}
+
+	// if plugin does not support STAGE_UNSTAGE but CSI volume path is staged
+	// it must mean this was placeholder staging performed by k8s and not CSI staging
+	// in which case we should return from here so as volume can be node published
+	// before we can resize
+	if !stageUnstageSet && resizeOptions.CSIVolumePhase == volume.CSIVolumeStaged {
+		return false, nil
+	}
+
+	_, err = csiClient.NodeExpandVolume(ctx, csiSource.VolumeHandle, resizeOptions.DeviceMountPath, resizeOptions.NewSize)
+	if err != nil {
+		return false, fmt.Errorf("Expander.NodeExpand failed to expand the volume : %v", err)
+	}
+	return true, nil
 }

pkg/volume/flexvolume/expander-defaults.go

@@ -38,9 +38,9 @@ func (e *expanderDefaults) ExpandVolumeDevice(spec *volume.Spec, newSize resourc
 
 // the defaults for NodeExpand return a generic resize indicator that will trigger the operation executor to go ahead with
 // generic filesystem resize
-func (e *expanderDefaults) NodeExpand(spec *volume.Spec, devicePath, deviceMountPath string, _, _ resource.Quantity) (bool, error) {
-	klog.Warning(logPrefix(e.plugin), "using default filesystem resize for volume ", spec.Name(), ", at ", devicePath)
-	_, err := util.GenericResizeFS(e.plugin.host, e.plugin.GetPluginName(), devicePath, deviceMountPath)
+func (e *expanderDefaults) NodeExpand(rsOpt volume.NodeResizeOptions) (bool, error) {
+	klog.Warning(logPrefix(e.plugin), "using default filesystem resize for volume ", rsOpt.VolumeSpec.Name(), ", at ", rsOpt.DevicePath)
+	_, err := util.GenericResizeFS(e.plugin.host, e.plugin.GetPluginName(), rsOpt.DevicePath, rsOpt.DeviceMountPath)
 	if err != nil {
 		return false, err
 	}

pkg/volume/flexvolume/expander.go

@@ -43,26 +43,26 @@ func (plugin *flexVolumePlugin) ExpandVolumeDevice(spec *volume.Spec, newSize re
 	return newSize, err
 }
 
-func (plugin *flexVolumePlugin) NodeExpand(spec *volume.Spec, devicePath, deviceMountPath string, newSize, oldSize resource.Quantity) (bool, error) {
+func (plugin *flexVolumePlugin) NodeExpand(rsOpt volume.NodeResizeOptions) (bool, error) {
 	// This method is called after we spec.PersistentVolume.Spec.Capacity
 	// has been updated to the new size. The underlying driver thus sees
 	// the _new_ (requested) size and can find out the _current_ size from
 	// its underlying storage implementation
 
-	if spec.PersistentVolume == nil {
-		return false, fmt.Errorf("PersistentVolume not found for spec: %s", spec.Name())
+	if rsOpt.VolumeSpec.PersistentVolume == nil {
+		return false, fmt.Errorf("PersistentVolume not found for spec: %s", rsOpt.VolumeSpec.Name())
 	}
 
 	call := plugin.NewDriverCall(expandFSCmd)
-	call.AppendSpec(spec, plugin.host, nil)
-	call.Append(devicePath)
-	call.Append(deviceMountPath)
-	call.Append(strconv.FormatInt(newSize.Value(), 10))
-	call.Append(strconv.FormatInt(oldSize.Value(), 10))
+	call.AppendSpec(rsOpt.VolumeSpec, plugin.host, nil)
+	call.Append(rsOpt.DevicePath)
+	call.Append(rsOpt.DeviceMountPath)
+	call.Append(strconv.FormatInt(rsOpt.NewSize.Value(), 10))
+	call.Append(strconv.FormatInt(rsOpt.OldSize.Value(), 10))
 
 	_, err := call.Run()
 	if isCmdNotSupportedErr(err) {
-		return newExpanderDefaults(plugin).NodeExpand(spec, devicePath, deviceMountPath, newSize, oldSize)
+		return newExpanderDefaults(plugin).NodeExpand(rsOpt)
 	}
 	if err != nil {
 		return false, err

pkg/volume/gcepd/gce_pd.go

@@ -290,8 +290,8 @@ func (plugin *gcePersistentDiskPlugin) ExpandVolumeDevice(
 	return updatedQuantity, nil
 }
 
-func (plugin *gcePersistentDiskPlugin) NodeExpand(spec *volume.Spec, devicePath, deviceMountPath string, _, _ resource.Quantity) (bool, error) {
-	_, err := util.GenericResizeFS(plugin.host, plugin.GetPluginName(), devicePath, deviceMountPath)
+func (plugin *gcePersistentDiskPlugin) NodeExpand(resizeOptions volume.NodeResizeOptions) (bool, error) {
+	_, err := util.GenericResizeFS(plugin.host, plugin.GetPluginName(), resizeOptions.DevicePath, resizeOptions.DeviceMountPath)
 	if err != nil {
 		return false, err
 	}

pkg/volume/plugins.go

@@ -46,6 +46,9 @@ type ProbeEvent struct {
 	Op         ProbeOperation // The operation to the plugin
 }
 
+// CSIVolumePhaseType stores information about CSI volume path.
+type CSIVolumePhaseType string
+
 const (
 	// Common parameter which can be specified in StorageClass to specify the desired FSType
 	// Provisioners SHOULD implement support for this if they are block device based
@@ -55,6 +58,8 @@ const (
 
 	ProbeAddOrUpdate ProbeOperation = 1 << iota
 	ProbeRemove
+	CSIVolumeStaged    CSIVolumePhaseType = "staged"
+	CSIVolumePublished CSIVolumePhaseType = "published"
 )
 
 // VolumeOptions contains option information about a volume.
@@ -85,6 +90,26 @@ type VolumeOptions struct {
 	Parameters map[string]string
 }
 
+// NodeResizeOptions contain options to be passed for node expansion.
+type NodeResizeOptions struct {
+	VolumeSpec *Spec
+
+	// DevicePath - location of actual device on the node. In case of CSI
+	// this just could be volumeID
+	DevicePath string
+
+	// DeviceMountPath location where device is mounted on the node. If volume type
+	// is attachable - this would be global mount path otherwise
+	// it would be location where volume was mounted for the pod
+	DeviceMountPath string
+
+	NewSize resource.Quantity
+	OldSize resource.Quantity
+
+	// CSIVolumePhase contains volume phase on the node
+	CSIVolumePhase CSIVolumePhaseType
+}
+
 type DynamicPluginProber interface {
 	Init() error
 
@@ -232,8 +257,7 @@ type NodeExpandableVolumePlugin interface {
 	VolumePlugin
 	RequiresFSResize() bool
 	// NodeExpand expands volume on given deviceMountPath and returns true if resize is successful.
-	// devicePath can be set to empty string if unavailable.
-	NodeExpand(spec *Spec, devicePath, deviceMountPath string, newSize, oldSize resource.Quantity) (bool, error)
+	NodeExpand(resizeOptions NodeResizeOptions) (bool, error)
 }
 
 // VolumePluginWithAttachLimits is an extended interface of VolumePlugin that restricts number of

pkg/volume/rbd/rbd.go

@@ -201,8 +201,8 @@ func (plugin *rbdPlugin) ExpandVolumeDevice(spec *volume.Spec, newSize resource.
 	}
 }
 
-func (plugin *rbdPlugin) NodeExpand(spec *volume.Spec, devicePath, deviceMountPath string, _, _ resource.Quantity) (bool, error) {
-	_, err := volutil.GenericResizeFS(plugin.host, plugin.GetPluginName(), devicePath, deviceMountPath)
+func (plugin *rbdPlugin) NodeExpand(resizeOptions volume.NodeResizeOptions) (bool, error) {
+	_, err := volutil.GenericResizeFS(plugin.host, plugin.GetPluginName(), resizeOptions.DevicePath, resizeOptions.DeviceMountPath)
 	if err != nil {
 		return false, err
 	}

pkg/volume/testing/testing.go

@@ -537,7 +537,7 @@ func (plugin *FakeVolumePlugin) RequiresFSResize() bool {
 	return true
 }
 
-func (plugin *FakeVolumePlugin) NodeExpand(spec *Spec, devicePath, deviceMountPath string, _, _ resource.Quantity) (bool, error) {
+func (plugin *FakeVolumePlugin) NodeExpand(resizeOptions NodeResizeOptions) (bool, error) {
 	return true, nil
 }
 

pkg/volume/util/operationexecutor/operation_generator.go

@@ -597,8 +597,12 @@ func (og *operationGenerator) GenerateMountVolumeFunc(
 
 			klog.Infof(volumeToMount.GenerateMsgDetailed("MountVolume.WaitForAttach succeeded", fmt.Sprintf("DevicePath %q", devicePath)))
 		}
+
 		var resizeDone bool
 		var resizeError error
+		resizeOptions := volume.NodeResizeOptions{
+			DevicePath: devicePath,
+		}
 
 		if volumeDeviceMounter != nil {
 			deviceMountPath, err :=
@@ -628,9 +632,12 @@ func (og *operationGenerator) GenerateMountVolumeFunc(
 				return volumeToMount.GenerateError("MountVolume.MarkDeviceAsMounted failed", markDeviceMountedErr)
 			}
 
+			resizeOptions.DeviceMountPath = deviceMountPath
+			resizeOptions.CSIVolumePhase = volume.CSIVolumeStaged
+
 			// resizeFileSystem will resize the file system if user has requested a resize of
 			// underlying persistent volume and is allowed to do so.
-			resizeDone, resizeError = og.resizeFileSystem(volumeToMount, devicePath, deviceMountPath, volumePluginName)
+			resizeDone, resizeError = og.resizeFileSystem(volumeToMount, resizeOptions, volumePluginName)
 
 			if resizeError != nil {
 				return volumeToMount.GenerateError("MountVolume.MountDevice failed while expanding volume", resizeError)
@@ -659,9 +666,11 @@ func (og *operationGenerator) GenerateMountVolumeFunc(
 			verbosity = klog.Level(4)
 		}
 		klog.V(verbosity).Infof(detailedMsg)
+		resizeOptions.DeviceMountPath = volumeMounter.GetPath()
+		resizeOptions.CSIVolumePhase = volume.CSIVolumePublished
 
 		if !resizeDone {
-			resizeDone, resizeError = og.resizeFileSystem(volumeToMount, "", volumeMounter.GetPath(), volumePluginName)
+			resizeDone, resizeError = og.resizeFileSystem(volumeToMount, resizeOptions, volumePluginName)
 			if resizeError != nil {
 				return volumeToMount.GenerateError("MountVolume.Setup failed while expanding volume", resizeError)
 			}
@@ -698,7 +707,7 @@ func (og *operationGenerator) GenerateMountVolumeFunc(
 	}
 }
 
-func (og *operationGenerator) resizeFileSystem(volumeToMount VolumeToMount, devicePath, deviceMountPath, pluginName string) (bool, error) {
+func (og *operationGenerator) resizeFileSystem(volumeToMount VolumeToMount, rsOpts volume.NodeResizeOptions, pluginName string) (bool, error) {
 	if !utilfeature.DefaultFeatureGate.Enabled(features.ExpandPersistentVolumes) {
 		klog.V(4).Infof("Resizing is not enabled for this volume %s", volumeToMount.VolumeName)
 		return true, nil
@@ -730,7 +739,10 @@ func (og *operationGenerator) resizeFileSystem(volumeToMount VolumeToMount, devi
 				og.recorder.Eventf(volumeToMount.Pod, v1.EventTypeWarning, kevents.FileSystemResizeFailed, simpleMsg)
 				return true, nil
 			}
-			resizeDone, resizeErr := expandableVolumePlugin.NodeExpand(volumeToMount.VolumeSpec, devicePath, deviceMountPath, pvSpecCap, pvcStatusCap)
+			rsOpts.VolumeSpec = volumeToMount.VolumeSpec
+			rsOpts.NewSize = pvSpecCap
+			rsOpts.OldSize = pvcStatusCap
+			resizeDone, resizeErr := expandableVolumePlugin.NodeExpand(rsOpts)
 			if resizeErr != nil {
 				return false, fmt.Errorf("MountVolume.resizeFileSystem failed : %v", resizeErr)
 			}
@@ -1459,41 +1471,67 @@ func (og *operationGenerator) GenerateExpandVolumeFSWithoutUnmountingFunc(
 		return volumetypes.GeneratedOperations{}, volumeToMount.GenerateErrorDetailed("VolumeFSResize.FindPluginBySpec failed", err)
 	}
 
-	attachableVolumePlugin, err :=
-		og.volumePluginMgr.FindAttachablePluginBySpec(volumeToMount.VolumeSpec)
-	if err != nil || attachableVolumePlugin == nil {
-		if attachableVolumePlugin == nil {
-			err = fmt.Errorf("AttachableVolumePlugin is nil")
-		}
-		return volumetypes.GeneratedOperations{}, volumeToMount.GenerateErrorDetailed("VolumeFSResize.FindAttachablePluginBySpec failed", err)
-	}
-
-	volumeAttacher, err := attachableVolumePlugin.NewAttacher()
-	if err != nil || volumeAttacher == nil {
-		if volumeAttacher == nil {
-			err = fmt.Errorf("VolumeAttacher is nil")
+	fsResizeFunc := func() (error, error) {
+		var resizeDone bool
+		resizeOptions := volume.NodeResizeOptions{
+			VolumeSpec: volumeToMount.VolumeSpec,
 		}
-		return volumetypes.GeneratedOperations{}, volumeToMount.GenerateErrorDetailed("VolumeFSResize.NewAttacher failed", err)
-	}
 
-	deviceMountPath, err := volumeAttacher.GetDeviceMountPath(volumeToMount.VolumeSpec)
-	if err != nil {
-		return volumetypes.GeneratedOperations{}, volumeToMount.GenerateErrorDetailed("VolumeFSResize.GetDeviceMountPath failed", err)
-	}
+		attachableVolumePlugin, _ :=
+			og.volumePluginMgr.FindAttachablePluginBySpec(volumeToMount.VolumeSpec)
 
-	fsResizeFunc := func() (error, error) {
-		_, resizeError := og.resizeFileSystem(volumeToMount, volumeToMount.DevicePath, deviceMountPath, volumePlugin.GetPluginName())
+		if attachableVolumePlugin != nil {
+			volumeAttacher, _ := attachableVolumePlugin.NewAttacher()
+			if volumeAttacher != nil {
+				resizeOptions.CSIVolumePhase = volume.CSIVolumeStaged
+				resizeOptions.DevicePath = volumeToMount.DevicePath
+				dmp, err := volumeAttacher.GetDeviceMountPath(volumeToMount.VolumeSpec)
+				if err != nil {
+					return volumeToMount.GenerateError("VolumeFSResize.GetDeviceMountPath failed", err)
+				}
+				resizeOptions.DeviceMountPath = dmp
+				resizeDone, err = og.resizeFileSystem(volumeToMount, resizeOptions, volumePlugin.GetPluginName())
+				if err != nil {
+					return volumeToMount.GenerateError("VolumeFSResize.resizeFileSystem failed", err)
+				}
+				if resizeDone {
+					markFSResizedErr := actualStateOfWorld.MarkVolumeAsResized(volumeToMount.PodName, volumeToMount.VolumeName)
+					if markFSResizedErr != nil {
+						// On failure, return error. Caller will log and retry.
+						return volumeToMount.GenerateError("VolumeFSResize.MarkVolumeAsResized failed", markFSResizedErr)
+					}
+					return nil, nil
+				}
+			}
+		}
+		// 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("VolumeFSResize.NewMounter initialization failed", newMounterErr)
+		}
 
-		if resizeError != nil {
-			return volumeToMount.GenerateError("VolumeFSResize.resizeFileSystem failed", resizeError)
+		resizeOptions.DeviceMountPath = volumeMounter.GetPath()
+		resizeOptions.CSIVolumePhase = volume.CSIVolumePublished
+		resizeDone, err = og.resizeFileSystem(volumeToMount, resizeOptions, volumePlugin.GetPluginName())
+		if err != nil {
+			return volumeToMount.GenerateError("VolumeFSResize.resizeFileSystem failed", err)
 		}
-		markFSResizedErr := actualStateOfWorld.MarkVolumeAsResized(volumeToMount.PodName, volumeToMount.VolumeName)
-		if markFSResizedErr != nil {
-			// On failure, return error. Caller will log and retry.
-			return volumeToMount.GenerateError("VolumeFSResize.MarkVolumeAsResized failed", markFSResizedErr)
+		if resizeDone {
+			markFSResizedErr := actualStateOfWorld.MarkVolumeAsResized(volumeToMount.PodName, volumeToMount.VolumeName)
+			if markFSResizedErr != nil {
+				// On failure, return error. Caller will log and retry.
+				return volumeToMount.GenerateError("VolumeFSResize.MarkVolumeAsResized failed", markFSResizedErr)
+			}
+			return nil, nil
 		}
-		return nil, nil
+		// This is a placeholder error - we should NEVER reach here.
+		err := fmt.Errorf("volume resizing failed for unknown reason")
+		return volumeToMount.GenerateError("VolumeFSResize.resizeFileSystem failed to resize volume", err)
 	}
+
 	eventRecorderFunc := func(err *error) {
 		if *err != nil {
 			og.recorder.Eventf(volumeToMount.Pod, v1.EventTypeWarning, kevents.VolumeResizeFailed, (*err).Error())