rbd: check for clusterid mapping in genVolFromVolumeOptions()

This commit adds capability to genVolFromVolumeOptions() to fetch
mapped clusted-id & mon ips for mirrored PVC on secondary cluster
which may have different cluster-id.

This is required for NodeStageVolume().

We also don't need to check for mapping during volume create requests,
so it can be disabled by passing a bool checkClusterIDMapping as false.

GetMonsAndClusterID() is modified to accept bool checkClusterIDMapping
based on which clustermapping is checked to fetch mapped cluster-id and
mon-ips.

Signed-off-by: Rakshith R <rar@redhat.com>

internal/cephfs/volumeoptions.go

@@ -595,7 +595,7 @@ func genSnapFromOptions(ctx context.Context, req *csi.CreateSnapshotRequest) (sn
 	cephfsSnap.RequestName = req.GetName()
 	snapOptions := req.GetParameters()
 
-	cephfsSnap.Monitors, cephfsSnap.ClusterID, err = util.GetMonsAndClusterID(snapOptions)
+	cephfsSnap.Monitors, cephfsSnap.ClusterID, err = util.GetMonsAndClusterID(ctx, snapOptions, false)
 	if err != nil {
 		log.ErrorLog(ctx, "failed getting mons (%s)", err)
 

internal/rbd/controllerserver.go

@@ -123,7 +123,10 @@ func (cs *ControllerServer) parseVolCreateRequest(
 	}
 
 	// if it's NOT SINGLE_NODE_WRITER and it's BLOCK we'll set the parameter to ignore the in-use checks
-	rbdVol, err := genVolFromVolumeOptions(ctx, req.GetParameters(), req.GetSecrets(), (isMultiNode && isBlock))
+	rbdVol, err := genVolFromVolumeOptions(
+		ctx,
+		req.GetParameters(), req.GetSecrets(),
+		(isMultiNode && isBlock), false)
 	if err != nil {
 		return nil, status.Error(codes.InvalidArgument, err.Error())
 	}

internal/rbd/nodeserver.go

@@ -178,7 +178,7 @@ func populateRbdVol(
 		disableInUseChecks = true
 	}
 
-	rv, err := genVolFromVolumeOptions(ctx, req.GetVolumeContext(), req.GetSecrets(), disableInUseChecks)
+	rv, err := genVolFromVolumeOptions(ctx, req.GetVolumeContext(), req.GetSecrets(), disableInUseChecks, true)
 	if err != nil {
 		return nil, status.Error(codes.Internal, err.Error())
 	}

internal/rbd/rbd_util.go

@@ -866,7 +866,7 @@ func genSnapFromSnapID(
 	rbdSnap.ClusterID = vi.ClusterID
 	options["clusterID"] = rbdSnap.ClusterID
 
-	rbdSnap.Monitors, _, err = util.GetMonsAndClusterID(options)
+	rbdSnap.Monitors, _, err = util.GetMonsAndClusterID(ctx, options, false)
 	if err != nil {
 		log.ErrorLog(ctx, "failed getting mons (%s)", err)
 
@@ -950,12 +950,11 @@ func generateVolumeFromVolumeID(
 	//              Mounter, MultiNodeWritable
 	rbdVol = &rbdVolume{}
 	rbdVol.VolID = volumeID
-	// TODO check clusterID mapping exists
 
 	rbdVol.ClusterID = vi.ClusterID
 	options["clusterID"] = rbdVol.ClusterID
 
-	rbdVol.Monitors, _, err = util.GetMonsAndClusterID(options)
+	rbdVol.Monitors, _, err = util.GetMonsAndClusterID(ctx, options, false)
 	if err != nil {
 		log.ErrorLog(ctx, "failed getting mons (%s)", err)
 
@@ -1153,7 +1152,7 @@ func generateVolumeFromMapping(
 func genVolFromVolumeOptions(
 	ctx context.Context,
 	volOptions, credentials map[string]string,
-	disableInUseChecks bool) (*rbdVolume, error) {
+	disableInUseChecks, checkClusterIDMapping bool) (*rbdVolume, error) {
 	var (
 		ok         bool
 		err        error
@@ -1171,7 +1170,7 @@ func genVolFromVolumeOptions(
 		rbdVol.NamePrefix = namePrefix
 	}
 
-	rbdVol.Monitors, rbdVol.ClusterID, err = util.GetMonsAndClusterID(volOptions)
+	rbdVol.Monitors, rbdVol.ClusterID, err = util.GetMonsAndClusterID(ctx, volOptions, checkClusterIDMapping)
 	if err != nil {
 		log.ErrorLog(ctx, "failed getting mons (%s)", err)
 
@@ -1248,7 +1247,7 @@ func genSnapFromOptions(ctx context.Context, rbdVol *rbdVolume, snapOptions map[
 	rbdSnap.JournalPool = rbdVol.JournalPool
 	rbdSnap.RadosNamespace = rbdVol.RadosNamespace
 
-	rbdSnap.Monitors, rbdSnap.ClusterID, err = util.GetMonsAndClusterID(snapOptions)
+	rbdSnap.Monitors, rbdSnap.ClusterID, err = util.GetMonsAndClusterID(ctx, snapOptions, false)
 	if err != nil {
 		log.ErrorLog(ctx, "failed getting mons (%s)", err)
 

internal/util/csiconfig.go

@@ -17,6 +17,7 @@ limitations under the License.
 package util
 
 import (
+	"context"
 	"encoding/json"
 	"errors"
 	"fmt"
@@ -131,11 +132,19 @@ func CephFSSubvolumeGroup(pathToConfig, clusterID string) (string, error) {
 
 // GetMonsAndClusterID returns monitors and clusterID information read from
 // configfile.
-func GetMonsAndClusterID(options map[string]string) (string, string, error) {
+func GetMonsAndClusterID(ctx context.Context, options map[string]string, checkClusterIDMapping bool) (string, string, error) {
 	clusterID, ok := options["clusterID"]
 	if !ok {
 		return "", "", errors.New("clusterID must be set")
 	}
+	if checkClusterIDMapping {
+		monitors, mappedClusterID, err := FetchMappedClusterIDAndMons(ctx, clusterID)
+		if err != nil {
+			return "", "", err
+		}
+
+		return monitors, mappedClusterID, nil
+	}
 
 	monitors, err := Mons(CsiConfigFile, clusterID)
 	if err != nil {