Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

rbd: get volumegroup in secondary cluster #5118

Open
wants to merge 1 commit into
base: devel
Choose a base branch
from
+109 −12
Open

rbd: get volumegroup in secondary cluster #5118

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
121 changes: 109 additions & 12 deletions internal/rbd/group/util.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@ import (
"context"
"errors"
"fmt"
"strconv"
"time"

"github.com/ceph/go-ceph/rados"
Expand Down Expand Up @@ -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
Expand All @@ -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) {
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

you need to make use of the following function instead of singular pool not found check ?

ceph-csi/internal/rbd/rbd_util.go

Line 1308 in c45bc7d

func shouldRetryVolumeGeneration(err error) bool {

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
Expand Down