Implement CSI migration logic for volume resize

* Using PVC annotation for sychronication between in-tree resizer and extternal resizer
* Add unit tests

pkg/controller/controller.go

@@ -249,7 +249,7 @@ func (ctrl *resizeController) pvcNeedResize(pvc *v1.PersistentVolumeClaim) bool
 
 // pvNeedResize returns true if a pv supports and also requests resize.
 func (ctrl *resizeController) pvNeedResize(pvc *v1.PersistentVolumeClaim, pv *v1.PersistentVolume) bool {
-	if !ctrl.resizer.CanSupport(pv) {
+	if !ctrl.resizer.CanSupport(pv, pvc) {
 		klog.V(4).Infof("Resizer %q doesn't support PV %q", ctrl.name, pv.Name)
 		return false
 	}

pkg/controller/controller_test.go

@@ -65,7 +65,7 @@ func TestController(t *testing.T) {
 			NodeResize:    true,
 		},
 	} {
-		client := csi.NewMockClient(test.NodeResize, true, true)
+		client := csi.NewMockClient("mock", test.NodeResize, true, true)
 		driverName, _ := client.GetDriverName(context.TODO())
 
 		initialObjects := []runtime.Object{}

pkg/csi/mock_client.go

@@ -3,11 +3,12 @@ package csi
 import "context"
 
 func NewMockClient(
+	name string,
 	supportsNodeResize bool,
 	supportsControllerResize bool,
 	supportsPluginControllerService bool) *MockClient {
 	return &MockClient{
-		name:                            "mock",
+		name:                            name,
 		supportsNodeResize:              supportsNodeResize,
 		supportsControllerResize:        supportsControllerResize,
 		supportsPluginControllerService: supportsPluginControllerService,

pkg/resizer/csi_resizer.go

@@ -33,12 +33,19 @@ import (
 	"k8s.io/client-go/informers"
 	"k8s.io/client-go/kubernetes"
 	storagev1listers "k8s.io/client-go/listers/storage/v1"
+
+	csitranslationlib "k8s.io/csi-translation-lib"
 	"k8s.io/klog"
 )
 
 const (
 	resizerSecretNameKey      = "csi.storage.k8s.io/resizer-secret-name"
 	resizerSecretNamespaceKey = "csi.storage.k8s.io/resizer-secret-namespace"
+
+	// In-tree resizer populates the key with the value as the storage plugin name
+	// If CSI migration is enabled, the value will be CSI driver name
+	// Otherwise, it will be in-tree storage plugin name
+	volumeResizerKey = "volume.kubernetes.io/storage-resizer"
 )
 
 var (
@@ -118,7 +125,14 @@ func (r *csiResizer) Name() string {
 	return r.name
 }
 
-func (r *csiResizer) CanSupport(pv *v1.PersistentVolume) bool {
+// CanSupport returns whether the PV is supported by resizer
+// Resizer will resize the volume if it is CSI volume or is migration enabled in-tree volume
+func (r *csiResizer) CanSupport(pv *v1.PersistentVolume, pvc *v1.PersistentVolumeClaim) bool {
+	resizerName := pvc.Annotations[volumeResizerKey]
+	if csitranslationlib.IsMigratedCSIDriverByName(r.name) && resizerName == r.name {
+		return true
+	}
+
 	source := pv.Spec.CSI
 	if source == nil {
 		klog.V(4).Infof("PV %s is not a CSI volume, skip it", pv.Name)
@@ -131,14 +145,27 @@ func (r *csiResizer) CanSupport(pv *v1.PersistentVolume) bool {
 	return true
 }
 
+// Resize resizes the persistence volume given request size
+// It supports both CSI volume and migrated in-tree volume
 func (r *csiResizer) Resize(pv *v1.PersistentVolume, requestSize resource.Quantity) (resource.Quantity, bool, error) {
 	oldSize := pv.Spec.Capacity[v1.ResourceStorage]
+	var err error
+
+	if csitranslationlib.IsMigratedCSIDriverByName(r.name) {
+		// handle migrated in-tree volume
+		pv, err = csitranslationlib.TranslateInTreePVToCSI(pv)
+		if err != nil {
+			return oldSize, false, fmt.Errorf("failed to translate persistent volume: %v", err)
+		}
+	}
 
 	source := pv.Spec.CSI
 	if source == nil {
-		return oldSize, false, errors.New("not a CSI volume")
+		// in-tree volume that is not migrated yet
+		return oldSize, false, fmt.Errorf("volume %v is not migrated to CSI", pv.Name)
 	}
 	volumeID := source.VolumeHandle
+
 	if len(volumeID) == 0 {
 		return oldSize, false, errors.New("empty volume handle")
 	}

pkg/resizer/csi_resizer_test.go

@@ -1,9 +1,14 @@
 package resizer
 
 import (
+	"errors"
 	"testing"
 
 	"github.com/kubernetes-csi/external-resizer/pkg/csi"
+	"k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/resource"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/client-go/informers"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/kubernetes/fake"
@@ -51,7 +56,7 @@ func TestNewResizer(t *testing.T) {
 			Error: resizeNotSupportErr,
 		},
 	} {
-		client := csi.NewMockClient(c.SupportsNodeResize, c.SupportsControllerResize, c.SupportsPluginControllerService)
+		client := csi.NewMockClient("mock", c.SupportsNodeResize, c.SupportsControllerResize, c.SupportsPluginControllerService)
 		k8sClient, informerFactory := fakeK8s()
 		resizer, err := NewResizerFromClient(client, 0, k8sClient, informerFactory)
 		if err != c.Error {
@@ -66,8 +71,218 @@ func TestNewResizer(t *testing.T) {
 	}
 }
 
-func fakeK8s() (kubernetes.Interface, informers.SharedInformerFactory) {
-	client := fake.NewSimpleClientset()
+func TestResizeMigratedPV(t *testing.T) {
+	testCases := []struct {
+		name               string
+		driverName         string
+		pv                 *v1.PersistentVolume
+		nodeResizeRequired bool
+		err                error
+	}{
+		{
+			name:               "Test AWS EBS CSI Driver",
+			driverName:         "ebs.csi.aws.com",
+			pv:                 createEBSPV(1),
+			nodeResizeRequired: true,
+		},
+		{
+			name:               "Test GCE PD Driver",
+			driverName:         "pd.csi.storage.gke.io",
+			pv:                 createGCEPDPV(1),
+			nodeResizeRequired: true,
+		},
+		{
+			name:               "Test unknonwn driver",
+			driverName:         "unknown",
+			pv:                 createEBSPV(1),
+			nodeResizeRequired: true,
+			err:                errors.New("volume testEBSPV is not migrated to CSI"),
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			driverName := tc.driverName
+			client := csi.NewMockClient(driverName, true, true, true)
+			k8sClient, informerFactory := fakeK8s()
+			resizer, err := NewResizerFromClient(client, 0, k8sClient, informerFactory)
+			if err != nil {
+				t.Fatalf("Failed to create resizer: %v", err)
+			}
+
+			pv := tc.pv
+			expectedSize := quantityGB(2)
+			newSize, nodeResizeRequired, err := resizer.Resize(pv, expectedSize)
+
+			if tc.err != nil {
+				if err == nil {
+					t.Fatalf("Got wrong error, wanted: %v, got: %v", tc.err, err)
+				}
+			} else {
+				if err != nil {
+					t.Fatalf("Failed to resize the PV: %v", err)
+				}
+
+				if newSize != expectedSize {
+					t.Fatalf("newSize mismatches, wanted: %v, got: %v", expectedSize, newSize)
+				}
+				if nodeResizeRequired != tc.nodeResizeRequired {
+					t.Fatalf("nodeResizeRequired mismatches, wanted: %v, got: %v", tc.nodeResizeRequired, nodeResizeRequired)
+				}
+			}
+		})
+	}
+}
+
+func TestCanSupport(t *testing.T) {
+	testCases := []struct {
+		name       string
+		driverName string
+		pv         *v1.PersistentVolume
+		pvc        *v1.PersistentVolumeClaim
+		canSupport bool
+	}{
+		{
+			name:       "EBS PV/PVC is supported",
+			driverName: "ebs.csi.aws.com",
+			pv:         createEBSPV(1),
+			pvc:        createPVC("ebs.csi.aws.com"),
+			canSupport: true,
+		},
+		{
+			name:       "EBS PV/PVC is not supported when migartion is disabled",
+			driverName: "ebs.csi.aws.com",
+			pv:         createEBSPV(1),
+			pvc:        createPVC("kubernetes.io/aws-ebs"),
+			canSupport: false,
+		},
+		{
+			name:       "PD PV/PVC is supported",
+			driverName: "pd.csi.storage.gke.io",
+			pv:         createGCEPDPV(1),
+			pvc:        createPVC("pd.csi.storage.gke.io"),
+			canSupport: true,
+		},
+		{
+			name:       "unknown PV/PVC is not supported",
+			driverName: "ebs.csi.aws.com",
+			pv:         createEBSPV(1),
+			pvc:        createPVC("unknown"),
+			canSupport: false,
+		},
+	}
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			driverName := tc.driverName
+			client := csi.NewMockClient(driverName, true, true, true)
+			k8sClient, informerFactory := fakeK8s()
+			resizer, err := NewResizerFromClient(client, 0, k8sClient, informerFactory)
+			if err != nil {
+				t.Fatalf("Failed to create resizer: %v", err)
+			}
+
+			canSupport := resizer.CanSupport(tc.pv, tc.pvc)
+			if canSupport != tc.canSupport {
+				t.Fatalf("Wrong canSupport, wanted: %v got: %v", tc.canSupport, canSupport)
+			}
+		})
+	}
+}
+
+func quantityGB(i int) resource.Quantity {
+	q := resource.NewQuantity(int64(i*1024*1024), resource.BinarySI)
+	return *q
+}
+
+func createPVC(resizerName string) *v1.PersistentVolumeClaim {
+	request := quantityGB(2)
+	capacity := quantityGB(1)
+
+	return &v1.PersistentVolumeClaim{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "testPVC",
+			Namespace: "test",
+			Annotations: map[string]string{
+				"volume.kubernetes.io/storage-resizer": resizerName,
+			},
+		},
+		Spec: v1.PersistentVolumeClaimSpec{
+			Resources: v1.ResourceRequirements{
+				Requests: map[v1.ResourceName]resource.Quantity{
+					v1.ResourceStorage: request,
+				},
+			},
+			VolumeName: "testPV",
+		},
+		Status: v1.PersistentVolumeClaimStatus{
+			Phase: v1.ClaimBound,
+			Capacity: map[v1.ResourceName]resource.Quantity{
+				v1.ResourceStorage: capacity,
+			},
+		},
+	}
+}
+
+func createPV(capacityGB int) *v1.PersistentVolume {
+	capacity := quantityGB(capacityGB)
+
+	return &v1.PersistentVolume{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "testPV",
+		},
+		Spec: v1.PersistentVolumeSpec{
+			Capacity: map[v1.ResourceName]resource.Quantity{
+				v1.ResourceStorage: capacity,
+			},
+			PersistentVolumeSource: v1.PersistentVolumeSource{
+				CSI: &v1.CSIPersistentVolumeSource{
+					Driver:       "foo",
+					VolumeHandle: "foo",
+				},
+			},
+		},
+	}
+}
+func createEBSPV(capacityGB int) *v1.PersistentVolume {
+	capacity := quantityGB(capacityGB)
+
+	return &v1.PersistentVolume{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "testEBSPV",
+		},
+		Spec: v1.PersistentVolumeSpec{
+			Capacity: map[v1.ResourceName]resource.Quantity{
+				v1.ResourceStorage: capacity,
+			},
+			PersistentVolumeSource: v1.PersistentVolumeSource{
+				AWSElasticBlockStore: &v1.AWSElasticBlockStoreVolumeSource{
+					VolumeID: "testVolumeId",
+				},
+			},
+		},
+	}
+}
+
+func createGCEPDPV(capacityGB int) *v1.PersistentVolume {
+	capacity := quantityGB(capacityGB)
+
+	return &v1.PersistentVolume{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "testPDPV",
+		},
+		Spec: v1.PersistentVolumeSpec{
+			Capacity: map[v1.ResourceName]resource.Quantity{
+				v1.ResourceStorage: capacity,
+			},
+			PersistentVolumeSource: v1.PersistentVolumeSource{
+				GCEPersistentDisk: &v1.GCEPersistentDiskVolumeSource{},
+			},
+		},
+	}
+}
+
+func fakeK8s(objects ...runtime.Object) (kubernetes.Interface, informers.SharedInformerFactory) {
+	client := fake.NewSimpleClientset(objects...)
 	informerFactory := informers.NewSharedInformerFactory(client, 0)
 	return client, informerFactory
 }

pkg/resizer/resizer.go

@@ -25,8 +25,9 @@ import (
 type Resizer interface {
 	// Name returns the resizer's name.
 	Name() string
-	// CanSupport returns true if resizer supports resize operation of this PV.
-	CanSupport(pv *v1.PersistentVolume) bool
+	// CanSupport returns true if resizer supports resize operation of this PV
+	// with its corresponding PVC.
+	CanSupport(pv *v1.PersistentVolume, pvc *v1.PersistentVolumeClaim) bool
 	// Resize executes the resize operation of this PV.
 	Resize(pv *v1.PersistentVolume, requestSize resource.Quantity) (newSize resource.Quantity, fsResizeRequired bool, err error)
 }