Add IsMigratableToCSI to volume plugin interface

pkg/controller/volume/attachdetach/testing/testvolumespec.go

@@ -253,6 +253,10 @@ func (plugin *TestPlugin) CanSupport(spec *volume.Spec) bool {
 	return true
 }
 
+func (plugin *TestPlugin) IsMigratedToCSI() bool {
+	return false
+}
+
 func (plugin *TestPlugin) RequiresRemount() bool {
 	return false
 }

pkg/controller/volume/persistentvolume/framework_test.go

@@ -1258,6 +1258,10 @@ func (plugin *mockVolumePlugin) CanSupport(spec *vol.Spec) bool {
 	return true
 }
 
+func (plugin *mockVolumePlugin) IsMigratedToCSI() bool {
+	return false
+}
+
 func (plugin *mockVolumePlugin) RequiresRemount() bool {
 	return false
 }

pkg/volume/awsebs/aws_ebs.go

@@ -86,6 +86,11 @@ func (plugin *awsElasticBlockStorePlugin) CanSupport(spec *volume.Spec) bool {
 		(spec.Volume != nil && spec.Volume.AWSElasticBlockStore != nil)
 }
 
+func (plugin *awsElasticBlockStorePlugin) IsMigratedToCSI() bool {
+	return utilfeature.DefaultFeatureGate.Enabled(features.CSIMigration) &&
+		utilfeature.DefaultFeatureGate.Enabled(features.CSIMigrationAWS)
+}
+
 func (plugin *awsElasticBlockStorePlugin) RequiresRemount() bool {
 	return false
 }

pkg/volume/azure_dd/azure_dd.go

@@ -121,6 +121,10 @@ func (plugin *azureDataDiskPlugin) CanSupport(spec *volume.Spec) bool {
 		(spec.Volume != nil && spec.Volume.AzureDisk != nil)
 }
 
+func (plugin *azureDataDiskPlugin) IsMigratedToCSI() bool {
+	return false
+}
+
 func (plugin *azureDataDiskPlugin) RequiresRemount() bool {
 	return false
 }

pkg/volume/azure_file/azure_file.go

@@ -80,6 +80,10 @@ func (plugin *azureFilePlugin) CanSupport(spec *volume.Spec) bool {
 		(spec.Volume != nil && spec.Volume.AzureFile != nil)
 }
 
+func (plugin *azureFilePlugin) IsMigratedToCSI() bool {
+	return false
+}
+
 func (plugin *azureFilePlugin) RequiresRemount() bool {
 	return false
 }

pkg/volume/cephfs/cephfs.go

@@ -71,6 +71,10 @@ func (plugin *cephfsPlugin) CanSupport(spec *volume.Spec) bool {
 	return (spec.Volume != nil && spec.Volume.CephFS != nil) || (spec.PersistentVolume != nil && spec.PersistentVolume.Spec.CephFS != nil)
 }
 
+func (plugin *cephfsPlugin) IsMigratedToCSI() bool {
+	return false
+}
+
 func (plugin *cephfsPlugin) RequiresRemount() bool {
 	return false
 }

pkg/volume/cinder/cinder.go

@@ -108,6 +108,10 @@ func (plugin *cinderPlugin) CanSupport(spec *volume.Spec) bool {
 	return (spec.Volume != nil && spec.Volume.Cinder != nil) || (spec.PersistentVolume != nil && spec.PersistentVolume.Spec.Cinder != nil)
 }
 
+func (plugin *cinderPlugin) IsMigratedToCSI() bool {
+	return false
+}
+
 func (plugin *cinderPlugin) RequiresRemount() bool {
 	return false
 }

pkg/volume/configmap/configmap.go

@@ -77,6 +77,10 @@ func (plugin *configMapPlugin) CanSupport(spec *volume.Spec) bool {
 	return spec.Volume != nil && spec.Volume.ConfigMap != nil
 }
 
+func (plugin *configMapPlugin) IsMigratedToCSI() bool {
+	return false
+}
+
 func (plugin *configMapPlugin) RequiresRemount() bool {
 	return true
 }

pkg/volume/csi/csi_plugin.go

@@ -279,6 +279,10 @@ func (p *csiPlugin) CanSupport(spec *volume.Spec) bool {
 	return spec.PersistentVolume != nil && spec.PersistentVolume.Spec.CSI != nil
 }
 
