rbd: Add rados namespace support

Make sure to operate within the namespace if any given when dealing
with rbd images and snapshots and volume and snapshot journals.

Re-run the entire e2e tests one more time using a namespace.

Closes: #798

Signed-off-by: Mehdy Khoshnoody <[email protected]>

.travis.yml

@@ -116,13 +116,27 @@ jobs:
         - make image-cephcsi || travis_terminate 1;
         - scripts/travis-functest.sh v1.16.9 || travis_terminate 1;
 
+    - stage: e2e testing
+      name: cephcsi with kube 1.16.9 and using rbd namespace
+      script:
+        - scripts/skip-doc-change.sh || travis_terminate 0;
+        - make image-cephcsi || travis_terminate 1;
+        - scripts/travis-functest.sh v1.16.9 rbd-ns || travis_terminate 1;
+
     - stage: e2e testing
       name: cephcsi with kube 1.17.5
       script:
         - scripts/skip-doc-change.sh || travis_terminate 0;
         - make image-cephcsi || travis_terminate 1;
         - scripts/travis-functest.sh v1.17.5 || travis_terminate 1;
 
+    - stage: e2e testing
+      name: cephcsi with kube 1.17.5 and using rbd namespace
+      script:
+        - scripts/skip-doc-change.sh || travis_terminate 0;
+        - make image-cephcsi || travis_terminate 1;
+        - scripts/travis-functest.sh v1.17.5 rbd-ns || travis_terminate 1;
+
     - stage: e2e testing
       name: cephcsi helm charts with kube 1.17.5
       script:

e2e/e2e_test.go

