feat: Add rados namespace support to rbd plugin

e2e/rbd.go

@@ -958,6 +958,100 @@ var _ = Describe("RBD", func() {
 				}
 			})
 
+			By("ensuring all operations will work within a rados namespace", func() {
+				updateConfigMap := func(radosNS string) {
+					radosNamespace = radosNS
+					deleteConfigMap(rbdDirPath)
+					createConfigMap(rbdDirPath, f.ClientSet, f)
+					createRadosNamespace(f)
+
+					// delete csi pods
+					err := deletePodWithLabel("app in (ceph-csi-rbd, csi-rbdplugin, csi-rbdplugin-provisioner)",
+						cephCSINamespace, false)
+					if err != nil {
+						Fail(err.Error())
+					}
+					// wait for csi pods to come up
+					err = waitForDaemonSets(rbdDaemonsetName, cephCSINamespace, f.ClientSet, deployTimeout)
+					if err != nil {
+						Fail(err.Error())
+					}
+					err = waitForDeploymentComplete(rbdDeploymentName, cephCSINamespace, f.ClientSet, deployTimeout)
+					if err != nil {
+						Fail(err.Error())
+					}
+				}
+
+				updateConfigMap("e2e-ns")
+
+				// Create a PVC and Bind it to an app within the namesapce
+				validatePVCAndAppBinding(pvcPath, appPath, f)
+
+				v, err := f.ClientSet.Discovery().ServerVersion()
+				if err != nil {
+					e2elog.Logf("failed to get server version with error %v", err)
+					Fail(err.Error())
+				}
+
+				// Resize Block PVC and check Device size within the namespace
+				// Block PVC resize is supported in kubernetes 1.16+
+				if v.Major > "1" || (v.Major == "1" && v.Minor >= "16") {
+					err = resizePVCAndValidateSize(rawPvcPath, rawAppPath, f)
+					if err != nil {
+						e2elog.Logf("failed to resize block PVC %v", err)
+						Fail(err.Error())
+					}
+				}
+
+				// Create a PVC clone and bind it to an app within the namespace
+				// snapshot beta is only supported from v1.17+
+				if v.Major > "1" || (v.Major == "1" && v.Minor >= "17") {
+					pvc, err := loadPVC(pvcPath)
+					if err != nil {
+						Fail(err.Error())
+					}
+
+					pvc.Namespace = f.UniqueName
+					e2elog.Logf("The PVC  template %+v", pvc)
+					err = createPVCAndvalidatePV(f.ClientSet, pvc, deployTimeout)
+					if err != nil {
+						Fail(err.Error())
+					}
+					// validate created backend rbd images
+					images := listRBDImages(f)
+					if len(images) != 1 {
+						e2elog.Logf("backend image count %d expected image count %d", len(images), 1)
+						Fail("validate backend image failed")
+					}
+					snap := getSnapshot(snapshotPath)
+					snap.Namespace = f.UniqueName
+					snap.Spec.Source.PersistentVolumeClaimName = &pvc.Name
+					err = createSnapshot(&snap, deployTimeout)
+					if err != nil {
+						Fail(err.Error())
+					}
+					expectedImages := len(images) + 1
+					images = listRBDImages(f)
+					if len(images) != expectedImages {
+						e2elog.Logf("backend images not matching kubernetes resource count,image count %d kubernetes resource count %d", len(images), 2)
+						Fail("validate backend images failed")
+					}
+
+					validatePVCAndAppBinding(pvcClonePath, appClonePath, f)
+
+					err = deleteSnapshot(&snap, deployTimeout)
+					if err != nil {
+						Fail(err.Error())
+					}
+					err = deletePVCAndValidatePV(f.ClientSet, pvc, deployTimeout)
+					if err != nil {
+						Fail(err.Error())
+					}
+				}
+
+				updateConfigMap("")
+			})
+
 			By("Mount pvc as readonly in pod", func() {
 				// create pvc and bind it to an app
 				pvc, err := loadPVC(pvcPath)

e2e/staticpvc.go

@@ -99,7 +99,7 @@ func validateRBDStaticPV(f *framework.Framework, appPath string, isBlock bool) e
 	fsID = strings.Trim(fsID, "\n")
 	size := "4Gi"
 	// create rbd image
-	cmd := fmt.Sprintf("rbd create %s --size=%d --pool=%s --image-feature=layering", rbdImageName, 4096, defaultRBDPool)
+	cmd := fmt.Sprintf("rbd create %s --size=%d --image-feature=layering %s", rbdImageName, 4096, rbdOptions(defaultRBDPool))
 
 	_, e = execCommandInToolBoxPod(f, cmd, rookNamespace)
 	if e != "" {
@@ -109,6 +109,9 @@ func validateRBDStaticPV(f *framework.Framework, appPath string, isBlock bool) e
 	opt["imageFeatures"] = "layering"
 	opt["pool"] = defaultRBDPool
 	opt["staticVolume"] = "true"
+	if radosNamespace != "" {
+		opt["radosNamespace"] = radosNamespace
+	}
 
 	pv := getStaticPV(pvName, rbdImageName, size, "csi-rbd-secret", cephCSINamespace, sc, "rbd.csi.ceph.com", isBlock, opt)
 
@@ -151,7 +154,7 @@ func validateRBDStaticPV(f *framework.Framework, appPath string, isBlock bool) e
 		return err
 	}
 
-	cmd = fmt.Sprintf("rbd rm %s --pool=%s", rbdImageName, defaultRBDPool)
+	cmd = fmt.Sprintf("rbd rm %s %s", rbdImageName, rbdOptions(defaultRBDPool))
 	execCommandInToolBoxPod(f, cmd, rookNamespace)
 	return nil
 }

e2e/utils.go

@@ -59,6 +59,7 @@ var (
 	upgradeVersion   string
 	cephCSINamespace string
 	rookNamespace    string
+	radosNamespace   string
 	ns               string
 	vaultAddr        string
 	poll             = 2 * time.Second
@@ -330,6 +331,25 @@ func createRBDStorageClass(c kubernetes.Interface, f *framework.Framework, scOpt
 	Expect(err).Should(BeNil())
 }
 
+func createRadosNamespace(f *framework.Framework) {
+	stdOut, stdErr := execCommandInToolBoxPod(f,
+		fmt.Sprintf("rbd namespace ls --pool=%s", defaultRBDPool), rookNamespace)
+	Expect(stdErr).Should(BeEmpty())
+	if !strings.Contains(stdOut, radosNamespace) {
+		_, stdErr = execCommandInToolBoxPod(f,
+			fmt.Sprintf("rbd namespace create %s", rbdOptions(defaultRBDPool)), rookNamespace)
+		Expect(stdErr).Should(BeEmpty())
+	}
+	stdOut, stdErr = execCommandInToolBoxPod(f,
+		fmt.Sprintf("rbd namespace ls --pool=%s", rbdTopologyPool), rookNamespace)
+	Expect(stdErr).Should(BeEmpty())
+	if !strings.Contains(stdOut, radosNamespace) {
+		_, stdErr = execCommandInToolBoxPod(f,
+			fmt.Sprintf("rbd namespace create %s", rbdOptions(rbdTopologyPool)), rookNamespace)
+		Expect(stdErr).Should(BeEmpty())
+	}
+}
+
 func deleteConfigMap(pluginPath string) {
 	path := pluginPath + configMap
 	_, err := framework.RunKubectl(cephCSINamespace, "delete", "-f", path, ns)
@@ -351,8 +371,9 @@ func createConfigMap(pluginPath string, c kubernetes.Interface, f *framework.Fra
 	// get mon list
 	mons := getMons(rookNamespace, c)
 	conmap := []util.ClusterInfo{{
-		ClusterID: fsID,
-		Monitors:  mons,
+		ClusterID:      fsID,
+		Monitors:       mons,
+		RadosNamespace: radosNamespace,
 	}}
 	if upgradeTesting {
 		subvolumegroup = "csi"
@@ -780,7 +801,7 @@ func validateEncryptedPVCAndAppBinding(pvcPath, appPath, kms string, f *framewor
 	if err != nil {
 		Fail(err.Error())
 	}
-	rbdImageSpec := fmt.Sprintf("%s/%s", defaultRBDPool, imageData.imageName)
+	rbdImageSpec := imageSpec(defaultRBDPool, imageData.imageName)
 	encryptedState, err := getImageMeta(rbdImageSpec, ".rbd.csi.ceph.com/encrypted", f)
 	if err != nil {
 		Fail(err.Error())
@@ -916,7 +937,8 @@ func deleteBackingCephFSVolume(f *framework.Framework, pvc *v1.PersistentVolumeC
 }
 
 func listRBDImages(f *framework.Framework) []string {
-	stdout, stdErr := execCommandInToolBoxPod(f, fmt.Sprintf("rbd ls --pool=%s --format=json", defaultRBDPool), rookNamespace)
+	stdout, stdErr := execCommandInToolBoxPod(f,
+		fmt.Sprintf("rbd ls --format=json %s", rbdOptions(defaultRBDPool)), rookNamespace)
 	Expect(stdErr).Should(BeEmpty())
 	var imgInfos []string
 
@@ -1003,7 +1025,7 @@ func deleteBackingRBDImage(f *framework.Framework, pvc *v1.PersistentVolumeClaim
 		return err
 	}
 
-	cmd := fmt.Sprintf("rbd rm %s --pool=%s", imageData.imageName, defaultRBDPool)
+	cmd := fmt.Sprintf("rbd rm %s %s", rbdOptions(defaultRBDPool), imageData.imageName)
 	execCommandInToolBoxPod(f, cmd, rookNamespace)
 	return nil
 }
@@ -1132,10 +1154,15 @@ func getPVCImageInfoInPool(f *framework.Framework, pvc *v1.PersistentVolumeClaim
 		return "", err
 	}
 
-	stdOut, stdErr := execCommandInToolBoxPod(f, "rbd info "+pool+"/"+imageData.imageName, rookNamespace)
+	stdOut, stdErr := execCommandInToolBoxPod(f,
+		fmt.Sprintf("rbd info %s", imageSpec(pool, imageData.imageName)), rookNamespace)
 	Expect(stdErr).Should(BeEmpty())
 
-	e2elog.Logf("found image %s in pool %s", imageData.imageName, pool)
+	if radosNamespace != "" {
+		e2elog.Logf("found image %s in pool %s namespace %s", imageData.imageName, pool, radosNamespace)
+	} else {
+		e2elog.Logf("found image %s in pool %s", imageData.imageName, pool)
+	}
 
 	return stdOut, nil
 }
@@ -1165,10 +1192,15 @@ func checkPVCImageJournalInPool(f *framework.Framework, pvc *v1.PersistentVolume
 		return err
 	}
 
-	_, stdErr := execCommandInToolBoxPod(f, "rados listomapkeys -p "+pool+" csi.volume."+imageData.imageID, rookNamespace)
+	_, stdErr := execCommandInToolBoxPod(f,
+		fmt.Sprintf("rados listomapkeys %s csi.volume.%s", rbdOptions(pool), imageData.imageID), rookNamespace)
 	Expect(stdErr).Should(BeEmpty())
 
-	e2elog.Logf("found image journal %s in pool %s", "csi.volume."+imageData.imageID, pool)
+	if radosNamespace != "" {
+		e2elog.Logf("found image journal %s in pool %s namespace %s", "csi.volume."+imageData.imageID, pool, radosNamespace)
+	} else {
+		e2elog.Logf("found image journal %s in pool %s", "csi.volume."+imageData.imageID, pool)
+	}
 
 	return nil
 }
@@ -1179,10 +1211,15 @@ func checkPVCCSIJournalInPool(f *framework.Framework, pvc *v1.PersistentVolumeCl
 		return err
 	}
 
-	_, stdErr := execCommandInToolBoxPod(f, "rados getomapval -p "+pool+" csi.volumes.default csi.volume."+imageData.pvName, rookNamespace)
+	_, stdErr := execCommandInToolBoxPod(f,
+		fmt.Sprintf("rados getomapval %s csi.volumes.default csi.volume.%s", rbdOptions(pool), imageData.pvName), rookNamespace)
 	Expect(stdErr).Should(BeEmpty())
 
-	e2elog.Logf("found CSI journal entry %s in pool %s", "csi.volume."+imageData.pvName, pool)
+	if radosNamespace != "" {
+		e2elog.Logf("found CSI journal entry %s in pool %s namespace %s", "csi.volume."+imageData.pvName, pool, radosNamespace)
+	} else {
+		e2elog.Logf("found CSI journal entry %s in pool %s", "csi.volume."+imageData.pvName, pool)
+	}
 
 	return nil
 }
@@ -1291,3 +1328,17 @@ func validateSubvolumegroup(f *framework.Framework, subvolgrp string) error {
 	}
 	return nil
 }
+
+func imageSpec(pool, image string) string {
+	if radosNamespace != "" {
+		return pool + "/" + radosNamespace + "/" + image
+	}
+	return pool + "/" + image
+}
+
+func rbdOptions(pool string) string {
+	if radosNamespace != "" {
+		return "--pool=" + pool + " --namespace " + radosNamespace
+	}
+	return "--pool=" + pool
+}

examples/csi-config-map-sample.yaml

@@ -12,6 +12,12 @@ kind: ConfigMap
 # each such cluster in use.
 # To add more clusters or edit MON addresses in an existing configmap, use
 # the `kubectl replace` command.
+# The <rados-namespace> is optional and represents a radosNamespace in the pool.
+# If any given all the of the rbd images, snapshots and other metadata
+# will be store within the radosNamespace.
+# NOTE: The given radosNamespace must already exists in the pool.
+# NOTE: Make sure you don't add radosNamespace option to a currently in use
+# configuration as it will cause issues.
 # The field "cephFS.subvolumeGroup" is optional and defaults to "csi".
 # NOTE: Changes to the configmap is automatically updated in the running pods,
 # thus restarting existing pods using the configmap is NOT required on edits
@@ -21,6 +27,7 @@ data:
     [
       {
         "clusterID": "<cluster-id>",
+        "radosNamespace": "<rados-namespace>",
         "monitors": [
           "<MONValue1>",
           "<MONValue2>",

internal/cephfs/fsjournal.go

@@ -65,7 +65,8 @@ func checkVolExists(ctx context.Context,
 	sID *snapshotIdentifier,
 	cr *util.Credentials) (*volumeIdentifier, error) {
 	var vid volumeIdentifier
-	j, err := volJournal.Connect(volOptions.Monitors, cr)
+	// Connect to cephfs' default radosNamespace (csi)
+	j, err := volJournal.Connect(volOptions.Monitors, radosNamespace, cr)
 	if err != nil {
 		return nil, err
 	}
@@ -174,7 +175,8 @@ func undoVolReservation(ctx context.Context, volOptions *volumeOptions, vid volu
 	}
 	defer cr.DeleteCredentials()
 
-	j, err := volJournal.Connect(volOptions.Monitors, cr)
+	// Connect to cephfs' default radosNamespace (csi)
+	j, err := volJournal.Connect(volOptions.Monitors, radosNamespace, cr)
 	if err != nil {
 		return err
 	}
@@ -220,7 +222,8 @@ func reserveVol(ctx context.Context, volOptions *volumeOptions, secret map[strin
 		return nil, err
 	}
 
-	j, err := volJournal.Connect(volOptions.Monitors, cr)
+	// Connect to cephfs' default radosNamespace (csi)
+	j, err := volJournal.Connect(volOptions.Monitors, radosNamespace, cr)
 	if err != nil {
 		return nil, err
 	}
@@ -256,7 +259,8 @@ func reserveSnap(ctx context.Context, volOptions *volumeOptions, parentSubVolNam
 		err       error
 	)
 
-	j, err := snapJournal.Connect(volOptions.Monitors, cr)
+	// Connect to cephfs' default radosNamespace (csi)
+	j, err := snapJournal.Connect(volOptions.Monitors, radosNamespace, cr)
 	if err != nil {
 		return nil, err
 	}
@@ -285,7 +289,8 @@ func reserveSnap(ctx context.Context, volOptions *volumeOptions, parentSubVolNam
 
 // undoSnapReservation is a helper routine to undo a name reservation for a CSI SnapshotName.
 func undoSnapReservation(ctx context.Context, volOptions *volumeOptions, vid snapshotIdentifier, snapName string, cr *util.Credentials) error {
-	j, err := snapJournal.Connect(volOptions.Monitors, cr)
+	// Connect to cephfs' default radosNamespace (csi)
+	j, err := snapJournal.Connect(volOptions.Monitors, radosNamespace, cr)
 	if err != nil {
 		return err
 	}
@@ -316,7 +321,8 @@ func checkSnapExists(
 	parentSubVolName string,
 	snap *cephfsSnapshot,
 	cr *util.Credentials) (*snapshotIdentifier, *snapshotInfo, error) {
-	j, err := snapJournal.Connect(volOptions.Monitors, cr)
+	// Connect to cephfs' default radosNamespace (csi)
+	j, err := snapJournal.Connect(volOptions.Monitors, radosNamespace, cr)
 	if err != nil {
 		return nil, nil, err
 	}

internal/cephfs/volumeoptions.go

@@ -261,7 +261,8 @@ func newVolumeOptionsFromVolID(ctx context.Context, volID string, volOpt, secret
 		return nil, nil, err
 	}
 
-	j, err := volJournal.Connect(volOptions.Monitors, cr)
+	// Connect to cephfs' default radosNamespace (csi)
+	j, err := volJournal.Connect(volOptions.Monitors, radosNamespace, cr)
 	if err != nil {
 		return nil, nil, err
 	}
@@ -448,7 +449,8 @@ func newSnapshotOptionsFromID(ctx context.Context, snapID string, cr *util.Crede
 		return &volOptions, nil, &sid, err
 	}
 
-	j, err := snapJournal.Connect(volOptions.Monitors, cr)
+	// Connect to cephfs' default radosNamespace (csi)
+	j, err := snapJournal.Connect(volOptions.Monitors, radosNamespace, cr)
 	if err != nil {
 		return &volOptions, nil, &sid, err
 	}

internal/journal/voljournal.go

@@ -235,7 +235,8 @@ type Connection struct {
 }
 
 // Connect establishes a new connection to a ceph cluster for journal metadata.
-func (cj *Config) Connect(monitors string, cr *util.Credentials) (*Connection, error) {
+func (cj *Config) Connect(monitors, namespace string, cr *util.Credentials) (*Connection, error) {
+	cj.namespace = namespace
 	cc := &util.ClusterConnection{}
 	if err := cc.Connect(monitors, cr); err != nil {
 		return nil, err

internal/rbd/clone.go

@@ -164,7 +164,7 @@ func (rv *rbdVolume) createCloneFromImage(ctx context.Context, parentVol *rbdVol
 	)
 	var j = &journal.Connection{}
 
-	j, err = volJournal.Connect(rv.Monitors, rv.conn.Creds)
+	j, err = volJournal.Connect(rv.Monitors, rv.RadosNamespace, rv.conn.Creds)
 	if err != nil {
 		return status.Error(codes.Internal, err.Error())
 	}