Merge pull request #67851 from aniket-s-kulkarni/flexvolume-resize-implementation

Flexvolume resize implementation

pkg/controller/volume/expand/expand_controller.go

@@ -208,13 +208,16 @@ func (expc *expandController) pvcUpdate(oldObj, newObj interface{}) {
 			return
 		}
 
-		// Filter PVCs for which the corresponding volume plugins don't allow expansion.
 		volumeSpec := volume.NewSpecFromPersistentVolume(pv, false)
 		volumePlugin, err := expc.volumePluginMgr.FindExpandablePluginBySpec(volumeSpec)
 		if err != nil || volumePlugin == nil {
 			err = fmt.Errorf("didn't find a plugin capable of expanding the volume; " +
 				"waiting for an external controller to process this PVC")
-			expc.recorder.Event(newPVC, v1.EventTypeNormal, events.ExternalExpanding,
+			eventType := v1.EventTypeNormal
+			if err != nil {
+				eventType = v1.EventTypeWarning
+			}
+			expc.recorder.Event(newPVC, eventType, events.ExternalExpanding,
 				fmt.Sprintf("Ignoring the PVC: %v.", err))
 			glog.V(3).Infof("Ignoring the PVC %q (uid: %q) : %v.",
 				util.GetPersistentVolumeClaimQualifiedName(newPVC), newPVC.UID, err)

pkg/controller/volume/expand/pvc_populator.go

@@ -104,7 +104,11 @@ func (populator *pvcPopulator) Sync() {
 		if (err != nil || volumePlugin == nil) && pvcStatusSize.Cmp(pvcSize) < 0 {
 			err = fmt.Errorf("didn't find a plugin capable of expanding the volume; " +
 				"waiting for an external controller to process this PVC")
-			populator.recorder.Event(pvc, v1.EventTypeNormal, events.ExternalExpanding,
+			eventType := v1.EventTypeNormal
+			if err != nil {
+				eventType = v1.EventTypeWarning
+			}
+			populator.recorder.Event(pvc, eventType, events.ExternalExpanding,
 				fmt.Sprintf("Ignoring the PVC: %v.", err))
 			glog.V(3).Infof("Ignoring the PVC %q (uid: %q) : %v.",
 				util.GetPersistentVolumeClaimQualifiedName(pvc), pvc.UID, err)

pkg/controller/volume/expand/sync_volume_resize.go

@@ -77,7 +77,6 @@ func (rc *syncResize) Sync() {
 			glog.V(10).Infof("Operation for PVC %v is already pending", pvcWithResizeRequest.QualifiedName())
 			continue
 		}
-		glog.V(5).Infof("Starting opsExecutor.ExpandVolume for volume %s", pvcWithResizeRequest.QualifiedName())
 		growFuncError := rc.opsExecutor.ExpandVolume(pvcWithResizeRequest, rc.resizeMap)
 		if growFuncError != nil && !exponentialbackoff.IsExponentialBackoff(growFuncError) {
 			glog.Errorf("Error growing pvc %s with %v", pvcWithResizeRequest.QualifiedName(), growFuncError)

pkg/volume/BUILD

@@ -9,6 +9,7 @@ go_library(
         "metrics_errors.go",
         "metrics_nil.go",
         "metrics_statfs.go",
+        "noop_expandable_plugin.go",
         "plugins.go",
         "volume.go",
         "volume_linux.go",

pkg/volume/awsebs/aws_ebs.go

@@ -312,6 +312,12 @@ func (plugin *awsElasticBlockStorePlugin) ExpandVolumeDevice(
 	return awsVolume.ResizeDisk(volumeID, oldSize, newSize)
 }
 
+func (plugin *awsElasticBlockStorePlugin) ExpandFS(spec *volume.Spec, devicePath, deviceMountPath string, _, _ resource.Quantity) error {
+	_, err := util.GenericResizeFS(plugin.host, plugin.GetPluginName(), devicePath, deviceMountPath)
+	return err
+}
+
+var _ volume.FSResizableVolumePlugin = &awsElasticBlockStorePlugin{}
 var _ volume.ExpandableVolumePlugin = &awsElasticBlockStorePlugin{}
 var _ volume.VolumePluginWithAttachLimits = &awsElasticBlockStorePlugin{}
 

pkg/volume/azure_dd/azure_dd.go

@@ -301,6 +301,13 @@ func (plugin *azureDataDiskPlugin) ExpandVolumeDevice(
 	return diskController.ResizeDisk(spec.PersistentVolume.Spec.AzureDisk.DataDiskURI, oldSize, newSize)
 }
 
+func (plugin *azureDataDiskPlugin) ExpandFS(spec *volume.Spec, devicePath, deviceMountPath string, _, _ resource.Quantity) error {
+	_, err := util.GenericResizeFS(plugin.host, plugin.GetPluginName(), devicePath, deviceMountPath)
+	return err
+}
+
+var _ volume.FSResizableVolumePlugin = &azureDataDiskPlugin{}
+
 func (plugin *azureDataDiskPlugin) ConstructVolumeSpec(volumeName, mountPath string) (*volume.Spec, error) {
 	mounter := plugin.host.GetMounter(plugin.GetPluginName())
 	pluginDir := plugin.host.GetPluginDir(plugin.GetPluginName())

pkg/volume/cinder/cinder.go

@@ -267,6 +267,13 @@ func (plugin *cinderPlugin) ExpandVolumeDevice(spec *volume.Spec, newSize resour
 	return expandedSize, nil
 }
 
+func (plugin *cinderPlugin) ExpandFS(spec *volume.Spec, devicePath, deviceMountPath string, _, _ resource.Quantity) error {
+	_, err := util.GenericResizeFS(plugin.host, plugin.GetPluginName(), devicePath, deviceMountPath)
+	return err
+}
+
+var _ volume.FSResizableVolumePlugin = &cinderPlugin{}
+
 func (plugin *cinderPlugin) RequiresFSResize() bool {
 	return true
 }

pkg/volume/flexvolume/BUILD

@@ -14,6 +14,8 @@ go_library(
         "detacher.go",
         "detacher-defaults.go",
         "driver-call.go",
+        "expander.go",
+        "expander-defaults.go",
         "fake_watcher.go",
         "mounter.go",
         "mounter-defaults.go",
@@ -33,6 +35,7 @@ go_library(
         "//pkg/volume:go_default_library",
         "//pkg/volume/util:go_default_library",
         "//staging/src/k8s.io/api/core/v1:go_default_library",
+        "//staging/src/k8s.io/apimachinery/pkg/api/resource:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/types:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/util/errors:go_default_library",
         "//vendor/github.com/fsnotify/fsnotify:go_default_library",

pkg/volume/flexvolume/common_test.go

@@ -76,11 +76,11 @@ func fakeResultOutput(result interface{}) fakeexec.FakeCombinedOutputAction {
 }
 
 func successOutput() fakeexec.FakeCombinedOutputAction {
-	return fakeResultOutput(&DriverStatus{StatusSuccess, "", "", "", true, nil})
+	return fakeResultOutput(&DriverStatus{StatusSuccess, "", "", "", true, nil, 0})
 }
 
 func notSupportedOutput() fakeexec.FakeCombinedOutputAction {
-	return fakeResultOutput(&DriverStatus{StatusNotSupported, "", "", "", false, nil})
+	return fakeResultOutput(&DriverStatus{StatusNotSupported, "", "", "", false, nil, 0})
 }
 
 func sameArgs(args, expectedArgs []string) bool {

pkg/volume/flexvolume/driver-call.go

@@ -44,6 +44,9 @@ const (
 	mountCmd   = "mount"
 	unmountCmd = "unmount"
 
+	expandVolumeCmd = "expandvolume"
+	expandFSCmd     = "expandfs"
+
 	// Option keys
 	optionFSType         = "kubernetes.io/fsType"
 	optionReadWrite      = "kubernetes.io/readwrite"
@@ -221,22 +224,26 @@ type DriverStatus struct {
 	// By default we assume all the capabilities are supported.
 	// If the plugin does not support a capability, it can return false for that capability.
 	Capabilities *DriverCapabilities `json:",omitempty"`
+	// Returns the actual size of the volume after resizing is done, the size is in bytes.
+	ActualVolumeSize int64 `json:"volumeNewSize,omitempty"`
 }
 
 // DriverCapabilities represents what driver can do
 type DriverCapabilities struct {
-	Attach          bool `json:"attach"`
-	SELinuxRelabel  bool `json:"selinuxRelabel"`
-	SupportsMetrics bool `json:"supportsMetrics"`
-	FSGroup         bool `json:"fsGroup"`
+	Attach           bool `json:"attach"`
+	SELinuxRelabel   bool `json:"selinuxRelabel"`
+	SupportsMetrics  bool `json:"supportsMetrics"`
+	FSGroup          bool `json:"fsGroup"`
+	RequiresFSResize bool `json:"requiresFSResize"`
 }
 
 func defaultCapabilities() *DriverCapabilities {
 	return &DriverCapabilities{
-		Attach:          true,
-		SELinuxRelabel:  true,
-		SupportsMetrics: false,
-		FSGroup:         true,
+		Attach:           true,
+		SELinuxRelabel:   true,
+		SupportsMetrics:  false,
+		FSGroup:          true,
+		RequiresFSResize: true,
 	}
 }
 

pkg/volume/flexvolume/expander-defaults.go

@@ -0,0 +1,45 @@
+/*
+Copyright 2018 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package flexvolume
+
+import (
+	"github.com/golang/glog"
+	"k8s.io/apimachinery/pkg/api/resource"
+	"k8s.io/kubernetes/pkg/volume"
+	"k8s.io/kubernetes/pkg/volume/util"
+)
+
+type expanderDefaults struct {
+	plugin *flexVolumePlugin
+}
+
+func newExpanderDefaults(plugin *flexVolumePlugin) *expanderDefaults {
+	return &expanderDefaults{plugin}
+}
+
+func (e *expanderDefaults) ExpandVolumeDevice(spec *volume.Spec, newSize resource.Quantity, oldSize resource.Quantity) (resource.Quantity, error) {
+	glog.Warning(logPrefix(e.plugin), "using default expand for volume ", spec.Name(), ", to size ", newSize, " from ", oldSize)
+	return newSize, nil
+}
+
+// the defaults for ExpandFS return a generic resize indicator that will trigger the operation executor to go ahead with
+// generic filesystem resize
+func (e *expanderDefaults) ExpandFS(spec *volume.Spec, devicePath, deviceMountPath string, _, _ resource.Quantity) error {
+	glog.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)
+	return err
+}

pkg/volume/flexvolume/expander.go

@@ -0,0 +1,67 @@
+/*
+Copyright 2018 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package flexvolume
+
+import (
+	"fmt"
+	"k8s.io/apimachinery/pkg/api/resource"
+	"k8s.io/kubernetes/pkg/volume"
+	"strconv"
+)
+
+func (plugin *flexVolumePlugin) ExpandVolumeDevice(spec *volume.Spec, newSize resource.Quantity, oldSize resource.Quantity) (resource.Quantity, error) {
+	call := plugin.NewDriverCall(expandVolumeCmd)
+	call.AppendSpec(spec, plugin.host, nil)
+
+	devicePath, err := plugin.getDeviceMountPath(spec)
+	if err != nil {
+		return newSize, err
+	}
+	call.Append(devicePath)
+	call.Append(strconv.FormatInt(newSize.Value(), 10))
+	call.Append(strconv.FormatInt(oldSize.Value(), 10))
+
+	_, err = call.Run()
+	if isCmdNotSupportedErr(err) {
+		return newExpanderDefaults(plugin).ExpandVolumeDevice(spec, newSize, oldSize)
+	}
+	return newSize, err
+}
+
+func (plugin *flexVolumePlugin) ExpandFS(spec *volume.Spec, devicePath, deviceMountPath string, newSize, oldSize resource.Quantity) 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 fmt.Errorf("PersistentVolume not found for spec: %s", spec.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))
+
+	_, err := call.Run()
+	if isCmdNotSupportedErr(err) {
+		return newExpanderDefaults(plugin).ExpandFS(spec, devicePath, deviceMountPath, newSize, oldSize)
+	}
+	return err
+}

pkg/volume/flexvolume/plugin.go

@@ -57,6 +57,8 @@ type flexVolumeAttachablePlugin struct {
 
 var _ volume.AttachableVolumePlugin = &flexVolumeAttachablePlugin{}
 var _ volume.PersistentVolumePlugin = &flexVolumePlugin{}
+var _ volume.FSResizableVolumePlugin = &flexVolumePlugin{}
+var _ volume.ExpandableVolumePlugin = &flexVolumePlugin{}
 
 var _ volume.DeviceMountableVolumePlugin = &flexVolumeAttachablePlugin{}
 
@@ -305,3 +307,7 @@ func (plugin *flexVolumePlugin) getDeviceMountPath(spec *volume.Spec) (string, e
 	mountsDir := path.Join(plugin.host.GetPluginDir(flexVolumePluginName), plugin.driverName, "mounts")
 	return path.Join(mountsDir, volumeName), nil
 }
+
+func (plugin *flexVolumePlugin) RequiresFSResize() bool {
+	return plugin.capabilities.RequiresFSResize
+}

pkg/volume/gcepd/gce_pd.go

@@ -284,6 +284,13 @@ func (plugin *gcePersistentDiskPlugin) ExpandVolumeDevice(
 	return updatedQuantity, nil
 }
 
+func (plugin *gcePersistentDiskPlugin) ExpandFS(spec *volume.Spec, devicePath, deviceMountPath string, _, _ resource.Quantity) error {
+	_, err := util.GenericResizeFS(plugin.host, plugin.GetPluginName(), devicePath, deviceMountPath)
+	return err
+}
+
+var _ volume.FSResizableVolumePlugin = &gcePersistentDiskPlugin{}
+
 func (plugin *gcePersistentDiskPlugin) ConstructVolumeSpec(volumeName, mountPath string) (*volume.Spec, error) {
 	mounter := plugin.host.GetMounter(plugin.GetPluginName())
 	pluginDir := plugin.host.GetPluginDir(plugin.GetPluginName())

pkg/volume/noop_expandable_plugin.go

@@ -0,0 +1,77 @@
+/*
+Copyright 2018 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package volume
+
+import (
+	"k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/resource"
+	"k8s.io/apimachinery/pkg/types"
+)
+
+type noopExpandableVolumePluginInstance struct {
+	spec *Spec
+}
+
+var _ ExpandableVolumePlugin = &noopExpandableVolumePluginInstance{}
+
+func (n *noopExpandableVolumePluginInstance) ExpandVolumeDevice(spec *Spec, newSize resource.Quantity, oldSize resource.Quantity) (resource.Quantity, error) {
+	return newSize, nil
+}
+
+func (n *noopExpandableVolumePluginInstance) Init(host VolumeHost) error {
+	return nil
+}
+
+func (n *noopExpandableVolumePluginInstance) GetPluginName() string {
+	return n.spec.KubeletExpandablePluginName()
+}
+
+func (n *noopExpandableVolumePluginInstance) GetVolumeName(spec *Spec) (string, error) {
+	return n.spec.Name(), nil
+}
+
+func (n *noopExpandableVolumePluginInstance) CanSupport(spec *Spec) bool {
+	return true
+}
+
+func (n *noopExpandableVolumePluginInstance) RequiresRemount() bool {
+	return false
+}
+
+func (n *noopExpandableVolumePluginInstance) NewMounter(spec *Spec, podRef *v1.Pod, opts VolumeOptions) (Mounter, error) {
+	return nil, nil
+}
+
+func (n *noopExpandableVolumePluginInstance) NewUnmounter(name string, podUID types.UID) (Unmounter, error) {
+	return nil, nil
+}
+
+func (n *noopExpandableVolumePluginInstance) ConstructVolumeSpec(volumeName, mountPath string) (*Spec, error) {
+	return n.spec, nil
+}
+
+func (n *noopExpandableVolumePluginInstance) SupportsMountOption() bool {
+	return true
+}
+
+func (n *noopExpandableVolumePluginInstance) SupportsBulkVolumeVerification() bool {
+	return false
+}
+
+func (n *noopExpandableVolumePluginInstance) RequiresFSResize() bool {
+	return true
+}

pkg/volume/plugins.go

@@ -225,6 +225,13 @@ type ExpandableVolumePlugin interface {
 	RequiresFSResize() bool
 }
 
+// FSResizableVolumePlugin is an extension of ExpandableVolumePlugin and is used for volumes (flex)
+// that require extra steps on nodes for expansion to complete
+type FSResizableVolumePlugin interface {
+	ExpandableVolumePlugin
+	ExpandFS(spec *Spec, devicePath, deviceMountPath string, newSize, oldSize resource.Quantity) error
+}
+
 // VolumePluginWithAttachLimits is an extended interface of VolumePlugin that restricts number of
 // volumes that can be attached to a node.
 type VolumePluginWithAttachLimits interface {
@@ -388,6 +395,36 @@ func (spec *Spec) Name() string {
 	}
 }
 
+// IsKubeletExpandable returns true for volume types that can be expanded only by the node
+// and not the controller. Currently Flex volume is the only one in this category since
+// it is typically not installed on the controller
+func (spec *Spec) IsKubeletExpandable() bool {
+	switch {
+	case spec.Volume != nil:
+		return spec.Volume.FlexVolume != nil
+	case spec.PersistentVolume != nil:
+		return spec.PersistentVolume.Spec.FlexVolume != nil
+	default:
+		return false
+
+	}
+}
+
+// KubeletExpandablePluginName creates and returns a name for the plugin
+// this is used in context on the controller where the plugin lookup fails
+// as volume expansion on controller isn't supported, but a plugin name is
+// required
+func (spec *Spec) KubeletExpandablePluginName() string {
+	switch {
+	case spec.Volume != nil && spec.Volume.FlexVolume != nil:
+		return spec.Volume.FlexVolume.Driver
+	case spec.PersistentVolume != nil && spec.PersistentVolume.Spec.FlexVolume != nil:
+		return spec.PersistentVolume.Spec.FlexVolume.Driver
+	default:
+		return ""
+	}
+}
+
 // VolumeConfig is how volume plugins receive configuration.  An instance
 // specific to the plugin will be passed to the plugin's
 // ProbeVolumePlugins(config) func.  Reasonable defaults will be provided by
@@ -797,6 +834,13 @@ func (pm *VolumePluginMgr) FindDeviceMountablePluginByName(name string) (DeviceM
 func (pm *VolumePluginMgr) FindExpandablePluginBySpec(spec *Spec) (ExpandableVolumePlugin, error) {
 	volumePlugin, err := pm.FindPluginBySpec(spec)
 	if err != nil {
+		if spec.IsKubeletExpandable() {
+			// for kubelet expandable volumes, return a noop plugin that
+			// returns success for expand on the controller
+			glog.Warningf("FindExpandablePluginBySpec(%s) -> returning noopExpandableVolumePluginInstance", spec.Name())
+			return &noopExpandableVolumePluginInstance{spec}, nil
+		}
+		glog.Warningf("FindExpandablePluginBySpec(%s) -> err:%v", spec.Name(), err)
 		return nil, err
 	}
 
@@ -845,6 +889,32 @@ func (pm *VolumePluginMgr) FindMapperPluginByName(name string) (BlockVolumePlugi
 	return nil, nil
 }
 
+// FindFSResizablePluginBySpec fetches a persistent volume plugin by spec
+func (pm *VolumePluginMgr) FindFSResizablePluginBySpec(spec *Spec) (FSResizableVolumePlugin, error) {
+	volumePlugin, err := pm.FindPluginBySpec(spec)
+	if err != nil {
+		return nil, err
+	}
+	if fsResizablePlugin, ok := volumePlugin.(FSResizableVolumePlugin); ok {
+		return fsResizablePlugin, nil
+	}
+	return nil, nil
+}
+
+// FindFSResizablePluginByName fetches a persistent volume plugin by name
+func (pm *VolumePluginMgr) FindFSResizablePluginByName(name string) (FSResizableVolumePlugin, error) {
+	volumePlugin, err := pm.FindPluginByName(name)
+	if err != nil {
+		return nil, err
+	}
+
+	if fsResizablePlugin, ok := volumePlugin.(FSResizableVolumePlugin); ok {
+		return fsResizablePlugin, nil
+	}
+
+	return nil, nil
+}
+
 // NewPersistentVolumeRecyclerPodTemplate creates a template for a recycler
 // pod.  By default, a recycler pod simply runs "rm -rf" on a volume and tests
 // for emptiness.  Most attributes of the template will be correct for most

pkg/volume/rbd/rbd.go

@@ -197,6 +197,13 @@ func (plugin *rbdPlugin) ExpandVolumeDevice(spec *volume.Spec, newSize resource.
 	}
 }
 
+func (plugin *rbdPlugin) ExpandFS(spec *volume.Spec, devicePath, deviceMountPath string, _, _ resource.Quantity) error {
+	_, err := volutil.GenericResizeFS(plugin.host, plugin.GetPluginName(), devicePath, deviceMountPath)
+	return err
+}
+
+var _ volume.FSResizableVolumePlugin = &rbdPlugin{}
+
 func (expander *rbdVolumeExpander) ResizeImage(oldSize resource.Quantity, newSize resource.Quantity) (resource.Quantity, error) {
 	return expander.manager.ExpandImage(expander, oldSize, newSize)
 }

pkg/volume/testing/testing.go

@@ -262,6 +262,7 @@ var _ ProvisionableVolumePlugin = &FakeVolumePlugin{}
 var _ AttachableVolumePlugin = &FakeVolumePlugin{}
 var _ VolumePluginWithAttachLimits = &FakeVolumePlugin{}
 var _ DeviceMountableVolumePlugin = &FakeVolumePlugin{}
+var _ FSResizableVolumePlugin = &FakeVolumePlugin{}
 
 func (plugin *FakeVolumePlugin) getFakeVolume(list *[]*FakeVolume) *FakeVolume {
 	volume := &FakeVolume{}
@@ -480,6 +481,10 @@ func (plugin *FakeVolumePlugin) RequiresFSResize() bool {
 	return true
 }
 
+func (plugin *FakeVolumePlugin) ExpandFS(spec *Spec, devicePath, deviceMountPath string, _, _ resource.Quantity) error {
+	return nil
+}
+
 func (plugin *FakeVolumePlugin) GetVolumeLimits() (map[string]int64, error) {
 	return plugin.VolumeLimits, plugin.VolumeLimitsError
 }

pkg/volume/util/BUILD

@@ -25,6 +25,7 @@ go_library(
         "//pkg/features:go_default_library",
         "//pkg/kubelet/apis:go_default_library",
         "//pkg/util/mount:go_default_library",
+        "//pkg/util/resizefs:go_default_library",
         "//pkg/util/strings:go_default_library",
         "//pkg/volume:go_default_library",
         "//pkg/volume/util/types:go_default_library",

pkg/volume/util/operationexecutor/BUILD

@@ -18,7 +18,6 @@ go_library(
         "//pkg/features:go_default_library",
         "//pkg/kubelet/events:go_default_library",
         "//pkg/util/mount:go_default_library",
-        "//pkg/util/resizefs:go_default_library",
         "//pkg/volume:go_default_library",
         "//pkg/volume/util:go_default_library",
         "//pkg/volume/util/nestedpendingoperations:go_default_library",

pkg/volume/util/operationexecutor/operation_generator.go

@@ -34,7 +34,6 @@ import (
 	"k8s.io/kubernetes/pkg/features"
 	kevents "k8s.io/kubernetes/pkg/kubelet/events"
 	"k8s.io/kubernetes/pkg/util/mount"
-	"k8s.io/kubernetes/pkg/util/resizefs"
 	"k8s.io/kubernetes/pkg/volume"
 	"k8s.io/kubernetes/pkg/volume/util"
 	volumetypes "k8s.io/kubernetes/pkg/volume/util/types"
@@ -604,10 +603,9 @@ func (og *operationGenerator) resizeFileSystem(volumeToMount VolumeToMount, devi
 		return nil, nil
 	}
 
-	mounter := og.volumePluginMgr.Host.GetMounter(pluginName)
 	// Get expander, if possible
 	expandableVolumePlugin, _ :=
-		og.volumePluginMgr.FindExpandablePluginBySpec(volumeToMount.VolumeSpec)
+		og.volumePluginMgr.FindFSResizablePluginBySpec(volumeToMount.VolumeSpec)
 
 	if expandableVolumePlugin != nil &&
 		expandableVolumePlugin.RequiresFSResize() &&
@@ -631,25 +629,12 @@ func (og *operationGenerator) resizeFileSystem(volumeToMount VolumeToMount, devi
 				og.recorder.Eventf(volumeToMount.Pod, v1.EventTypeWarning, kevents.FileSystemResizeFailed, simpleMsg)
 				return nil, nil
 			}
-
-			diskFormatter := &mount.SafeFormatAndMount{
-				Interface: mounter,
-				Exec:      og.volumePluginMgr.Host.GetExec(expandableVolumePlugin.GetPluginName()),
-			}
-
-			resizer := resizefs.NewResizeFs(diskFormatter)
-			resizeStatus, resizeErr := resizer.Resize(devicePath, deviceMountPath)
-
-			if resizeErr != nil {
+			if resizeErr := expandableVolumePlugin.ExpandFS(volumeToMount.VolumeSpec, devicePath, deviceMountPath, pvSpecCap, pvcStatusCap); resizeErr != nil {
 				return volumeToMount.GenerateError("MountVolume.resizeFileSystem failed", resizeErr)
 			}
-
-			if resizeStatus {
-				simpleMsg, detailedMsg := volumeToMount.GenerateMsg("MountVolume.resizeFileSystem succeeded", "")
-				og.recorder.Eventf(volumeToMount.Pod, v1.EventTypeNormal, kevents.FileSystemResizeSuccess, simpleMsg)
-				glog.Infof(detailedMsg)
-			}
-
+			simpleMsg, detailedMsg := volumeToMount.GenerateMsg("MountVolume.resizeFileSystem succeeded", "")
+			og.recorder.Eventf(volumeToMount.Pod, v1.EventTypeNormal, kevents.FileSystemResizeSuccess, simpleMsg)
+			glog.Infof(detailedMsg)
 			// File system resize succeeded, now update the PVC's Capacity to match the PV's
 			err = util.MarkFSResizeFinished(pvc, pv.Spec.Capacity, og.kubeClient)
 			if err != nil {
@@ -1270,6 +1255,7 @@ func (og *operationGenerator) GenerateExpandVolumeFunc(
 	if err != nil {
 		return volumetypes.GeneratedOperations{}, fmt.Errorf("Error finding plugin for expanding volume: %q with error %v", pvcWithResizeRequest.QualifiedName(), err)
 	}
+
 	if volumePlugin == nil {
 		return volumetypes.GeneratedOperations{}, fmt.Errorf("Can not find plugin for expanding volume: %q", pvcWithResizeRequest.QualifiedName())
 	}
@@ -1284,9 +1270,10 @@ func (og *operationGenerator) GenerateExpandVolumeFunc(
 				pvcWithResizeRequest.CurrentSize)
 
 			if expandErr != nil {
-				detailedErr := fmt.Errorf("Error expanding volume %q of plugin %s : %v", pvcWithResizeRequest.QualifiedName(), volumePlugin.GetPluginName(), expandErr)
+				detailedErr := fmt.Errorf("error expanding volume %q of plugin %q: %v", pvcWithResizeRequest.QualifiedName(), volumePlugin.GetPluginName(), expandErr)
 				return detailedErr, detailedErr
 			}
+
 			glog.Infof("ExpandVolume succeeded for volume %s", pvcWithResizeRequest.QualifiedName())
 			newSize = updatedSize
 			// k8s doesn't have transactions, we can't guarantee that after updating PV - updating PVC will be
@@ -1371,6 +1358,7 @@ func (og *operationGenerator) GenerateExpandVolumeFSWithoutUnmountingFunc(
 
 	fsResizeFunc := func() (error, error) {
 		resizeSimpleError, resizeDetailedError := og.resizeFileSystem(volumeToMount, volumeToMount.DevicePath, deviceMountPath, volumePlugin.GetPluginName())
+
 		if resizeSimpleError != nil || resizeDetailedError != nil {
 			return resizeSimpleError, resizeDetailedError
 		}

pkg/volume/util/resize_util.go

@@ -24,6 +24,9 @@ import (
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/strategicpatch"
 	clientset "k8s.io/client-go/kubernetes"
+	"k8s.io/kubernetes/pkg/util/mount"
+	"k8s.io/kubernetes/pkg/util/resizefs"
+	"k8s.io/kubernetes/pkg/volume"
 )
 
 var (
@@ -123,3 +126,14 @@ func MergeResizeConditionOnPVC(
 	pvc.Status.Conditions = newConditions
 	return pvc
 }
+
+// GenericResizeFS : call generic filesystem resizer for plugins that don't have any special filesystem resize requirements
+func GenericResizeFS(host volume.VolumeHost, pluginName, devicePath, deviceMountPath string) (bool, error) {
+	mounter := host.GetMounter(pluginName)
+	diskFormatter := &mount.SafeFormatAndMount{
+		Interface: mounter,
+		Exec:      host.GetExec(pluginName),
+	}
+	resizer := resizefs.NewResizeFs(diskFormatter)
+	return resizer.Resize(devicePath, deviceMountPath)
+}

test/e2e/storage/BUILD

@@ -8,6 +8,7 @@ go_library(
         "empty_dir_wrapper.go",
         "ephemeral_volume.go",
         "flexvolume.go",
+        "flexvolume_online_resize.go",
         "generic_persistent_volume-disruptive.go",
         "in_tree_volumes.go",
         "mounted_volume_resize.go",

test/e2e/storage/flexvolume_online_resize.go

@@ -0,0 +1,174 @@
+/*
+Copyright 2018 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package storage
+
+import (
+	"fmt"
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+	"k8s.io/api/core/v1"
+	storage "k8s.io/api/storage/v1"
+	"k8s.io/apimachinery/pkg/api/resource"
+	utilerrors "k8s.io/apimachinery/pkg/util/errors"
+	clientset "k8s.io/client-go/kubernetes"
+	"k8s.io/kubernetes/test/e2e/framework"
+	"k8s.io/kubernetes/test/e2e/storage/utils"
+	"path"
+)
+
+func createStorageClass(ns string, c clientset.Interface) (*storage.StorageClass, error) {
+	bindingMode := storage.VolumeBindingImmediate
+	stKlass := getStorageClass("flex-expand", map[string]string{}, &bindingMode, ns, "resizing")
+	allowExpansion := true
+	stKlass.AllowVolumeExpansion = &allowExpansion
+
+	var err error
+	stKlass, err = c.StorageV1().StorageClasses().Create(stKlass)
+	return stKlass, err
+}
+
+var _ = utils.SIGDescribe("Mounted flexvolume volume expand [Slow] [Feature:ExpandInUsePersistentVolumes]", func() {
+	var (
+		c                 clientset.Interface
+		ns                string
+		err               error
+		pvc               *v1.PersistentVolumeClaim
+		resizableSc       *storage.StorageClass
+		nodeName          string
+		isNodeLabeled     bool
+		nodeKeyValueLabel map[string]string
+		nodeLabelValue    string
+		nodeKey           string
+		nodeList          *v1.NodeList
+	)
+
+	f := framework.NewDefaultFramework("mounted-flexvolume-expand")
+	BeforeEach(func() {
+		framework.SkipUnlessProviderIs("aws", "gce", "local")
+		c = f.ClientSet
+		ns = f.Namespace.Name
+		framework.ExpectNoError(framework.WaitForAllNodesSchedulable(c, framework.TestContext.NodeSchedulableTimeout))
+
+		nodeList = framework.GetReadySchedulableNodesOrDie(f.ClientSet)
+		if len(nodeList.Items) == 0 {
+			framework.Failf("unable to find ready and schedulable Node")
+		}
+		nodeName = nodeList.Items[0].Name
+
+		nodeKey = "mounted_flexvolume_expand"
+
+		if !isNodeLabeled {
+			nodeLabelValue = ns
+			nodeKeyValueLabel = make(map[string]string)
+			nodeKeyValueLabel[nodeKey] = nodeLabelValue
+			framework.AddOrUpdateLabelOnNode(c, nodeName, nodeKey, nodeLabelValue)
+			isNodeLabeled = true
+		}
+
+		resizableSc, err = createStorageClass(ns, c)
+		if err != nil {
+			fmt.Printf("storage class creation error: %v\n", err)
+		}
+		Expect(err).NotTo(HaveOccurred(), "Error creating resizable storage class: %v", err)
+		Expect(*resizableSc.AllowVolumeExpansion).To(BeTrue())
+
+		pvc = getClaim("2Gi", ns)
+		pvc.Spec.StorageClassName = &resizableSc.Name
+		pvc, err = c.CoreV1().PersistentVolumeClaims(pvc.Namespace).Create(pvc)
+		Expect(err).NotTo(HaveOccurred(), "Error creating pvc: %v", err)
+
+	})
+
+	framework.AddCleanupAction(func() {
+		if len(nodeLabelValue) > 0 {
+			framework.RemoveLabelOffNode(c, nodeName, nodeKey)
+		}
+	})
+
+	AfterEach(func() {
+		framework.Logf("AfterEach: Cleaning up resources for mounted volume resize")
+
+		if c != nil {
+			if errs := framework.PVPVCCleanup(c, ns, nil, pvc); len(errs) > 0 {
+				framework.Failf("AfterEach: Failed to delete PVC and/or PV. Errors: %v", utilerrors.NewAggregate(errs))
+			}
+			pvc, nodeName, isNodeLabeled, nodeLabelValue = nil, "", false, ""
+			nodeKeyValueLabel = make(map[string]string)
+		}
+	})
+
+	It("should be resizable when mounted", func() {
+		driver := "dummy-attachable"
+
+		node := nodeList.Items[0]
+		By(fmt.Sprintf("installing flexvolume %s on node %s as %s", path.Join(driverDir, driver), node.Name, driver))
+		installFlex(c, &node, "k8s", driver, path.Join(driverDir, driver))
+
+		pv := framework.MakePersistentVolume(framework.PersistentVolumeConfig{
+			PVSource: v1.PersistentVolumeSource{
+				FlexVolume: &v1.FlexPersistentVolumeSource{
+					Driver: "k8s/" + driver,
+				}},
+			NamePrefix:       "pv-",
+			StorageClassName: resizableSc.Name,
+			VolumeMode:       pvc.Spec.VolumeMode,
+		})
+
+		pv, err = framework.CreatePV(c, pv)
+		Expect(err).NotTo(HaveOccurred(), "Error creating pv %v", err)
+
+		By("Waiting for PVC to be in bound phase")
+		pvcClaims := []*v1.PersistentVolumeClaim{pvc}
+		var pvs []*v1.PersistentVolume
+
+		pvs, err = framework.WaitForPVClaimBoundPhase(c, pvcClaims, framework.ClaimProvisionTimeout)
+		Expect(err).NotTo(HaveOccurred(), "Failed waiting for PVC to be bound %v", err)
+		Expect(len(pvs)).To(Equal(1))
+
+		var pod *v1.Pod
+		By("Creating pod")
+		pod, err = framework.CreateNginxPod(c, ns, nodeKeyValueLabel, pvcClaims)
+		Expect(err).NotTo(HaveOccurred(), "Failed to create pod %v", err)
+		defer framework.DeletePodWithWait(f, c, pod)
+
+		By("Waiting for pod to go to 'running' state")
+		err = f.WaitForPodRunning(pod.ObjectMeta.Name)
+		Expect(err).NotTo(HaveOccurred(), "Pod didn't go to 'running' state %v", err)
+
+		By("Expanding current pvc")
+		newSize := resource.MustParse("6Gi")
+		pvc, err = expandPVCSize(pvc, newSize, c)
+		Expect(err).NotTo(HaveOccurred(), "While updating pvc for more size")
+		Expect(pvc).NotTo(BeNil())
+
+		pvcSize := pvc.Spec.Resources.Requests[v1.ResourceStorage]
+		if pvcSize.Cmp(newSize) != 0 {
+			framework.Failf("error updating pvc size %q", pvc.Name)
+		}
+
+		By("Waiting for cloudprovider resize to finish")
+		err = waitForControllerVolumeResize(pvc, c)
+		Expect(err).NotTo(HaveOccurred(), "While waiting for pvc resize to finish")
+
+		By("Waiting for file system resize to finish")
+		pvc, err = waitForFSResize(pvc, c)
+		Expect(err).NotTo(HaveOccurred(), "while waiting for fs resize to finish")
+
+		pvcConditions := pvc.Status.Conditions
+		Expect(len(pvcConditions)).To(Equal(0), "pvc should not have conditions")
+	})
+})

test/e2e/testing-manifests/flexvolume/dummy-attachable

@@ -90,11 +90,23 @@ unmountdevice() {
 	exit 0
 }
 
+expandvolume() {
+	debug "expandvolume $@"
+	log "{\"status\":\"Success\"}"
+	exit 0
+}
+
+expandfs() {
+	debug "expandfs $@"
+	log "{\"status\":\"Success\"}"
+	exit 0
+}
+
 op=$1
 
 if [ "$op" = "init" ]; then
 	debug "init $@"
-	log "{\"status\":\"Success\",\"capabilities\":{\"attach\":true}}"
+	log "{\"status\":\"Success\",\"capabilities\":{\"attach\":true, \"requiresFSResize\":true}}"
 	exit 0
 fi
 
@@ -119,6 +131,12 @@ case "$op" in
 	unmountdevice)
 		unmountdevice $*
 		;;
+	expandvolume)
+		expandvolume $*
+		;;
+	expandfs)
+		expandfs $*
+		;;
 	*)
 		log "{\"status\":\"Not supported\"}"
 		exit 0