Repository: ceph/ceph-csi
Pull request #2374 (2021-08-09 04:41:16)

RBD Regional DR: Implement clusterID mapping  (backport #2314)
This is an automatic backport of pull request #2314 done by [Mergify](https://mergify.io).
Cherry-pick of fce5a181d05b4880db3ece754489403fc9b7c9e1 has failed:
```
On branch mergify/bp/release-v3.4/pr-2314
Your branch is up to date with 'origin/release-v3.4'.

You are currently cherry-picking commit fce5a181.
  (fix conflicts and run "git cherry-pick --continue")
  (use "git cherry-pick --skip" to skip this patch)
  (use "git cherry-pick --abort" to cancel the cherry-pick operation)

Unmerged paths:
  (use "git add/rm <file>..." as appropriate to mark resolution)
	deleted by us:   docs/design/proposals/clusterid-mapping.md

no changes added to commit (use "git add" and/or "git commit -a")
```


To fix up this pull request, you can check it out locally. See documentation: https://docs.github.com/en/github/collaborating-with-pull-requests/reviewing-changes-in-pull-requests/checking-out-pull-requests-locally

---


<details>
<summary>Mergify commands and options</summary>

<br />

More conditions and actions can be found in the [documentation](https://docs.mergify.io/).

You can also trigger Mergify actions by commenting on this pull request:

- `@Mergifyio refresh` will re-evaluate the rules
- `@Mergifyio rebase` will rebase this PR on its base branch
- `@Mergifyio update` will merge the base branch into this PR
- `@Mergifyio backport <destination>` will backport this PR on `<destination>` branch

Additionally, on Mergify [dashboard](https://dashboard.mergify.io/) you can:

- look at your merge queues
- generate the Mergify configuration with the config editor.

Finally, you can contact us on https://mergify.io/
</details>


Madhu-1: resolved conflict by adding cherry-picking the design doc which was not backported to release-v3.4
Madhu-1: /test ci/centos/mini-e2e-helm/k8s-1.20

Patch:
diff --git a/charts/ceph-csi-rbd/README.md b/charts/ceph-csi-rbd/README.md
index 246348bfa..9f26025ed 100644
--- a/charts/ceph-csi-rbd/README.md
+++ b/charts/ceph-csi-rbd/README.md
@@ -85,6 +85,7 @@ charts and their default values.
 | `serviceAccounts.provisioner.create`           | Specifies whether a provisioner ServiceAccount should be created                                                                                     | `true`                                             |
 | `serviceAccounts.provisioner.name`             | The name of the provisioner ServiceAccount to use. If not set and create is true, a name is generated using the fullname                             | ""                                                 |
 | `csiConfig`                                    | Configuration for the CSI to connect to the cluster                                                                                                  | []                                                 |
+| `csiMapping`                                   | Configuration details of clusterID,PoolID,FscID mapping                                                                                              | []                                                 |
 | `encryptionKMSConfig`                          | Configuration for the encryption KMS                                                                                                                 | `{}`                                               |
 | `logLevel`                                     | Set logging level for csi containers. Supported values from 0 to 5. 0 for general useful logs, 5 for trace level verbosity.                          | `5`                                                |
 | `nodeplugin.name`                              | Specifies the nodeplugins name                                                                                                                       | `nodeplugin`                                       |
diff --git a/charts/ceph-csi-rbd/templates/csiplugin-configmap.yaml b/charts/ceph-csi-rbd/templates/csiplugin-configmap.yaml
index 67cb22f1a..3cec53e6e 100644
--- a/charts/ceph-csi-rbd/templates/csiplugin-configmap.yaml
+++ b/charts/ceph-csi-rbd/templates/csiplugin-configmap.yaml
@@ -13,4 +13,6 @@ metadata:
 data:
   config.json: |-
 {{ toJson .Values.csiConfig | indent 4 -}}
+  cluster-mapping.json: |-
+{{ toJson .Values.csiMapping | indent 4 -}}
 {{- end }}
diff --git a/charts/ceph-csi-rbd/values.yaml b/charts/ceph-csi-rbd/values.yaml
index 782c2ac0c..546e59b04 100644
--- a/charts/ceph-csi-rbd/values.yaml
+++ b/charts/ceph-csi-rbd/values.yaml
@@ -27,6 +27,16 @@ serviceAccounts:
 #       - "<MONValue2>"
 csiConfig: []
 
+# Configuration details of clusterID,PoolID and FscID mapping
+# csiMapping:
+#  - clusterIDMapping:
+#     clusterID on site1: clusterID on site2
+#    RBDPoolIDMapping:
+#    - poolID on site1: poolID on site2
+#    CephFSFscIDMapping:
+#    - CephFS FscID on site1: CephFS FscID on site2
+csiMapping: []
+
 # Configuration for the encryption KMS
 # Ref: https://github.com/ceph/ceph-csi/blob/devel/docs/deploy-rbd.md
 # Example:
diff --git a/docs/design/proposals/clusterid-mapping.md b/docs/design/proposals/clusterid-mapping.md
new file mode 100644
index 000000000..acb734bcb
--- /dev/null
+++ b/docs/design/proposals/clusterid-mapping.md
@@ -0,0 +1,176 @@
+# Design to handle clusterID and poolID for DR
+
+During disaster recovery/migration of a cluster, as part of the failover, the
+kubernetes artifacts like deployment, PVC, PV, etc will be restored to a new
+cluster by the admin. Even if the kubernetes objects are restored the
+corresponding RBD/CephFS subvolume cannot be retrieved during CSI operations as
+the clusterID and poolID are not the same in both clusters. Let's see the
+problem in more detail below.
+
+`0001-0009-rook-ceph-0000000000000002-b0285c97-a0ce-11eb-8c66-0242ac110002`
+
+The above is the sample volumeID sent back in response to the CreateVolume
+operation and added as a volumeHandle in the PV spec. CO (Kubernetes) uses
+above as the identifier for other operations on the volume/PVC.
+
+The VolumeID is encoded as,
+
+```text
+0001 -->                              [csi_id_version=1:4byte] + [-:1byte]
+0009 -->                              [length of clusterID=1:4byte] + [-:1byte]
+rook-ceph -->                         [clusterID:36bytes (MAX)] + [-:1byte]
+0000000000000002 -->                  [poolID:16bytes] + [-:1byte]
+b0285c97-a0ce-11eb-8c66-0242ac110002 --> [ObjectUUID:36bytes]
+Total of constant field lengths, including '-' field separators would hence be,
+4+1+4+1+1+16+1+36 = 64
+```
+
+When mirroring is enabled volume which is `csi-vol-ObjectUUID` is mirrored to
+the other cluster.
+
+> `csi-vol` is const name and over has the option to override it in
+> storageclass.
+
+During the Disaster Recovery (failover operation) the PVC and PV will be
+recreated on the other cluster. When Ceph-CSI receives the request for
+operations like (NodeStage, ExpandVolume, DeleteVolume, etc) the volumeID is
+sent in the request which will help to identify the volume.
+
+```yaml=
+apiVersion: v1
+kind: ConfigMap
+data:
+  config.json: |-
+    [
+      {
+       "clusterID": "rook-ceph",
+       "radosNamespace": "<rados-namespace>",
+       "monitors": [
+         "192.168.39.82:6789"
+       ],
+       "cephFS": {
+         "subvolumeGroup": "<subvolumegroup for cephfs volumes>"
+       }
+      },
+      {
+       "clusterID": "fs-id",
+       "radosNamespace": "<rados-namespace>",
+       "monitors": [
+         "192.168.39.83:6789"
+       ],
+       "cephFS": {
+         "subvolumeGroup": "<subvolumegroup for cephfs volumes>"
+       }
+      }
+    ]
+metadata:
+  name: ceph-csi-config
+```
+
+During CSI/Replication operations, Ceph-CSI will decode the volumeID and gets
+the monitor configuration from the configmap and by the poolID will get the
+pool Name and retrieves the OMAP data stored in the rados OMAP and finally
+check the volume is present in the pool.
+
+## Problems with volumeID Replication
+
+* The clusterID can be different
+  * as the clusterID is the namespace where rook is deployed, the Rook might be
+    deployed in the different namespace on a secondary cluster
+  * In standalone Ceph-CSI the clusterID is fsID and fsID is unique per
+    cluster
+
+* The poolID can be different
+  * PoolID which is encoded in the volumeID won't remain the same across
+    clusters
+
+To solve this problem we need to have a new mapping between clusterID's and the
+poolID's.
+
+Example configmap Need to be created before failover to `site2-storage` from
+`site1-storage` and `site3-storage`.
+
+```yaml=
+apiVersion: v1
+kind: ConfigMap
+data:
+  cluster-mapping.json: |-
+  [{
+    "clusterIDMapping": {
+    "site1-storage" (clusterID on site1): "site2-storage" (clusterID on site2)
+   },
+    "RBDPoolIDMapping": [{
+    "1" (poolID on site1): "2" (poolID on site2),
+    "11": "12"
+   }],
+    "CephFSFscIDMapping": [{
+    "13" (FscID on site1): "34" (FscID on site2),
+    "3": "4"
+   }]
+  }, {
+   "clusterIDMapping": {
+   "site3-storage"  (clusterID on site3): "site2-storage" (clusterID on site2)
+   },
+   "RBDPoolIDMapping": [{
+   "5" (poolID on site3): "2" (poolID on site2),
+   "16": "12"
+   }],
+   "CephFSFscIDMapping": [{
+   "3"(FscID on site3): "34" (FscID on site2),
+   "4": "4"
+   }]
+ }]
+metadata:
+  name: ceph-csi-config
+```
+
+**Note:-** the configmap will be mounted as a volume to the CSI (provisioner
+and node plugin) pods.
+
+The above configmap will get created as it is or updated (if new Pools are
+created on the existing cluster) with new entries when the admin choose to
+failover/failback the cluster.
+
+Whenever Ceph-CSI receives a CSI/Replication request it will first decode the
+volumeHandle and try to get the required OMAP details. If it is not able to
+retrieve the poolID or clusterID details from the decoded volumeHandle, Ceph-CSI
+will check for the clusterID and PoolID mapping.
+
+If the old volumeID
+`0001-00013-site1-storage-0000000000000001-b0285c97-a0ce-11eb-8c66-0242ac110002`
+contains the `site1-storage` as the clusterID, now Ceph-CSI will look for the
+corresponding clusterID `site2-storage` from the above configmap. If the
+clusterID mapping is found now Ceph-CSI will look for the poolID mapping ie
+mapping between `1` and `2`.
+
+Example:- pool with the same name exists on both the clusters with different IDs
+Replicapool with ID `1` on site1 and Replicapool with ID `2` on site2.
+
+After getting the required mapping Ceph-CSI has the required information to get
+more details from the rados OMAP. If we have multiple clusterID mapping it will
+loop through all the mapping and checks the corresponding pool to get the OMAP
+data. If the clusterID mapping does not exist Ceph-CSI will return a `Not
+Found` error message to the caller.
+
+After failover to the cluster `site2-storage`, the admin might have created new
+PVCs on the primary cluster `site2-storage`. Later after recovering the
+cluster `site1-storage`, the admin might choose to failback from
+`site2-storage` to `site1-storage`. Now admin needs to copy all the newly
+created kubernetes artifacts to the failback cluster. For clusterID mapping, the
+admin needs to copy the above-created configmap `ceph-clusterid-mapping` to
+the failback cluster. When Ceph-CSI receives a CSI/Replication request for
+the volumes created on the `site2-storage` it will decode the volumeID and
+retrieves the clusterID ie `site2-storage`. In the above configmap
+`ceph-clusterid-mapping` the `site2-storage` is the value and `site1-storage`
+is the key in the `clusterIDMapping` entry.
+
+Ceph-CSI will check both `key` and `value` to check the clusterID mapping. If it
+is found in `key` it will consider `value` as the  corresponding mapping, if it
+is found in `value` place it will treat `key` as the corresponding mapping and
+retrieves all the poolID details of the cluster.
+
+This mapping on the remote cluster is only required when we are doing a
+failover operation from the primary cluster to a remote cluster. The existing
+volumes that are created on the remote cluster does not require
+any mapping as the volumeHandle already contains the required information about
+the local cluster (clusterID, poolID etc).
diff --git a/examples/csi-config-map-sample.yaml b/examples/csi-config-map-sample.yaml
index 524671c69..806abee2f 100644
--- a/examples/csi-config-map-sample.yaml
+++ b/examples/csi-config-map-sample.yaml
@@ -3,6 +3,7 @@
 # as required by the CSI plugins.
 apiVersion: v1
 kind: ConfigMap
+# Lets see the different configuration under config.json key.
 # The <cluster-id> is used by the CSI plugin to uniquely identify and use a
 # Ceph cluster, the value MUST match the value provided as `clusterID` in the
 # StorageClass
@@ -22,6 +23,14 @@ kind: ConfigMap
 # NOTE: Changes to the configmap is automatically updated in the running pods,
 # thus restarting existing pods using the configmap is NOT required on edits
 # to the configmap.
+
+# Lets see the different configuration under cluster-mapping.json key.
+# This configuration is needed when volumes are mirrored using the Ceph-CSI.
+# clusterIDMapping holds the mapping between two clusterId's of storage
+# clusters.
+# RBDPoolIDMapping holds the mapping between two poolId's of storage clusters.
+# CephFSFscIDMapping holds the mapping between two FscId's of storage
+# clusters.
 data:
   config.json: |-
     [
@@ -39,5 +48,21 @@ data:
         }
       }
     ]
+  cluster-mapping.json: |-
+    [
+      {
+        "clusterIDMapping": {
+          "clusterID on site1": "clusterID on site2"
+        },
+        "RBDPoolIDMapping": [{
+          "poolID on site1": "poolID on site2"
+          ...
+        }],
+        "CephFSFscIDMapping": [{
+          "CephFS FscID on site1": "CephFS FscID on site2"
+          ...
+        }]
+      }
+    ]
 metadata:
   name: ceph-csi-config
diff --git a/internal/rbd/rbd_util.go b/internal/rbd/rbd_util.go
index 858cce1bd..8a6a24352 100644
--- a/internal/rbd/rbd_util.go
+++ b/internal/rbd/rbd_util.go
@@ -932,11 +932,11 @@ func genSnapFromSnapID(
 func generateVolumeFromVolumeID(
 	ctx context.Context,
 	volumeID string,
+	vi util.CSIIdentifier,
 	cr *util.Credentials,
 	secrets map[string]string) (*rbdVolume, error) {
 	var (
 		options map[string]string
-		vi      util.CSIIdentifier
 		rbdVol  *rbdVolume
 		err     error
 	)
@@ -946,13 +946,6 @@ func generateVolumeFromVolumeID(
 	//              Mounter, MultiNodeWritable
 	rbdVol = &rbdVolume{}
 	rbdVol.VolID = volumeID
-
-	err = vi.DecomposeCSIID(rbdVol.VolID)
-	if err != nil {
-		return rbdVol, fmt.Errorf("%w: error decoding volume ID (%s) (%s)",
-			ErrInvalidVolID, err, rbdVol.VolID)
-	}
-
 	// TODO check clusterID mapping exists
 
 	rbdVol.ClusterID = vi.ClusterID
@@ -1031,12 +1024,37 @@ func genVolFromVolID(
 	volumeID string,
 	cr *util.Credentials,
 	secrets map[string]string) (*rbdVolume, error) {
-	vol, err := generateVolumeFromVolumeID(ctx, volumeID, cr, secrets)
+	var (
+		vi  util.CSIIdentifier
+		vol *rbdVolume
+	)
+
+	err := vi.DecomposeCSIID(volumeID)
+	if err != nil {
+		return vol, fmt.Errorf("%w: error decoding volume ID (%s) (%s)",
+			ErrInvalidVolID, err, volumeID)
+	}
+
+	vol, err = generateVolumeFromVolumeID(ctx, volumeID, vi, cr, secrets)
 	if !errors.Is(err, util.ErrKeyNotFound) && !errors.Is(err, util.ErrPoolNotFound) &&
 		!errors.Is(err, ErrImageNotFound) {
 		return vol, err
 	}
 
+	// Check clusterID mapping exists
+	mapping, mErr := util.GetClusterMappingInfo(vi.ClusterID)
+	if mErr != nil {
+		return vol, mErr
+	}
+	if mapping != nil {
+		rbdVol, vErr := generateVolumeFromMapping(ctx, mapping, volumeID, vi, cr, secrets)
+		if !errors.Is(vErr, util.ErrKeyNotFound) && !errors.Is(vErr, util.ErrPoolNotFound) &&
+			!errors.Is(vErr, ErrImageNotFound) {
+			return rbdVol, vErr
+		}
+	}
+	// TODO: remove extracting volumeID from PV annotations.
+
 	// If the volume details are not found in the OMAP it can be a mirrored RBD
 	// image and the OMAP is already generated and the volumeHandle might not
 	// be the same in the PV.Spec.CSI.VolumeHandle. Check the PV annotation for
@@ -1054,8 +1072,13 @@ func genVolFromVolID(
 		if pvlist.Items[i].Spec.CSI != nil && pvlist.Items[i].Spec.CSI.VolumeHandle == volumeID {
 			if v, ok := pvlist.Items[i].Annotations[PVVolumeHandleAnnotationKey]; ok {
 				util.UsefulLog(ctx, "found new volumeID %s for existing volumeID %s", v, volumeID)
+				err = vi.DecomposeCSIID(v)
+				if err != nil {
+					return vol, fmt.Errorf("%w: error decoding volume ID (%s) (%s)",
+						ErrInvalidVolID, err, v)
+				}
 
-				return generateVolumeFromVolumeID(ctx, v, cr, secrets)
+				return generateVolumeFromVolumeID(ctx, v, vi, cr, secrets)
 			}
 		}
 	}
@@ -1063,6 +1086,76 @@ func genVolFromVolID(
 	return vol, err
 }
 
+// generateVolumeFromMapping checks the clusterID and poolID mapping and
+// generates retrieves the OMAP information from the poolID got from the
+// mapping.
+func generateVolumeFromMapping(
+	ctx context.Context,
+	mapping *[]util.ClusterMappingInfo,
+	volumeID string,
+	vi util.CSIIdentifier,
+	cr *util.Credentials,
+	secrets map[string]string) (*rbdVolume, error) {
+	nvi := vi
+	vol := &rbdVolume{}
+	// extract clusterID mapping
+	for _, cm := range *mapping {
+		for key, val := range cm.ClusterIDMapping {
+			mappedClusterID := getMappedID(key, val, vi.ClusterID)
+			if mappedClusterID == "" {
+				continue
+			}
+
+			util.DebugLog(ctx,
+				"found new clusterID mapping %s for existing clusterID %s",
+				mappedClusterID,
+				vi.ClusterID)
+			// Add mapping clusterID to Identifier
+			nvi.ClusterID = mappedClusterID
+			poolID := fmt.Sprintf("%d", (vi.LocationID))
+			for _, pools := range cm.RBDpoolIDMappingInfo {
+				for key, val := range pools {
+					mappedPoolID := getMappedID(key, val, poolID)
+					if mappedPoolID == "" {
+						continue
+					}
+					util.DebugLog(ctx,
+						"found new poolID mapping %s for existing pooID %s",
+						mappedPoolID,
+						poolID)
+					pID, err := strconv.ParseInt(mappedPoolID, 10, 64)
+					if err != nil {
+						return vol, err
+					}
+					// Add mapping poolID to Identifier
+					nvi.LocationID = pID
+					vol, err = generateVolumeFromVolumeID(ctx, volumeID, nvi, cr, secrets)
+					if !errors.Is(err, util.ErrKeyNotFound) && !errors.Is(err, util.ErrPoolNotFound) &&
+						!errors.Is(err, ErrImageNotFound) {
+						return vol, err
+					}
+				}
+			}
+		}
+	}
+
+	return vol, util.ErrPoolNotFound
+}
+
+// getMappedID check the input id is matching key or value.
+// If key==id the value will be returned.
+// If value==id the key will be returned.
+func getMappedID(key, value, id string) string {
+	if key == id {
+		return value
+	}
+	if value == id {
+		return key
+	}
+
+	return ""
+}
+
 func genVolFromVolumeOptions(
 	ctx context.Context,
 	volOptions, credentials map[string]string,
diff --git a/internal/util/cluster_mapping.go b/internal/util/cluster_mapping.go
new file mode 100644
index 000000000..51af34f9a
--- /dev/null
+++ b/internal/util/cluster_mapping.go
@@ -0,0 +1,116 @@
+/*
+Copyright 2021 The Ceph-CSI 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 util
+
+import (
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io/ioutil"
+	"os"
+)
+
+// clusterMappingConfigFile is the location of the cluster mapping config file.
+var clusterMappingConfigFile = "/etc/ceph-csi-config/cluster-mapping.json"
+
+// ClusterMappingInfo holds the details of clusterID mapping and poolID mapping.
+type ClusterMappingInfo struct {
+	// ClusterIDMapping holds the details of clusterID mapping
+	ClusterIDMapping map[string]string `json:"clusterIDMapping"`
+	// rbdpoolIDMappingInfo holds the details of RBD poolID mapping.
+	RBDpoolIDMappingInfo []map[string]string `json:"RBDPoolIDMapping"`
+	// cephFSpoolIDMappingInfo holds the details of CephFS Fscid mapping.
+	CephFSFscIDMappingInfo []map[string]string `json:"CephFSFscIDMapping"`
+}
+
+// Expected JSON structure in the passed in config file is,
+// [{
+// 	"clusterIDMapping": {
+// 		"site1-storage": "site2-storage"
+// 	},
+// 	"RBDPoolIDMapping": [{
+// 		"1": "2",
+// 		"11": "12"
+// 	}],
+// 	"CephFSFscIDMapping": [{
+// 		"13": "34",
+// 		"3": "4"
+// 	}]
+// }, {
+// 	"clusterIDMapping": {
+// 		"site3-storage": "site2-storage"
+// 	},
+// 	"RBDPoolIDMapping": [{
+// 		"5": "2",
+// 		"16": "12"
+// 	}],
+// 	"CephFSFscIDMapping": [{
+// 		"3": "34",
+// 		"4": "4"
+// 	}]
+// ...
+// }]
+
+func readClusterMappingInfo() (*[]ClusterMappingInfo, error) {
+	var info []ClusterMappingInfo
+	content, err := ioutil.ReadFile(clusterMappingConfigFile)
+	if err != nil {
+		err = fmt.Errorf("error fetching clusterID mapping %w", err)
+
+		return nil, err
+	}
+
+	err = json.Unmarshal(content, &info)
+	if err != nil {
+		return nil, fmt.Errorf("unmarshal failed (%w), raw buffer response: %s",
+			err, string(content))
+	}
+
+	return &info, nil
+}
+
+// GetClusterMappingInfo returns corresponding cluster details like clusterID's
+// poolID,fscID lists read from configfile.
+func GetClusterMappingInfo(clusterID string) (*[]ClusterMappingInfo, error) {
+	var mappingInfo []ClusterMappingInfo
+	info, err := readClusterMappingInfo()
+	if err != nil {
+		// discard not found error as this file is expected to be created by
+		// the admin in case of failover.
+		if errors.Is(err, os.ErrNotExist) {
+			return nil, nil
+		}
+
+		return nil, fmt.Errorf("failed to fetch cluster mapping: %w", err)
+	}
+	for _, i := range *info {
+		for key, val := range i.ClusterIDMapping {
+			// Same file will be copied to the failover cluster check for both
+			// key and value to check clusterID mapping exists
+			if key == clusterID || val == clusterID {
+				mappingInfo = append(mappingInfo, i)
+			}
+		}
+	}
+
+	// if the mapping is not found return response as nil
+	if len(mappingInfo) == 0 {
+		return nil, nil
+	}
+
+	return &mappingInfo, nil
+}


Test patch:
diff --git a/internal/rbd/rbd_util_test.go b/internal/rbd/rbd_util_test.go
index 705463bba..1dfa35d7d 100644
--- a/internal/rbd/rbd_util_test.go
+++ b/internal/rbd/rbd_util_test.go
@@ -137,3 +137,55 @@ func TestValidateImageFeatures(t *testing.T) {
 		assert.Nil(t, err)
 	}
 }
+
+func TestGetMappedID(t *testing.T) {
+	t.Parallel()
+	type args struct {
+		key   string
+		value string
+		id    string
+	}
+	tests := []struct {
+		name     string
+		args     args
+		expected string
+	}{
+		{
+			name: "test for matching key",
+			args: args{
+				key:   "cluster1",
+				value: "cluster2",
+				id:    "cluster1",
+			},
+			expected: "cluster2",
+		},
+		{
+			name: "test for matching value",
+			args: args{
+				key:   "cluster1",
+				value: "cluster2",
+				id:    "cluster2",
+			},
+			expected: "cluster1",
+		},
+		{
+			name: "test for invalid match",
+			args: args{
+				key:   "cluster1",
+				value: "cluster2",
+				id:    "cluster3",
+			},
+			expected: "",
+		},
+	}
+	for _, tt := range tests {
+		tt := tt
+		t.Run(tt.name, func(t *testing.T) {
+			t.Parallel()
+			val := getMappedID(tt.args.key, tt.args.value, tt.args.id)
+			if val != tt.expected {
+				t.Errorf("getMappedID() got = %v, expected %v", val, tt.expected)
+			}
+		})
+	}
+}
diff --git a/internal/util/cluster_mapping_test.go b/internal/util/cluster_mapping_test.go
new file mode 100644
index 000000000..15c66ed6c
--- /dev/null
+++ b/internal/util/cluster_mapping_test.go
@@ -0,0 +1,249 @@
+/*
+Copyright 2021 The Ceph-CSI 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 util
+
+import (
+	"encoding/json"
+	"fmt"
+	"io/ioutil"
+	"reflect"
+	"testing"
+)
+
+func TestGetClusterMappingInfo(t *testing.T) {
+	t.Parallel()
+	mappingBasePath := t.TempDir()
+
+	// clusterID,poolID on site1
+	clusterIDOfSite1 := "site1-storage"
+	rbdPoolIDOfSite1 := "1"
+	cephfsFscIDOfSite1 := "11"
+
+	// clusterID,poolID on site2
+	clusterIDOfSite2 := "site2-storage"
+	rbdPoolIDOfSite2 := "3"
+	cephfsFscIDOfSite2 := "5"
+
+	// clusterID,poolID on site3
+	clusterIDOfSite3 := "site3-storage"
+	rbdPoolIDOfSite3 := "8"
+	cephfsFscIDOfSite3 := "10"
+
+	clusterMappingInfos := make([]ClusterMappingInfo, 2)
+
+	// create mapping between site1 and site2
+	clusterIDmappingOfSite1To2 := make(map[string]string)
+	clusterIDmappingOfSite1To2[clusterIDOfSite1] = clusterIDOfSite2
+	rbdMappingOfSite1To2 := make([]map[string]string, 1)
+	rbdMappingOfSite1To2[0] = map[string]string{rbdPoolIDOfSite1: rbdPoolIDOfSite2}
+	cephFSMappingOfSite1To2 := make([]map[string]string, 1)
+	cephFSMappingOfSite1To2[0] = map[string]string{cephfsFscIDOfSite1: cephfsFscIDOfSite2}
+	mappingOfSite1To2 := ClusterMappingInfo{
+		clusterIDmappingOfSite1To2,
+		rbdMappingOfSite1To2,
+		cephFSMappingOfSite1To2,
+	}
+
+	// create mapping between site3 and site2
+	clusterIDmappingOfSite3To2 := make(map[string]string)
+	clusterIDmappingOfSite3To2[clusterIDOfSite3] = clusterIDOfSite2
+	rbdMappingOfSite3To2 := make([]map[string]string, 1)
+	rbdMappingOfSite3To2[0] = map[string]string{rbdPoolIDOfSite3: rbdPoolIDOfSite2}
+	cephFSMappingOfSite3To2 := make([]map[string]string, 1)
+	cephFSMappingOfSite3To2[0] = map[string]string{cephfsFscIDOfSite3: cephfsFscIDOfSite2}
+	mappingOfSite3To2 := ClusterMappingInfo{
+		clusterIDmappingOfSite3To2,
+		rbdMappingOfSite3To2,
+		cephFSMappingOfSite3To2,
+	}
+
+	clusterMappingInfos[0] = mappingOfSite1To2
+	clusterMappingInfos[1] = mappingOfSite3To2
+
+	mappingFileContent, err := json.Marshal(clusterMappingInfos)
+	if err != nil {
+		t.Errorf("failed to marshal mapping info %v", err)
+	}
+	// expected output of mapping
+	expectedSite2Data := clusterMappingInfos
+	expectedSite1To2Data := clusterMappingInfos[:1]
+	expectedSite3To2Data := clusterMappingInfos[1:]
+
+	tests := []struct {
+		name                    string
+		clusterID               string
+		mappingFilecontent      []byte
+		expectedData            *[]ClusterMappingInfo
+		createMappingConfigFile bool
+		expectErr               bool
+	}{
+		{
+			name:                    "mapping file not found",
+			clusterID:               "site-a-clusterid",
+			createMappingConfigFile: false,
+			mappingFilecontent:      []byte{},
+			expectedData:            nil,
+			expectErr:               false,
+		},
+		{
+			name:                    "mapping file found with empty data",
+			clusterID:               "site-a-clusterid",
+			createMappingConfigFile: true,
+			mappingFilecontent:      []byte{},
+			expectedData:            nil,
+			expectErr:               false,
+		},
+		{
+			name:                    "cluster-id mapping not found",
+			clusterID:               "site-a-clusterid",
+			createMappingConfigFile: true,
+			mappingFilecontent:      mappingFileContent,
+			expectedData:            nil,
+			expectErr:               false,
+		},
+		{
+			name:                    "site2-storage cluster-id mapping",
+			clusterID:               clusterIDOfSite2,
+			createMappingConfigFile: true,
+			mappingFilecontent:      mappingFileContent,
+			expectedData:            &expectedSite2Data,
+			expectErr:               false,
+		},
+		{
+			name:                    "site1-storage cluster-id mapping",
+			clusterID:               clusterIDOfSite1,
+			createMappingConfigFile: true,
+			mappingFilecontent:      mappingFileContent,
+			expectedData:            &expectedSite1To2Data,
+			expectErr:               false,
+		},
+		{
+			name:                    "site3-storage cluster-id mapping",
+			clusterID:               clusterIDOfSite3,
+			createMappingConfigFile: true,
+			mappingFilecontent:      mappingFileContent,
+			expectedData:            &expectedSite3To2Data,
+			expectErr:               false,
+		},
+	}
+	for _, tt := range tests {
+		tt := tt
+		t.Run(tt.name, func(t *testing.T) {
+			t.Parallel()
+			if tt.createMappingConfigFile {
+				clusterMappingConfigFile = fmt.Sprintf("%s/mapping.json", mappingBasePath)
+			}
+			if len(tt.mappingFilecontent) != 0 {
+				err = ioutil.WriteFile(clusterMappingConfigFile, tt.mappingFilecontent, 0o600)
+				if err != nil {
+					t.Errorf("GetClusterMappingInfo() error = %v", err)
+				}
+			}
+			data, mErr := GetClusterMappingInfo(tt.clusterID)
+			if (mErr != nil) != tt.expectErr {
+				t.Errorf("GetClusterMappingInfo() error = %v, expected Error %v", mErr, tt.expectErr)
+			}
+			if !reflect.DeepEqual(data, tt.expectedData) {
+				t.Errorf("GetClusterMappingInfo() = %v, expected data %v", data, tt.expectedData)
+			}
+		})
+	}
+
+	clusterMappingConfigFile = fmt.Sprintf("%s/mapping.json", mappingBasePath)
+	err = ioutil.WriteFile(clusterMappingConfigFile, mappingFileContent, 0o600)
+	if err != nil {
+		t.Errorf("failed to write mapping content error = %v", err)
+	}
+
+	// validate site-3 to site-2 and site-1 to site-2 mappings when failover to
+	// site-2.
+	// The volumeId's from site-1 looks like `0001-0013-site1-storage-xyz` we
+	// need to have validate `site1-storage` to `site2-storage` mapping exists
+	// The volumeId's from site-3 looks like `0001-0013-site3-storage-xyz` we
+	// need to have validate `site3-storage` to `site2-storage` mapping exists
+	mappedClusterCount := 2
+	err = validateMapping(t, clusterIDOfSite2, rbdPoolIDOfSite2, cephfsFscIDOfSite2, mappedClusterCount)
+	if err != nil {
+		t.Error(err)
+	}
+	// validate site-2 and site-1 mappings when failback to site-1.
+	// The volumeId's from site-2 looks like `0001-0013-site2-storage-xyz` we
+	// need to have validate `site2-storage` to `site3-storage` mapping exists
+	mappedClusterCount = 1
+	err = validateMapping(t, clusterIDOfSite1, rbdPoolIDOfSite1, cephfsFscIDOfSite1, mappedClusterCount)
+	if err != nil {
+		t.Error(err)
+	}
+	// validate site-2 and site-3 mappings when failback to site-3
+	// The volumeId's from site-2 looks like `0001-0013-site2-storage-xyz` we
+	// need to have validate `site2-storage` to `site3-storage` mapping exists
+	mappedClusterCount = 1
+	err = validateMapping(t, clusterIDOfSite3, rbdPoolIDOfSite3, cephfsFscIDOfSite3, mappedClusterCount)
+	if err != nil {
+		t.Error(err)
+	}
+}
+
+func validateMapping(t *testing.T, clusterID, rbdPoolID, cephFSPoolID string, mappingCount int) error {
+	t.Helper()
+
+	mapping, err := GetClusterMappingInfo(clusterID)
+	if err != nil {
+		return fmt.Errorf("failed to retrieve mapping %w", err)
+	}
+	// verify we are able to retrieve both site-1:site2 and site-2:site3 mapping
+	if mapping == nil || len(*mapping) != mappingCount {
+		return fmt.Errorf(
+			"clusterID mapping got length=%v, expected length=%v",
+			len(*mapping),
+			mappingCount)
+	}
+	// check mapping rbd pool mapping exists in mapping
+	foundRBDPoolMappingCount := 0
+	foundCephFSPoolMappingCount := 0
+	for _, c := range *mapping {
+		for _, rp := range c.RBDpoolIDMappingInfo {
+			for k, v := range rp {
+				if k == rbdPoolID || v == rbdPoolID {
+					foundRBDPoolMappingCount++
+				}
+			}
+		}
+
+		for _, cp := range c.CephFSFscIDMappingInfo {
+			for k, v := range cp {
+				if k == cephFSPoolID || v == cephFSPoolID {
+					foundCephFSPoolMappingCount++
+				}
+			}
+		}
+	}
+	if foundRBDPoolMappingCount != mappingCount {
+		return fmt.Errorf(
+			"rbd pool mapping got length= %v, expected length=%v",
+			foundRBDPoolMappingCount,
+			mappingCount)
+	}
+	if foundCephFSPoolMappingCount != mappingCount {
+		return fmt.Errorf(
+			"cephFS filesystem mapping got length= %v, expected length=%v",
+			foundCephFSPoolMappingCount,
+			mappingCount)
+	}
+
+	return nil
+}