Changed admission controller to allow volume expansion for all volume plugins

There are two motivations for this change:
(1) CSI plugins are soon going to support volume expansion. For such
plugins, admission controller doesn't know whether the plugins are
capabale of supporting volume expansion or not.
(2) Currently, admission controller rejects PVC updates for in-tree plugins
that don't support volume expansion (e.g., NFS, iSCSI). This change allows
external controllers to expand volumes similar to how external provisioners
operate.

pkg/controller/volume/events/event.go

@@ -30,4 +30,5 @@ const (
 	ProvisioningCleanupFailed = "ProvisioningCleanupFailed"
 	ProvisioningSucceeded     = "ProvisioningSucceeded"
 	WaitForFirstConsumer      = "WaitForFirstConsumer"
+	ExternalExpanding         = "ExternalExpanding"
 )

pkg/controller/volume/expand/BUILD

@@ -16,6 +16,7 @@ go_library(
     deps = [
         "//pkg/cloudprovider:go_default_library",
         "//pkg/controller:go_default_library",
+        "//pkg/controller/volume/events:go_default_library",
         "//pkg/controller/volume/expand/cache:go_default_library",
         "//pkg/util/goroutinemap/exponentialbackoff:go_default_library",
         "//pkg/util/mount:go_default_library",

pkg/controller/volume/expand/cache/BUILD

@@ -11,7 +11,6 @@ go_library(
     srcs = ["volume_resize_map.go"],
     importpath = "k8s.io/kubernetes/pkg/controller/volume/expand/cache",
     deps = [
-        "//pkg/util/strings:go_default_library",
         "//pkg/volume/util:go_default_library",
         "//pkg/volume/util/types:go_default_library",
         "//staging/src/k8s.io/api/core/v1:go_default_library",

pkg/controller/volume/expand/cache/volume_resize_map.go

@@ -28,7 +28,6 @@ import (
 	commontypes "k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/strategicpatch"
 	clientset "k8s.io/client-go/kubernetes"
-	"k8s.io/kubernetes/pkg/util/strings"
 	"k8s.io/kubernetes/pkg/volume/util"
 	"k8s.io/kubernetes/pkg/volume/util/types"
 )
@@ -78,7 +77,7 @@ func (pvcr *PVCWithResizeRequest) UniquePVCKey() types.UniquePVCName {
 
 // QualifiedName returns namespace and name combination of the PVC
 func (pvcr *PVCWithResizeRequest) QualifiedName() string {
-	return strings.JoinQualifiedName(pvcr.PVC.Namespace, pvcr.PVC.Name)
+	return util.GetPersistentVolumeClaimQualifiedName(pvcr.PVC)
 }
 
 // NewVolumeResizeMap returns new VolumeResizeMap which acts as a cache

pkg/controller/volume/expand/expand_controller.go

@@ -39,9 +39,11 @@ import (
 	"k8s.io/client-go/tools/record"
 	"k8s.io/kubernetes/pkg/cloudprovider"
 	"k8s.io/kubernetes/pkg/controller"
+	"k8s.io/kubernetes/pkg/controller/volume/events"
 	"k8s.io/kubernetes/pkg/controller/volume/expand/cache"
 	"k8s.io/kubernetes/pkg/util/mount"
 	"k8s.io/kubernetes/pkg/volume"
+	"k8s.io/kubernetes/pkg/volume/util"
 	"k8s.io/kubernetes/pkg/volume/util/operationexecutor"
 	"k8s.io/kubernetes/pkg/volume/util/volumepathhandler"
 )
@@ -117,13 +119,13 @@ func NewExpandController(
 	eventBroadcaster := record.NewBroadcaster()
 	eventBroadcaster.StartLogging(glog.Infof)
 	eventBroadcaster.StartRecordingToSink(&v1core.EventSinkImpl{Interface: kubeClient.CoreV1().Events("")})
-	recorder := eventBroadcaster.NewRecorder(scheme.Scheme, v1.EventSource{Component: "volume_expand"})
+	expc.recorder = eventBroadcaster.NewRecorder(scheme.Scheme, v1.EventSource{Component: "volume_expand"})
 	blkutil := volumepathhandler.NewBlockVolumePathHandler()
 
 	expc.opExecutor = operationexecutor.NewOperationExecutor(operationexecutor.NewOperationGenerator(
 		kubeClient,
 		&expc.volumePluginMgr,
-		recorder,
+		expc.recorder,
 		false,
 		blkutil))
 
@@ -140,6 +142,7 @@ func NewExpandController(
 		expc.resizeMap,
 		expc.pvcLister,
 		expc.pvLister,
+		&expc.volumePluginMgr,
 		kubeClient)
 	return expc, nil
 }
@@ -179,9 +182,9 @@ func (expc *expandController) deletePVC(obj interface{}) {
 }
 
 func (expc *expandController) pvcUpdate(oldObj, newObj interface{}) {
-	oldPvc, ok := oldObj.(*v1.PersistentVolumeClaim)
+	oldPVC, ok := oldObj.(*v1.PersistentVolumeClaim)
 
-	if oldPvc == nil || !ok {
+	if oldPVC == nil || !ok {
 		return
 	}
 
@@ -192,7 +195,7 @@ func (expc *expandController) pvcUpdate(oldObj, newObj interface{}) {
 	}
 
 	newSize := newPVC.Spec.Resources.Requests[v1.ResourceStorage]
-	oldSize := oldPvc.Spec.Resources.Requests[v1.ResourceStorage]
+	oldSize := oldPVC.Spec.Resources.Requests[v1.ResourceStorage]
 
 	// We perform additional checks inside resizeMap.AddPVCUpdate function
 	// this check here exists to ensure - we do not consider every
@@ -200,7 +203,20 @@ func (expc *expandController) pvcUpdate(oldObj, newObj interface{}) {
 	if newSize.Cmp(oldSize) > 0 {
 		pv, err := getPersistentVolume(newPVC, expc.pvLister)
 		if err != nil {
-			glog.V(5).Infof("Error getting Persistent Volume for pvc %q : %v", newPVC.UID, err)
+			glog.V(5).Infof("Error getting Persistent Volume for PVC %q : %v", newPVC.UID, err)
+			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,
+				fmt.Sprintf("Ignoring the PVC: %v.", err))
+			glog.V(3).Infof("Ignoring the PVC %q (uid: %q) : %v.",
+				util.GetPersistentVolumeClaimQualifiedName(newPVC), newPVC.UID, err)
 			return
 		}
 		expc.resizeMap.AddPVCUpdate(newPVC, pv)

pkg/controller/volume/expand/pvc_populator.go

@@ -21,15 +21,23 @@ limitations under the License.
 package expand
 
 import (
+	"fmt"
 	"time"
 
 	"github.com/golang/glog"
 
+	"k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/util/wait"
 	clientset "k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/kubernetes/scheme"
+	v1core "k8s.io/client-go/kubernetes/typed/core/v1"
 	corelisters "k8s.io/client-go/listers/core/v1"
+	"k8s.io/client-go/tools/record"
+	"k8s.io/kubernetes/pkg/controller/volume/events"
 	"k8s.io/kubernetes/pkg/controller/volume/expand/cache"
+	"k8s.io/kubernetes/pkg/volume"
+	"k8s.io/kubernetes/pkg/volume/util"
 )
 
 // PVCPopulator iterates through PVCs and checks if for bound PVCs
@@ -39,11 +47,13 @@ type PVCPopulator interface {
 }
 
 type pvcPopulator struct {
-	loopPeriod time.Duration
-	resizeMap  cache.VolumeResizeMap
-	pvcLister  corelisters.PersistentVolumeClaimLister
-	pvLister   corelisters.PersistentVolumeLister
-	kubeClient clientset.Interface
+	loopPeriod      time.Duration
+	resizeMap       cache.VolumeResizeMap
+	pvcLister       corelisters.PersistentVolumeClaimLister
+	pvLister        corelisters.PersistentVolumeLister
+	kubeClient      clientset.Interface
+	volumePluginMgr *volume.VolumePluginMgr
+	recorder        record.EventRecorder
 }
 
 func NewPVCPopulator(
@@ -51,14 +61,19 @@ func NewPVCPopulator(
 	resizeMap cache.VolumeResizeMap,
 	pvcLister corelisters.PersistentVolumeClaimLister,
 	pvLister corelisters.PersistentVolumeLister,
+	volumePluginMgr *volume.VolumePluginMgr,
 	kubeClient clientset.Interface) PVCPopulator {
 	populator := &pvcPopulator{
-		loopPeriod: loopPeriod,
-		pvcLister:  pvcLister,
-		pvLister:   pvLister,
-		resizeMap:  resizeMap,
-		kubeClient: kubeClient,
+		loopPeriod:      loopPeriod,
+		pvcLister:       pvcLister,
+		pvLister:        pvLister,
+		resizeMap:       resizeMap,
+		volumePluginMgr: volumePluginMgr,
+		kubeClient:      kubeClient,
 	}
+	eventBroadcaster := record.NewBroadcaster()
+	eventBroadcaster.StartRecordingToSink(&v1core.EventSinkImpl{Interface: kubeClient.CoreV1().Events("")})
+	populator.recorder = eventBroadcaster.NewRecorder(scheme.Scheme, v1.EventSource{Component: "volume_expand"})
 	return populator
 }
 
@@ -77,9 +92,25 @@ func (populator *pvcPopulator) Sync() {
 		pv, err := getPersistentVolume(pvc, populator.pvLister)
 
 		if err != nil {
-			glog.V(5).Infof("Error getting persistent volume for pvc %q : %v", pvc.UID, err)
+			glog.V(5).Infof("Error getting persistent volume for PVC %q : %v", pvc.UID, err)
 			continue
 		}
+
+		// Filter PVCs for which the corresponding volume plugins don't allow expansion.
+		pvcSize := pvc.Spec.Resources.Requests[v1.ResourceStorage]
+		pvcStatusSize := pvc.Status.Capacity[v1.ResourceStorage]
+		volumeSpec := volume.NewSpecFromPersistentVolume(pv, false)
+		volumePlugin, err := populator.volumePluginMgr.FindExpandablePluginBySpec(volumeSpec)
+		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,
+				fmt.Sprintf("Ignoring the PVC: %v.", err))
+			glog.V(3).Infof("Ignoring the PVC %q (uid: %q) : %v.",
+				util.GetPersistentVolumeClaimQualifiedName(pvc), pvc.UID, err)
+			continue
+		}
+
 		// We are only going to add PVCs which are:
 		//  - bound
 		//  - pvc.Spec.Size > pvc.Status.Size

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/strings:go_default_library",
         "//pkg/volume:go_default_library",
         "//pkg/volume/util/types:go_default_library",
         "//pkg/volume/util/volumepathhandler:go_default_library",

pkg/volume/util/util.go

@@ -39,6 +39,7 @@ import (
 	"k8s.io/kubernetes/pkg/features"
 	kubeletapis "k8s.io/kubernetes/pkg/kubelet/apis"
 	"k8s.io/kubernetes/pkg/util/mount"
+	utilstrings "k8s.io/kubernetes/pkg/util/strings"
 	"k8s.io/kubernetes/pkg/volume"
 
 	"reflect"
@@ -815,6 +816,11 @@ func GetPersistentVolumeClaimVolumeMode(claim *v1.PersistentVolumeClaim) (v1.Per
 	return "", fmt.Errorf("cannot get volumeMode from pvc: %v", claim.Name)
 }
 
+// GetPersistentVolumeClaimQualifiedName returns a qualified name for pvc.
+func GetPersistentVolumeClaimQualifiedName(claim *v1.PersistentVolumeClaim) string {
+	return utilstrings.JoinQualifiedName(claim.GetNamespace(), claim.GetName())
+}
+
 // CheckVolumeModeFilesystem checks VolumeMode.
 // If the mode is Filesystem, return true otherwise return false.
 func CheckVolumeModeFilesystem(volumeSpec *volume.Spec) (bool, error) {

plugin/pkg/admission/storage/persistentvolume/resize/admission.go

@@ -117,15 +117,6 @@ func (pvcr *persistentVolumeClaimResize) Validate(a admission.Attributes) error
 			"the storageclass that provisions the pvc must support resize"))
 	}
 
-	// volume plugin must support resize
-	pv, err := pvcr.pvLister.Get(pvc.Spec.VolumeName)
-	if err != nil {
-		return admission.NewForbidden(a, fmt.Errorf("Error updating persistent volume claim because fetching associated persistent volume failed"))
-	}
-
-	if !pvcr.checkVolumePlugin(pv) {
-		return admission.NewForbidden(a, fmt.Errorf("volume plugin does not support resize"))
-	}
 	return nil
 }
 
@@ -146,27 +137,3 @@ func (pvcr *persistentVolumeClaimResize) allowResize(pvc, oldPvc *api.Persistent
 	}
 	return false
 }
-
-// checkVolumePlugin checks whether the volume plugin supports resize
-func (pvcr *persistentVolumeClaimResize) checkVolumePlugin(pv *api.PersistentVolume) bool {
-	if pv.Spec.Glusterfs != nil || pv.Spec.Cinder != nil || pv.Spec.RBD != nil || pv.Spec.PortworxVolume != nil {
-		return true
-	}
-
-	if pv.Spec.GCEPersistentDisk != nil {
-		return true
-	}
-
-	if pv.Spec.AWSElasticBlockStore != nil {
-		return true
-	}
-
-	if pv.Spec.AzureFile != nil {
-		return true
-	}
-
-	if pv.Spec.AzureDisk != nil {
-		return true
-	}
-	return false
-}

plugin/pkg/admission/storage/persistentvolume/resize/admission_test.go

@@ -72,9 +72,6 @@ func TestPVCResizeAdmission(t *testing.T) {
 		return strings.Contains(err.Error(), "only dynamically provisioned pvc can be resized and "+
 			"the storageclass that provisions the pvc must support resize")
 	}
-	expectVolumePluginError := func(err error) bool {
-		return strings.Contains(err.Error(), "volume plugin does not support resize")
-	}
 	tests := []struct {
 		name        string
 		resource    schema.GroupVersionResource
@@ -115,37 +112,6 @@ func TestPVCResizeAdmission(t *testing.T) {
 			},
 			checkError: expectNoError,
 		},
-		{
-			name:     "pvc-resize, update, volume plugin error",
-			resource: api.SchemeGroupVersion.WithResource("persistentvolumeclaims"),
-			oldObj: &api.PersistentVolumeClaim{
-				Spec: api.PersistentVolumeClaimSpec{
-					VolumeName: "volume2",
-					Resources: api.ResourceRequirements{
-						Requests: getResourceList("1Gi"),
-					},
-					StorageClassName: &goldClassName,
-				},
-				Status: api.PersistentVolumeClaimStatus{
-					Capacity: getResourceList("1Gi"),
-					Phase:    api.ClaimBound,
-				},
-			},
-			newObj: &api.PersistentVolumeClaim{
-				Spec: api.PersistentVolumeClaimSpec{
-					VolumeName: "volume2",
-					Resources: api.ResourceRequirements{
-						Requests: getResourceList("2Gi"),
-					},
-					StorageClassName: &goldClassName,
-				},
-				Status: api.PersistentVolumeClaimStatus{
-					Capacity: getResourceList("2Gi"),
-					Phase:    api.ClaimBound,
-				},
-			},
-			checkError: expectVolumePluginError,
-		},
 		{
 			name:     "pvc-resize, update, dynamically provisioned error",
 			resource: api.SchemeGroupVersion.WithResource("persistentvolumeclaims"),
@@ -290,18 +256,9 @@ func TestPVCResizeAdmission(t *testing.T) {
 			StorageClassName: goldClassName,
 		},
 	}
-	pv2 := &api.PersistentVolume{
-		ObjectMeta: metav1.ObjectMeta{Name: "volume2"},
-		Spec: api.PersistentVolumeSpec{
-			PersistentVolumeSource: api.PersistentVolumeSource{
-				HostPath: &api.HostPathVolumeSource{},
-			},
-			StorageClassName: goldClassName,
-		},
-	}
 
 	pvs := []*api.PersistentVolume{}
-	pvs = append(pvs, pv1, pv2)
+	pvs = append(pvs, pv1)
 
 	for _, pv := range pvs {
 		err := informerFactory.Core().InternalVersion().PersistentVolumes().Informer().GetStore().Add(pv)