+func (p *csiPlugin) IsMigratedToCSI() bool {
+	return false
+}
+
 func (p *csiPlugin) RequiresRemount() bool {
 	return false
 }

pkg/volume/downwardapi/downwardapi.go

@@ -81,6 +81,10 @@ func (plugin *downwardAPIPlugin) CanSupport(spec *volume.Spec) bool {
 	return spec.Volume != nil && spec.Volume.DownwardAPI != nil
 }
 
+func (plugin *downwardAPIPlugin) IsMigratedToCSI() bool {
+	return false
+}
+
 func (plugin *downwardAPIPlugin) RequiresRemount() bool {
 	return true
 }

pkg/volume/emptydir/empty_dir.go

@@ -89,6 +89,10 @@ func (plugin *emptyDirPlugin) CanSupport(spec *volume.Spec) bool {
 	return false
 }
 
+func (plugin *emptyDirPlugin) IsMigratedToCSI() bool {
+	return false
+}
+
 func (plugin *emptyDirPlugin) RequiresRemount() bool {
 	return false
 }

pkg/volume/fc/fc.go

@@ -85,6 +85,10 @@ func (plugin *fcPlugin) CanSupport(spec *volume.Spec) bool {
 	return (spec.Volume != nil && spec.Volume.FC != nil) || (spec.PersistentVolume != nil && spec.PersistentVolume.Spec.FC != nil)
 }
 
+func (plugin *fcPlugin) IsMigratedToCSI() bool {
+	return false
+}
+
 func (plugin *fcPlugin) RequiresRemount() bool {
 	return false
 }

pkg/volume/flexvolume/plugin.go

@@ -149,6 +149,10 @@ func (plugin *flexVolumePlugin) CanSupport(spec *volume.Spec) bool {
 	return sourceDriver == plugin.driverName
 }
 
