Repository: ceph/ceph-csi
Pull request #2314 (2021-07-22 11:03:03)

RBD Regional DR: Implement clusterID mapping 
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 receives a `Not Found` error message and Ceph-CSI will check for the clusterID 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 clusterID mapping file. If the clusterID mapping is found now Ceph-CSI will look for the poolID mapping ie mapping between `1` and `2`. Example:- pool with 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.

Fixes: #2118
Depends-on: #2292

## Note to Reviewers

Up for initial review

## TODO:

- [x] Add example configmap yaml
- [x] Update required deployment templates to mount the configmap as volume
- [x] E2E testing

Madhu-1: /retest ci/centos/upgrade-tests-rbd
Madhu-1: > /retest ci/centos/upgrade-tests-rbd

`X Exiting due to GUEST_PROVISION: Failed to start host: creating host: create: Error creating machine: Error in driver during machine creation: IP not available after waiting: machine minikube didn't return IP after 1 minute
`
Madhu-1: @nixpanic manually rebased the PR and we store FsID not pool ID for cephfs volumes. Updated the examples now PTAL.
Madhu-1: /test ci/centos/upgrade-tests-cephfs
Madhu-1: > /test ci/centos/upgrade-tests-cephfs

`X Exiting due to GUEST_PROVISION: Failed to start host: creating host: create: Error creating machine: Error in driver during machine creation: IP not available after waiting: machine minikube didn't return IP after 1 minute
`
Madhu-1: @humblec @nixpanic PTAL Thanks.
Madhu-1: @ShyamsundarR @agarwal-mudit can you please help with a review here?
Madhu-1: @mergifyio rebase
mergify[bot]: **Command `rebase`: success**
> **Branch has been successfully rebased**
> 


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 7993927dc..19a6e134f 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
index 04f37c9b1..acb734bcb 100644
--- a/docs/design/proposals/clusterid-mapping.md
+++ b/docs/design/proposals/clusterid-mapping.md
@@ -103,8 +103,8 @@ data:
     "1" (poolID on site1): "2" (poolID on site2),
     "11": "12"
    }],
-    "CephFSFsIDMapping": [{
-    "13" (FsID on site1): "34" (FsID on site2),
+    "CephFSFscIDMapping": [{
+    "13" (FscID on site1): "34" (FscID on site2),
     "3": "4"
    }]
   }, {
@@ -115,8 +115,8 @@ data:
    "5" (poolID on site3): "2" (poolID on site2),
    "16": "12"
    }],
-   "CephFSFsIDMapping": [{
-   "3"(FsID on site3): "34" (FsID on site2),
+   "CephFSFscIDMapping": [{
+   "3"(FscID on site3): "34" (FscID on site2),
    "4": "4"
    }]
  }]
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
+}