rbd: use VolumeGroup.GetName() to check if it exists already

When repeated CreateVolumeGroup requests come in, it is needed to check
if the VolumeGroup already exists in the backend. If it does, there is
no need to create it again (and fail).

Signed-off-by: Niels de Vos <ndevos@ibm.com>

internal/rbd/manager.go

@@ -202,9 +202,16 @@ func (mgr *rbdManager) CreateVolumeGroup(ctx context.Context, name string) (type
 		return nil, fmt.Errorf("failed to get volume group %q at cluster %q: %w", name, clusterID, err)
 	}
 
-	err = vg.Create(ctx, vgName)
-	if err != nil {
-		return nil, fmt.Errorf("failed to create volume group %q: %w", name, err)
+	// check if the volume group exists in the backend
+	existingName, err := vg.GetName(ctx)
+	if err != nil {
+		// the volume group does not exist yet
+		err = vg.Create(ctx, vgName)
+		if err != nil {
+			return nil, fmt.Errorf("failed to create volume group %q: %w", name, err)
+		}
+	} else if existingName != vgName {
+		return nil, fmt.Errorf("volume group id %q has a name mismatch, expected %q, not %q", name, vgName, existingName)
 	}
 
 	// TODO: create the volume group in the journal

internal/rbd_group/volume_group.go

@@ -77,6 +77,8 @@ var (
 
 // GetVolumeGroup initializes a new VolumeGroup object that can be used
 // to manage an `rbd group`.
+// If the .GetName() function returns an error, the VolumeGroup does not exist
+// yet. It is needed to call .Create() in that case first.
 func GetVolumeGroup(
 	ctx context.Context,
 	id string,
@@ -107,13 +109,22 @@ func GetVolumeGroup(
 
 	attrs, err := j.GetVolumeGroupAttributes(ctx, pool, csiID.ObjectUUID)
 	if err != nil {
-		return nil, fmt.Errorf("failed to get attributes for volume group id %q: %w", id, err)
+		if !errors.Is(err, util.ErrKeyNotFound) && !errors.Is(err, util.ErrPoolNotFound) {
+			return nil, fmt.Errorf("failed to get attributes for volume group id %q: %w", id, err)
+		}
+
+		attrs = &journal.VolumeGroupAttributes{}
 	}
 
 	var volumes []types.Volume
 	for volID := range attrs.VolumeMap {
 		vol, err := volumeResolver.GetVolumeByID(ctx, volID)
 		if err != nil {
+			// free the previously allocated volumes
+			for _, v := range volumes {
+				v.Destroy(ctx)
+			}
+
 			return nil, fmt.Errorf("failed to get attributes for volume group id %q: %w", id, err)
 		}
 
@@ -285,18 +296,24 @@ func (vg *volumeGroup) Create(ctx context.Context, name string) error {
 		return fmt.Errorf("failed to create volume group %q: %w", name, err)
 	}
 
-	log.DebugLog(ctx, "volume group %q has been created", name)
+	vg.name = name
+	log.DebugLog(ctx, "volume group %q has been created", vg)
 
 	return nil
 }
 
 func (vg *volumeGroup) Delete(ctx context.Context) error {
+	name, err := vg.GetName(ctx)
+	if err != nil {
+		return err
+	}
+
 	ioctx, err := vg.GetIOContext(ctx)
 	if err != nil {
 		return err
 	}
 
-	err = librbd.GroupRemove(ioctx, vg.name)
+	err = librbd.GroupRemove(ioctx, name)
 	if err != nil {
 		return fmt.Errorf("failed to remove volume group %q: %w", vg, err)
 	}