@@ -22,6 +22,7 @@ func init() {
 	flag.BoolVar(&deployRBD, "deploy-rbd", true, "deploy rbd csi driver")
 	flag.StringVar(&cephCSINamespace, "cephcsi-namespace", defaultNs, "namespace in which cephcsi deployed")
 	flag.StringVar(&rookNamespace, "rook-namespace", "rook-ceph", "namespace in which rook is deployed")
+	flag.StringVar(&rbdNamespace, "rbd-namespace", "", "namespace in which rbd images are stored")
 	setDefaultKubeconfig()
 
 	// Register framework flags, then handle flags

e2e/utils.go

@@ -51,6 +51,7 @@ var (
 	deployRBD        bool
 	cephCSINamespace string
 	rookNamespace    string
+	rbdNamespace     string
 	ns               string
 	vaultAddr        string
 	poll             = 2 * time.Second
@@ -369,6 +370,7 @@ func deleteConfigMap(pluginPath string) {
 type clusterInfo struct {
 	ClusterID string   `json:"clusterID"`
 	Monitors  []string `json:"monitors"`
+	Namespace string   `json:"namespace"`
 	CephFS    struct {
 		SubvolumeGroup string `json:"subvolumeGroup"`
 	} `json:"cephFS"`
@@ -386,11 +388,21 @@ func createConfigMap(pluginPath string, c kubernetes.Interface, f *framework.Fra
 	Expect(stdErr).Should(BeEmpty())
 	// remove new line present in fsID
 	fsID = strings.Trim(fsID, "\n")
+	// create rbd namespace if required
+	if rbdNamespace != "" {
+		stdOut, stdErr := execCommandInPod(f, "rbd namespace ls --pool=replicapool", rookNamespace, &opt)
+		Expect(stdErr).Should(BeEmpty())
+		if !strings.Contains(stdOut, rbdNamespace) {
+			_, stdErr := execCommandInPod(f, "rbd namespace create "+rbdOptions("replicapool"), rookNamespace, &opt)
+			Expect(stdErr).Should(BeEmpty())
+		}
+	}
 	// get mon list
 	mons := getMons(rookNamespace, c)
 	conmap := []clusterInfo{{
 		ClusterID: fsID,
 		Monitors:  mons,
+		Namespace: rbdNamespace,
 	}}
 	conmap[0].CephFS.SubvolumeGroup = "e2e"
 	data, err := json.Marshal(conmap)
@@ -821,7 +833,7 @@ func validateEncryptedPVCAndAppBinding(pvcPath, appPath, kms string, f *framewor
 	if err != nil {
 		Fail(err.Error())
 	}
-	rbdImageSpec := fmt.Sprintf("replicapool/%s", imageData.imageName)
+	rbdImageSpec := imageSpec("replicapool", imageData.imageName)
 	encryptedState, err := getImageMeta(rbdImageSpec, ".rbd.csi.ceph.com/encrypted", f)
 	if err != nil {
 		Fail(err.Error())
@@ -1030,7 +1042,7 @@ func listRBDImages(f *framework.Framework) []string {
 	opt := metav1.ListOptions{
 		LabelSelector: "app=rook-ceph-tools",
 	}
-	stdout, stdErr := execCommandInPod(f, "rbd ls --pool=replicapool --format=json", rookNamespace, &opt)
+	stdout, stdErr := execCommandInPod(f, "rbd ls --format=json "+rbdOptions("replicapool"), rookNamespace, &opt)
 	Expect(stdErr).Should(BeEmpty())
 	var imgInfos []string
 
@@ -1113,7 +1125,7 @@ func deleteBackingRBDImage(f *framework.Framework, pvc *v1.PersistentVolumeClaim
 		LabelSelector: "app=rook-ceph-tools",
 	}
 
-	cmd := fmt.Sprintf("rbd rm %s --pool=replicapool", imageData.imageName)
+	cmd := fmt.Sprintf("rbd rm %s "+rbdOptions("replicapool"), imageData.imageName)
 	execCommandInPod(f, cmd, rookNamespace, &opt)
 	return nil
 }
@@ -1249,10 +1261,14 @@ func getPVCImageInfoInPool(f *framework.Framework, pvc *v1.PersistentVolumeClaim
 		LabelSelector: "app=rook-ceph-tools",
 	}
 
-	stdOut, stdErr := execCommandInPod(f, "rbd info "+pool+"/"+imageData.imageName, rookNamespace, &opt)
+	stdOut, stdErr := execCommandInPod(f, "rbd info "+imageSpec(pool, imageData.imageName), rookNamespace, &opt)
 	Expect(stdErr).Should(BeEmpty())
 
-	e2elog.Logf("found image %s in pool %s", imageData.imageName, pool)
+	if rbdNamespace != "" {
+		e2elog.Logf("found image %s in pool %s namespace %s", imageData.imageName, pool, rbdNamespace)
+	} else {
+		e2elog.Logf("found image %s in pool %s", imageData.imageName, pool)
+	}
 
 	return stdOut, nil
 }
@@ -1285,11 +1301,14 @@ func checkPVCImageJournalInPool(f *framework.Framework, pvc *v1.PersistentVolume
 	opt := metav1.ListOptions{
 		LabelSelector: "app=rook-ceph-tools",
 	}
-
-	_, stdErr := execCommandInPod(f, "rados listomapkeys -p "+pool+" csi.volume."+imageData.imageID, rookNamespace, &opt)
+	_, stdErr := execCommandInPod(f, "rados listomapkeys "+rbdOptions(pool)+" csi.volume."+imageData.imageID, rookNamespace, &opt)
 	Expect(stdErr).Should(BeEmpty())
 
-	e2elog.Logf("found image journal %s in pool %s", "csi.volume."+imageData.imageID, pool)
+	if rbdNamespace != "" {
+		e2elog.Logf("found image journal %s in pool %s namespace %s", "csi.volume."+imageData.imageID, pool, rbdNamespace)
+	} else {
+		e2elog.Logf("found image journal %s in pool %s", "csi.volume."+imageData.imageID, pool)
+	}
 
 	return nil
 }
@@ -1304,10 +1323,14 @@ func checkPVCCSIJournalInPool(f *framework.Framework, pvc *v1.PersistentVolumeCl
 		LabelSelector: "app=rook-ceph-tools",
 	}
 
-	_, stdErr := execCommandInPod(f, "rados getomapval -p "+pool+" csi.volumes.default csi.volume."+imageData.pvName, rookNamespace, &opt)
+	_, stdErr := execCommandInPod(f, "rados getomapval "+rbdOptions(pool)+" csi.volumes.default csi.volume."+imageData.pvName, rookNamespace, &opt)
 	Expect(stdErr).Should(BeEmpty())
 
-	e2elog.Logf("found CSI journal entry %s in pool %s", "csi.volume."+imageData.pvName, pool)
+	if rbdNamespace != "" {
+		e2elog.Logf("found CSI journal entry %s in pool %s namespace %s", "csi.volume."+imageData.pvName, pool, rbdNamespace)
+	} else {
+		e2elog.Logf("found CSI journal entry %s in pool %s", "csi.volume."+imageData.pvName, pool)
+	}
 
 	return nil
 }
@@ -1365,3 +1388,17 @@ func addTopologyDomainsToDSYaml(template, labels string) string {
 	return strings.ReplaceAll(template, "# - \"--domainlabels=failure-domain/region,failure-domain/zone\"",
 		"- \"--domainlabels="+labels+"\"")
 }
+
+func imageSpec(pool, image string) string {
+	if rbdNamespace != "" {
+		return pool + "/" + rbdNamespace + "/" + image
+	}
+	return pool + "/" + image
+}
+
+func rbdOptions(pool string) string {
+	if rbdNamespace != "" {
+		return "--pool=" + pool + " --namespace " + rbdNamespace
+	}
+	return "--pool=" + pool
+}

examples/csi-config-map-sample.yaml

@@ -12,6 +12,8 @@ kind: ConfigMap
 # each such cluster in use.
 # To add more clusters or edit MON addresses in an existing config map, use
 # the `kubectl replace` command.
+# The <namespace> is optional and represents the rbd namespace in the pool.
+# NOTE: The given namespace must already exists in the pool.
 # The field "cephFS.subvolumeGroup" is optional and defaults to "csi".
 # NOTE: Changes to the config map is automatically updated in the running pods,
 # thus restarting existing pods using the config map is NOT required on edits
@@ -21,6 +23,7 @@ data:
     [
       {
         "clusterID": "<cluster-id>",
+        "namespace": "<namespace>",
         "monitors": [
           "<MONValue1>",
           "<MONValue2>",

internal/rbd/controllerserver.go

@@ -71,6 +71,9 @@ func (cs *ControllerServer) validateVolumeReq(ctx context.Context, req *csi.Crea
 	if value, ok := options["dataPool"]; ok && value == "" {
 		return status.Error(codes.InvalidArgument, "empty datapool name to provision volume from")
 	}
+	if value, ok := options["namespace"]; ok && value == "" {
+		return status.Error(codes.InvalidArgument, "empty namespace name to provision volume from")
+	}
 	if value, ok := options["volumeNamePrefix"]; ok && value == "" {
 		return status.Error(codes.InvalidArgument, "empty volume name prefix to provision volume from")
 	}
@@ -141,6 +144,7 @@ func buildCreateVolumeResponse(ctx context.Context, req *csi.CreateVolumeRequest
 	volumeContext := req.GetParameters()
 	volumeContext["pool"] = rbdVol.Pool
 	volumeContext["journalPool"] = rbdVol.JournalPool
+	volumeContext["namespace"] = rbdVol.Namespace
 	volumeContext["imageName"] = rbdVol.RbdImageName
 	volume := &csi.Volume{
 		VolumeId:      rbdVol.VolID,
@@ -230,11 +234,11 @@ func (cs *ControllerServer) CreateVolume(ctx context.Context, req *csi.CreateVol
 	if rbdVol.Encrypted {
 		err = rbdVol.ensureEncryptionMetadataSet(rbdImageRequiresEncryption)
 		if err != nil {
-			klog.Errorf(util.Log(ctx, "failed to save encryption status, deleting image %s"),
-				rbdVol.RbdImageName)
-			if deleteErr := deleteImage(ctx, rbdVol, cr); err != nil {
-				klog.Errorf(util.Log(ctx, "failed to delete rbd image: %s/%s with error: %v"),
-					rbdVol.Pool, rbdVol.RbdImageName, deleteErr)
+			klog.Errorf(util.Log(ctx, "failed to save encryption status, deleting image %s: %s"),
+				rbdVol.RbdImageName, err)
+			if deleteErr := deleteImage(ctx, rbdVol, cr); deleteErr != nil {
+				klog.Errorf(util.Log(ctx, "failed to delete rbd image: %s with error: %v"),
+					getImageSpec(rbdVol.Pool, rbdVol.Namespace, rbdVol.RbdImageName), deleteErr)
 				return nil, deleteErr
 			}
 			return nil, err
@@ -244,6 +248,7 @@ func (cs *ControllerServer) CreateVolume(ctx context.Context, req *csi.CreateVol
 	volumeContext := req.GetParameters()
 	volumeContext["pool"] = rbdVol.Pool
 	volumeContext["journalPool"] = rbdVol.JournalPool
+	volumeContext["namespace"] = rbdVol.Namespace
 	volumeContext["imageName"] = rbdVol.RbdImageName
 	volume := &csi.Volume{
 		VolumeId:      rbdVol.VolID,
@@ -452,8 +457,8 @@ func (cs *ControllerServer) DeleteVolume(ctx context.Context, req *csi.DeleteVol
 	// Deleting rbd image
 	klog.V(4).Infof(util.Log(ctx, "deleting image %s"), rbdVol.RbdImageName)
 	if err = deleteImage(ctx, rbdVol, cr); err != nil {
-		klog.Errorf(util.Log(ctx, "failed to delete rbd image: %s/%s with error: %v"),
-			rbdVol.Pool, rbdVol.RbdImageName, err)
+		klog.Errorf(util.Log(ctx, "failed to delete rbd image: %s with error: %v"),
+			getImageSpec(rbdVol.Pool, rbdVol.Namespace, rbdVol.RbdImageName), err)
 		return nil, status.Error(codes.Internal, err.Error())
 	}
 
@@ -735,20 +740,22 @@ func (cs *ControllerServer) DeleteSnapshot(ctx context.Context, req *csi.DeleteS
 	}
 	defer cs.SnapshotLocks.Release(rbdSnap.RequestName)
 
+	snapSpec := fmt.Sprintf("%s@%s",
+		getImageSpec(rbdSnap.Pool, rbdSnap.Namespace, rbdSnap.RbdImageName),
+		rbdSnap.RbdSnapName)
+
 	// Unprotect snapshot
 	err = unprotectSnapshot(ctx, rbdSnap, cr)
 	if err != nil {
 		return nil, status.Errorf(codes.FailedPrecondition,
-			"failed to unprotect snapshot: %s/%s with error: %v",
-			rbdSnap.Pool, rbdSnap.RbdSnapName, err)
+			"failed to unprotect snapshot: %s with error: %v", snapSpec, err)
 	}
 
 	// Deleting snapshot
-	klog.V(4).Infof(util.Log(ctx, "deleting Snaphot %s"), rbdSnap.RbdSnapName)
+	klog.V(4).Infof(util.Log(ctx, "deleting Snaphot %s"), snapSpec)
 	if err := deleteSnapshot(ctx, rbdSnap, cr); err != nil {
 		return nil, status.Errorf(codes.FailedPrecondition,
-			"failed to delete snapshot: %s/%s with error: %v",
-			rbdSnap.Pool, rbdSnap.RbdSnapName, err)
+			"failed to delete snapshot: %s with error: %v", snapSpec, err)
 	}
 
 	return &csi.DeleteSnapshotResponse{}, nil
@@ -800,8 +807,8 @@ func (cs *ControllerServer) ControllerExpandVolume(ctx context.Context, req *csi
 	defer rbdVol.Destroy()
 
 	if rbdVol.Encrypted {
-		return nil, status.Errorf(codes.InvalidArgument, "encrypted volumes do not support resize (%s/%s)",
-			rbdVol.Pool, rbdVol.RbdImageName)
+		return nil, status.Errorf(codes.InvalidArgument, "encrypted volumes do not support resize (%s)",
+			getImageSpec(rbdVol.Pool, rbdVol.Namespace, rbdVol.RbdImageName))
 	}
 
 	// always round up the request size in bytes to the nearest MiB/GiB
@@ -810,12 +817,12 @@ func (cs *ControllerServer) ControllerExpandVolume(ctx context.Context, req *csi
 	// resize volume if required
 	nodeExpansion := false
 	if rbdVol.VolSize < volSize {
-		klog.V(4).Infof(util.Log(ctx, "rbd volume %s/%s size is %v,resizing to %v"), rbdVol.Pool, rbdVol.RbdImageName, rbdVol.VolSize, volSize)
+		klog.V(4).Infof(util.Log(ctx, "rbd volume %s size is %v,resizing to %v"), getImageSpec(rbdVol.Pool, rbdVol.Namespace, rbdVol.RbdImageName), rbdVol.VolSize, volSize)
 		rbdVol.VolSize = volSize
 		nodeExpansion = true
 		err = resizeRBDImage(rbdVol, cr)
 		if err != nil {
-			klog.Errorf(util.Log(ctx, "failed to resize rbd image: %s/%s with error: %v"), rbdVol.Pool, rbdVol.RbdImageName, err)
+			klog.Errorf(util.Log(ctx, "failed to resize rbd image: %s with error: %v"), getImageSpec(rbdVol.Pool, rbdVol.Namespace, rbdVol.RbdImageName), err)
 			return nil, status.Error(codes.Internal, err.Error())
 		}
 	}

internal/rbd/errors.go

@@ -16,7 +16,7 @@ limitations under the License.
 
 package rbd
 
-// ErrImageNotFound is returned when image name is not found in the cluster on the given pool
+// ErrImageNotFound is returned when image name is not found in the cluster on the given pool and/or namespace
 type ErrImageNotFound struct {
 	imageName string
 	err       error

internal/rbd/nodeserver.go

@@ -152,6 +152,7 @@ func (ns *NodeServer) NodeStageVolume(ctx context.Context, req *csi.NodeStageVol
 			err = fmt.Errorf("error decoding volume ID (%s) (%s)", err, volID)
 			return nil, status.Error(codes.Internal, err.Error())
 		}
+		volJournal.SetNamespace(volOptions.Namespace)
 
 		imageAttributes, err = volJournal.GetImageAttributes(ctx, volOptions.Monitors, cr,
 			volOptions.Pool, vi.ObjectUUID, false)
@@ -608,7 +609,7 @@ func (ns *NodeServer) NodeUnstageVolume(ctx context.Context, req *csi.NodeUnstag
 	}
 
 	// Unmapping rbd device
-	imageSpec := imgInfo.Pool + "/" + imgInfo.ImageName
+	imageSpec := getImageSpec(imgInfo.Pool, imgInfo.Namespace, imgInfo.ImageName)
 	if err = detachRBDImageOrDeviceSpec(ctx, imageSpec, true, imgInfo.NbdAccess, imgInfo.Encrypted, req.GetVolumeId()); err != nil {
 		klog.Errorf(util.Log(ctx, "error unmapping volume (%s) from staging path (%s): (%v)"), req.GetVolumeId(), stagingTargetPath, err)
 		return nil, status.Error(codes.Internal, err.Error())
@@ -678,7 +679,7 @@ func getDevicePath(ctx context.Context, volumePath string) (string, error) {
 	if err != nil {
 		klog.Errorf(util.Log(ctx, "failed to find image metadata: %v"), err)
 	}
-	device, found := findDeviceMappingImage(ctx, imgInfo.Pool, imgInfo.ImageName, imgInfo.NbdAccess)
+	device, found := findDeviceMappingImage(ctx, imgInfo.Pool, imgInfo.Namespace, imgInfo.ImageName, imgInfo.NbdAccess)
 	if found {
 		return device, nil
 	}
@@ -715,7 +716,7 @@ func (ns *NodeServer) NodeGetCapabilities(ctx context.Context, req *csi.NodeGetC
 }
 
 func (ns *NodeServer) processEncryptedDevice(ctx context.Context, volOptions *rbdVolume, devicePath string) (string, error) {
-	imageSpec := volOptions.Pool + "/" + volOptions.RbdImageName
+	imageSpec := getImageSpec(volOptions.Pool, volOptions.Namespace, volOptions.RbdImageName)
 	encrypted, err := volOptions.checkRbdImageEncrypted(ctx)
 	if err != nil {
 		klog.Errorf(util.Log(ctx, "failed to get encryption status for rbd image %s: %v"),
@@ -765,13 +766,13 @@ func (ns *NodeServer) processEncryptedDevice(ctx context.Context, volOptions *rb
 func encryptDevice(ctx context.Context, rbdVol *rbdVolume, devicePath string) error {
 	passphrase, err := util.GetCryptoPassphrase(ctx, rbdVol.VolID, rbdVol.KMS)
 	if err != nil {
-		klog.Errorf(util.Log(ctx, "failed to get crypto passphrase for %s/%s: %v"),
-			rbdVol.Pool, rbdVol.RbdImageName, err)
+		klog.Errorf(util.Log(ctx, "failed to get crypto passphrase for %s: %v"),
+			getImageSpec(rbdVol.Pool, rbdVol.Namespace, rbdVol.RbdImageName), err)
 		return err
 	}
 
 	if err = util.EncryptVolume(ctx, devicePath, passphrase); err != nil {
-		err = fmt.Errorf("failed to encrypt volume %s/%s: %v", rbdVol.Pool, rbdVol.RbdImageName, err)
+		err = fmt.Errorf("failed to encrypt volume %s: %v", getImageSpec(rbdVol.Pool, rbdVol.Namespace, rbdVol.RbdImageName), err)
 		klog.Errorf(util.Log(ctx, err.Error()))
 		return err
 	}
@@ -788,8 +789,8 @@ func encryptDevice(ctx context.Context, rbdVol *rbdVolume, devicePath string) er
 func openEncryptedDevice(ctx context.Context, volOptions *rbdVolume, devicePath string) (string, error) {
 	passphrase, err := util.GetCryptoPassphrase(ctx, volOptions.VolID, volOptions.KMS)
 	if err != nil {
-		klog.Errorf(util.Log(ctx, "failed to get passphrase for encrypted device %s/%s: %v"),
-			volOptions.Pool, volOptions.RbdImageName, err)
+		klog.Errorf(util.Log(ctx, "failed to get passphrase for encrypted device %s: %v"),
+			getImageSpec(volOptions.Pool, volOptions.Namespace, volOptions.RbdImageName), err)
 		return "", status.Error(codes.Internal, err.Error())
 	}
 
@@ -805,8 +806,8 @@ func openEncryptedDevice(ctx context.Context, volOptions *rbdVolume, devicePath
 	} else {
 		err = util.OpenEncryptedVolume(ctx, devicePath, mapperFile, passphrase)
 		if err != nil {
-			klog.Errorf(util.Log(ctx, "failed to open device %s/%s: %v"),
-				volOptions.Pool, volOptions.RbdImageName, err)
+			klog.Errorf(util.Log(ctx, "failed to open device %s: %v"),
+				getImageSpec(volOptions.Pool, volOptions.Namespace, volOptions.RbdImageName), err)
 			return devicePath, err
 		}
 	}