Add CSI volume resizing tests

pkg/volume/csi/BUILD

@@ -10,6 +10,7 @@ go_library(
         "csi_mounter.go",
         "csi_plugin.go",
         "csi_util.go",
+        "expander.go",
     ],
     importpath = "k8s.io/kubernetes/pkg/volume/csi",
     visibility = ["//visibility:public"],
@@ -22,6 +23,7 @@ go_library(
         "//staging/src/k8s.io/api/core/v1:go_default_library",
         "//staging/src/k8s.io/api/storage/v1:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/api/errors:go_default_library",
+        "//staging/src/k8s.io/apimachinery/pkg/api/resource:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/types:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/util/runtime:go_default_library",
@@ -50,6 +52,7 @@ go_test(
         "csi_drivers_store_test.go",
         "csi_mounter_test.go",
         "csi_plugin_test.go",
+        "expander_test.go",
     ],
     embed = [":go_default_library"],
     deps = [

pkg/volume/csi/csi_client.go

@@ -56,7 +56,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, volumeid, volumePath string, newSize resource.Quantity) (resource.Quantity, error)
 	NodeUnpublishVolume(
 		ctx context.Context,
 		volID string,
@@ -307,14 +307,25 @@ func (c *csiDriverClient) NodePublishVolume(
 
 }
 
-func (c *csiDriverClient) NodeExpandVolume(ctx context.Context, volumeID, volumePath string, newSize resource.Quantity) (*resource.Quantity, error) {
+func (c *csiDriverClient) NodeExpandVolume(ctx context.Context, volumeID, volumePath string, newSize resource.Quantity) (resource.Quantity, error) {
 	if c.nodeV1ClientCreator == nil {
-		return nil, fmt.Errorf("version of CSI driver does not support volume expansion")
+		return newSize, fmt.Errorf("version of CSI driver does not support volume expansion")
+	}
+
+	if volumeID == "" {
+		return newSize, errors.New("missing volume id")
+	}
+	if volumePath == "" {
+		return newSize, errors.New("missing volume path")
+	}
+
+	if newSize.Value() < 0 {
+		return newSize, errors.New("size can not be less than 0")
 	}
 
 	nodeClient, closer, err := c.nodeV1ClientCreator(c.addr)
 	if err != nil {
-		return nil, err
+		return newSize, err
 	}
 	defer closer.Close()
 
@@ -325,10 +336,10 @@ func (c *csiDriverClient) NodeExpandVolume(ctx context.Context, volumeID, volume
 	}
 	resp, err := nodeClient.NodeExpandVolume(ctx, req)
 	if err != nil {
-		return nil, err
+		return newSize, err
 	}
 	updatedQuantity := resource.NewQuantity(resp.CapacityBytes, resource.BinarySI)
-	return updatedQuantity, nil
+	return *updatedQuantity, nil
 }
 
 func (c *csiDriverClient) nodePublishVolumeV1(

pkg/volume/csi/csi_client_test.go

@@ -25,6 +25,7 @@ import (
 
 	csipbv1 "github.com/container-storage-interface/spec/lib/go/csi"
 	api "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/resource"
 	"k8s.io/kubernetes/pkg/volume/csi/fake"
 )
 
@@ -40,6 +41,13 @@ func newFakeCsiDriverClient(t *testing.T, stagingCapable bool) *fakeCsiDriverCli
 	}
 }
 
+func newFakeCsiDriverClientWithExpansion(t *testing.T, stagingCapable bool, expansionSet bool) *fakeCsiDriverClient {
+	return &fakeCsiDriverClient{
+		t:          t,
+		nodeClient: fake.NewNodeClientWithExpansion(stagingCapable, expansionSet),
+	}
+}
+
 func (c *fakeCsiDriverClient) NodeGetInfo(ctx context.Context) (
 	nodeID string,
 	maxVolumePerNode int64,
@@ -144,6 +152,28 @@ func (c *fakeCsiDriverClient) NodeUnstageVolume(ctx context.Context, volID, stag
 	return err
 }
 
+func (c *fakeCsiDriverClient) NodeSupportsNodeExpand(ctx context.Context) (bool, error) {
+	c.t.Log("calling fake.NodeSupportsNodeExpand...")
+	req := &csipbv1.NodeGetCapabilitiesRequest{}
+
+	resp, err := c.nodeClient.NodeGetCapabilities(ctx, req)
+	if err != nil {
+		return false, err
+	}
+
+	capabilities := resp.GetCapabilities()
+
+	if capabilities == nil {
+		return false, nil
+	}
+	for _, capability := range capabilities {
+		if capability.GetRpc().GetType() == csipbv1.NodeServiceCapability_RPC_EXPAND_VOLUME {
+			return true, nil
+		}
+	}
+	return false, nil
+}
+
 func (c *fakeCsiDriverClient) NodeSupportsStageUnstage(ctx context.Context) (bool, error) {
 	c.t.Log("calling fake.NodeGetCapabilities for NodeSupportsStageUnstage...")
 	req := &csipbv1.NodeGetCapabilitiesRequest{}
@@ -166,10 +196,29 @@ 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) {
+	c.t.Log("calling fake.NodeExpandVolume")
+	req := &csipbv1.NodeExpandVolumeRequest{
+		VolumeId:      volumeid,
+		VolumePath:    volumePath,
+		CapacityRange: &csipbv1.CapacityRange{RequiredBytes: newSize.Value()},
+	}
+	resp, err := c.nodeClient.NodeExpandVolume(ctx, req)
+	if err != nil {
+		return newSize, err
+	}
+	updatedQuantity := resource.NewQuantity(resp.CapacityBytes, resource.BinarySI)
+	return *updatedQuantity, nil
+}
+
 func setupClient(t *testing.T, stageUnstageSet bool) csiClient {
 	return newFakeCsiDriverClient(t, stageUnstageSet)
 }
 
+func setupClientWithExpansion(t *testing.T, stageUnstageSet bool, expansionSet bool) csiClient {
+	return newFakeCsiDriverClientWithExpansion(t, stageUnstageSet, expansionSet)
+}
+
 func checkErr(t *testing.T, expectedAnError bool, actualError error) {
 	t.Helper()
 
@@ -415,3 +464,59 @@ func TestClientNodeUnstageVolume(t *testing.T) {
 		}
 	}
 }
+
+func TestNodeExpandVolume(t *testing.T) {
+	testCases := []struct {
+		name       string
+		volID      string
+		volumePath string
+		newSize    resource.Quantity
+		mustFail   bool
+		err        error
+	}{
+		{
+			name:       "with all correct values",
+			volID:      "vol-abcde",
+			volumePath: "/foo/bar",
+			newSize:    resource.MustParse("10Gi"),
+			mustFail:   false,
+		},
+		{
+			name:       "with missing volume-id",
+			volumePath: "/foo/bar",
+			newSize:    resource.MustParse("10Gi"),
+			mustFail:   true,
+		},
+		{
+			name:     "with missing volume path",
+			volID:    "vol-1234",
+			newSize:  resource.MustParse("10Gi"),
+			mustFail: true,
+		},
+		{
+			name:       "with invalid quantity",
+			volID:      "vol-1234",
+			volumePath: "/foo/bar",
+			newSize:    *resource.NewQuantity(-10, resource.DecimalSI),
+			mustFail:   true,
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Logf("Running test cases : %s", tc.name)
+		fakeCloser := fake.NewCloser(t)
+		client := &csiDriverClient{
+			driverName: "Fake Driver Name",
+			nodeV1ClientCreator: func(addr csiAddr) (csipbv1.NodeClient, io.Closer, error) {
+				nodeClient := fake.NewNodeClient(false /* stagingCapable */)
+				nodeClient.SetNextError(tc.err)
+				return nodeClient, fakeCloser, nil
+			},
+		}
+		_, err := client.NodeExpandVolume(context.Background(), tc.volID, tc.volumePath, tc.newSize)
+		checkErr(t, tc.mustFail, err)
+		if !tc.mustFail {
+			fakeCloser.Check()
+		}
+	}
+}

pkg/volume/csi/expander.go

@@ -18,9 +18,9 @@ package csi
 
 import (
 	"context"
-	"errors"
 	"fmt"
 
+	api "k8s.io/api/core/v1"
 	utilfeature "k8s.io/apiserver/pkg/util/feature"
 	"k8s.io/klog"
 	"k8s.io/kubernetes/pkg/features"
@@ -34,7 +34,7 @@ func (c *csiPlugin) RequiresFSResize() bool {
 	// 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")
+		klog.V(4).Infof("Resizing is not enabled for CSI volume")
 		return false
 	}
 	return true
@@ -42,40 +42,40 @@ func (c *csiPlugin) RequiresFSResize() bool {
 
 func (c *csiPlugin) NodeExpand(resizeOptions volume.NodeResizeOptions) (bool, error) {
 	klog.V(4).Infof(log("Expander.NodeExpand(%s)", resizeOptions.DeviceMountPath))
-	pvSource, err := getCSISourceFromSpec(resizeOptions.VolumeSpec)
+	csiSource, err := getCSISourceFromSpec(resizeOptions.VolumeSpec)
 	if err != nil {
+		klog.Error(log("Expander.NodeExpand failed to get CSI persistent source: %v", err))
 		return false, err
 	}
-	k8s := c.host.GetKubeClient()
-	if k8s == nil {
-		klog.Error(log("failed to get a kubernetes client"))
-		return false, errors.New("failed to get a Kubernetes client")
-	}
 
-	csiClient, err := newCsiDriverClient(csiDriverName(pvSource.Driver))
+	csClient, err := newCsiDriverClient(csiDriverName(csiSource.Driver))
 	if err != nil {
 		return false, err
 	}
 
-	csiSource, err := getCSISourceFromSpec(resizeOptions.VolumeSpec)
-	if err != nil {
-		klog.Error(log("Expander.NodeExpand failed to get CSI persistent source: %v", err))
-		return false, err
-	}
+	return c.nodeExpandWithClient(resizeOptions, csiSource, csClient)
+}
+
+func (c *csiPlugin) nodeExpandWithClient(
+	resizeOptions volume.NodeResizeOptions,
+	csiSource *api.CSIPersistentVolumeSource,
+	csClient csiClient) (bool, error) {
+	driverName := csiSource.Driver
+
 	ctx, cancel := context.WithTimeout(context.Background(), csiTimeout)
 	defer cancel()
 
-	nodeExpandSet, err := csiClient.NodeSupportsNodeExpand(ctx)
+	nodeExpandSet, err := csClient.NodeSupportsNodeExpand(ctx)
 	if err != nil {
 		return false, fmt.Errorf("Expander.NodeExpand failed to check if node supports expansion : %v", err)
 	}
 
 	if !nodeExpandSet {
-		return false, fmt.Errorf("Expander.NodeExpand found CSI plugin %s to not support node expansion", c.GetPluginName())
+		return false, fmt.Errorf("Expander.NodeExpand found CSI plugin %s/%s to not support node expansion", c.GetPluginName(), driverName)
 	}
 
 	// Check whether "STAGE_UNSTAGE_VOLUME" is set
-	stageUnstageSet, err := csiClient.NodeSupportsStageUnstage(ctx)
+	stageUnstageSet, err := csClient.NodeSupportsStageUnstage(ctx)
 	if err != nil {
 		return false, fmt.Errorf("Expander.NodeExpand failed to check if plugins supports stage_unstage %v", err)
 	}
@@ -88,7 +88,7 @@ func (c *csiPlugin) NodeExpand(resizeOptions volume.NodeResizeOptions) (bool, er
 		return false, nil
 	}
 
-	_, err = csiClient.NodeExpandVolume(ctx, csiSource.VolumeHandle, resizeOptions.DeviceMountPath, resizeOptions.NewSize)
+	_, err = csClient.NodeExpandVolume(ctx, csiSource.VolumeHandle, resizeOptions.DeviceMountPath, resizeOptions.NewSize)
 	if err != nil {
 		return false, fmt.Errorf("Expander.NodeExpand failed to expand the volume : %v", err)
 	}

pkg/volume/csi/expander_test.go

+/*
+Copyright 2019 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 csi
+
+import (
+	"os"
+	"testing"
+
+	"k8s.io/apimachinery/pkg/api/resource"
+	"k8s.io/kubernetes/pkg/volume"
+)
+
+func TestNodeExpand(t *testing.T) {
+	tests := []struct {
+		name          string
+		nodeExpansion bool
+		nodeStageSet  bool
+		volumePhase   volume.CSIVolumePhaseType
+		success       bool
+	}{
+		{
+			name:    "when node expansion is not set",
+			success: false,
+		},
+		{
+			name:          "when nodeExpansion=on, nodeStage=on, volumePhase=staged",
+			nodeExpansion: true,
+			nodeStageSet:  true,
+			volumePhase:   volume.CSIVolumeStaged,
+			success:       true,
+		},
+		{
+			name:          "when nodeExpansion=on, nodeStage=off, volumePhase=staged",
+			nodeExpansion: true,
+			volumePhase:   volume.CSIVolumeStaged,
+			success:       false,
+		},
+		{
+			name:          "when nodeExpansion=on, nodeStage=on, volumePhase=published",
+			nodeExpansion: true,
+			nodeStageSet:  true,
+			volumePhase:   volume.CSIVolumePublished,
+			success:       true,
+		},
+		{
+			name:          "when nodeExpansion=on, nodeStage=off, volumePhase=published",
+			nodeExpansion: true,
+			volumePhase:   volume.CSIVolumePublished,
+			success:       true,
+		},
+	}
+	for _, tc := range tests {
+		plug, tmpDir := newTestPlugin(t, nil)
+		defer os.RemoveAll(tmpDir)
+
+		spec := volume.NewSpecFromPersistentVolume(makeTestPV("test-pv", 10, "expandable", "test-vol"), false)
+
+		newSize, _ := resource.ParseQuantity("20Gi")
+
+		resizeOptions := volume.NodeResizeOptions{
+			VolumeSpec:      spec,
+			NewSize:         newSize,
+			DeviceMountPath: "/foo/bar",
+			CSIVolumePhase:  tc.volumePhase,
+		}
+		csiSource, _ := getCSISourceFromSpec(resizeOptions.VolumeSpec)
+
+		csClient := setupClientWithExpansion(t, tc.nodeStageSet, tc.nodeExpansion)
+
+		ok, err := plug.nodeExpandWithClient(resizeOptions, csiSource, csClient)
+		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)
+			}
+
+		}
+	}
+}

pkg/volume/csi/fake/fake_client.go

@@ -67,6 +67,7 @@ type NodeClient struct {
 	nodePublishedVolumes map[string]CSIVolume
 	nodeStagedVolumes    map[string]CSIVolume
 	stageUnstageSet      bool
+	expansionSet         bool
 	nodeGetInfoResp      *csipb.NodeGetInfoResponse
 	nextErr              error
 }
@@ -80,6 +81,15 @@ func NewNodeClient(stageUnstageSet bool) *NodeClient {
 	}
 }
 
+func NewNodeClientWithExpansion(stageUnstageSet bool, expansionSet bool) *NodeClient {
+	return &NodeClient{
+		nodePublishedVolumes: make(map[string]CSIVolume),
+		nodeStagedVolumes:    make(map[string]CSIVolume),
+		stageUnstageSet:      stageUnstageSet,
+		expansionSet:         expansionSet,
+	}
+}
+
 // SetNextError injects next expected error
 func (f *NodeClient) SetNextError(err error) {
 	f.nextErr = err
@@ -195,6 +205,29 @@ func (f *NodeClient) NodeUnstageVolume(ctx context.Context, req *csipb.NodeUnsta
 	return &csipb.NodeUnstageVolumeResponse{}, nil
 }
 
+// NodeExpandVolume implements csi method
+func (f *NodeClient) NodeExpandVolume(ctx context.Context, req *csipb.NodeExpandVolumeRequest, opts ...grpc.CallOption) (*csipb.NodeExpandVolumeResponse, error) {
+	if f.nextErr != nil {
+		return nil, f.nextErr
+	}
+
+	if req.GetVolumeId() == "" {
+		return nil, errors.New("missing volume id")
+	}
+	if req.GetVolumePath() == "" {
+		return nil, errors.New("missing volume path")
+	}
+
+	if req.GetCapacityRange().RequiredBytes <= 0 {
+		return nil, errors.New("required bytes should be greater than 0")
+	}
+
+	resp := &csipb.NodeExpandVolumeResponse{
+		CapacityBytes: req.GetCapacityRange().RequiredBytes,
+	}
+	return resp, nil
+}
+
 // NodeGetId implements csi method
 func (f *NodeClient) NodeGetInfo(ctx context.Context, in *csipb.NodeGetInfoRequest, opts ...grpc.CallOption) (*csipb.NodeGetInfoResponse, error) {
 	if f.nextErr != nil {
@@ -206,20 +239,27 @@ func (f *NodeClient) NodeGetInfo(ctx context.Context, in *csipb.NodeGetInfoReque
 // NodeGetCapabilities implements csi method
 func (f *NodeClient) NodeGetCapabilities(ctx context.Context, in *csipb.NodeGetCapabilitiesRequest, opts ...grpc.CallOption) (*csipb.NodeGetCapabilitiesResponse, error) {
 	resp := &csipb.NodeGetCapabilitiesResponse{
-		Capabilities: []*csipb.NodeServiceCapability{
-			{
-				Type: &csipb.NodeServiceCapability_Rpc{
-					Rpc: &csipb.NodeServiceCapability_RPC{
-						Type: csipb.NodeServiceCapability_RPC_STAGE_UNSTAGE_VOLUME,
-					},
+		Capabilities: []*csipb.NodeServiceCapability{},
+	}
+	if f.stageUnstageSet {
+		resp.Capabilities = append(resp.Capabilities, &csipb.NodeServiceCapability{
+			Type: &csipb.NodeServiceCapability_Rpc{
+				Rpc: &csipb.NodeServiceCapability_RPC{
+					Type: csipb.NodeServiceCapability_RPC_STAGE_UNSTAGE_VOLUME,
 				},
 			},
-		},
+		})
 	}
-	if f.stageUnstageSet {
-		return resp, nil
+	if f.expansionSet {
+		resp.Capabilities = append(resp.Capabilities, &csipb.NodeServiceCapability{
+			Type: &csipb.NodeServiceCapability_Rpc{
+				Rpc: &csipb.NodeServiceCapability_RPC{
+					Type: csipb.NodeServiceCapability_RPC_EXPAND_VOLUME,
+				},
+			},
+		})
 	}
-	return nil, nil
+	return resp, nil
 }
 
 // NodeGetVolumeStats implements csi method

pkg/volume/util/operationexecutor/operation_generator.go

@@ -640,6 +640,7 @@ func (og *operationGenerator) GenerateMountVolumeFunc(
 			resizeDone, resizeError = og.resizeFileSystem(volumeToMount, resizeOptions, volumePluginName)
 
 			if resizeError != nil {
+				klog.Errorf("MountVolume.resizeFileSystem failed with %v", resizeError)
 				return volumeToMount.GenerateError("MountVolume.MountDevice failed while expanding volume", resizeError)
 			}
 		}
@@ -669,9 +670,14 @@ func (og *operationGenerator) GenerateMountVolumeFunc(
 		resizeOptions.DeviceMountPath = volumeMounter.GetPath()
 		resizeOptions.CSIVolumePhase = volume.CSIVolumePublished
 
+		// We need to call resizing here again in case resizing was not done during device mount. There could be
+		// two reasons of that:
+		//	- Volume does not support DeviceMounter interface.
+		//	- In case of CSI the volume does not have node stage_unstage capability.
 		if !resizeDone {
 			resizeDone, resizeError = og.resizeFileSystem(volumeToMount, resizeOptions, volumePluginName)
 			if resizeError != nil {
+				klog.Errorf("MountVolume.resizeFileSystem failed with %v", resizeError)
 				return volumeToMount.GenerateError("MountVolume.Setup failed while expanding volume", resizeError)
 			}
 		}
@@ -746,6 +752,9 @@ func (og *operationGenerator) resizeFileSystem(volumeToMount VolumeToMount, rsOp
 			if resizeErr != nil {
 				return false, fmt.Errorf("MountVolume.resizeFileSystem failed : %v", resizeErr)
 			}
+			// Volume resizing is not done but it did not error out. This could happen if a CSI volume
+			// does not have node stage_unstage capability but was asked to resize the volume before
+			// node publish. In which case - we must retry resizing after node publish.
 			if !resizeDone {
 				return false, nil
 			}
@@ -1473,6 +1482,7 @@ func (og *operationGenerator) GenerateExpandVolumeFSWithoutUnmountingFunc(
 
 	fsResizeFunc := func() (error, error) {
 		var resizeDone bool
+		var simpleErr, detailedErr error
 		resizeOptions := volume.NodeResizeOptions{
 			VolumeSpec: volumeToMount.VolumeSpec,
 		}
@@ -1490,16 +1500,11 @@ func (og *operationGenerator) GenerateExpandVolumeFSWithoutUnmountingFunc(
 					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)
+				resizeDone, simpleErr, detailedErr = og.doOnlineExpansion(volumeToMount, actualStateOfWorld, resizeOptions, volumePlugin.GetPluginName())
+				if simpleErr != nil || detailedErr != nil {
+					return simpleErr, detailedErr
 				}
 				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
 				}
 			}
@@ -1515,16 +1520,11 @@ func (og *operationGenerator) GenerateExpandVolumeFSWithoutUnmountingFunc(
 
 		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)
+		resizeDone, simpleErr, detailedErr = og.doOnlineExpansion(volumeToMount, actualStateOfWorld, resizeOptions, volumePlugin.GetPluginName())
+		if simpleErr != nil || detailedErr != nil {
+			return simpleErr, detailedErr
 		}
 		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
 		}
 		// This is a placeholder error - we should NEVER reach here.
@@ -1545,6 +1545,28 @@ func (og *operationGenerator) GenerateExpandVolumeFSWithoutUnmountingFunc(
 	}, nil
 }
 
+func (og *operationGenerator) doOnlineExpansion(volumeToMount VolumeToMount,
+	actualStateOfWorld ActualStateOfWorldMounterUpdater,
+	resizeOptions volume.NodeResizeOptions,
+	pluginName string) (bool, error, error) {
+	resizeDone, err := og.resizeFileSystem(volumeToMount, resizeOptions, pluginName)
+	if err != nil {
+		klog.Errorf("VolumeFSResize.resizeFileSystem failed : %v", err)
+		e1, e2 := volumeToMount.GenerateError("VolumeFSResize.resizeFileSystem failed", err)
+		return false, e1, e2
+	}
+	if resizeDone {
+		markFSResizedErr := actualStateOfWorld.MarkVolumeAsResized(volumeToMount.PodName, volumeToMount.VolumeName)
+		if markFSResizedErr != nil {
+			// On failure, return error. Caller will log and retry.
+			e1, e2 := volumeToMount.GenerateError("VolumeFSResize.MarkVolumeAsResized failed", markFSResizedErr)
+			return false, e1, e2
+		}
+		return true, nil, nil
+	}
+	return false, nil, nil
+}
+
 func checkMountOptionSupport(og *operationGenerator, volumeToMount VolumeToMount, plugin volume.VolumePlugin) error {
 	mountOptions := util.MountOptionFromSpec(volumeToMount.VolumeSpec)
 

test/e2e/storage/drivers/csi.go

@@ -169,36 +169,52 @@ func (h *hostpathCSIDriver) PrepareTest(f *framework.Framework) (*testsuites.Per
 
 // mockCSI
 type mockCSIDriver struct {
-	driverInfo  testsuites.DriverInfo
-	manifests   []string
-	podInfo     *bool
-	attachable  bool
-	attachLimit int
+	driverInfo          testsuites.DriverInfo
+	manifests           []string
+	podInfo             *bool
+	attachable          bool
+	attachLimit         int
+	enableNodeExpansion bool
+}
+
+// CSIMockDriverOpts defines options used for csi driver
+type CSIMockDriverOpts struct {
+	RegisterDriver      bool
+	DisableAttach       bool
+	PodInfo             *bool
+	AttachLimit         int
+	EnableResizing      bool
+	EnableNodeExpansion bool
 }
 
 var _ testsuites.TestDriver = &mockCSIDriver{}
 var _ testsuites.DynamicPVTestDriver = &mockCSIDriver{}
 
 // InitMockCSIDriver returns a mockCSIDriver that implements TestDriver interface
-func InitMockCSIDriver(registerDriver, driverAttachable bool, podInfo *bool, attachLimit int) testsuites.TestDriver {
+func InitMockCSIDriver(driverOpts CSIMockDriverOpts) testsuites.TestDriver {
 	driverManifests := []string{
 		"test/e2e/testing-manifests/storage-csi/cluster-driver-registrar/rbac.yaml",
 		"test/e2e/testing-manifests/storage-csi/driver-registrar/rbac.yaml",
 		"test/e2e/testing-manifests/storage-csi/external-attacher/rbac.yaml",
 		"test/e2e/testing-manifests/storage-csi/external-provisioner/rbac.yaml",
+		"test/e2e/testing-manifests/storage-csi/external-resizer/rbac.yaml",
 		"test/e2e/testing-manifests/storage-csi/mock/csi-mock-rbac.yaml",
 		"test/e2e/testing-manifests/storage-csi/mock/csi-storageclass.yaml",
 		"test/e2e/testing-manifests/storage-csi/mock/csi-mock-driver.yaml",
 	}
 
-	if registerDriver {
+	if driverOpts.RegisterDriver {
 		driverManifests = append(driverManifests, "test/e2e/testing-manifests/storage-csi/mock/csi-mock-cluster-driver-registrar.yaml")
 	}
 
-	if driverAttachable {
+	if !driverOpts.DisableAttach {
 		driverManifests = append(driverManifests, "test/e2e/testing-manifests/storage-csi/mock/csi-mock-driver-attacher.yaml")
 	}
 
+	if driverOpts.EnableResizing {
+		driverManifests = append(driverManifests, "test/e2e/testing-manifests/storage-csi/mock/csi-mock-driver-resizer.yaml")
+	}
+
 	return &mockCSIDriver{
 		driverInfo: testsuites.DriverInfo{
 			Name:        "csi-mock",
@@ -213,10 +229,11 @@ func InitMockCSIDriver(registerDriver, driverAttachable bool, podInfo *bool, att
 				testsuites.CapExec:        false,
 			},
 		},
-		manifests:   driverManifests,
-		podInfo:     podInfo,
-		attachable:  driverAttachable,
-		attachLimit: attachLimit,
+		manifests:           driverManifests,
+		podInfo:             driverOpts.PodInfo,
+		attachable:          !driverOpts.DisableAttach,
+		attachLimit:         driverOpts.AttachLimit,
+		enableNodeExpansion: driverOpts.EnableNodeExpansion,
 	}
 }
 
@@ -264,6 +281,10 @@ func (m *mockCSIDriver) PrepareTest(f *framework.Framework) (*testsuites.PerTest
 		containerArgs = append(containerArgs, "--attach-limit", strconv.Itoa(m.attachLimit))
 	}
 
+	if m.enableNodeExpansion {
+		containerArgs = append(containerArgs, "--node-expand-required=true")
+	}
+
 	// TODO (?): the storage.csi.image.version and storage.csi.image.registry
 	// settings are ignored for this test. We could patch the image definitions.
 	o := utils.PatchCSIOptions{

test/e2e/storage/flexvolume_mounted_volume_resize.go

@@ -159,7 +159,7 @@ var _ = utils.SIGDescribe("Mounted flexvolume expand[Slow]", func() {
 		}
 
 		By("Waiting for cloudprovider resize to finish")
-		err = waitForControllerVolumeResize(pvc, c)
+		err = waitForControllerVolumeResize(pvc, c, totalResizeWaitPeriod)
 		Expect(err).NotTo(HaveOccurred(), "While waiting for pvc resize to finish")
 
 		By("Getting a pod from deployment")

test/e2e/storage/flexvolume_online_resize.go

@@ -164,7 +164,7 @@ var _ = utils.SIGDescribe("Mounted flexvolume volume expand [Slow] [Feature:Expa
 		}
 
 		By("Waiting for cloudprovider resize to finish")
-		err = waitForControllerVolumeResize(pvc, c)
+		err = waitForControllerVolumeResize(pvc, c, totalResizeWaitPeriod)
 		Expect(err).NotTo(HaveOccurred(), "While waiting for pvc resize to finish")
 
 		By("Waiting for file system resize to finish")

test/e2e/storage/mounted_volume_resize.go

@@ -136,7 +136,7 @@ var _ = utils.SIGDescribe("Mounted volume expand", func() {
 		}
 
 		By("Waiting for cloudprovider resize to finish")
-		err = waitForControllerVolumeResize(pvc, c)
+		err = waitForControllerVolumeResize(pvc, c, totalResizeWaitPeriod)
 		Expect(err).NotTo(HaveOccurred(), "While waiting for pvc resize to finish")
 
 		By("Getting a pod from deployment")

test/e2e/storage/volume_expand.go

@@ -136,7 +136,7 @@ var _ = utils.SIGDescribe("Volume expand", func() {
 		}
 
 		By("Waiting for cloudprovider resize to finish")
-		err = waitForControllerVolumeResize(pvc, c)
+		err = waitForControllerVolumeResize(pvc, c, totalResizeWaitPeriod)
 		Expect(err).NotTo(HaveOccurred(), "While waiting for pvc resize to finish")
 
 		By("Checking for conditions on pvc")
@@ -198,9 +198,29 @@ func expandPVCSize(origPVC *v1.PersistentVolumeClaim, size resource.Quantity, c
 	return updatedPVC, waitErr
 }
 
-func waitForControllerVolumeResize(pvc *v1.PersistentVolumeClaim, c clientset.Interface) error {
+func waitForResizingCondition(pvc *v1.PersistentVolumeClaim, c clientset.Interface, duration time.Duration) error {
+	waitErr := wait.PollImmediate(resizePollInterval, duration, func() (bool, error) {
+		var err error
+		updatedPVC, err := c.CoreV1().PersistentVolumeClaims(pvc.Namespace).Get(pvc.Name, metav1.GetOptions{})
+
+		if err != nil {
+			return false, fmt.Errorf("error fetching pvc %q for checking for resize status : %v", pvc.Name, err)
+		}
+
+		pvcConditions := updatedPVC.Status.Conditions
+		if len(pvcConditions) > 0 {
+			if pvcConditions[0].Type == v1.PersistentVolumeClaimResizing {
+				return true, nil
+			}
+		}
+		return false, nil
+	})
+	return waitErr
+}
+
+func waitForControllerVolumeResize(pvc *v1.PersistentVolumeClaim, c clientset.Interface, duration time.Duration) error {
 	pvName := pvc.Spec.VolumeName
-	return wait.PollImmediate(resizePollInterval, totalResizeWaitPeriod, func() (bool, error) {
+	return wait.PollImmediate(resizePollInterval, duration, func() (bool, error) {
 		pvcSize := pvc.Spec.Resources.Requests[v1.ResourceStorage]
 
 		pv, err := c.CoreV1().PersistentVolumes().Get(pvName, metav1.GetOptions{})

test/e2e/testing-manifests/storage-csi/external-resizer/README.md

+The original file is (or will be) https://github.com/kubernetes-csi/external-resizer/blob/master/deploy/kubernetes/rbac.yaml

test/e2e/testing-manifests/storage-csi/external-resizer/rbac.yaml

+# This YAML file contains all RBAC objects that are necessary to run external
+# CSI resizer.
+#
+# In production, each CSI driver deployment has to be customized:
+# - to avoid conflicts, use non-default namespace and different names
+#   for non-namespaced entities like the ClusterRole
+# - decide whether the deployment replicates the external CSI
+#   resizer, in which case leadership election must be enabled;
+#   this influences the RBAC setup, see below
+
+apiVersion: v1
+kind: ServiceAccount
+metadata:
+  name: csi-resizer
+  # replace with non-default namespace name
+  namespace: default
+
+---
+# Resizer must be able to work with PVCs, PVs, SCs.
+kind: ClusterRole
+apiVersion: rbac.authorization.k8s.io/v1
+metadata:
+  name: external-resizer-runner
+rules:
+  # The following rule should be uncommented for plugins that require secrets
+  # for provisioning.
+  # - apiGroups: [""]
+  #   resources: ["secrets"]
+  #   verbs: ["get", "list", "watch"]
+  - apiGroups: [""]
+    resources: ["persistentvolumes"]
+    verbs: ["get", "list", "watch", "update", "patch"]
+  - apiGroups: [""]
+    resources: ["persistentvolumeclaims"]
+    verbs: ["get", "list", "watch"]
+  - apiGroups: [""]
+    resources: ["persistentvolumeclaims/status"]
+    verbs: ["update", "patch"]
+  - apiGroups: ["storage.k8s.io"]
+    resources: ["storageclasses"]
+    verbs: ["get", "list", "watch"]
+  - apiGroups: [""]
+    resources: ["events"]
+    verbs: ["list", "watch", "create", "update", "patch"]
+
+---
+kind: ClusterRoleBinding
+apiVersion: rbac.authorization.k8s.io/v1
+metadata:
+  name: csi-resizer-role
+subjects:
+  - kind: ServiceAccount
+    name: csi-resizer
+    # replace with non-default namespace name
+    namespace: default
+roleRef:
+  kind: ClusterRole
+  name: external-resizer-runner
+  apiGroup: rbac.authorization.k8s.io
+
+---
+# Resizer must be able to work with end point in current namespace
+# if (and only if) leadership election is enabled
+kind: Role
+apiVersion: rbac.authorization.k8s.io/v1
+metadata:
+  # replace with non-default namespace name
+  namespace: default
+  name: external-resizer-cfg
+rules:
+- apiGroups: [""]
+  resources: ["endpoints"]
+  verbs: ["get", "watch", "list", "delete", "update", "create"]
+
+---
+kind: RoleBinding
+apiVersion: rbac.authorization.k8s.io/v1
+metadata:
+  name: csi-resizer-role-cfg
+  # replace with non-default namespace name
+  namespace: default
+subjects:
+  - kind: ServiceAccount
+    name: csi-resizer
+    # replace with non-default namespace name
+    namespace: default
+roleRef:
+  kind: Role
+  name: external-resizer-cfg
+  apiGroup: rbac.authorization.k8s.io

test/e2e/testing-manifests/storage-csi/mock/csi-mock-driver-resizer.yaml

+kind: StatefulSet
+apiVersion: apps/v1
+metadata:
+  name: csi-mockplugin-resizer
+spec:
+  selector:
+    matchLabels:
+      app: csi-mockplugin-resizer
+  replicas: 1
+  template:
+    metadata:
+      labels:
+        app: csi-mockplugin-resizer
+    spec:
+      serviceAccountName: csi-mock
+      containers:
+        - name: csi-resizer
+          image: quay.io/k8scsi/csi-resizer:canary
+          args:
+            - "--v=5"
+            - "--csi-address=$(ADDRESS)"
+          env:
+            - name: ADDRESS
+              value: /csi/csi.sock
+          imagePullPolicy: Always
+          volumeMounts:
+          - mountPath: /csi
+            name: socket-dir
+      volumes:
+        - hostPath:
+            path: /var/lib/kubelet/plugins/csi-mock
+            type: DirectoryOrCreate
+          name: socket-dir

test/e2e/testing-manifests/storage-csi/mock/csi-mock-driver.yaml

@@ -54,7 +54,7 @@ spec:
 
 
         - name: mock
-          image: quay.io/k8scsi/mock-driver:v1.0.0-1
+          image: quay.io/k8scsi/mock-driver:v1.1.1
           env:
             - name: CSI_ENDPOINT
               value: /csi/csi.sock

test/e2e/testing-manifests/storage-csi/mock/csi-mock-rbac.yaml

@@ -59,3 +59,16 @@ roleRef:
   kind: ClusterRole
   name: e2e-test-privileged-psp
   apiGroup: rbac.authorization.k8s.io
+---
+kind: ClusterRoleBinding
+apiVersion: rbac.authorization.k8s.io/v1
+metadata:
+  name: csi-controller-resizer-role
+subjects:
+  - kind: ServiceAccount
+    name: csi-mock
+    namespace: default
+roleRef:
+  kind: ClusterRole
+  name: external-resizer-runner
+  apiGroup: rbac.authorization.k8s.io