From 13817b75c4620de018d13817f3377b17235eeca3 Mon Sep 17 00:00:00 2001
From: Praveen M <m.praveen@ibm.com>
Date: Sun, 19 Jan 2025 21:55:09 +0530
Subject: [PATCH] rbd: get volumegroup in secondary cluster

Currently, `GetVolumeGroup()` fetches the RBD group from the
pool using the clusterID & poolID encoded in the VolumeGroupHandle.
However, this approach may fail in a secondary mirrored cluster,
where the clusterID & poolID could differ.

This commit ensures that `GetVolumeGroup` leverages the
clusterIDMapping and RBDPoolIDMapping to locate the RBD group in the
appropriate  pool if it is not found in the pool corresponding
to the poolID encoded in the VolumeGroupHandle.

Signed-off-by: Praveen M <m.praveen@ibm.com>
---
 internal/rbd/group/util.go | 121 +++++++++++++++++++++++++++++++++----
 1 file changed, 109 insertions(+), 12 deletions(-)

diff --git a/internal/rbd/group/util.go b/internal/rbd/group/util.go
index 6c4e56d5dd9..7a037973ede 100644
--- a/internal/rbd/group/util.go
+++ b/internal/rbd/group/util.go
@@ -20,6 +20,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"strconv"
 	"time"
 
 	"github.com/ceph/go-ceph/rados"
@@ -63,31 +64,27 @@ type commonVolumeGroup struct {
 	journal journal.VolumeGroupJournal
 }
 
-func (cvg *commonVolumeGroup) initCommonVolumeGroup(
-	ctx context.Context,
+// generateVolumeGroup generates a commonVolumeGroup structure from the volumeGroup identifier.
+func generateVolumeGroup(
 	id string,
 	csiDriver string,
 	creds *util.Credentials,
-) error {
-	csiID := util.CSIIdentifier{}
-	err := csiID.DecomposeCSIID(id)
-	if err != nil {
-		return fmt.Errorf("failed to decompose volume group id %q: %w", id, err)
-	}
-
+	csiID util.CSIIdentifier,
+) (*commonVolumeGroup, error) {
+	cvg := &commonVolumeGroup{}
 	mons, err := util.Mons(util.CsiConfigFile, csiID.ClusterID)
 	if err != nil {
-		return fmt.Errorf("failed to get MONs for cluster id %q: %w", csiID.ClusterID, err)
+		return nil, fmt.Errorf("failed to get MONs for cluster id %q: %w", csiID.ClusterID, err)
 	}
 
 	namespace, err := util.GetRBDRadosNamespace(util.CsiConfigFile, csiID.ClusterID)
 	if err != nil {
-		return fmt.Errorf("failed to get RADOS namespace for cluster id %q: %w", csiID.ClusterID, err)
+		return nil, fmt.Errorf("failed to get RADOS namespace for cluster id %q: %w", csiID.ClusterID, err)
 	}
 
 	pool, err := util.GetPoolName(mons, creds, csiID.LocationID)
 	if err != nil {
-		return fmt.Errorf("failed to get pool for volume group id %q: %w", id, err)
+		return nil, fmt.Errorf("failed to get pool for volume group id %q: %w", id, err)
 	}
 
 	cvg.csiDriver = csiDriver
@@ -99,6 +96,106 @@ func (cvg *commonVolumeGroup) initCommonVolumeGroup(
 	cvg.pool = pool
 	cvg.namespace = namespace
 
+	return cvg, nil
+}
+
+// generateVolumeGroupFromMapping checks the clusterID and poolID mapping and
+// generates commonVolumeGroup structure for the mapped clusterID and poolID.
+func generateVolumeGroupFromMapping(
+	ctx context.Context,
+	id string,
+	csiDriver string,
+	creds *util.Credentials,
+	csiID util.CSIIdentifier,
+	mapping *[]util.ClusterMappingInfo,
+) (*commonVolumeGroup, error) {
+	cvg := &commonVolumeGroup{}
+	mcsiID := csiID
+	existingClusterID := csiID.ClusterID
+	existingPoolID := strconv.FormatInt(csiID.LocationID, 10)
+
+	for _, cm := range *mapping {
+		for key, val := range cm.ClusterIDMapping {
+			mappedClusterID := util.GetMappedID(key, val, csiID.ClusterID)
+			if mappedClusterID == "" {
+				continue
+			}
+
+			log.DebugLog(ctx,
+				"found new clusterID mapping %s for existing clusterID %s", mappedClusterID, existingClusterID)
+
+			// Add mapped clusterID to Identifier
+			mcsiID.ClusterID = mappedClusterID
+			for _, pools := range cm.RBDpoolIDMappingInfo {
+				for key, val := range pools {
+					mappedPoolID := util.GetMappedID(key, val, existingPoolID)
+					if mappedPoolID == "" {
+						continue
+					}
+					log.DebugLog(ctx,
+						"found new poolID mapping %s for existing poolID %s", mappedPoolID, existingPoolID)
+
+					mPID, err := strconv.ParseInt(mappedPoolID, 10, 64)
+					if err != nil {
+						return cvg, err
+					}
+					mcsiID.LocationID = mPID
+					cvg, err = generateVolumeGroup(id, csiDriver, creds, mcsiID)
+					if err != nil && !errors.Is(err, util.ErrPoolNotFound) {
+						return cvg, err
+					}
+					// If the pool is found, return the volume group
+					if cvg != nil {
+						return cvg, nil
+					}
+				}
+			}
+		}
+	}
+
+	return nil, util.ErrPoolNotFound
+}
+
+func (cvg *commonVolumeGroup) initCommonVolumeGroup(
+	ctx context.Context,
+	id string,
+	csiDriver string,
+	creds *util.Credentials,
+) error {
+	csiID := util.CSIIdentifier{}
+
+	err := csiID.DecomposeCSIID(id)
+	if err != nil {
+		return fmt.Errorf("failed to decompose volume group id %q: %w", id, err)
+	}
+
+	vg, err := generateVolumeGroup(id, csiDriver, creds, csiID)
+	if err != nil && !errors.Is(err, util.ErrPoolNotFound) {
+		return err
+	}
+
+	if err != nil && errors.Is(err, util.ErrPoolNotFound) {
+		mapping, err := util.GetClusterMappingInfo(csiID.ClusterID)
+		if err != nil {
+			return err
+		}
+		if mapping != nil {
+			vg, err = generateVolumeGroupFromMapping(ctx, id, csiDriver, creds, csiID, mapping)
+			if err != nil {
+				return err
+			}
+		}
+	}
+
+	cvg.csiDriver = vg.csiDriver
+	cvg.credentials = vg.credentials
+	cvg.id = vg.id
+	cvg.clusterID = vg.clusterID
+	cvg.objectUUID = vg.objectUUID
+	cvg.monitors = vg.monitors
+	cvg.pool = vg.pool
+	cvg.namespace = vg.namespace
+
 	log.DebugLog(ctx, "object for volume group %q has been initialized", cvg.id)
 
 	return nil