Merge pull request #95361 from humblec/rbd-migration

RBD in-tree plugin migration to CSI driver using migration translation lib

cmd/kube-controller-manager/app/plugins.go

@@ -42,7 +42,6 @@ import (
 	"k8s.io/kubernetes/pkg/volume/nfs"
 	"k8s.io/kubernetes/pkg/volume/portworx"
 	"k8s.io/kubernetes/pkg/volume/quobyte"
-	"k8s.io/kubernetes/pkg/volume/rbd"
 	"k8s.io/kubernetes/pkg/volume/storageos"
 	volumeutil "k8s.io/kubernetes/pkg/volume/util"
 
@@ -67,7 +66,6 @@ func ProbeAttachableVolumePlugins() ([]volume.VolumePlugin, error) {
 	allPlugins = append(allPlugins, storageos.ProbeVolumePlugins()...)
 	allPlugins = append(allPlugins, fc.ProbeVolumePlugins()...)
 	allPlugins = append(allPlugins, iscsi.ProbeVolumePlugins()...)
-	allPlugins = append(allPlugins, rbd.ProbeVolumePlugins()...)
 	allPlugins = append(allPlugins, csi.ProbeVolumePlugins()...)
 	return allPlugins, nil
 }
@@ -89,7 +87,6 @@ func ProbeExpandableVolumePlugins(config persistentvolumeconfig.VolumeConfigurat
 	}
 	allPlugins = append(allPlugins, portworx.ProbeVolumePlugins()...)
 	allPlugins = append(allPlugins, glusterfs.ProbeVolumePlugins()...)
-	allPlugins = append(allPlugins, rbd.ProbeVolumePlugins()...)
 	allPlugins = append(allPlugins, storageos.ProbeVolumePlugins()...)
 	allPlugins = append(allPlugins, fc.ProbeVolumePlugins()...)
 	return allPlugins, nil
@@ -132,7 +129,6 @@ func ProbeControllerVolumePlugins(cloud cloudprovider.Interface, config persiste
 	allPlugins = append(allPlugins, nfs.ProbeVolumePlugins(nfsConfig)...)
 	allPlugins = append(allPlugins, glusterfs.ProbeVolumePlugins()...)
 	// add rbd provisioner
-	allPlugins = append(allPlugins, rbd.ProbeVolumePlugins()...)
 	allPlugins = append(allPlugins, quobyte.ProbeVolumePlugins()...)
 	var err error
 	allPlugins, err = appendExpandableLegacyProviderVolumes(allPlugins, utilfeature.DefaultFeatureGate)

cmd/kube-controller-manager/app/plugins_providers.go

@@ -31,6 +31,7 @@ import (
 	"k8s.io/kubernetes/pkg/volume/cinder"
 	"k8s.io/kubernetes/pkg/volume/csimigration"
 	"k8s.io/kubernetes/pkg/volume/gcepd"
+	"k8s.io/kubernetes/pkg/volume/rbd"
 	"k8s.io/kubernetes/pkg/volume/vsphere_volume"
 )
 
@@ -67,7 +68,7 @@ func appendAttachableLegacyProviderVolumes(allPlugins []volume.VolumePlugin, fea
 	pluginMigrationStatus[plugins.CinderInTreePluginName] = pluginInfo{pluginMigrationFeature: features.CSIMigrationOpenStack, pluginUnregisterFeature: features.InTreePluginOpenStackUnregister, pluginProbeFunction: cinder.ProbeVolumePlugins}
 	pluginMigrationStatus[plugins.AzureDiskInTreePluginName] = pluginInfo{pluginMigrationFeature: features.CSIMigrationAzureDisk, pluginUnregisterFeature: features.InTreePluginAzureDiskUnregister, pluginProbeFunction: azuredd.ProbeVolumePlugins}
 	pluginMigrationStatus[plugins.VSphereInTreePluginName] = pluginInfo{pluginMigrationFeature: features.CSIMigrationvSphere, pluginUnregisterFeature: features.InTreePluginvSphereUnregister, pluginProbeFunction: vsphere_volume.ProbeVolumePlugins}
-
+	pluginMigrationStatus[plugins.RBDVolumePluginName] = pluginInfo{pluginMigrationFeature: features.CSIMigrationRBD, pluginUnregisterFeature: features.InTreePluginRBDUnregister, pluginProbeFunction: rbd.ProbeVolumePlugins}
 	var err error
 	for pluginName, pluginInfo := range pluginMigrationStatus {
 		allPlugins, err = appendPluginBasedOnFeatureFlags(allPlugins, pluginName, featureGate, pluginInfo)

cmd/kubelet/app/plugins.go

@@ -40,7 +40,6 @@ import (
 	"k8s.io/kubernetes/pkg/volume/portworx"
 	"k8s.io/kubernetes/pkg/volume/projected"
 	"k8s.io/kubernetes/pkg/volume/quobyte"
-	"k8s.io/kubernetes/pkg/volume/rbd"
 	"k8s.io/kubernetes/pkg/volume/secret"
 	"k8s.io/kubernetes/pkg/volume/storageos"
 
@@ -70,7 +69,6 @@ func ProbeVolumePlugins(featureGate featuregate.FeatureGate) ([]volume.VolumePlu
 	allPlugins = append(allPlugins, secret.ProbeVolumePlugins()...)
 	allPlugins = append(allPlugins, iscsi.ProbeVolumePlugins()...)
 	allPlugins = append(allPlugins, glusterfs.ProbeVolumePlugins()...)
-	allPlugins = append(allPlugins, rbd.ProbeVolumePlugins()...)
 	allPlugins = append(allPlugins, quobyte.ProbeVolumePlugins()...)
 	allPlugins = append(allPlugins, cephfs.ProbeVolumePlugins()...)
 	allPlugins = append(allPlugins, downwardapi.ProbeVolumePlugins()...)

cmd/kubelet/app/plugins_providers.go

@@ -36,6 +36,7 @@ import (
 	"k8s.io/kubernetes/pkg/volume/cinder"
 	"k8s.io/kubernetes/pkg/volume/csimigration"
 	"k8s.io/kubernetes/pkg/volume/gcepd"
+	"k8s.io/kubernetes/pkg/volume/rbd"
 	"k8s.io/kubernetes/pkg/volume/vsphere_volume"
 )
 
@@ -74,7 +75,7 @@ func appendLegacyProviderVolumes(allPlugins []volume.VolumePlugin, featureGate f
 	pluginMigrationStatus[plugins.AzureDiskInTreePluginName] = pluginInfo{pluginMigrationFeature: features.CSIMigrationAzureDisk, pluginUnregisterFeature: features.InTreePluginAzureDiskUnregister, pluginProbeFunction: azuredd.ProbeVolumePlugins}
 	pluginMigrationStatus[plugins.AzureFileInTreePluginName] = pluginInfo{pluginMigrationFeature: features.CSIMigrationAzureFile, pluginUnregisterFeature: features.InTreePluginAzureFileUnregister, pluginProbeFunction: azure_file.ProbeVolumePlugins}
 	pluginMigrationStatus[plugins.VSphereInTreePluginName] = pluginInfo{pluginMigrationFeature: features.CSIMigrationvSphere, pluginUnregisterFeature: features.InTreePluginvSphereUnregister, pluginProbeFunction: vsphere_volume.ProbeVolumePlugins}
-
+	pluginMigrationStatus[plugins.RBDVolumePluginName] = pluginInfo{pluginMigrationFeature: features.CSIMigrationRBD, pluginUnregisterFeature: features.InTreePluginRBDUnregister, pluginProbeFunction: rbd.ProbeVolumePlugins}
 	var err error
 	for pluginName, pluginInfo := range pluginMigrationStatus {
 		allPlugins, err = appendPluginBasedOnFeatureFlags(allPlugins, pluginName, featureGate, pluginInfo)

pkg/features/kube_features.go

@@ -330,6 +330,18 @@ const (
 	// Disables the OpenStack Cinder in-tree driver.
 	InTreePluginOpenStackUnregister featuregate.Feature = "InTreePluginOpenStackUnregister"
 
+	// owner: @humblec
+	// alpha: v1.23
+	//
+	// Enables the RBD in-tree driver to RBD CSI Driver  migration feature.
+	CSIMigrationRBD featuregate.Feature = "csiMigrationRBD"
+
+	// owner: @humblec
+	// alpha: v1.23
+	//
+	// Disables the RBD in-tree driver.
+	InTreePluginRBDUnregister featuregate.Feature = "InTreePluginRBDUnregister"
+
 	// owner: @huffmanca, @dobsonj
 	// alpha: v1.19
 	// beta: v1.20
@@ -843,6 +855,8 @@ var defaultKubernetesFeatureGates = map[featuregate.Feature]featuregate.FeatureS
 	InTreePluginvSphereUnregister:                  {Default: false, PreRelease: featuregate.Alpha},
 	CSIMigrationOpenStack:                          {Default: true, PreRelease: featuregate.Beta},
 	InTreePluginOpenStackUnregister:                {Default: false, PreRelease: featuregate.Alpha},
+	CSIMigrationRBD:                                {Default: false, PreRelease: featuregate.Alpha}, // Off by default (requires RBD CSI driver)
+	InTreePluginRBDUnregister:                      {Default: false, PreRelease: featuregate.Alpha},
 	ConfigurableFSGroupPolicy:                      {Default: true, PreRelease: featuregate.GA, LockToDefault: true}, // remove in 1.25
 	CSIInlineVolume:                                {Default: true, PreRelease: featuregate.Beta},
 	CSIStorageCapacity:                             {Default: true, PreRelease: featuregate.Beta},

pkg/scheduler/framework/plugins/nodevolumelimits/utils.go

@@ -59,6 +59,10 @@ func isCSIMigrationOn(csiNode *storagev1.CSINode, pluginName string) bool {
 		if !utilfeature.DefaultFeatureGate.Enabled(features.CSIMigrationOpenStack) {
 			return false
 		}
+	case csilibplugins.RBDVolumePluginName:
+		if !utilfeature.DefaultFeatureGate.Enabled(features.CSIMigrationRBD) {
+			return false
+		}
 	default:
 		return false
 	}

pkg/scheduler/framework/plugins/volumebinding/binder.go

@@ -1025,6 +1025,8 @@ func isCSIMigrationOnForPlugin(pluginName string) bool {
 		return utilfeature.DefaultFeatureGate.Enabled(features.CSIMigrationAzureDisk)
 	case csiplugins.CinderInTreePluginName:
 		return utilfeature.DefaultFeatureGate.Enabled(features.CSIMigrationOpenStack)
+	case csiplugins.RBDVolumePluginName:
+		return utilfeature.DefaultFeatureGate.Enabled(features.CSIMigrationRBD)
 	}
 	return false
 }

pkg/volume/csi/csi_plugin.go

@@ -234,6 +234,9 @@ func (p *csiPlugin) Init(host volume.VolumeHost) error {
 		csitranslationplugins.VSphereInTreePluginName: func() bool {
 			return utilfeature.DefaultFeatureGate.Enabled(features.CSIMigration) && utilfeature.DefaultFeatureGate.Enabled(features.CSIMigrationvSphere)
 		},
+		csitranslationplugins.RBDVolumePluginName: func() bool {
+			return utilfeature.DefaultFeatureGate.Enabled(features.CSIMigration) && utilfeature.DefaultFeatureGate.Enabled(features.CSIMigrationRBD)
+		},
 	}
 
 	// Initializing the label management channels

pkg/volume/csimigration/plugin_manager.go

@@ -72,6 +72,8 @@ func (pm PluginManager) IsMigrationCompleteForPlugin(pluginName string) bool {
 		return pm.featureGate.Enabled(features.InTreePluginOpenStackUnregister)
 	case csilibplugins.VSphereInTreePluginName:
 		return pm.featureGate.Enabled(features.InTreePluginvSphereUnregister)
+	case csilibplugins.RBDVolumePluginName:
+		return pm.featureGate.Enabled(features.InTreePluginRBDUnregister)
 	default:
 		return false
 	}
@@ -98,6 +100,8 @@ func (pm PluginManager) IsMigrationEnabledForPlugin(pluginName string) bool {
 		return pm.featureGate.Enabled(features.CSIMigrationOpenStack)
 	case csilibplugins.VSphereInTreePluginName:
 		return pm.featureGate.Enabled(features.CSIMigrationvSphere)
+	case csilibplugins.RBDVolumePluginName:
+		return pm.featureGate.Enabled(features.CSIMigrationRBD)
 	default:
 		return false
 	}

pkg/volume/rbd/rbd.go

@@ -19,6 +19,8 @@ package rbd
 import (
 	"context"
 	"fmt"
+	utilfeature "k8s.io/apiserver/pkg/util/feature"
+	"k8s.io/kubernetes/pkg/features"
 	"os"
 	"path/filepath"
 	"regexp"
@@ -78,6 +80,11 @@ func getPath(uid types.UID, volName string, host volume.VolumeHost) string {
 	return host.GetPodVolumeDir(uid, utilstrings.EscapeQualifiedName(rbdPluginName), volName)
 }
 
+func (plugin *rbdPlugin) IsMigratedToCSI() bool {
+	return utilfeature.DefaultFeatureGate.Enabled(features.CSIMigration) &&
+		utilfeature.DefaultFeatureGate.Enabled(features.CSIMigrationRBD)
+}
+
 func (plugin *rbdPlugin) Init(host volume.VolumeHost) error {
 	plugin.host = host
 	return nil

staging/src/k8s.io/csi-translation-lib/plugins/rbd.go

@@ -0,0 +1,325 @@
+/*
+Copyright 2021 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 plugins
+
+import (
+	"crypto/md5"
+	"encoding/hex"
+	"fmt"
+	"strings"
+
+	"k8s.io/api/core/v1"
+	storagev1 "k8s.io/api/storage/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+const (
+	RBDVolumePluginName           = "kubernetes.io/rbd"
+	RBDDriverName                 = "rbd.csi.ceph.com"
+	defaultAdminSecretNamespace   = "default"
+	defaultImgFeatureVal          = "layering"
+	defaultAdminUser              = "admin"
+	defaultPoolVal                = "rbd"
+	defaultIntreeImagePfx         = "kubernetes-dynamic-pvc-"
+	defaultMigKey                 = "migration"
+	defaultMigStaticVal           = "true"
+	CSIRBDVolHandleAnnKey         = "rbd.csi.ceph.com/volume-handle"
+	imgFeatureKey                 = "imageFeatures"
+	imgFmtKey                     = "imageFormat"
+	imgNameKey                    = "imageName"
+	clusterIDKey                  = "clusterID"
+	journalPoolKey                = "journalPool"
+	poolKey                       = "pool"
+	monsKey                       = "monitors"
+	adminIDKey                    = "adminId"
+	staticVolKey                  = "staticVolume"
+	monsPfx                       = "mons-"
+	imgPfx                        = "image-"
+	migVolPfx                     = "mig"
+	provSecretNameKey             = "csi.storage.k8s.io/provisioner-secret-name"
+	nodeStageSecretNameKey        = "csi.storage.k8s.io/node-stage-secret-name"
+	cntrlExpandSecretNameKey      = "csi.storage.k8s.io/controller-expand-secret-name"
+	provSecretNamespaceKey        = "csi.storage.k8s.io/provisioner-secret-namespace"
+	nodeStageSecretNamespaceKey   = "csi.storage.k8s.io/node-stage-secret-namespace"
+	cntrlExpandSecretNamespaceKey = "csi.storage.k8s.io/controller-expand-secret-namespace"
+)
+
+var _ InTreePlugin = &rbdCSITranslator{}
+
+type rbdCSITranslator struct{}
+
+func NewRBDCSITranslator() InTreePlugin {
+	return &rbdCSITranslator{}
+}
+
+// TranslateInTreeStorageClassToCSI takes in-tree storage class used by in-tree plugin
+// and translates them to a storage class consumable by CSI plugin
+func (p rbdCSITranslator) TranslateInTreeStorageClassToCSI(sc *storagev1.StorageClass) (*storagev1.StorageClass, error) {
+	if sc == nil {
+		return nil, fmt.Errorf("sc is nil")
+	}
+
+	var params = map[string]string{}
+
+	fillDefaultSCParams(params)
+	for k, v := range sc.Parameters {
+		switch strings.ToLower(k) {
+		case fsTypeKey:
+			params[csiFsTypeKey] = v
+		case "imagefeatures":
+			params[imgFeatureKey] = v
+		case poolKey:
+			params[poolKey] = v
+		case "imageformat":
+			params[imgFmtKey] = v
+		case "adminid":
+			params[adminIDKey] = v
+		case "adminsecretname":
+			params[provSecretNameKey] = v
+			params[nodeStageSecretNameKey] = v
+			params[cntrlExpandSecretNameKey] = v
+		case "adminsecretnamespace":
+			params[provSecretNamespaceKey] = v
+			params[nodeStageSecretNamespaceKey] = v
+			params[cntrlExpandSecretNamespaceKey] = v
+		case monsKey:
+			arr := strings.Split(v, ",")
+			if len(arr) < 1 {
+				return nil, fmt.Errorf("missing Ceph monitors")
+			}
+			params[monsKey] = v
+			params[clusterIDKey] = fmt.Sprintf("%x", md5.Sum([]byte(v)))
+		}
+	}
+
+	if params[provSecretNameKey] == "" {
+		return nil, fmt.Errorf("missing Ceph admin secret name")
+	}
+	if params[monsKey] == "" {
+		return nil, fmt.Errorf("missing Ceph monitors")
+	}
+	sc.Provisioner = RBDDriverName
+	sc.Parameters = params
+	return sc, nil
+}
+
+// TranslateInTreeInlineVolumeToCSI takes an inline volume and will translate
+// the in-tree inline volume source to a CSIPersistentVolumeSource
+func (p rbdCSITranslator) TranslateInTreeInlineVolumeToCSI(volume *v1.Volume, podNamespace string) (*v1.PersistentVolume, error) {
+	if volume == nil || volume.RBD == nil {
+		return nil, fmt.Errorf("volume is nil or RBDVolume not defined on volume")
+	}
+
+	var am v1.PersistentVolumeAccessMode
+	if volume.RBD.ReadOnly {
+		am = v1.ReadOnlyMany
+	} else {
+		am = v1.ReadWriteOnce
+	}
+	secRef := &v1.SecretReference{}
+	if volume.RBD.SecretRef != nil {
+		secRef.Name = volume.RBD.SecretRef.Name
+		secRef.Namespace = podNamespace
+	}
+	volumeAttr := make(map[string]string)
+	volumeAttr[clusterIDKey] = fmt.Sprintf("%x", md5.Sum([]byte(strings.Join(volume.RBD.CephMonitors, ","))))
+	volumeAttr[poolKey] = defaultPoolVal
+	if volume.RBD.RBDPool != "" {
+		volumeAttr[poolKey] = volume.RBD.RBDPool
+	}
+	volumeAttr[staticVolKey] = defaultMigStaticVal
+	volumeAttr[imgFeatureKey] = defaultImgFeatureVal
+	pv := &v1.PersistentVolume{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: fmt.Sprintf("%s-%s", RBDDriverName, volume.RBD.RBDImage),
+		},
+		Spec: v1.PersistentVolumeSpec{
+			PersistentVolumeSource: v1.PersistentVolumeSource{
+				CSI: &v1.CSIPersistentVolumeSource{
+					Driver:                    RBDDriverName,
+					VolumeHandle:              volume.RBD.RBDImage,
+					FSType:                    volume.RBD.FSType,
+					VolumeAttributes:          volumeAttr,
+					NodeStageSecretRef:        secRef,
+					ControllerExpandSecretRef: secRef,
+				},
+			},
+			AccessModes: []v1.PersistentVolumeAccessMode{am},
+		},
+	}
+	return pv, nil
+}
+
+// TranslateInTreePVToCSI takes a RBD persistent volume and will translate
+// the in-tree pv source to a CSI Source
+func (p rbdCSITranslator) TranslateInTreePVToCSI(pv *v1.PersistentVolume) (*v1.PersistentVolume, error) {
+	if pv == nil || pv.Spec.RBD == nil {
+		return nil, fmt.Errorf("pv is nil or RBD Volume not defined on pv")
+	}
+	var volID string
+	volumeAttributes := make(map[string]string)
+
+	if pv.Annotations[CSIRBDVolHandleAnnKey] != "" {
+		volID = pv.Annotations[CSIRBDVolHandleAnnKey]
+		volumeAttributes[clusterIDKey] = pv.Annotations[clusterIDKey]
+	} else {
+		mons := strings.Join(pv.Spec.RBD.CephMonitors, ",")
+		pool := pv.Spec.RBD.RBDPool
+		image := pv.Spec.RBD.RBDImage
+		volumeAttributes[staticVolKey] = defaultMigStaticVal
+		volumeAttributes[clusterIDKey] = fmt.Sprintf("%x", md5.Sum([]byte(mons)))
+		volID = composeMigVolID(mons, pool, image)
+	}
+
+	err := fillVolAttrsForRequest(pv, volumeAttributes)
+	if err != nil {
+		return nil, err
+	}
+	if volumeAttributes[imgFeatureKey] == "" {
+		volumeAttributes[imgFeatureKey] = defaultImgFeatureVal
+	}
+	var am v1.PersistentVolumeAccessMode
+	if pv.Spec.RBD.ReadOnly {
+		am = v1.ReadOnlyMany
+	} else {
+		am = v1.ReadWriteOnce
+	}
+	pv.Spec.AccessModes = []v1.PersistentVolumeAccessMode{am}
+	csiSource := &v1.CSIPersistentVolumeSource{
+		Driver:                    RBDDriverName,
+		FSType:                    pv.Spec.RBD.FSType,
+		VolumeHandle:              volID,
+		VolumeAttributes:          volumeAttributes,
+		NodeStageSecretRef:        pv.Spec.RBD.SecretRef,
+		ControllerExpandSecretRef: pv.Spec.RBD.SecretRef,
+	}
+	pv.Spec.RBD = nil
+	pv.Spec.CSI = csiSource
+	return pv, nil
+}
+
+// TranslateCSIPVToInTree takes a PV with a CSI PersistentVolume Source and will translate
+// it to an in-tree Persistent Volume Source for the in-tree volume
+func (p rbdCSITranslator) TranslateCSIPVToInTree(pv *v1.PersistentVolume) (*v1.PersistentVolume, error) {
+	if pv == nil || pv.Spec.CSI == nil {
+		return nil, fmt.Errorf("pv is nil or CSI source not defined on pv")
+	}
+	var rbdImageName string
+	var monSlice []string
+	csiSource := pv.Spec.CSI
+
+	rbdImageName = csiSource.VolumeAttributes[imgNameKey]
+	rbdPool := csiSource.VolumeAttributes[poolKey]
+	radosUser := csiSource.VolumeAttributes[adminIDKey]
+	if radosUser == "" {
+		radosUser = defaultAdminUser
+	}
+
+	RBDSource := &v1.RBDPersistentVolumeSource{
+		CephMonitors: monSlice,
+		RBDImage:     rbdImageName,
+		FSType:       csiSource.FSType,
+		RBDPool:      rbdPool,
+		RadosUser:    radosUser,
+		ReadOnly:     csiSource.ReadOnly,
+	}
+
+	if pv.Annotations == nil {
+		pv.Annotations = make(map[string]string)
+	}
+	fillAnnotationsFromCSISource(pv, csiSource)
+	nodeSecret := csiSource.NodeStageSecretRef
+	if nodeSecret != nil {
+		RBDSource.SecretRef = &v1.SecretReference{Name: nodeSecret.Name, Namespace: nodeSecret.Namespace}
+	}
+	pv.Spec.CSI = nil
+	pv.Spec.RBD = RBDSource
+
+	return pv, nil
+}
+
+// CanSupport tests whether the plugin supports a given persistent volume
+// specification from the API.
+func (p rbdCSITranslator) CanSupport(pv *v1.PersistentVolume) bool {
+	return pv != nil && pv.Spec.RBD != nil
+}
+
+// CanSupportInline tests whether the plugin supports a given inline volume
+// specification from the API.
+func (p rbdCSITranslator) CanSupportInline(volume *v1.Volume) bool {
+	return volume != nil && volume.RBD != nil
+}
+
+// GetInTreePluginName returns the in-tree plugin name this migrates
+func (p rbdCSITranslator) GetInTreePluginName() string {
+	return RBDVolumePluginName
+}
+
+// GetCSIPluginName returns the name of the CSI plugin that supersedes the in-tree plugin
+func (p rbdCSITranslator) GetCSIPluginName() string {
+	return RBDDriverName
+}
+
+// RepairVolumeHandle generates a correct volume handle based on node ID information.
+func (p rbdCSITranslator) RepairVolumeHandle(volumeHandle, nodeID string) (string, error) {
+	return volumeHandle, nil
+}
+
+// fillDefaultSCParams fills some sc parameters with default values
+func fillDefaultSCParams(params map[string]string) {
+	params[defaultMigKey] = defaultMigStaticVal
+	params[poolKey] = defaultPoolVal
+	params[provSecretNamespaceKey] = defaultAdminSecretNamespace
+	params[cntrlExpandSecretNamespaceKey] = defaultAdminSecretNamespace
+	params[nodeStageSecretNamespaceKey] = defaultAdminSecretNamespace
+}
+
+// composeMigVolID composes migration handle for RBD PV
+// mig_mons-afcca55bc1bdd3f479be1e8281c13ab1_image-e0b45b52-7e09-47d3-8f1b-806995fa4412_7265706c696361706f6f6c
+func composeMigVolID(mons string, pool string, image string) string {
+	clusterIDInHandle := md5.Sum([]byte(mons))
+	clusterField := monsPfx + fmt.Sprintf("%x", clusterIDInHandle)
+	poolHashInHandle := hex.EncodeToString([]byte(pool))
+	imageHashInHandle := strings.Split(image, defaultIntreeImagePfx)[1]
+	imageField := imgPfx + imageHashInHandle
+	volHash := strings.Join([]string{migVolPfx, clusterField, imageField, poolHashInHandle}, "_")
+	return volHash
+}
+
+// fillVolAttrsForRequest fill the volume attributes for node operations
+func fillVolAttrsForRequest(pv *v1.PersistentVolume, volumeAttributes map[string]string) error {
+	if pv == nil || pv.Spec.RBD == nil {
+		return fmt.Errorf("pv is nil or RBD Volume not defined on pv")
+	}
+	volumeAttributes[imgNameKey] = pv.Spec.RBD.RBDImage
+	volumeAttributes[poolKey] = pv.Spec.RBD.RBDPool
+	volumeAttributes[imgFeatureKey] = pv.Annotations[imgFeatureKey]
+	volumeAttributes[imgFmtKey] = pv.Annotations[imgFmtKey]
+	volumeAttributes[journalPoolKey] = pv.Annotations[journalPoolKey]
+	volumeAttributes[defaultMigKey] = defaultMigStaticVal
+	volumeAttributes["tryOtherMounters"] = defaultMigStaticVal
+	return nil
+}
+
+// fillAnnotationsFromCSISource capture required information from csi source
+func fillAnnotationsFromCSISource(pv *v1.PersistentVolume, csiSource *v1.CSIPersistentVolumeSource) {
+	pv.Annotations[CSIRBDVolHandleAnnKey] = csiSource.VolumeHandle
+	pv.Annotations[clusterIDKey] = csiSource.VolumeAttributes[clusterIDKey]
+	pv.Annotations[journalPoolKey] = csiSource.VolumeAttributes[journalPoolKey]
+	pv.Annotations[imgFeatureKey] = csiSource.VolumeAttributes[imgFeatureKey]
+	pv.Annotations[imgFmtKey] = csiSource.VolumeAttributes[imgFmtKey]
+}

staging/src/k8s.io/csi-translation-lib/plugins/rbd_test.go

@@ -0,0 +1,454 @@
+/*
+Copyright 2021 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 plugins
+
+import (
+	v1 "k8s.io/api/core/v1"
+	storage "k8s.io/api/storage/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"reflect"
+	"testing"
+)
+
+func TestTranslateRBDInTreeStorageClassToCSI(t *testing.T) {
+	translator := NewRBDCSITranslator()
+	testCases := []struct {
+		name     string
+		inTreeSC *storage.StorageClass
+		csiSC    *storage.StorageClass
+		errorExp bool
+	}{
+		{
+			name: "correct",
+			inTreeSC: &storage.StorageClass{
+				Provisioner: RBDVolumePluginName,
+				Parameters: map[string]string{
+					"adminId":              "kubeadmin",
+					"monitors":             "10.70.53.126:6789,10.70.53.156:6789",
+					"pool":                 "replicapool",
+					"adminSecretName":      "ceph-admin-secret",
+					"adminSecretNamespace": "default",
+				},
+			},
+			csiSC: &storage.StorageClass{
+				Provisioner: RBDDriverName,
+				Parameters: map[string]string{
+					"adminId":   "kubeadmin",
+					"pool":      "replicapool",
+					"migration": "true",
+					"clusterID": "7982de6a23b77bce50b1ba9f2e879cce",
+					"monitors":  "10.70.53.126:6789,10.70.53.156:6789",
+					"csi.storage.k8s.io/controller-expand-secret-name":      "ceph-admin-secret",
+					"csi.storage.k8s.io/controller-expand-secret-namespace": "default",
+					"csi.storage.k8s.io/node-stage-secret-name":             "ceph-admin-secret",
+					"csi.storage.k8s.io/node-stage-secret-namespace":        "default",
+					"csi.storage.k8s.io/provisioner-secret-name":            "ceph-admin-secret",
+					"csi.storage.k8s.io/provisioner-secret-namespace":       "default",
+				},
+			},
+			errorExp: false,
+		},
+		{
+			name: "missing monitor",
+			inTreeSC: &storage.StorageClass{
+				Provisioner: RBDVolumePluginName,
+				Parameters: map[string]string{
+					"adminId":              "kubeadmin",
+					"monitors":             "",
+					"pool":                 "replicapool",
+					"adminSecretName":      "ceph-admin-secret",
+					"adminSecretNamespace": "default",
+				},
+			},
+			csiSC:    nil,
+			errorExp: true,
+		},
+		{
+			name: "monitor unavailable",
+			inTreeSC: &storage.StorageClass{
+				Provisioner: RBDVolumePluginName,
+				Parameters: map[string]string{
+					"adminId":              "kubeadmin",
+					"pool":                 "replicapool",
+					"adminSecretName":      "ceph-admin-secret",
+					"adminSecretNamespace": "default",
+				},
+			},
+			csiSC:    nil,
+			errorExp: true,
+		},
+		{
+			name: "admin secret unavailable",
+			inTreeSC: &storage.StorageClass{
+				Provisioner: RBDVolumePluginName,
+				Parameters: map[string]string{
+					"adminId":              "kubeadmin",
+					"pool":                 "replicapool",
+					"monitors":             "10.70.53.126:6789,10.70.53.156:6789",
+					"adminSecretNamespace": "default",
+				},
+			},
+			csiSC:    nil,
+			errorExp: true,
+		},
+
+		{
+			name:     "nil, err expected",
+			inTreeSC: nil,
+			csiSC:    nil,
+			errorExp: true,
+		},
+	}
+	for _, tc := range testCases {
+		t.Logf("Testing %v", tc.name)
+		result, err := translator.TranslateInTreeStorageClassToCSI(tc.inTreeSC)
+		if err != nil && !tc.errorExp {
+			t.Errorf("Did not expect error but got: %v", err)
+		}
+		if err == nil && tc.errorExp {
+			t.Errorf("Expected error, but did not get one.")
+		}
+		if !reflect.DeepEqual(result, tc.csiSC) {
+			t.Errorf("Got parameters: %v\n, expected :%v", result, tc.csiSC)
+		}
+	}
+}
+
+func TestTranslateRBDInTreeInlineVolumeToCSI(t *testing.T) {
+	translator := NewRBDCSITranslator()
+	testCases := []struct {
+		name        string
+		inLine      *v1.Volume
+		csiVol      *v1.PersistentVolume
+		errExpected bool
+	}{
+		{
+			name: "normal",
+			inLine: &v1.Volume{
+				Name: "rbdVol",
+				VolumeSource: v1.VolumeSource{
+					RBD: &v1.RBDVolumeSource{
+						CephMonitors: []string{"10.70.53.126:6789,10.70.53.156:6789"},
+						RBDPool:      "replicapool",
+						RBDImage:     "kubernetes-dynamic-pvc-e4111eb6-4088-11ec-b823-0242ac110003",
+						RadosUser:    "admin",
+						SecretRef:    &v1.LocalObjectReference{Name: "ceph-secret"},
+						FSType:       "ext4",
+						ReadOnly:     false,
+					},
+				},
+			},
+
+			csiVol: &v1.PersistentVolume{
+				ObjectMeta: metav1.ObjectMeta{
+					// Must be unique per disk as it is used as the unique part of the
+					// staging path
+					Name: "rbd.csi.ceph.com-kubernetes-dynamic-pvc-e4111eb6-4088-11ec-b823-0242ac110003",
+				},
+				Spec: v1.PersistentVolumeSpec{
+					PersistentVolumeSource: v1.PersistentVolumeSource{
+						CSI: &v1.CSIPersistentVolumeSource{
+							Driver:       RBDDriverName,
+							VolumeHandle: "kubernetes-dynamic-pvc-e4111eb6-4088-11ec-b823-0242ac110003",
+							FSType:       "ext4",
+							VolumeAttributes: map[string]string{
+								"clusterID":     "7982de6a23b77bce50b1ba9f2e879cce",
+								"imageFeatures": "layering",
+								"pool":          "replicapool",
+								"staticVolume":  "true",
+							},
+							NodeStageSecretRef:        &v1.SecretReference{Name: "ceph-secret", Namespace: "ns"},
+							ControllerExpandSecretRef: &v1.SecretReference{Name: "ceph-secret", Namespace: "ns"},
+						},
+					},
+					AccessModes: []v1.PersistentVolumeAccessMode{
+						v1.ReadWriteOnce,
+					},
+				},
+			},
+			errExpected: false,
+		},
+		{
+			name:        "nil",
+			inLine:      nil,
+			csiVol:      nil,
+			errExpected: true,
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Logf("Testing %v", tc.name)
+		result, err := translator.TranslateInTreeInlineVolumeToCSI(tc.inLine, "ns")
+		if err != nil && !tc.errExpected {
+			t.Errorf("Did not expect error but got: %v", err)
+		}
+		if err == nil && tc.errExpected {
+			t.Errorf("Expected error, but did not get one.")
+		}
+		if !reflect.DeepEqual(result, tc.csiVol) {
+			t.Errorf("Got parameters: %v\n, expected :%v", result, tc.csiVol)
+		}
+	}
+}
+
+func TestTranslateRBDInTreePVToCSI(t *testing.T) {
+	translator := NewRBDCSITranslator()
+	testCases := []struct {
+		name        string
+		inTree      *v1.PersistentVolume
+		csi         *v1.PersistentVolume
+		errExpected bool
+	}{
+		{
+			name: "no RBD volume",
+			inTree: &v1.PersistentVolume{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: "rbd.csi.ceph.com",
+				},
+				Spec: v1.PersistentVolumeSpec{
+					AccessModes: []v1.PersistentVolumeAccessMode{
+						v1.ReadWriteOnce,
+					},
+					ClaimRef: &v1.ObjectReference{
+						Name:      "test-pvc",
+						Namespace: "default",
+					},
+				},
+			},
+			csi:         nil,
+			errExpected: true,
+		},
+		{
+			name: "normal",
+			inTree: &v1.PersistentVolume{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: RBDDriverName,
+				},
+				Spec: v1.PersistentVolumeSpec{
+					AccessModes: []v1.PersistentVolumeAccessMode{
+						v1.ReadWriteOnce,
+					},
+					ClaimRef: &v1.ObjectReference{
+						Name:      "test-pvc",
+						Namespace: "default",
+					},
+					PersistentVolumeSource: v1.PersistentVolumeSource{
+						RBD: &v1.RBDPersistentVolumeSource{
+							CephMonitors: []string{"10.70.53.126:6789"},
+							RBDPool:      "replicapool",
+							RBDImage:     "kubernetes-dynamic-pvc-e4111eb6-4088-11ec-b823-0242ac110003",
+							RadosUser:    "admin",
+							FSType:       "ext4",
+							ReadOnly:     false,
+							SecretRef: &v1.SecretReference{
+								Name:      "ceph-secret",
+								Namespace: "default",
+							},
+						},
+					},
+				},
+			},
+			csi: &v1.PersistentVolume{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: RBDDriverName,
+				},
+				Spec: v1.PersistentVolumeSpec{
+					AccessModes: []v1.PersistentVolumeAccessMode{
+						v1.ReadWriteOnce,
+					},
+					ClaimRef: &v1.ObjectReference{
+						Name:      "test-pvc",
+						Namespace: "default",
+					},
+					PersistentVolumeSource: v1.PersistentVolumeSource{
+						CSI: &v1.CSIPersistentVolumeSource{
+							Driver:       RBDDriverName,
+							VolumeHandle: "mig_mons-b7f67366bb43f32e07d8a261a7840da9_image-e4111eb6-4088-11ec-b823-0242ac110003_7265706c696361706f6f6c",
+							ReadOnly:     false,
+							FSType:       "ext4",
+							VolumeAttributes: map[string]string{
+								"clusterID":        "b7f67366bb43f32e07d8a261a7840da9",
+								"imageFeatures":    "layering",
+								"imageFormat":      "",
+								"imageName":        "kubernetes-dynamic-pvc-e4111eb6-4088-11ec-b823-0242ac110003",
+								"journalPool":      "",
+								"migration":        "true",
+								"pool":             "replicapool",
+								"staticVolume":     "true",
+								"tryOtherMounters": "true",
+							},
+							NodeStageSecretRef: &v1.SecretReference{
+								Name:      "ceph-secret",
+								Namespace: "default",
+							},
+							ControllerExpandSecretRef: &v1.SecretReference{
+								Name:      "ceph-secret",
+								Namespace: "default",
+							},
+						},
+					},
+				},
+			},
+			errExpected: false,
+		},
+		{
+			name:        "nil PV",
+			inTree:      nil,
+			csi:         nil,
+			errExpected: true,
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Logf("Testing %v", tc.name)
+		result, err := translator.TranslateInTreePVToCSI(tc.inTree)
+		if err != nil && !tc.errExpected {
+			t.Errorf("Did not expect error but got: %v", err)
+		}
+		if err == nil && tc.errExpected {
+			t.Errorf("Expected error, but did not get one.")
+		}
+		if !reflect.DeepEqual(result, tc.csi) {
+			t.Errorf("Got parameters: %v\n, expected :%v", result, tc.csi)
+		}
+	}
+}
+func TestTranslateCSIPvToInTree(t *testing.T) {
+	translator := NewRBDCSITranslator()
+
+	testCases := []struct {
+		name        string
+		csi         *v1.PersistentVolume
+		inTree      *v1.PersistentVolume
+		errExpected bool
+	}{
+		{
+			name: "no CSI section",
+			csi: &v1.PersistentVolume{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: RBDDriverName,
+				},
+				Spec: v1.PersistentVolumeSpec{
+					AccessModes: []v1.PersistentVolumeAccessMode{
+						v1.ReadWriteOnce,
+					},
+					ClaimRef: &v1.ObjectReference{
+						Name:      "test-pvc",
+						Namespace: "default",
+					},
+				},
+			},
+			inTree:      nil,
+			errExpected: true,
+		},
+		{
+			name: "normal",
+			csi: &v1.PersistentVolume{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: RBDDriverName,
+				},
+				Spec: v1.PersistentVolumeSpec{
+					AccessModes: []v1.PersistentVolumeAccessMode{
+						v1.ReadWriteOnce,
+					},
+					ClaimRef: &v1.ObjectReference{
+						Name:      "test-pvc",
+						Namespace: "default",
+					},
+					PersistentVolumeSource: v1.PersistentVolumeSource{
+						CSI: &v1.CSIPersistentVolumeSource{
+							Driver:       RBDDriverName,
+							VolumeHandle: "dummy",
+							ReadOnly:     false,
+							FSType:       "ext4",
+							VolumeAttributes: map[string]string{
+								"clusterID":        "b7f67366bb43f32e07d8a261a7840da9",
+								"imageFeatures":    "layering",
+								"imageFormat":      "1",
+								"imageName":        "kubernetes-dynamic-pvc-e4111eb6-4088-11ec-b823-0242ac110003",
+								"journalPool":      "some",
+								"migration":        "true",
+								"pool":             "replicapool",
+								"staticVolume":     "true",
+								"tryOtherMounters": "true",
+							},
+							NodeStageSecretRef: &v1.SecretReference{
+								Name:      "ceph-secret",
+								Namespace: "default",
+							},
+						},
+					},
+				},
+			},
+			inTree: &v1.PersistentVolume{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: RBDDriverName,
+					Annotations: map[string]string{
+						"clusterID":                      "b7f67366bb43f32e07d8a261a7840da9",
+						"imageFeatures":                  "layering",
+						"imageFormat":                    "1",
+						"journalPool":                    "some",
+						"rbd.csi.ceph.com/volume-handle": "dummy",
+					},
+				},
+				Spec: v1.PersistentVolumeSpec{
+					AccessModes: []v1.PersistentVolumeAccessMode{
+						v1.ReadWriteOnce,
+					},
+					ClaimRef: &v1.ObjectReference{
+						Name:      "test-pvc",
+						Namespace: "default",
+					},
+					PersistentVolumeSource: v1.PersistentVolumeSource{
+						RBD: &v1.RBDPersistentVolumeSource{
+							CephMonitors: nil,
+							RBDPool:      "replicapool",
+							RBDImage:     "kubernetes-dynamic-pvc-e4111eb6-4088-11ec-b823-0242ac110003",
+							RadosUser:    "admin",
+							FSType:       "ext4",
+							ReadOnly:     false,
+							SecretRef: &v1.SecretReference{
+								Name:      "ceph-secret",
+								Namespace: "default",
+							},
+						},
+					},
+				},
+			},
+			errExpected: false,
+		},
+		{
+			name:        "nil PV",
+			inTree:      nil,
+			csi:         nil,
+			errExpected: true,
+		},
+	}
+	for _, tc := range testCases {
+		t.Logf("Testing %v", tc.name)
+		result, err := translator.TranslateCSIPVToInTree(tc.csi)
+		if err != nil && !tc.errExpected {
+			t.Errorf("Did not expect error but got: %v", err)
+		}
+		if err == nil && tc.errExpected {
+			t.Errorf("Expected error, but did not get one.")
+		}
+		if !reflect.DeepEqual(result, tc.inTree) {
+			t.Errorf("Got parameters: %v\n, expected :%v", result, tc.inTree)
+		}
+	}
+}

staging/src/k8s.io/csi-translation-lib/translate.go

@@ -33,6 +33,7 @@ var (
 		plugins.AzureDiskDriverName: plugins.NewAzureDiskCSITranslator(),
 		plugins.AzureFileDriverName: plugins.NewAzureFileCSITranslator(),
 		plugins.VSphereDriverName:   plugins.NewvSphereCSITranslator(),
+		plugins.RBDDriverName:       plugins.NewRBDCSITranslator(),
 	}
 )
 

staging/src/k8s.io/csi-translation-lib/translate_test.go

@@ -439,6 +439,12 @@ func generateUniqueVolumeSource(driverName string) (v1.VolumeSource, error) {
 				FSType:     "ext4",
 			},
 		}, nil
+	case plugins.RBDDriverName:
+		return v1.VolumeSource{
+			RBD: &v1.RBDVolumeSource{
+				RBDImage: string(uuid.NewUUID()),
+			},
+		}, nil
 	default:
 		return v1.VolumeSource{}, fmt.Errorf("couldn't find logic for driver: %v", driverName)
 	}
@@ -460,6 +466,11 @@ func TestPluginNameMappings(t *testing.T) {
 			inTreePluginName: "kubernetes.io/aws-ebs",
 			csiPluginName:    "ebs.csi.aws.com",
 		},
+		{
+			name:             "RBD plugin name",
+			inTreePluginName: "kubernetes.io/rbd",
+			csiPluginName:    "rbd.csi.ceph.com",
+		},
 	}
 	for _, test := range testCases {
 		t.Logf("Testing %v", test.name)