Introduce feature gate for volume expansion

Update CSI library version

Godeps/Godeps.json

@@ -975,8 +975,8 @@
 		},
 		{
 			"ImportPath": "github.com/container-storage-interface/spec/lib/go/csi",
-			"Comment": "v1.0.0",
-			"Rev": "ed0bb0e1557548aa028307f48728767cfe8f6345"
+			"Comment": "v1.0.0-15-g915ae314723e53",
+			"Rev": "915ae314723e53f501b9ce4d01dc3c023f869ea0"
 		},
 		{
 			"ImportPath": "github.com/containerd/console",

pkg/features/kube_features.go

@@ -109,6 +109,11 @@ const (
 	// Ability to expand persistent volumes' file system without unmounting volumes.
 	ExpandInUsePersistentVolumes utilfeature.Feature = "ExpandInUsePersistentVolumes"
 
+	// owner: @gnufied
+	// alpha: v1.14
+	// Ability to expand CSI volumes
+	ExpandCSIVolumes utilfeature.Feature = "ExpandCSIVolumes"
+
 	// owner: @verb
 	// alpha: v1.10
 	//
@@ -450,6 +455,7 @@ var defaultKubernetesFeatureGates = map[utilfeature.Feature]utilfeature.FeatureS
 	QOSReserved:                                 {Default: false, PreRelease: utilfeature.Alpha},
 	ExpandPersistentVolumes:                     {Default: true, PreRelease: utilfeature.Beta},
 	ExpandInUsePersistentVolumes:                {Default: false, PreRelease: utilfeature.Alpha},
+	ExpandCSIVolumes:                            {Default: false, PreRelease: utilfeature.Alpha},
 	AttachVolumeLimit:                           {Default: true, PreRelease: utilfeature.Beta},
 	CPUManager:                                  {Default: true, PreRelease: utilfeature.Beta},
 	CPUCFSQuotaPeriod:                           {Default: false, PreRelease: utilfeature.Alpha},

pkg/volume/csi/csi_client.go

@@ -28,6 +28,7 @@ import (
 	csipbv1 "github.com/container-storage-interface/spec/lib/go/csi"
 	"google.golang.org/grpc"
 	api "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/resource"
 	utilversion "k8s.io/apimachinery/pkg/util/version"
 	"k8s.io/apimachinery/pkg/util/wait"
 	utilfeature "k8s.io/apiserver/pkg/util/feature"
@@ -55,6 +56,7 @@ type csiClient interface {
 		fsType string,
 		mountOptions []string,
 	) error
+	NodeExpandVolume(ctx context.Context, volumeid, volumePath string, newSize resource.Quantity) (*resource.Quantity, error)
 	NodeUnpublishVolume(
 		ctx context.Context,
 		volID string,
@@ -71,6 +73,7 @@ type csiClient interface {
 	) error
 	NodeUnstageVolume(ctx context.Context, volID, stagingTargetPath string) error
 	NodeSupportsStageUnstage(ctx context.Context) (bool, error)
+	NodeSupportsNodeExpand(ctx context.Context) (bool, error)
 }
 
 // Strongly typed address
@@ -304,6 +307,30 @@ func (c *csiDriverClient) NodePublishVolume(
 
 }
 
+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")
+	}
+
+	nodeClient, closer, err := c.nodeV1ClientCreator(c.addr)
+	if err != nil {
+		return nil, err
+	}
+	defer closer.Close()
+
+	req := &csipbv1.NodeExpandVolumeRequest{
+		VolumeId:      volumeID,
+		VolumePath:    volumePath,
+		CapacityRange: &csipbv1.CapacityRange{RequiredBytes: newSize.Value()},
+	}
+	resp, err := nodeClient.NodeExpandVolume(ctx, req)
+	if err != nil {
+		return nil, err
+	}
+	updatedQuantity := resource.NewQuantity(resp.CapacityBytes, resource.BinarySI)
+	return updatedQuantity, nil
+}
+
 func (c *csiDriverClient) nodePublishVolumeV1(
 	ctx context.Context,
 	volID string,
@@ -624,6 +651,41 @@ func (c *csiDriverClient) nodeUnstageVolumeV0(ctx context.Context, volID, stagin
 	return err
 }
 
+func (c *csiDriverClient) NodeSupportsNodeExpand(ctx context.Context) (bool, error) {
+	klog.V(4).Info(log("calling NodeGetCapabilities rpc to determine if Node has EXPAND_VOLUME capability"))
+
+	if c.nodeV1ClientCreator != nil {
+		nodeClient, closer, err := c.nodeV1ClientCreator(c.addr)
+		if err != nil {
+			return false, err
+		}
+		defer closer.Close()
+
+		req := &csipbv1.NodeGetCapabilitiesRequest{}
+		resp, err := nodeClient.NodeGetCapabilities(ctx, req)
+		if err != nil {
+			return false, err
+		}
+
+		capabilities := resp.GetCapabilities()
+
+		nodeExpandSet := false
+		if capabilities == nil {
+			return false, nil
+		}
+		for _, capability := range capabilities {
+			if capability.GetRpc().GetType() == csipbv1.NodeServiceCapability_RPC_EXPAND_VOLUME {
+				nodeExpandSet = true
+			}
+		}
+		return nodeExpandSet, nil
+	} else if c.nodeV0ClientCreator != nil {
+		return false, nil
+	}
+	return false, fmt.Errorf("failed to call NodeSupportsNodeExpand. Both nodeV1ClientCreator and nodeV0ClientCreator are nil")
+
+}
+
 func (c *csiDriverClient) NodeSupportsStageUnstage(ctx context.Context) (bool, error) {
 	klog.V(4).Info(log("calling NodeGetCapabilities rpc to determine if NodeSupportsStageUnstage"))
 

pkg/volume/csi/expander.go

@@ -0,0 +1,66 @@
+/*
+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 (
+	"context"
+	"errors"
+	"fmt"
+
+	"k8s.io/apimachinery/pkg/api/resource"
+	"k8s.io/klog"
+	"k8s.io/kubernetes/pkg/volume"
+)
+
+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.
+	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)
+	if err != nil {
+		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))
+	if err != nil {
+		return false, err
+	}
+
+	csiSource, err := getCSISourceFromSpec(spec)
+	if err != nil {
+		klog.Error(log("Expander.NodeExpand failed to get CSI persistent source: %v", err))
+		return false, err
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), csiTimeout)
+	defer cancel()
+
+	nodeExpandSet, err := csiClient.NodeSupportsNodeExpand(ctx)
+	if err != nil {
+		return false, fmt.Errorf("Expander.NodeExpand failed to check if node supports expansion : %v", err)
+	}
+	return false, nil
+}

pkg/volume/plugins.go

@@ -219,17 +219,18 @@ type DeviceMountableVolumePlugin interface {
 }
 
 // ExpandableVolumePlugin is an extended interface of VolumePlugin and is used for volumes that can be
-// expanded
+// expanded via control-plane ExpandVolumeDevice call.
 type ExpandableVolumePlugin interface {
 	VolumePlugin
 	ExpandVolumeDevice(spec *Spec, newSize resource.Quantity, oldSize resource.Quantity) (resource.Quantity, error)
 	RequiresFSResize() bool
 }
 
-// NodeExpandableVolumePlugin is an extension of ExpandableVolumePlugin and is used for volumes (flex)
-// that require extra steps on nodes for expansion to complete
+// NodeExpandableVolumePlugin is an expanded interface of VolumePlugin and is used for volumes that
+// require expansion on the node via NodeExpand call.
 type NodeExpandableVolumePlugin interface {
-	ExpandableVolumePlugin
+	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)
@@ -965,8 +966,8 @@ func (pm *VolumePluginMgr) FindNodeExpandablePluginBySpec(spec *Spec) (NodeExpan
 	return nil, nil
 }
 
-// FindFSResizablePluginByName fetches a persistent volume plugin by name
-func (pm *VolumePluginMgr) FindFSResizablePluginByName(name string) (NodeExpandableVolumePlugin, error) {
+// FindNodeExpandablePluginByName fetches a persistent volume plugin by name
+func (pm *VolumePluginMgr) FindNodeExpandablePluginByName(name string) (NodeExpandableVolumePlugin, error) {
 	volumePlugin, err := pm.FindPluginByName(name)
 	if err != nil {
 		return nil, err