diff --git a/.travis.yml b/.travis.yml
index 1c0d43356f1c..4eee54c8ba84 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -116,6 +116,13 @@ 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:
@@ -123,6 +130,13 @@ jobs:
         - 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:
diff --git a/e2e/e2e_test.go b/e2e/e2e_test.go
index a3c700992d75..c2d2e6b831bf 100644
--- a/e2e/e2e_test.go
+++ b/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
diff --git a/e2e/utils.go b/e2e/utils.go
index e1d9e5e3aca4..cfa7f0a532bf 100644
--- a/e2e/utils.go
+++ b/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
+}
diff --git a/examples/csi-config-map-sample.yaml b/examples/csi-config-map-sample.yaml
index 77965a2a6c3f..448756304be4 100644
--- a/examples/csi-config-map-sample.yaml
+++ b/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>",
diff --git a/internal/rbd/controllerserver.go b/internal/rbd/controllerserver.go
index f9f53207ed23..e77579f0f04f 100644
--- a/internal/rbd/controllerserver.go
+++ b/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())
 		}
 	}
diff --git a/internal/rbd/errors.go b/internal/rbd/errors.go
index 865292e53961..75e9f223c733 100644
--- a/internal/rbd/errors.go
+++ b/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
diff --git a/internal/rbd/nodeserver.go b/internal/rbd/nodeserver.go
index 663a9c6e0ae2..29468636cac0 100644
--- a/internal/rbd/nodeserver.go
+++ b/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
 		}
 	}
