Add annotation updating for migration for PVs and PVCs

pkg/controller/volume/persistentvolume/BUILD

@@ -80,6 +80,7 @@ go_test(
         "//pkg/controller/volume/persistentvolume/util:go_default_library",
         "//pkg/features:go_default_library",
         "//pkg/volume:go_default_library",
+        "//pkg/volume/csimigration:go_default_library",
         "//pkg/volume/util:go_default_library",
         "//pkg/volume/util/recyclerclient:go_default_library",
         "//staging/src/k8s.io/api/core/v1:go_default_library",
@@ -101,6 +102,7 @@ go_test(
         "//staging/src/k8s.io/client-go/tools/cache:go_default_library",
         "//staging/src/k8s.io/client-go/tools/record:go_default_library",
         "//staging/src/k8s.io/client-go/tools/reference:go_default_library",
+        "//staging/src/k8s.io/component-base/featuregate:go_default_library",
         "//staging/src/k8s.io/component-base/featuregate/testing:go_default_library",
         "//staging/src/k8s.io/csi-translation-lib:go_default_library",
         "//vendor/k8s.io/klog:go_default_library",

pkg/controller/volume/persistentvolume/framework_test.go

@@ -407,8 +407,12 @@ func newClaimArray(name, claimUID, capacity, boundToVolume string, phase v1.Pers
 	}
 }
 
-// claimWithAnnotation saves given annotation into given claims.
-// Meant to be used to compose claims specified inline in a test.
+// claimWithAnnotation saves given annotation into given claims. Meant to be
+// used to compose claims specified inline in a test.
+// TODO(refactor): This helper function (and other helpers related to claim
+// arrays) could use some cleaning up (most assume an array size of one)-
+// replace with annotateClaim at all callsites. The tests require claimArrays
+// but mostly operate on single claims
 func claimWithAnnotation(name, value string, claims []*v1.PersistentVolumeClaim) []*v1.PersistentVolumeClaim {
 	if claims[0].Annotations == nil {
 		claims[0].Annotations = map[string]string{name: value}
@@ -418,6 +422,16 @@ func claimWithAnnotation(name, value string, claims []*v1.PersistentVolumeClaim)
 	return claims
 }
 
+func annotateClaim(claim *v1.PersistentVolumeClaim, ann map[string]string) *v1.PersistentVolumeClaim {
+	if claim.Annotations == nil {
+		claim.Annotations = map[string]string{}
+	}
+	for key, val := range ann {
+		claim.Annotations[key] = val
+	}
+	return claim
+}
+
 // volumeWithAnnotation saves given annotation into given volume.
 // Meant to be used to compose volume specified inline in a test.
 func volumeWithAnnotation(name, value string, volume *v1.PersistentVolume) *v1.PersistentVolume {
@@ -523,7 +537,7 @@ func wrapTestWithProvisionCalls(expectedProvisionCalls []provisionCall, toWrap t
 type fakeCSINameTranslator struct{}
 
 func (t fakeCSINameTranslator) GetCSINameFromInTreeName(pluginName string) (string, error) {
-	return "vendor.com/MockCSIPlugin", nil
+	return "vendor.com/MockCSIDriver", nil
 }
 
 type fakeCSIMigratedPluginManager struct{}

pkg/controller/volume/persistentvolume/provision_test.go

@@ -449,8 +449,14 @@ func TestProvisionSync(t *testing.T) {
 			novolumes,
 			novolumes,
 			newClaimArray("claim11-21", "uid11-21", "1Gi", "", v1.ClaimPending, &classGold),
-			claimWithAnnotation(pvutil.AnnStorageProvisioner, "vendor.com/MockCSIPlugin",
-				newClaimArray("claim11-21", "uid11-21", "1Gi", "", v1.ClaimPending, &classGold)),
+			[]*v1.PersistentVolumeClaim{
+				annotateClaim(
+					newClaim("claim11-21", "uid11-21", "1Gi", "", v1.ClaimPending, &classGold),
+					map[string]string{
+						pvutil.AnnStorageProvisioner: "vendor.com/MockCSIDriver",
+						pvutil.AnnMigratedTo:         "vendor.com/MockCSIDriver",
+					}),
+			},
 			[]string{"Normal ExternalProvisioning"},
 			noerrors, wrapTestWithCSIMigrationProvisionCalls(testSyncClaim),
 		},

pkg/controller/volume/persistentvolume/pv_controller.go

@@ -244,6 +244,16 @@ type PersistentVolumeController struct {
 func (ctrl *PersistentVolumeController) syncClaim(claim *v1.PersistentVolumeClaim) error {
 	klog.V(4).Infof("synchronizing PersistentVolumeClaim[%s]: %s", claimToClaimKey(claim), getClaimStatusForLogging(claim))
 
+	// Set correct "migrated-to" annotations on PVC and update in API server if
+	// necessary
+	newClaim, err := ctrl.updateClaimMigrationAnnotations(claim)
+	if err != nil {
+		// Nothing was saved; we will fall back into the same
+		// condition in the next call to this method
+		return err
+	}
+	claim = newClaim
+
 	if !metav1.HasAnnotation(claim.ObjectMeta, pvutil.AnnBindCompleted) {
 		return ctrl.syncUnboundClaim(claim)
 	} else {
@@ -492,6 +502,16 @@ func (ctrl *PersistentVolumeController) syncBoundClaim(claim *v1.PersistentVolum
 func (ctrl *PersistentVolumeController) syncVolume(volume *v1.PersistentVolume) error {
 	klog.V(4).Infof("synchronizing PersistentVolume[%s]: %s", volume.Name, getVolumeStatusForLogging(volume))
 
+	// Set correct "migrated-to" annotations on PV and update in API server if
+	// necessary
+	newVolume, err := ctrl.updateVolumeMigrationAnnotations(volume)
+	if err != nil {
+		// Nothing was saved; we will fall back into the same
+		// condition in the next call to this method
+		return err
+	}
+	volume = newVolume
+
 	// [Unit test set 4]
 	if volume.Spec.ClaimRef == nil {
 		// Volume is unused
@@ -1010,6 +1030,11 @@ func (ctrl *PersistentVolumeController) unbindVolume(volume *v1.PersistentVolume
 // reclaimVolume implements volume.Spec.PersistentVolumeReclaimPolicy and
 // starts appropriate reclaim action.
 func (ctrl *PersistentVolumeController) reclaimVolume(volume *v1.PersistentVolume) error {
+	if migrated := volume.Annotations[pvutil.AnnMigratedTo]; len(migrated) > 0 {
+		// PV is Migrated. The PV controller should stand down and the external
+		// provisioner will handle this PV
+		return nil
+	}
 	switch volume.Spec.PersistentVolumeReclaimPolicy {
 	case v1.PersistentVolumeReclaimRetain:
 		klog.V(4).Infof("reclaimVolume[%s]: policy is Retain, nothing to do", volume.Name)

pkg/controller/volume/persistentvolume/pv_controller_base.go

@@ -309,6 +309,89 @@ func (ctrl *PersistentVolumeController) Run(stopCh <-chan struct{}) {
 	<-stopCh
 }
 
+func (ctrl *PersistentVolumeController) updateClaimMigrationAnnotations(claim *v1.PersistentVolumeClaim) (*v1.PersistentVolumeClaim, error) {
+	// TODO: update[Claim|Volume]MigrationAnnotations can be optimized to not
+	// copy the claim/volume if no modifications are required. Though this
+	// requires some refactoring as well as an interesting change in the
+	// semantics of the function which may be undesirable. If no copy is made
+	// when no modifications are required this function could sometimes return a
+	// copy of the volume and sometimes return a ref to the original
+	claimClone := claim.DeepCopy()
+	modified := updateMigrationAnnotations(ctrl.csiMigratedPluginManager, ctrl.translator, claimClone.Annotations, pvutil.AnnStorageProvisioner)
+	if !modified {
+		return claimClone, nil
+	}
+	newClaim, err := ctrl.kubeClient.CoreV1().PersistentVolumeClaims(claimClone.Namespace).Update(claimClone)
+	if err != nil {
+		return nil, fmt.Errorf("persistent Volume Controller can't anneal migration annotations: %v", err)
+	}
+	_, err = ctrl.storeClaimUpdate(newClaim)
+	if err != nil {
+		return nil, fmt.Errorf("persistent Volume Controller can't anneal migration annotations: %v", err)
+	}
+	return newClaim, nil
+}
+
+func (ctrl *PersistentVolumeController) updateVolumeMigrationAnnotations(volume *v1.PersistentVolume) (*v1.PersistentVolume, error) {
+	volumeClone := volume.DeepCopy()
+	modified := updateMigrationAnnotations(ctrl.csiMigratedPluginManager, ctrl.translator, volumeClone.Annotations, pvutil.AnnDynamicallyProvisioned)
+	if !modified {
+		return volumeClone, nil
+	}
+	newVol, err := ctrl.kubeClient.CoreV1().PersistentVolumes().Update(volumeClone)
+	if err != nil {
+		return nil, fmt.Errorf("persistent Volume Controller can't anneal migration annotations: %v", err)
+	}
+	_, err = ctrl.storeVolumeUpdate(newVol)
+	if err != nil {
+		return nil, fmt.Errorf("persistent Volume Controller can't anneal migration annotations: %v", err)
+	}
+	return newVol, nil
+
+}
+
+// updateMigrationAnnotations takes an Annotations map and checks for a
+// provisioner name using the provisionerKey. It will then add a
+// "volume.beta.kubernetes.io/migrated-to" annotation if migration with the CSI
+// driver name for that provisioner is "on" based on feature flags, it will also
+// remove the annotation is migration is "off" for that provisioner in rollback
+// scenarios. Returns true if the annotations map was modified and false otherwise.
+func updateMigrationAnnotations(cmpm CSIMigratedPluginManager, translator CSINameTranslator, ann map[string]string, provisionerKey string) bool {
+	var csiDriverName string
+	var err error
+
+	if ann == nil {
+		// No annotations so we can't get the provisioner and don't know whether
+		// this is migrated - no change
+		return false
+	}
+	provisioner, ok := ann[provisionerKey]
+	if !ok {
+		// Volume not dynamically provisioned. Ignore
+		return false
+	}
+
+	migratedToDriver := ann[pvutil.AnnMigratedTo]
+	if cmpm.IsMigrationEnabledForPlugin(provisioner) {
+		csiDriverName, err = translator.GetCSINameFromInTreeName(provisioner)
+		if err != nil {
+			klog.Errorf("Could not update volume migration annotations. Migration enabled for plugin %s but could not find corresponding driver name: %v", provisioner, err)
+			return false
+		}
+		if migratedToDriver != csiDriverName {
+			ann[pvutil.AnnMigratedTo] = csiDriverName
+			return true
+		}
+	} else {
+		if migratedToDriver != "" {
+			// Migration annotation exists but the driver isn't migrated currently
+			delete(ann, pvutil.AnnMigratedTo)
+			return true
+		}
+	}
+	return false
+}
+
 // volumeWorker processes items from volumeQueue. It must run only once,
 // syncVolume is not assured to be reentrant.
 func (ctrl *PersistentVolumeController) volumeWorker() {
@@ -461,6 +544,7 @@ func (ctrl *PersistentVolumeController) setClaimProvisioner(claim *v1.Persistent
 	// modify these, therefore create a copy.
 	claimClone := claim.DeepCopy()
 	metav1.SetMetaDataAnnotation(&claimClone.ObjectMeta, pvutil.AnnStorageProvisioner, provisionerName)
+	updateMigrationAnnotations(ctrl.csiMigratedPluginManager, ctrl.translator, claimClone.Annotations, pvutil.AnnStorageProvisioner)
 	newClaim, err := ctrl.kubeClient.CoreV1().PersistentVolumeClaims(claim.Namespace).Update(claimClone)
 	if err != nil {
 		return newClaim, err

pkg/controller/volume/persistentvolume/pv_controller_test.go

@@ -18,23 +18,29 @@ package persistentvolume
 
 import (
 	"errors"
+	"reflect"
 	"testing"
 	"time"
 
 	v1 "k8s.io/api/core/v1"
 	storagev1 "k8s.io/api/storage/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/watch"
+	utilfeature "k8s.io/apiserver/pkg/util/feature"
 	"k8s.io/client-go/informers"
 	"k8s.io/client-go/kubernetes/fake"
 	storagelisters "k8s.io/client-go/listers/storage/v1"
 	core "k8s.io/client-go/testing"
 	"k8s.io/client-go/tools/cache"
+	"k8s.io/component-base/featuregate"
+	featuregatetesting "k8s.io/component-base/featuregate/testing"
 	csitrans "k8s.io/csi-translation-lib"
 	"k8s.io/klog"
 	"k8s.io/kubernetes/pkg/controller"
 	pvtesting "k8s.io/kubernetes/pkg/controller/volume/persistentvolume/testing"
 	pvutil "k8s.io/kubernetes/pkg/controller/volume/persistentvolume/util"
+	"k8s.io/kubernetes/pkg/features"
+	"k8s.io/kubernetes/pkg/volume/csimigration"
 )
 
 var (
@@ -461,3 +467,110 @@ func TestDelayBindingMode(t *testing.T) {
 		}
 	}
 }
+
+func TestAnnealMigrationAnnotations(t *testing.T) {
+	defer featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.CSIMigration, true)()
+
+	const testPlugin = "non-migrated-plugin"
+	const gcePlugin = "kubernetes.io/gce-pd"
+	const gceDriver = "pd.csi.storage.gke.io"
+	tests := []struct {
+		name                 string
+		volumeAnnotations    map[string]string
+		expVolumeAnnotations map[string]string
+		claimAnnotations     map[string]string
+		expClaimAnnotations  map[string]string
+		migratedDriverGates  []featuregate.Feature
+	}{
+		{
+			name:                 "migration on for GCE",
+			volumeAnnotations:    map[string]string{pvutil.AnnDynamicallyProvisioned: gcePlugin},
+			expVolumeAnnotations: map[string]string{pvutil.AnnDynamicallyProvisioned: gcePlugin, pvutil.AnnMigratedTo: gceDriver},
+			claimAnnotations:     map[string]string{pvutil.AnnStorageProvisioner: gcePlugin},
+			expClaimAnnotations:  map[string]string{pvutil.AnnStorageProvisioner: gcePlugin, pvutil.AnnMigratedTo: gceDriver},
+			migratedDriverGates:  []featuregate.Feature{features.CSIMigrationGCE},
+		},
+		{
+			name:                 "migration off for GCE",
+			volumeAnnotations:    map[string]string{pvutil.AnnDynamicallyProvisioned: gcePlugin},
+			expVolumeAnnotations: map[string]string{pvutil.AnnDynamicallyProvisioned: gcePlugin},
+			claimAnnotations:     map[string]string{pvutil.AnnStorageProvisioner: gcePlugin},
+			expClaimAnnotations:  map[string]string{pvutil.AnnStorageProvisioner: gcePlugin},
+			migratedDriverGates:  []featuregate.Feature{},
+		},
+		{
+			name:                 "migration off for GCE removes migrated to (rollback)",
+			volumeAnnotations:    map[string]string{pvutil.AnnDynamicallyProvisioned: gcePlugin, pvutil.AnnMigratedTo: gceDriver},
+			expVolumeAnnotations: map[string]string{pvutil.AnnDynamicallyProvisioned: gcePlugin},
+			claimAnnotations:     map[string]string{pvutil.AnnStorageProvisioner: gcePlugin, pvutil.AnnMigratedTo: gceDriver},
+			expClaimAnnotations:  map[string]string{pvutil.AnnStorageProvisioner: gcePlugin},
+			migratedDriverGates:  []featuregate.Feature{},
+		},
+		{
+			name:                 "migration on for GCE other plugin not affected",
+			volumeAnnotations:    map[string]string{pvutil.AnnDynamicallyProvisioned: testPlugin},
+			expVolumeAnnotations: map[string]string{pvutil.AnnDynamicallyProvisioned: testPlugin},
+			claimAnnotations:     map[string]string{pvutil.AnnStorageProvisioner: testPlugin},
+			expClaimAnnotations:  map[string]string{pvutil.AnnStorageProvisioner: testPlugin},
+			migratedDriverGates:  []featuregate.Feature{features.CSIMigrationGCE},
+		},
+		{
+			name:                 "not dynamically provisioned migration off for GCE",
+			volumeAnnotations:    map[string]string{},
+			expVolumeAnnotations: map[string]string{},
+			claimAnnotations:     map[string]string{},
+			expClaimAnnotations:  map[string]string{},
+			migratedDriverGates:  []featuregate.Feature{},
+		},
+		{
+			name:                 "not dynamically provisioned migration on for GCE",
+			volumeAnnotations:    map[string]string{},
+			expVolumeAnnotations: map[string]string{},
+			claimAnnotations:     map[string]string{},
+			expClaimAnnotations:  map[string]string{},
+			migratedDriverGates:  []featuregate.Feature{features.CSIMigrationGCE},
+		},
+		{
+			name:                 "nil annotations migration off for GCE",
+			volumeAnnotations:    nil,
+			expVolumeAnnotations: nil,
+			claimAnnotations:     nil,
+			expClaimAnnotations:  nil,
+			migratedDriverGates:  []featuregate.Feature{},
+		},
+		{
+			name:                 "nil annotations migration on for GCE",
+			volumeAnnotations:    nil,
+			expVolumeAnnotations: nil,
+			claimAnnotations:     nil,
+			expClaimAnnotations:  nil,
+			migratedDriverGates:  []featuregate.Feature{features.CSIMigrationGCE},
+		},
+	}
+
+	translator := csitrans.New()
+	cmpm := csimigration.NewPluginManager(translator)
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			for _, f := range tc.migratedDriverGates {
+				defer featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, f, true)()
+			}
+			if tc.volumeAnnotations != nil {
+				ann := tc.volumeAnnotations
+				updateMigrationAnnotations(cmpm, translator, ann, pvutil.AnnDynamicallyProvisioned)
+				if !reflect.DeepEqual(tc.expVolumeAnnotations, ann) {
+					t.Errorf("got volume annoations: %v, but expected: %v", ann, tc.expVolumeAnnotations)
+				}
+			}
+			if tc.claimAnnotations != nil {
+				ann := tc.claimAnnotations
+				updateMigrationAnnotations(cmpm, translator, ann, pvutil.AnnStorageProvisioner)
+				if !reflect.DeepEqual(tc.expClaimAnnotations, ann) {
+					t.Errorf("got volume annoations: %v, but expected: %v", ann, tc.expVolumeAnnotations)
+				}
+			}
+
+		})
+	}
+}

pkg/controller/volume/persistentvolume/util/util.go

@@ -64,6 +64,13 @@ const (
 	// recognize dynamically provisioned PVs in its decisions).
 	AnnDynamicallyProvisioned = "pv.kubernetes.io/provisioned-by"
 
+	// AnnMigratedTo annotation is added to a PVC and PV that is supposed to be
+	// dynamically provisioned/deleted by by its corresponding CSI driver
+	// through the CSIMigration feature flags. When this annotation is set the
+	// Kubernetes components will "stand-down" and the external-provisioner will
+	// act on the objects
+	AnnMigratedTo = "volume.beta.kubernetes.io/migrated-to"
+
 	// AnnStorageProvisioner annotation is added to a PVC that is supposed to be dynamically
 	// provisioned. Its value is name of volume plugin that is supposed to provision
 	// a volume for this PVC.