+func (plugin *flexVolumePlugin) IsMigratedToCSI() bool {
+	return false
+}
+
 // RequiresRemount is part of the volume.VolumePlugin interface.
 func (plugin *flexVolumePlugin) RequiresRemount() bool {
 	return false

pkg/volume/flocker/flocker.go

@@ -108,6 +108,10 @@ func (p *flockerPlugin) CanSupport(spec *volume.Spec) bool {
 		(spec.Volume != nil && spec.Volume.Flocker != nil)
 }
 
+func (p *flockerPlugin) IsMigratedToCSI() bool {
+	return false
+}
+
 func (p *flockerPlugin) RequiresRemount() bool {
 	return false
 }

pkg/volume/gcepd/gce_pd.go

@@ -98,6 +98,11 @@ func (plugin *gcePersistentDiskPlugin) CanSupport(spec *volume.Spec) bool {
 		(spec.Volume != nil && spec.Volume.GCEPersistentDisk != nil)
 }
 
+func (plugin *gcePersistentDiskPlugin) IsMigratedToCSI() bool {
+	return utilfeature.DefaultFeatureGate.Enabled(features.CSIMigration) &&
+		utilfeature.DefaultFeatureGate.Enabled(features.CSIMigrationGCE)
+}
+
 func (plugin *gcePersistentDiskPlugin) RequiresRemount() bool {
 	return false
 }

pkg/volume/git_repo/git_repo.go

@@ -78,6 +78,10 @@ func (plugin *gitRepoPlugin) CanSupport(spec *volume.Spec) bool {
 	return spec.Volume != nil && spec.Volume.GitRepo != nil
 }
 
+func (plugin *gitRepoPlugin) IsMigratedToCSI() bool {
+	return false
+}
+
 func (plugin *gitRepoPlugin) RequiresRemount() bool {
 	return false
 }

pkg/volume/glusterfs/glusterfs.go

@@ -114,6 +114,10 @@ func (plugin *glusterfsPlugin) CanSupport(spec *volume.Spec) bool {
 		(spec.Volume != nil && spec.Volume.Glusterfs != nil)
 }
 
+func (plugin *glusterfsPlugin) IsMigratedToCSI() bool {
+	return false
+}
+
 func (plugin *glusterfsPlugin) RequiresRemount() bool {
 	return false
 }

pkg/volume/host_path/host_path.go

@@ -83,6 +83,10 @@ func (plugin *hostPathPlugin) CanSupport(spec *volume.Spec) bool {
 		(spec.Volume != nil && spec.Volume.HostPath != nil)
 }
 
+func (plugin *hostPathPlugin) IsMigratedToCSI() bool {
+	return false
+}
+
 func (plugin *hostPathPlugin) RequiresRemount() bool {
 	return false
 }

pkg/volume/iscsi/iscsi.go

@@ -76,6 +76,10 @@ func (plugin *iscsiPlugin) CanSupport(spec *volume.Spec) bool {
 	return (spec.Volume != nil && spec.Volume.ISCSI != nil) || (spec.PersistentVolume != nil && spec.PersistentVolume.Spec.ISCSI != nil)
 }
 
+func (plugin *iscsiPlugin) IsMigratedToCSI() bool {
+	return false
+}
+
 func (plugin *iscsiPlugin) RequiresRemount() bool {
 	return false
 }

pkg/volume/local/local.go

@@ -82,6 +82,10 @@ func (plugin *localVolumePlugin) CanSupport(spec *volume.Spec) bool {
 	return (spec.PersistentVolume != nil && spec.PersistentVolume.Spec.Local != nil)
 }
 
+func (plugin *localVolumePlugin) IsMigratedToCSI() bool {
+	return false
+}
+
 func (plugin *localVolumePlugin) RequiresRemount() bool {
 	return false
 }

pkg/volume/nfs/nfs.go

@@ -84,6 +84,10 @@ func (plugin *nfsPlugin) CanSupport(spec *volume.Spec) bool {
 		(spec.Volume != nil && spec.Volume.NFS != nil)
 }
 
+func (plugin *nfsPlugin) IsMigratedToCSI() bool {
+	return false
+}
+
 func (plugin *nfsPlugin) RequiresRemount() bool {
 	return false
 }

pkg/volume/noop_expandable_plugin.go

@@ -48,6 +48,10 @@ func (n *noopExpandableVolumePluginInstance) CanSupport(spec *Spec) bool {
 	return true
 }
 
+func (n *noopExpandableVolumePluginInstance) IsMigratedToCSI() bool {
+	return false
+}
+
 func (n *noopExpandableVolumePluginInstance) RequiresRemount() bool {
 	return false
 }

pkg/volume/photon_pd/photon_pd.go

@@ -74,6 +74,10 @@ func (plugin *photonPersistentDiskPlugin) CanSupport(spec *volume.Spec) bool {
 		(spec.Volume != nil && spec.Volume.PhotonPersistentDisk != nil)
 }
 
+func (plugin *photonPersistentDiskPlugin) IsMigratedToCSI() bool {
+	return false
+}
+
 func (plugin *photonPersistentDiskPlugin) RequiresRemount() bool {
 	return false
 }

pkg/volume/plugins.go

@@ -118,6 +118,10 @@ type VolumePlugin interface {
 	// const.
 	CanSupport(spec *Spec) bool
 
+	// IsMigratedToCSI tests whether a CSIDriver implements this plugin's
+	// functionality
+	IsMigratedToCSI() bool
+
 	// RequiresRemount returns true if this plugin requires mount calls to be
 	// reexecuted. Atomically updating volumes, like Downward API, depend on
 	// this to update the contents of the volume.
@@ -601,6 +605,37 @@ func (pm *VolumePluginMgr) FindPluginBySpec(spec *Spec) (VolumePlugin, error) {
 	return matches[0], nil
 }
 
+// IsPluginMigratableBySpec looks for a plugin that can support a given volume
+// specification and whether that plugin is Migratable. If no plugins can
+// support or more than one plugin can support it, return error.
+func (pm *VolumePluginMgr) IsPluginMigratableBySpec(spec *Spec) (bool, error) {
+	pm.mutex.Lock()
+	defer pm.mutex.Unlock()
+
+	if spec == nil {
+		return false, fmt.Errorf("could not find if plugin is migratable because volume spec is nil")
+	}
+
+	matchedPluginNames := []string{}
+	matches := []VolumePlugin{}
+	for k, v := range pm.plugins {
+		if v.CanSupport(spec) {
+			matchedPluginNames = append(matchedPluginNames, k)
+			matches = append(matches, v)
+		}
+	}
+
+	if len(matches) == 0 {
+		// Not a known plugin (flex) in which case it is not migratable
+		return false, nil
+	}
+	if len(matches) > 1 {
+		return false, fmt.Errorf("multiple volume plugins matched: %s", strings.Join(matchedPluginNames, ","))
+	}
+
+	return matches[0].IsMigratedToCSI(), nil
+}
+
 // FindPluginByName fetches a plugin by name or by legacy name.  If no plugin
 // is found, returns error.
 func (pm *VolumePluginMgr) FindPluginByName(name string) (VolumePlugin, error) {

pkg/volume/plugins_test.go

@@ -75,6 +75,10 @@ func (plugin *testPlugins) CanSupport(spec *Spec) bool {
 	return true
 }
 
+func (plugin *testPlugins) IsMigratedToCSI() bool {
+	return false
+}
+
 func (plugin *testPlugins) RequiresRemount() bool {
 	return false
 }

pkg/volume/portworx/portworx.go

@@ -95,6 +95,10 @@ func (plugin *portworxVolumePlugin) CanSupport(spec *volume.Spec) bool {
 		(spec.Volume != nil && spec.Volume.PortworxVolume != nil)
 }
 
+func (plugin *portworxVolumePlugin) IsMigratedToCSI() bool {
+	return false
+}
+
 func (plugin *portworxVolumePlugin) RequiresRemount() bool {
 	return false
 }

pkg/volume/projected/projected.go

@@ -96,6 +96,10 @@ func (plugin *projectedPlugin) CanSupport(spec *volume.Spec) bool {
 	return spec.Volume != nil && spec.Volume.Projected != nil
 }
 
+func (plugin *projectedPlugin) IsMigratedToCSI() bool {
+	return false
+}
+
 func (plugin *projectedPlugin) RequiresRemount() bool {
 	return true
 }

pkg/volume/quobyte/quobyte.go

@@ -110,6 +110,10 @@ func (plugin *quobytePlugin) CanSupport(spec *volume.Spec) bool {
 	return false
 }
 
+func (plugin *quobytePlugin) IsMigratedToCSI() bool {
+	return false
+}
+
 func (plugin *quobytePlugin) RequiresRemount() bool {
 	return false
 }

pkg/volume/rbd/rbd.go

@@ -107,6 +107,10 @@ func (plugin *rbdPlugin) CanSupport(spec *volume.Spec) bool {
 	return (spec.Volume != nil && spec.Volume.RBD != nil) || (spec.PersistentVolume != nil && spec.PersistentVolume.Spec.RBD != nil)
 }
 
+func (plugin *rbdPlugin) IsMigratedToCSI() bool {
+	return false
+}
+
 func (plugin *rbdPlugin) RequiresRemount() bool {
 	return false
 }

pkg/volume/scaleio/sio_plugin.go

@@ -72,6 +72,10 @@ func (p *sioPlugin) CanSupport(spec *volume.Spec) bool {
 		(spec.Volume != nil && spec.Volume.ScaleIO != nil)
 }
 
+func (p *sioPlugin) IsMigratedToCSI() bool {
+	return false
+}
+
 func (p *sioPlugin) RequiresRemount() bool {
 	return false
 }

pkg/volume/secret/secret.go

@@ -80,6 +80,10 @@ func (plugin *secretPlugin) CanSupport(spec *volume.Spec) bool {
 	return spec.Volume != nil && spec.Volume.Secret != nil
 }
 
+func (plugin *secretPlugin) IsMigratedToCSI() bool {
+	return false
+}
+
 func (plugin *secretPlugin) RequiresRemount() bool {
 	return true
 }

pkg/volume/storageos/storageos.go

@@ -91,6 +91,10 @@ func (plugin *storageosPlugin) CanSupport(spec *volume.Spec) bool {
 		(spec.Volume != nil && spec.Volume.StorageOS != nil)
 }
 
+func (plugin *storageosPlugin) IsMigratedToCSI() bool {
+	return false
+}
+
 func (plugin *storageosPlugin) RequiresRemount() bool {
 	return false
 }

pkg/volume/testing/testing.go

@@ -313,6 +313,10 @@ func (plugin *FakeVolumePlugin) CanSupport(spec *Spec) bool {
 	return true
 }
 
+func (plugin *FakeVolumePlugin) IsMigratedToCSI() bool {
+	return false
+}
+
 func (plugin *FakeVolumePlugin) RequiresRemount() bool {
 	return false
 }
@@ -526,6 +530,10 @@ func (f *FakeBasicVolumePlugin) CanSupport(spec *Spec) bool {
 	return strings.HasPrefix(spec.Name(), f.GetPluginName())
 }
 
+func (plugin *FakeBasicVolumePlugin) IsMigratedToCSI() bool {
+	return false
+}
+
 func (f *FakeBasicVolumePlugin) ConstructVolumeSpec(ame, mountPath string) (*Spec, error) {
 	return f.Plugin.ConstructVolumeSpec(ame, mountPath)
 }
@@ -611,6 +619,10 @@ func (plugin *FakeFileVolumePlugin) CanSupport(spec *Spec) bool {
 	return true
 }
 
+func (plugin *FakeFileVolumePlugin) IsMigratedToCSI() bool {
+	return false
+}
+
 func (plugin *FakeFileVolumePlugin) RequiresRemount() bool {
 	return false
 }

pkg/volume/util/operationexecutor/BUILD

@@ -30,6 +30,7 @@ go_library(
         "//staging/src/k8s.io/apiserver/pkg/util/feature:go_default_library",
         "//staging/src/k8s.io/client-go/kubernetes:go_default_library",
         "//staging/src/k8s.io/client-go/tools/record:go_default_library",
+        "//staging/src/k8s.io/csi-translation-lib:go_default_library",
         "//vendor/k8s.io/klog:go_default_library",
     ],
 )

pkg/volume/util/operationexecutor/operation_generator.go

@@ -29,6 +29,7 @@ import (
 	utilfeature "k8s.io/apiserver/pkg/util/feature"
 	clientset "k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/tools/record"
+	csilib "k8s.io/csi-translation-lib"
 	"k8s.io/klog"
 	expandcache "k8s.io/kubernetes/pkg/controller/volume/expand/cache"
 	"k8s.io/kubernetes/pkg/features"
@@ -75,11 +76,11 @@ func NewOperationGenerator(kubeClient clientset.Interface,
 	blkUtil volumepathhandler.BlockVolumePathHandler) OperationGenerator {
 
 	return &operationGenerator{
-		kubeClient:                       kubeClient,
-		volumePluginMgr:                  volumePluginMgr,
-		recorder:                         recorder,
+		kubeClient:      kubeClient,
+		volumePluginMgr: volumePluginMgr,
+		recorder:        recorder,
 		checkNodeCapabilitiesBeforeMount: checkNodeCapabilitiesBeforeMount,
-		blkUtil:                          blkUtil,
+		blkUtil: blkUtil,
 	}
 }
 
@@ -1437,3 +1438,32 @@ func isDeviceOpened(deviceToDetach AttachedVolume, mounter mount.Interface) (boo
 	}
 	return deviceOpened, nil
 }
+
+// TODO(dyzz): need to also add logic to check CSINodeInfo for Kubelet migration status
+func useCSIPlugin(vpm *volume.VolumePluginMgr, spec *volume.Spec) bool {
+	if csilib.IsPVMigratable(spec.PersistentVolume) || csilib.IsInlineMigratable(spec.Volume) {
+		migratable, err := vpm.IsPluginMigratableBySpec(spec)
+		if err == nil && migratable {
+			return true
+		}
+	}
+	return false
+}
+
+func translateSpec(spec *volume.Spec) (*volume.Spec, error) {
+	if spec.PersistentVolume != nil {
+		// TranslateInTreePVToCSI will create a new PV
+		csiPV, err := csilib.TranslateInTreePVToCSI(spec.PersistentVolume)
+		if err != nil {
+			return nil, fmt.Errorf("failed to translate in tree pv to CSI: %v", err)
+		}
+		return &volume.Spec{
+			PersistentVolume: csiPV,
+			ReadOnly:         spec.ReadOnly,
+		}, nil
+	} else if spec.Volume != nil {
+		return &volume.Spec{}, fmt.Errorf("translation is not supported for in-line volumes yet")
+	} else {
+		return &volume.Spec{}, fmt.Errorf("not a valid volume spec")
+	}
+}

pkg/volume/vsphere_volume/vsphere_volume.go

@@ -81,6 +81,10 @@ func (plugin *vsphereVolumePlugin) CanSupport(spec *volume.Spec) bool {
 		(spec.Volume != nil && spec.Volume.VsphereVolume != nil)
 }
 
+func (plugin *vsphereVolumePlugin) IsMigratedToCSI() bool {
+	return false
+}
+
 func (plugin *vsphereVolumePlugin) RequiresRemount() bool {
 	return false
 }