diff --git a/internal/rbd/rbd_attach.go b/internal/rbd/rbd_attach.go
index fbcacc8b6d7e..f5bee2f22792 100644
--- a/internal/rbd/rbd_attach.go
+++ b/internal/rbd/rbd_attach.go
@@ -57,10 +57,11 @@ func init() {
 
 // rbdDeviceInfo strongly typed JSON spec for rbd device list output (of type krbd)
 type rbdDeviceInfo struct {
-	ID     string `json:"id"`
-	Pool   string `json:"pool"`
-	Name   string `json:"name"`
-	Device string `json:"device"`
+	ID        string `json:"id"`
+	Pool      string `json:"pool"`
+	Namespace string `json:"namespace"`
+	Name      string `json:"name"`
+	Device    string `json:"device"`
 }
 
 // nbdDeviceInfo strongly typed JSON spec for rbd-nbd device list output (of type nbd)
@@ -68,10 +69,11 @@ type rbdDeviceInfo struct {
 // requiring 2 different JSON structures to unmarshal the output.
 // NOTE: image key is "name" in krbd output and "image" in nbd output, which is another difference
 type nbdDeviceInfo struct {
-	ID     int64  `json:"id"`
-	Pool   string `json:"pool"`
-	Name   string `json:"image"`
-	Device string `json:"device"`
+	ID        int64  `json:"id"`
+	Pool      string `json:"pool"`
+	Namespace string `json:"namespace"`
+	Name      string `json:"image"`
+	Device    string `json:"device"`
 }
 
 // rbdGetDeviceList queries rbd about mapped devices and returns a list of rbdDeviceInfo
@@ -103,10 +105,11 @@ func rbdGetDeviceList(accessType string) ([]rbdDeviceInfo, error) {
 			rbdDeviceList = append(
 				rbdDeviceList,
 				rbdDeviceInfo{
-					ID:     strconv.FormatInt(device.ID, 10),
-					Pool:   device.Pool,
-					Name:   device.Name,
-					Device: device.Device,
+					ID:        strconv.FormatInt(device.ID, 10),
+					Pool:      device.Pool,
+					Namespace: device.Namespace,
+					Name:      device.Name,
+					Device:    device.Device,
 				})
 		}
 	}
@@ -114,21 +117,22 @@ func rbdGetDeviceList(accessType string) ([]rbdDeviceInfo, error) {
 	return rbdDeviceList, nil
 }
 
-// findDeviceMappingImage finds a devicePath, if available, based on image spec (pool/image) on the node.
-func findDeviceMappingImage(ctx context.Context, pool, image string, useNbdDriver bool) (string, bool) {
+// findDeviceMappingImage finds a devicePath, if available, based on image spec (pool/{namespace/}image) on the node.
+func findDeviceMappingImage(ctx context.Context, pool, namespace, image string, useNbdDriver bool) (string, bool) {
 	accessType := accessTypeKRbd
 	if useNbdDriver {
 		accessType = accessTypeNbd
 	}
 
+	imageSpec := getImageSpec(pool, namespace, image)
 	rbdDeviceList, err := rbdGetDeviceList(accessType)
 	if err != nil {
-		klog.Warningf(util.Log(ctx, "failed to determine if image (%s/%s) is mapped to a device (%v)"), pool, image, err)
+		klog.Warningf(util.Log(ctx, "failed to determine if image (%s) is mapped to a device (%v)"), imageSpec, err)
 		return "", false
 	}
 
 	for _, device := range rbdDeviceList {
-		if device.Name == image && device.Pool == pool {
+		if device.Name == image && device.Pool == pool && device.Namespace == namespace {
 			return device.Device, true
 		}
 	}
@@ -137,13 +141,13 @@ func findDeviceMappingImage(ctx context.Context, pool, image string, useNbdDrive
 }
 
 // Stat a path, if it doesn't exist, retry maxRetries times.
-func waitForPath(ctx context.Context, pool, image string, maxRetries int, useNbdDriver bool) (string, bool) {
+func waitForPath(ctx context.Context, pool, namespace, image string, maxRetries int, useNbdDriver bool) (string, bool) {
 	for i := 0; i < maxRetries; i++ {
 		if i != 0 {
 			time.Sleep(time.Second)
 		}
 
-		device, found := findDeviceMappingImage(ctx, pool, image, useNbdDriver)
+		device, found := findDeviceMappingImage(ctx, pool, namespace, image, useNbdDriver)
 		if found {
 			return device, found
 		}
@@ -181,7 +185,7 @@ func attachRBDImage(ctx context.Context, volOptions *rbdVolume, cr *util.Credent
 		useNBD = true
 	}
 
-	devicePath, found := waitForPath(ctx, volOptions.Pool, image, 1, useNBD)
+	devicePath, found := waitForPath(ctx, volOptions.Pool, volOptions.Namespace, image, 1, useNBD)
 	if !found {
 		backoff := wait.Backoff{
 			Duration: rbdImageWatcherInitDelay,
@@ -202,8 +206,7 @@ func attachRBDImage(ctx context.Context, volOptions *rbdVolume, cr *util.Credent
 
 func createPath(ctx context.Context, volOpt *rbdVolume, cr *util.Credentials) (string, error) {
 	isNbd := false
-	image := volOpt.RbdImageName
-	imagePath := fmt.Sprintf("%s/%s", volOpt.Pool, image)
+	imagePath := getImageSpec(volOpt.Pool, volOpt.Namespace, volOpt.RbdImageName)
 
 	klog.V(5).Infof(util.Log(ctx, "rbd: map mon %s"), volOpt.Monitors)
 
@@ -244,8 +247,7 @@ func createPath(ctx context.Context, volOpt *rbdVolume, cr *util.Credentials) (s
 }
 
 func waitForrbdImage(ctx context.Context, backoff wait.Backoff, volOptions *rbdVolume, cr *util.Credentials) error {
-	image := volOptions.RbdImageName
-	imagePath := fmt.Sprintf("%s/%s", volOptions.Pool, image)
+	imagePath := getImageSpec(volOptions.Pool, volOptions.Namespace, volOptions.RbdImageName)
 
 	err := wait.ExponentialBackoff(backoff, func() (bool, error) {
 		used, rbdOutput, err := rbdStatus(ctx, volOptions, cr)
diff --git a/internal/rbd/rbd_journal.go b/internal/rbd/rbd_journal.go
index ac6e48ce6300..a785982a25e9 100644
--- a/internal/rbd/rbd_journal.go
+++ b/internal/rbd/rbd_journal.go
@@ -114,6 +114,8 @@ func checkSnapExists(ctx context.Context, rbdSnap *rbdSnapshot, cr *util.Credent
 		return false, err
 	}
 
+	snapJournal.SetNamespace(rbdSnap.Namespace)
+
 	snapData, err := snapJournal.CheckReservation(ctx, rbdSnap.Monitors, cr, rbdSnap.JournalPool,
 		rbdSnap.RequestName, rbdSnap.NamePrefix, rbdSnap.RbdImageName, "")
 	if err != nil {
@@ -174,6 +176,8 @@ func (rv *rbdVolume) Exists(ctx context.Context) (bool, error) {
 		kmsID = rv.KMS.GetID()
 	}
 
+	volJournal.SetNamespace(rv.Namespace)
+
 	imageData, err := volJournal.CheckReservation(ctx, rv.Monitors, rv.conn.Creds, rv.JournalPool,
 		rv.RequestName, rv.NamePrefix, "", kmsID)
 	if err != nil {
@@ -246,6 +250,8 @@ func reserveSnap(ctx context.Context, rbdSnap *rbdSnapshot, cr *util.Credentials
 		return err
 	}
 
+	snapJournal.SetNamespace(rbdSnap.Namespace)
+
 	snapUUID, rbdSnap.RbdSnapName, err = snapJournal.ReserveName(ctx, rbdSnap.Monitors, cr, rbdSnap.JournalPool, journalPoolID,
 		rbdSnap.Pool, imagePoolID, rbdSnap.RequestName, rbdSnap.NamePrefix, rbdSnap.RbdImageName, "")
 	if err != nil {
@@ -318,6 +324,8 @@ func reserveVol(ctx context.Context, rbdVol *rbdVolume, rbdSnap *rbdSnapshot, cr
 		kmsID = rbdVol.KMS.GetID()
 	}
 
+	volJournal.SetNamespace(rbdVol.Namespace)
+
 	imageUUID, rbdVol.RbdImageName, err = volJournal.ReserveName(ctx, rbdVol.Monitors, cr, rbdVol.JournalPool, journalPoolID,
 		rbdVol.Pool, imagePoolID, rbdVol.RequestName, rbdVol.NamePrefix, "", kmsID)
 	if err != nil {
@@ -338,6 +346,7 @@ func reserveVol(ctx context.Context, rbdVol *rbdVolume, rbdSnap *rbdSnapshot, cr
 
 // undoSnapReservation is a helper routine to undo a name reservation for rbdSnapshot
 func undoSnapReservation(ctx context.Context, rbdSnap *rbdSnapshot, cr *util.Credentials) error {
+	snapJournal.SetNamespace(rbdSnap.Namespace)
 	err := snapJournal.UndoReservation(ctx, rbdSnap.Monitors, cr, rbdSnap.JournalPool, rbdSnap.Pool,
 		rbdSnap.RbdSnapName, rbdSnap.RequestName)
 
@@ -346,6 +355,7 @@ func undoSnapReservation(ctx context.Context, rbdSnap *rbdSnapshot, cr *util.Cre
 
 // undoVolReservation is a helper routine to undo a name reservation for rbdVolume
 func undoVolReservation(ctx context.Context, rbdVol *rbdVolume, cr *util.Credentials) error {
+	volJournal.SetNamespace(rbdVol.Namespace)
 	err := volJournal.UndoReservation(ctx, rbdVol.Monitors, cr, rbdVol.JournalPool, rbdVol.Pool,
 		rbdVol.RbdImageName, rbdVol.RequestName)
 
diff --git a/internal/rbd/rbd_util.go b/internal/rbd/rbd_util.go
index 96c9f18b35a1..38c18b008ecd 100644
--- a/internal/rbd/rbd_util.go
+++ b/internal/rbd/rbd_util.go
@@ -89,6 +89,7 @@ type rbdVolume struct {
 	JournalPool         string
 	Pool                string `json:"pool"`
 	DataPool            string
+	Namespace           string
 	ImageFeatures       string `json:"imageFeatures"`
 	AdminID             string `json:"adminId"`
 	UserID              string `json:"userId"`
@@ -123,6 +124,7 @@ type rbdSnapshot struct {
 	Monitors       string
 	JournalPool    string
 	Pool           string
+	Namespace      string
 	CreatedAt      *timestamp.Timestamp
 	SizeBytes      int64
 	ClusterID      string
@@ -156,6 +158,13 @@ func (rv *rbdVolume) Destroy() {
 	}
 }
 
+func getImageSpec(pool, namespace, imageName string) string {
+	if namespace != "" {
+		return fmt.Sprintf("%s/%s/%s", pool, namespace, imageName)
+	}
+	return fmt.Sprintf("%s/%s", pool, imageName)
+}
+
 // createImage creates a new ceph image with provision and volume options.
 func createImage(ctx context.Context, pOpts *rbdVolume, cr *util.Credentials) error {
 	volSzMiB := fmt.Sprintf("%dM", util.RoundOffVolSize(pOpts.VolSize))
@@ -168,6 +177,10 @@ func createImage(ctx context.Context, pOpts *rbdVolume, cr *util.Credentials) er
 		if err != nil {
 			return errors.Wrapf(err, "failed to set data pool")
 		}
+		logMsg += fmt.Sprintf("data pool %s ", pOpts.DataPool)
+	}
+	if pOpts.Namespace != "" {
+		logMsg += fmt.Sprintf("namespace %s", pOpts.Namespace)
 	}
 	klog.V(4).Infof(util.Log(ctx, logMsg),
 		pOpts.RbdImageName, volSzMiB, pOpts.ImageFeatures, pOpts.Monitors, pOpts.Pool)
@@ -191,6 +204,8 @@ func createImage(ctx context.Context, pOpts *rbdVolume, cr *util.Credentials) er
 	}
 	defer ioctx.Destroy()
 
+	ioctx.SetNamespace(pOpts.Namespace)
+
 	err = librbd.CreateImage(ioctx, pOpts.RbdImageName,
 		uint64(util.RoundOffVolSize(pOpts.VolSize)*helpers.MiB), options)
 	if err != nil {
@@ -208,8 +223,15 @@ func rbdStatus(ctx context.Context, pOpts *rbdVolume, cr *util.Credentials) (boo
 
 	image := pOpts.RbdImageName
 
-	klog.V(4).Infof(util.Log(ctx, "rbd: status %s using mon %s, pool %s"), image, pOpts.Monitors, pOpts.Pool)
+	logMsg := "rbd: status %s using mon %s, pool %s "
+	if pOpts.Namespace != "" {
+		logMsg += fmt.Sprintf("namespace %s", pOpts.Namespace)
+	}
+	klog.V(4).Infof(util.Log(ctx, logMsg), image, pOpts.Monitors, pOpts.Pool)
 	args := []string{"status", image, "--pool", pOpts.Pool, "-m", pOpts.Monitors, "--id", cr.ID, "--keyfile=" + cr.KeyFile}
+	if pOpts.Namespace != "" {
+		args = append(args, "--namespace", pOpts.Namespace)
+	}
 	cmd, err := execCommand("rbd", args)
 	output = string(cmd)
 
@@ -240,7 +262,7 @@ func rbdManagerTaskDeleteImage(ctx context.Context, pOpts *rbdVolume, cr *util.C
 	var output []byte
 
 	args := []string{"rbd", "task", "add", "remove",
-		pOpts.Pool + "/" + pOpts.RbdImageName,
+		getImageSpec(pOpts.Pool, pOpts.Namespace, pOpts.RbdImageName),
 		"--id", cr.ID,
 		"--keyfile=" + cr.KeyFile,
 		"-m", pOpts.Monitors,
@@ -268,6 +290,7 @@ func rbdManagerTaskDeleteImage(ctx context.Context, pOpts *rbdVolume, cr *util.C
 // deleteImage deletes a ceph image with provision and volume options.
 func deleteImage(ctx context.Context, pOpts *rbdVolume, cr *util.Credentials) error {
 	image := pOpts.RbdImageName
+	spec := getImageSpec(pOpts.Pool, pOpts.Namespace, image)
 	found, _, err := rbdStatus(ctx, pOpts, cr)
 	if err != nil {
 		return err
@@ -276,13 +299,16 @@ func deleteImage(ctx context.Context, pOpts *rbdVolume, cr *util.Credentials) er
 		klog.Errorf(util.Log(ctx, "rbd is still being used "), image)
 		return fmt.Errorf("rbd %s is still being used", image)
 	}
-
-	klog.V(4).Infof(util.Log(ctx, "rbd: rm %s using mon %s, pool %s"), image, pOpts.Monitors, pOpts.Pool)
+	logMsg := "rbd: rm %s using mon %s, pool %s "
+	if pOpts.Namespace != "" {
+		logMsg += fmt.Sprintf("namespace %s", pOpts.Namespace)
+	}
+	klog.V(4).Infof(util.Log(ctx, logMsg), image, pOpts.Monitors, pOpts.Pool)
 
 	// attempt to use Ceph manager based deletion support if available
 	rbdCephMgrSupported, err := rbdManagerTaskDeleteImage(ctx, pOpts, cr)
 	if rbdCephMgrSupported && err != nil {
-		klog.Errorf(util.Log(ctx, "failed to add task to delete rbd image: %s/%s, %v"), pOpts.Pool, image, err)
+		klog.Errorf(util.Log(ctx, "failed to add task to delete rbd image: %s, %v"), spec, err)
 		return err
 	}
 
@@ -294,10 +320,12 @@ func deleteImage(ctx context.Context, pOpts *rbdVolume, cr *util.Credentials) er
 		}
 		defer ioctx.Destroy()
 
+		ioctx.SetNamespace(pOpts.Namespace)
+
 		rbdImage := librbd.GetImage(ioctx, image)
 		err = rbdImage.Remove()
 		if err != nil {
-			klog.Errorf(util.Log(ctx, "failed to delete rbd image: %s/%s, error: %v"), pOpts.Pool, image, err)
+			klog.Errorf(util.Log(ctx, "failed to delete rbd image: %s, error: %v"), pOpts.Pool, image, err)
 		}
 	}
 
@@ -307,8 +335,7 @@ func deleteImage(ctx context.Context, pOpts *rbdVolume, cr *util.Credentials) er
 // updateSnapWithImageInfo updates provided rbdSnapshot with information from on-disk data
 // regarding the same
 func updateSnapWithImageInfo(ctx context.Context, rbdSnap *rbdSnapshot, cr *util.Credentials) error {
-	snapInfo, err := getSnapInfo(ctx, rbdSnap.Monitors, cr, rbdSnap.Pool,
-		rbdSnap.RbdImageName, rbdSnap.RbdSnapName)
+	snapInfo, err := getSnapInfo(ctx, rbdSnap.Monitors, cr, rbdSnap.Pool, rbdSnap.Namespace, rbdSnap.RbdImageName, rbdSnap.RbdSnapName)
 	if err != nil {
 		return err
 	}
@@ -328,7 +355,7 @@ func updateSnapWithImageInfo(ctx context.Context, rbdSnap *rbdSnapshot, cr *util
 // updateVolWithImageInfo updates provided rbdVolume with information from on-disk data
 // regarding the same
 func updateVolWithImageInfo(ctx context.Context, rbdVol *rbdVolume, cr *util.Credentials) error {
-	imageInfo, err := getImageInfo(ctx, rbdVol.Monitors, cr, rbdVol.Pool, rbdVol.RbdImageName)
+	imageInfo, err := getImageInfo(ctx, rbdVol.Monitors, cr, rbdVol.Pool, rbdVol.Namespace, rbdVol.RbdImageName)
 	if err != nil {
 		return err
 	}
@@ -370,6 +397,12 @@ func genSnapFromSnapID(ctx context.Context, rbdSnap *rbdSnapshot, snapshotID str
 	}
 	rbdSnap.JournalPool = rbdSnap.Pool
 
+	rbdSnap.Namespace, err = util.Namespace(csiConfigFile, rbdSnap.ClusterID)
+	if err != nil {
+		return err
+	}
+	snapJournal.SetNamespace(rbdSnap.Namespace)
+
 	imageAttributes, err := snapJournal.GetImageAttributes(ctx, rbdSnap.Monitors,
 		cr, rbdSnap.Pool, vi.ObjectUUID, true)
 	if err != nil {
@@ -432,6 +465,12 @@ func genVolFromVolID(ctx context.Context, volumeID string, cr *util.Credentials,
 	}
 	rbdVol.JournalPool = rbdVol.Pool
 
+	rbdVol.Namespace, err = util.Namespace(csiConfigFile, rbdVol.ClusterID)
+	if err != nil {
+		return nil, err
+	}
+	volJournal.SetNamespace(rbdVol.Namespace)
+
 	imageAttributes, err := volJournal.GetImageAttributes(ctx, rbdVol.Monitors, cr,
 		rbdVol.Pool, vi.ObjectUUID, false)
 	if err != nil {
@@ -573,6 +612,11 @@ func genVolFromVolumeOptions(ctx context.Context, volOptions, credentials map[st
 		if err != nil {
 			return nil, err
 		}
+
+		rbdVol.Namespace, err = util.Namespace(csiConfigFile, rbdVol.ClusterID)
+		if err != nil {
+			return nil, err
+		}
 	}
 
 	// if no image features is provided, it results in empty string
@@ -627,6 +671,7 @@ func genSnapFromOptions(ctx context.Context, rbdVol *rbdVolume, snapOptions map[
 	rbdSnap := &rbdSnapshot{}
 	rbdSnap.Pool = rbdVol.Pool
 	rbdSnap.JournalPool = rbdVol.JournalPool
+	rbdSnap.Namespace = rbdVol.Namespace
 
 	rbdSnap.Monitors, rbdSnap.ClusterID, err = getMonsAndClusterID(ctx, snapOptions)
 	if err != nil {
@@ -672,9 +717,16 @@ func protectSnapshot(ctx context.Context, pOpts *rbdSnapshot, cr *util.Credentia
 	image := pOpts.RbdImageName
 	snapName := pOpts.RbdSnapName
 
-	klog.V(4).Infof(util.Log(ctx, "rbd: snap protect %s using mon %s, pool %s "), image, pOpts.Monitors, pOpts.Pool)
+	logMsg := "rbd: snap protect %s using mon %s, pool %s "
+	if pOpts.Namespace != "" {
+		logMsg += fmt.Sprintf("namespace %s", pOpts.Namespace)
+	}
+	klog.V(4).Infof(util.Log(ctx, logMsg), image, pOpts.Monitors, pOpts.Pool)
 	args := []string{"snap", "protect", "--pool", pOpts.Pool, "--snap", snapName, image, "--id",
 		cr.ID, "-m", pOpts.Monitors, "--keyfile=" + cr.KeyFile}
+	if pOpts.Namespace != "" {
+		args = append(args, "--namespace", pOpts.Namespace)
+	}
 
 	output, err := execCommand("rbd", args)
 
@@ -691,9 +743,16 @@ func createSnapshot(ctx context.Context, pOpts *rbdSnapshot, cr *util.Credential
 	image := pOpts.RbdImageName
 	snapName := pOpts.RbdSnapName
 
-	klog.V(4).Infof(util.Log(ctx, "rbd: snap create %s using mon %s, pool %s"), image, pOpts.Monitors, pOpts.Pool)
+	logMsg := "rbd: snap create %s using mon %s, pool %s "
+	if pOpts.Namespace != "" {
+		logMsg += fmt.Sprintf("namespace %s", pOpts.Namespace)
+	}
+	klog.V(4).Infof(util.Log(ctx, logMsg), image, pOpts.Monitors, pOpts.Pool)
 	args := []string{"snap", "create", "--pool", pOpts.Pool, "--snap", snapName, image,
 		"--id", cr.ID, "-m", pOpts.Monitors, "--keyfile=" + cr.KeyFile}
+	if pOpts.Namespace != "" {
+		args = append(args, "--namespace", pOpts.Namespace)
+	}
 
 	output, err := execCommand("rbd", args)
 
@@ -710,9 +769,16 @@ func unprotectSnapshot(ctx context.Context, pOpts *rbdSnapshot, cr *util.Credent
 	image := pOpts.RbdImageName
 	snapName := pOpts.RbdSnapName
 
-	klog.V(4).Infof(util.Log(ctx, "rbd: snap unprotect %s using mon %s, pool %s"), image, pOpts.Monitors, pOpts.Pool)
+	logMsg := "rbd: snap unprotect %s using mon %s, pool %s "
+	if pOpts.Namespace != "" {
+		logMsg += fmt.Sprintf("namespace %s", pOpts.Namespace)
+	}
+	klog.V(4).Infof(util.Log(ctx, logMsg), image, pOpts.Monitors, pOpts.Pool)
 	args := []string{"snap", "unprotect", "--pool", pOpts.Pool, "--snap", snapName, image, "--id",
 		cr.ID, "-m", pOpts.Monitors, "--keyfile=" + cr.KeyFile}
+	if pOpts.Namespace != "" {
+		args = append(args, "--namespace", pOpts.Namespace)
+	}
 
 	output, err := execCommand("rbd", args)
 
@@ -729,9 +795,16 @@ func deleteSnapshot(ctx context.Context, pOpts *rbdSnapshot, cr *util.Credential
 	image := pOpts.RbdImageName
 	snapName := pOpts.RbdSnapName
 
-	klog.V(4).Infof(util.Log(ctx, "rbd: snap rm %s using mon %s, pool %s"), image, pOpts.Monitors, pOpts.Pool)
+	logMsg := "rbd: snap rm %s using mon %s, pool %s "
+	if pOpts.Namespace != "" {
+		logMsg += fmt.Sprintf("namespace %s", pOpts.Namespace)
+	}
+	klog.V(4).Infof(util.Log(ctx, logMsg), image, pOpts.Monitors, pOpts.Pool)
 	args := []string{"snap", "rm", "--pool", pOpts.Pool, "--snap", snapName, image, "--id",
 		cr.ID, "-m", pOpts.Monitors, "--keyfile=" + cr.KeyFile}
+	if pOpts.Namespace != "" {
+		args = append(args, "--namespace", pOpts.Namespace)
+	}
 
 	output, err := execCommand("rbd", args)
 
@@ -751,11 +824,14 @@ func restoreSnapshot(ctx context.Context, pVolOpts *rbdVolume, pSnapOpts *rbdSna
 	var output []byte
 
 	image := pVolOpts.RbdImageName
-	snapName := pSnapOpts.RbdSnapName
 
-	klog.V(4).Infof(util.Log(ctx, "rbd: clone %s using mon %s, pool %s"), image, pVolOpts.Monitors, pVolOpts.Pool)
-	args := []string{"clone", pSnapOpts.Pool + "/" + pSnapOpts.RbdImageName + "@" + snapName,
-		pVolOpts.Pool + "/" + image, "--id", cr.ID, "-m", pVolOpts.Monitors, "--keyfile=" + cr.KeyFile}
+	logMsg := "rbd: clone %s using mon %s, pool %s "
+	if pVolOpts.Namespace != "" {
+		logMsg += fmt.Sprintf("namespace %s", pVolOpts.Namespace)
+	}
+	klog.V(4).Infof(util.Log(ctx, logMsg), image, pVolOpts.Monitors, pVolOpts.Pool)
+	args := []string{"clone", fmt.Sprintf("%s@%s", getImageSpec(pVolOpts.Pool, pVolOpts.Namespace, pVolOpts.RbdImageName), pSnapOpts.RbdSnapName),
+		getImageSpec(pVolOpts.Pool, pVolOpts.Namespace, pVolOpts.RbdImageName), "--id", cr.ID, "-m", pVolOpts.Monitors, "--keyfile=" + cr.KeyFile}
 
 	output, err := execCommand("rbd", args)
 
@@ -772,7 +848,7 @@ func getSnapshotMetadata(ctx context.Context, pSnapOpts *rbdSnapshot, cr *util.C
 	imageName := pSnapOpts.RbdImageName
 	snapName := pSnapOpts.RbdSnapName
 
-	snapInfo, err := getSnapInfo(ctx, pSnapOpts.Monitors, cr, pSnapOpts.Pool, imageName, snapName)
+	snapInfo, err := getSnapInfo(ctx, pSnapOpts.Monitors, cr, pSnapOpts.Pool, pSnapOpts.Namespace, imageName, snapName)
 	if err != nil {
 		return err
 	}
@@ -802,11 +878,13 @@ type imageInfo struct {
 
 // getImageInfo queries rbd about the given image and returns its metadata, and returns
 // ErrImageNotFound if provided image is not found
-func getImageInfo(ctx context.Context, monitors string, cr *util.Credentials, poolName, imageName string) (imageInfo, error) {
+func getImageInfo(ctx context.Context, monitors string, cr *util.Credentials, poolName, namespace, imageName string) (imageInfo, error) {
 	// rbd --format=json info [image-spec | snap-spec]
 
 	var imgInfo imageInfo
 
+	imageSpec := getImageSpec(poolName, namespace, imageName)
+
 	stdout, stderr, err := util.ExecCommand(
 		"rbd",
 		"-m", monitors,
@@ -814,9 +892,9 @@ func getImageInfo(ctx context.Context, monitors string, cr *util.Credentials, po
 		"--keyfile="+cr.KeyFile,
 		"-c", util.CephConfigPath,
 		"--format="+"json",
-		"info", poolName+"/"+imageName)
+		"info", imageSpec)
 	if err != nil {
-		klog.Errorf(util.Log(ctx, "failed getting information for image (%s): (%s)"), poolName+"/"+imageName, err)
+		klog.Errorf(util.Log(ctx, "failed getting information for image (%s): (%s)"), imageSpec, err)
 		if strings.Contains(string(stderr), "rbd: error opening image "+imageName+
 			": (2) No such file or directory") {
 			return imgInfo, ErrImageNotFound{imageName, err}
@@ -827,7 +905,7 @@ func getImageInfo(ctx context.Context, monitors string, cr *util.Credentials, po
 	err = json.Unmarshal(stdout, &imgInfo)
 	if err != nil {
 		klog.Errorf(util.Log(ctx, "failed to parse JSON output of image info (%s): (%s)"),
-			poolName+"/"+imageName, err)
+			imageSpec, err)
 		return imgInfo, fmt.Errorf("unmarshal failed: %+v.  raw buffer response: %s",
 			err, string(stdout))
 	}
@@ -848,7 +926,7 @@ getSnapInfo queries rbd about the snapshots of the given image and returns its m
 returns ErrImageNotFound if provided image is not found, and ErrSnapNotFound if provided snap
 is not found in the images snapshot list
 */
-func getSnapInfo(ctx context.Context, monitors string, cr *util.Credentials, poolName, imageName, snapName string) (snapInfo, error) {
+func getSnapInfo(ctx context.Context, monitors string, cr *util.Credentials, poolName, namespace, imageName, snapName string) (snapInfo, error) {
 	// rbd --format=json snap ls [image-spec]
 
 	var (
@@ -856,6 +934,8 @@ func getSnapInfo(ctx context.Context, monitors string, cr *util.Credentials, poo
 		snaps   []snapInfo
 	)
 
+	imageSpec := getImageSpec(poolName, namespace, imageName)
+
 	stdout, stderr, err := util.ExecCommand(
 		"rbd",
 		"-m", monitors,
@@ -863,10 +943,10 @@ func getSnapInfo(ctx context.Context, monitors string, cr *util.Credentials, poo
 		"--keyfile="+cr.KeyFile,
 		"-c", util.CephConfigPath,
 		"--format="+"json",
-		"snap", "ls", poolName+"/"+imageName)
+		"snap", "ls", imageSpec)
 	if err != nil {
 		klog.Errorf(util.Log(ctx, "failed getting snap (%s) information from image (%s): (%s)"),
-			snapName, poolName+"/"+imageName, err)
+			snapName, imageSpec, err)
 		if strings.Contains(string(stderr), "rbd: error opening image "+imageName+
 			": (2) No such file or directory") {
 			return snpInfo, ErrImageNotFound{imageName, err}
@@ -877,7 +957,7 @@ func getSnapInfo(ctx context.Context, monitors string, cr *util.Credentials, poo
 	err = json.Unmarshal(stdout, &snaps)
 	if err != nil {
 		klog.Errorf(util.Log(ctx, "failed to parse JSON output of image snap list (%s): (%s)"),
-			poolName+"/"+imageName, err)
+			imageSpec, err)
 		return snpInfo, fmt.Errorf("unmarshal failed: %+v. raw buffer response: %s",
 			err, string(stdout))
 	}
@@ -889,13 +969,14 @@ func getSnapInfo(ctx context.Context, monitors string, cr *util.Credentials, poo
 	}
 
 	return snpInfo, ErrSnapNotFound{snapName, fmt.Errorf("snap (%s) for image (%s) not found",
-		snapName, poolName+"/"+imageName)}
+		snapName, imageSpec)}
 }
 
 // rbdImageMetadataStash strongly typed JSON spec for stashed RBD image metadata
 type rbdImageMetadataStash struct {
 	Version   int    `json:"Version"`
 	Pool      string `json:"pool"`
+	Namespace string `json:"namespace"`
 	ImageName string `json:"image"`
 	NbdAccess bool   `json:"accessType"`
 	Encrypted bool   `json:"encrypted"`
@@ -910,6 +991,7 @@ func stashRBDImageMetadata(volOptions *rbdVolume, path string) error {
 	var imgMeta = rbdImageMetadataStash{
 		Version:   2, // there are no checks for this at present
 		Pool:      volOptions.Pool,
+		Namespace: volOptions.Namespace,
 		ImageName: volOptions.RbdImageName,
 		Encrypted: volOptions.Encrypted,
 	}
@@ -921,6 +1003,11 @@ func stashRBDImageMetadata(volOptions *rbdVolume, path string) error {
 
 	encodedBytes, err := json.Marshal(imgMeta)
 	if err != nil {
+		if volOptions.Namespace != "" {
+			return fmt.Errorf(
+				"failed to marshall JSON image metadata for image (%s) in pool (%s) namespace (%s): (%v)",
+				volOptions.RbdImageName, volOptions.Pool, volOptions.Namespace, err)
+		}
 		return fmt.Errorf("failed to marshall JSON image metadata for image (%s) in pool (%s): (%v)",
 			volOptions.RbdImageName, volOptions.Pool, err)
 	}
@@ -928,6 +1015,11 @@ func stashRBDImageMetadata(volOptions *rbdVolume, path string) error {
 	fPath := filepath.Join(path, stashFileName)
 	err = ioutil.WriteFile(fPath, encodedBytes, 0600)
 	if err != nil {
+		if volOptions.Namespace != "" {
+			return fmt.Errorf(
+				"failed to stash JSON image metadata for image (%s) in pool (%s) namespace (%s) at path (%s): (%v)",
+				volOptions.RbdImageName, volOptions.Pool, volOptions.Namespace, fPath, err)
+		}
 		return fmt.Errorf("failed to stash JSON image metadata for image (%s) in pool (%s) at path (%s): (%v)",
 			volOptions.RbdImageName, volOptions.Pool, fPath, err)
 	}
@@ -976,6 +1068,9 @@ func resizeRBDImage(rbdVol *rbdVolume, cr *util.Credentials) error {
 	volSzMiB := fmt.Sprintf("%dM", util.RoundOffVolSize(rbdVol.VolSize))
 
 	args := []string{"resize", image, "--size", volSzMiB, "--pool", rbdVol.Pool, "--id", cr.ID, "-m", mon, "--keyfile=" + cr.KeyFile}
+	if rbdVol.Namespace != "" {
+		args = append(args, "--namespace", rbdVol.Namespace)
+	}
 	output, err := execCommand("rbd", args)
 
 	if err != nil {
@@ -991,6 +1086,8 @@ func (rv *rbdVolume) GetMetadata(key string) (string, error) {
 		return "", errors.Wrapf(err, "failed to get ioctx for %q", rv.RbdImageName)
 	}
 
+	ioctx.SetNamespace(rv.Namespace)
+
 	image, err := librbd.OpenImage(ioctx, rv.RbdImageName, librbd.NoSnapshot)
 	if err != nil {
 		return "", errors.Wrapf(err, "could not open image %q", rv.RbdImageName)
@@ -1006,6 +1103,8 @@ func (rv *rbdVolume) SetMetadata(key, value string) error {
 		return errors.Wrapf(err, "failed to get ioctx for %q", rv.RbdImageName)
 	}
 
+	ioctx.SetNamespace(rv.Namespace)
+
 	image, err := librbd.OpenImage(ioctx, rv.RbdImageName, librbd.NoSnapshot)
 	if err != nil {
 		return errors.Wrapf(err, "could not open image %q", rv.RbdImageName)
diff --git a/internal/util/csiconfig.go b/internal/util/csiconfig.go
index 4447d7224f35..d4f9c0feb314 100644
--- a/internal/util/csiconfig.go
+++ b/internal/util/csiconfig.go
@@ -33,6 +33,8 @@ const (
 type clusterInfo struct {
 	// ClusterID is used for unique identification
 	ClusterID string `json:"clusterID"`
+	// Namespace is the namespace in the pool
+	Namespace string `json:"namespace"`
 	// Monitors is monitor list for corresponding cluster ID
 	Monitors []string `json:"monitors"`
 	// CephFS contains CephFS specific options
@@ -46,6 +48,7 @@ type clusterInfo struct {
 // [
 // 	{
 // 		"clusterID": "<cluster-id>",
+//		"namespace": "<namespace>",
 // 		"monitors":
 // 			[
 // 				"<monitor-value>",
@@ -96,6 +99,15 @@ func Mons(pathToConfig, clusterID string) (string, error) {
 	return strings.Join(cluster.Monitors, ","), nil
 }
 
+// Namespace returns the namespace for the given clusterID
+func Namespace(pathToConfig, clusterID string) (string, error) {
+	cluster, err := readClusterInfo(pathToConfig, clusterID)
+	if err != nil {
+		return "", err
+	}
+	return cluster.Namespace, nil
+}
+
 // CephFSSubvolumeGroup returns the subvolumeGroup for CephFS volumes. If not set, it returns the default value "csi"
 func CephFSSubvolumeGroup(pathToConfig, clusterID string) (string, error) {
 	cluster, err := readClusterInfo(pathToConfig, clusterID)
diff --git a/scripts/travis-functest.sh b/scripts/travis-functest.sh
index 7322baf4a656..29b39feae9e2 100755
--- a/scripts/travis-functest.sh
+++ b/scripts/travis-functest.sh
@@ -4,6 +4,11 @@ set -e
 # This script will be used by travis to run functional test
 # against different kuberentes version
 export KUBE_VERSION=$1
+
+# If RBD_NAMESPACE is set to non-empty the tests will be
+# executed within that namespace 
+export RBD_NAMESPACE=$2
+
 sudo scripts/minikube.sh up
 sudo scripts/minikube.sh deploy-rook
 sudo scripts/minikube.sh create-block-pool
@@ -12,6 +17,5 @@ sudo scripts/minikube.sh cephcsi
 sudo scripts/minikube.sh k8s-sidecar
 sudo chown -R travis: "$HOME"/.minikube /usr/local/bin/kubectl
 # functional tests
-go test github.com/ceph/ceph-csi/e2e --deploy-timeout=10 -timeout=30m --cephcsi-namespace=cephcsi-e2e-$RANDOM -v -mod=vendor
-
-sudo scripts/minikube.sh clean
+go test github.com/ceph/ceph-csi/e2e --deploy-timeout=10 -timeout=30m --rbd-namespace=$RBD_NAMESPACE --cephcsi-namespace=cephcsi-e2e-$RANDOM -v -mod=vendor
+sudo scripts/minikube.sh clean
\ No newline at end of file