RBD: improvments to existing code to better align with volumegroup

go.mod

@@ -9,7 +9,7 @@ require (
 	github.com/ceph/ceph-csi/api v0.0.0-00010101000000-000000000000
 	github.com/ceph/go-ceph v0.28.0
 	github.com/container-storage-interface/spec v1.10.0
-	github.com/csi-addons/spec v0.2.1-0.20240718113938-dc98b454ba65
+	github.com/csi-addons/spec v0.2.1-0.20240730084235-3958a5b17d24
 	github.com/gemalto/kmip-go v0.0.10
 	github.com/golang/protobuf v1.5.4
 	github.com/google/fscrypt v0.3.6-0.20240502174735-068b9f8f5dec

go.sum

@@ -913,6 +913,8 @@ github.com/creack/pty v1.1.11/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ
 github.com/creack/pty v1.1.18/go.mod h1:MOBLtS5ELjhRRrroQr9kyvTxUAFNvYEK993ew/Vr4O4=
 github.com/csi-addons/spec v0.2.1-0.20240718113938-dc98b454ba65 h1:i9JGGQTEmRQXSpQQPR96+DV4D4o+V1+gjAWf+bpxQxk=
 github.com/csi-addons/spec v0.2.1-0.20240718113938-dc98b454ba65/go.mod h1:Mwq4iLiUV4s+K1bszcWU6aMsR5KPsbIYzzszJ6+56vI=
+github.com/csi-addons/spec v0.2.1-0.20240730084235-3958a5b17d24 h1:tJETaYbnnzlCSaqDXQzbszYyuAtG/sFzm6DargeVzJA=
+github.com/csi-addons/spec v0.2.1-0.20240730084235-3958a5b17d24/go.mod h1:Mwq4iLiUV4s+K1bszcWU6aMsR5KPsbIYzzszJ6+56vI=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc h1:U9qPSI2PIWSS1VwoXQT9A3Wy9MM3WgvqSxFWenqJduM=

internal/csi-addons/rbd/replication.go

@@ -282,16 +282,7 @@ func (rs *ReplicationServer) EnableVolumeReplication(ctx context.Context,
 
 	rbdVol, err := mgr.GetVolumeByID(ctx, volumeID)
 	if err != nil {
-		switch {
-		case errors.Is(err, corerbd.ErrImageNotFound):
-			err = status.Errorf(codes.NotFound, "volume %s not found", volumeID)
-		case errors.Is(err, util.ErrPoolNotFound):
-			err = status.Errorf(codes.NotFound, "pool %s not found for %s", rbdVol.GetPoolName(), volumeID)
-		default:
-			err = status.Errorf(codes.Internal, err.Error())
-		}
-
-		return nil, err
+		return nil, getGRPCError(err)
 	}
 	mirror, err := rbdVol.ToMirror()
 	if err != nil {
@@ -309,7 +300,7 @@ func (rs *ReplicationServer) EnableVolumeReplication(ctx context.Context,
 		return nil, err
 	}
 
-	info, err := mirror.GetMirroringInfo()
+	info, err := mirror.GetMirroringInfo(ctx)
 	if err != nil {
 		log.ErrorLog(ctx, err.Error())
 
@@ -322,7 +313,7 @@ func (rs *ReplicationServer) EnableVolumeReplication(ctx context.Context,
 
 			return nil, getGRPCError(err)
 		}
-		err = mirror.EnableMirroring(mirroringMode)
+		err = mirror.EnableMirroring(ctx, mirroringMode)
 		if err != nil {
 			log.ErrorLog(ctx, err.Error())
 
@@ -361,16 +352,7 @@ func (rs *ReplicationServer) DisableVolumeReplication(ctx context.Context,
 
 	rbdVol, err := mgr.GetVolumeByID(ctx, volumeID)
 	if err != nil {
-		switch {
-		case errors.Is(err, corerbd.ErrImageNotFound):
-			err = status.Errorf(codes.NotFound, "volume %s not found", volumeID)
-		case errors.Is(err, util.ErrPoolNotFound):
-			err = status.Errorf(codes.NotFound, "pool %s not found for %s", rbdVol.GetPoolName(), volumeID)
-		default:
-			err = status.Errorf(codes.Internal, err.Error())
-		}
-
-		return nil, err
+		return nil, getGRPCError(err)
 	}
 	mirror, err := rbdVol.ToMirror()
 	if err != nil {
@@ -383,7 +365,7 @@ func (rs *ReplicationServer) DisableVolumeReplication(ctx context.Context,
 		return nil, err
 	}
 
-	info, err := mirror.GetMirroringInfo()
+	info, err := mirror.GetMirroringInfo(ctx)
 	if err != nil {
 		log.ErrorLog(ctx, err.Error())
 
@@ -396,7 +378,7 @@ func (rs *ReplicationServer) DisableVolumeReplication(ctx context.Context,
 	case librbd.MirrorImageDisabling.String():
 		return nil, status.Errorf(codes.Aborted, "%s is in disabling state", volumeID)
 	case librbd.MirrorImageEnabled.String():
-		err = corerbd.DisableVolumeReplication(mirror, info.IsPrimary(), force)
+		err = corerbd.DisableVolumeReplication(mirror, ctx, info.IsPrimary(), force)
 		if err != nil {
 			return nil, getGRPCError(err)
 		}
@@ -438,23 +420,14 @@ func (rs *ReplicationServer) PromoteVolume(ctx context.Context,
 
 	rbdVol, err := mgr.GetVolumeByID(ctx, volumeID)
 	if err != nil {
-		switch {
-		case errors.Is(err, corerbd.ErrImageNotFound):
-			err = status.Errorf(codes.NotFound, "volume %s not found", volumeID)
-		case errors.Is(err, util.ErrPoolNotFound):
-			err = status.Errorf(codes.NotFound, "pool %s not found for %s", rbdVol.GetPoolName(), volumeID)
-		default:
-			err = status.Errorf(codes.Internal, err.Error())
-		}
-
-		return nil, err
+		return nil, getGRPCError(err)
 	}
 	mirror, err := rbdVol.ToMirror()
 	if err != nil {
 		return nil, status.Error(codes.Internal, err.Error())
 	}
 
-	info, err := mirror.GetMirroringInfo()
+	info, err := mirror.GetMirroringInfo(ctx)
 	if err != nil {
 		log.ErrorLog(ctx, err.Error())
 
@@ -474,9 +447,9 @@ func (rs *ReplicationServer) PromoteVolume(ctx context.Context,
 		if req.GetForce() {
 			// workaround for https://github.com/ceph/ceph-csi/issues/2736
 			// TODO: remove this workaround when the issue is fixed
-			err = mirror.ForcePromote(cr)
+			err = mirror.ForcePromote(ctx, cr)
 		} else {
-			err = mirror.Promote(req.GetForce())
+			err = mirror.Promote(ctx, req.GetForce())
 		}
 		if err != nil {
 			log.ErrorLog(ctx, err.Error())
@@ -540,16 +513,7 @@ func (rs *ReplicationServer) DemoteVolume(ctx context.Context,
 
 	rbdVol, err := mgr.GetVolumeByID(ctx, volumeID)
 	if err != nil {
-		switch {
-		case errors.Is(err, corerbd.ErrImageNotFound):
-			err = status.Errorf(codes.NotFound, "volume %s not found", volumeID)
-		case errors.Is(err, util.ErrPoolNotFound):
-			err = status.Errorf(codes.NotFound, "pool %s not found for %s", rbdVol.GetPoolName(), volumeID)
-		default:
-			err = status.Errorf(codes.Internal, err.Error())
-		}
-
-		return nil, err
+		return nil, getGRPCError(err)
 	}
 	mirror, err := rbdVol.ToMirror()
 	if err != nil {
@@ -563,7 +527,7 @@ func (rs *ReplicationServer) DemoteVolume(ctx context.Context,
 		return nil, status.Error(codes.Internal, err.Error())
 	}
 
-	info, err := mirror.GetMirroringInfo()
+	info, err := mirror.GetMirroringInfo(ctx)
 	if err != nil {
 		log.ErrorLog(ctx, err.Error())
 
@@ -592,7 +556,7 @@ func (rs *ReplicationServer) DemoteVolume(ctx context.Context,
 			return nil, status.Error(codes.Internal, err.Error())
 		}
 
-		err = mirror.Demote()
+		err = mirror.Demote(ctx)
 		if err != nil {
 			log.ErrorLog(ctx, err.Error())
 
@@ -659,23 +623,14 @@ func (rs *ReplicationServer) ResyncVolume(ctx context.Context,
 
 	rbdVol, err := mgr.GetVolumeByID(ctx, volumeID)
 	if err != nil {
-		switch {
-		case errors.Is(err, corerbd.ErrImageNotFound):
-			err = status.Errorf(codes.NotFound, "volume %s not found", volumeID)
-		case errors.Is(err, util.ErrPoolNotFound):
-			err = status.Errorf(codes.NotFound, "pool %s not found for %s", rbdVol.GetPoolName(), volumeID)
-		default:
-			err = status.Errorf(codes.Internal, err.Error())
-		}
-
-		return nil, err
+		return nil, getGRPCError(err)
 	}
 	mirror, err := rbdVol.ToMirror()
 	if err != nil {
 		return nil, status.Error(codes.Internal, err.Error())
 	}
 
-	info, err := mirror.GetMirroringInfo()
+	info, err := mirror.GetMirroringInfo(ctx)
 	if err != nil {
 		// in case of Resync the image will get deleted and gets recreated and
 		// it takes time for this operation.
@@ -693,7 +648,7 @@ func (rs *ReplicationServer) ResyncVolume(ctx context.Context,
 		return nil, status.Error(codes.InvalidArgument, "image is in primary state")
 	}
 
-	sts, err := mirror.GetGlobalMirroringStatus()
+	sts, err := mirror.GetGlobalMirroringStatus(ctx)
 	if err != nil {
 		// the image gets recreated after issuing resync
 		if errors.Is(err, corerbd.ErrImageNotFound) {
@@ -764,7 +719,7 @@ func (rs *ReplicationServer) ResyncVolume(ctx context.Context,
 		}
 		log.DebugLog(ctx, "image %s, savedImageTime=%v, currentImageTime=%v", rbdVol, st, creationTime.AsTime())
 		if req.GetForce() && st.Equal(creationTime.AsTime()) {
-			err = mirror.Resync()
+			err = mirror.Resync(ctx)
 			if err != nil {
 				return nil, getGRPCError(err)
 			}
@@ -830,6 +785,8 @@ func getGRPCError(err error) error {
 	}
 
 	errorStatusMap := map[error]codes.Code{
+		corerbd.ErrImageNotFound:      codes.NotFound,
+		util.ErrPoolNotFound:          codes.NotFound,
 		corerbd.ErrInvalidArgument:    codes.InvalidArgument,
 		corerbd.ErrFlattenInProgress:  codes.Aborted,
 		corerbd.ErrAborted:            codes.Aborted,
@@ -875,9 +832,9 @@ func (rs *ReplicationServer) GetVolumeReplicationInfo(ctx context.Context,
 	if err != nil {
 		switch {
 		case errors.Is(err, corerbd.ErrImageNotFound):
-			err = status.Errorf(codes.NotFound, "volume %s not found", volumeID)
+			err = status.Errorf(codes.NotFound, err.Error())
 		case errors.Is(err, util.ErrPoolNotFound):
-			err = status.Errorf(codes.NotFound, "pool %s not found for %s", rbdVol.GetPoolName(), volumeID)
+			err = status.Errorf(codes.NotFound, err.Error())
 		default:
 			err = status.Errorf(codes.Internal, err.Error())
 		}
@@ -889,7 +846,7 @@ func (rs *ReplicationServer) GetVolumeReplicationInfo(ctx context.Context,
 		return nil, status.Error(codes.Internal, err.Error())
 	}
 
-	info, err := mirror.GetMirroringInfo()
+	info, err := mirror.GetMirroringInfo(ctx)
 	if err != nil {
 		log.ErrorLog(ctx, err.Error())
 
@@ -905,7 +862,7 @@ func (rs *ReplicationServer) GetVolumeReplicationInfo(ctx context.Context,
 		return nil, status.Error(codes.InvalidArgument, "image is not in primary state")
 	}
 
-	mirrorStatus, err := mirror.GetGlobalMirroringStatus()
+	mirrorStatus, err := mirror.GetGlobalMirroringStatus(ctx)
 	if err != nil {
 		if errors.Is(err, corerbd.ErrImageNotFound) {
 			return nil, status.Error(codes.Aborted, err.Error())

internal/csi-addons/rbd/replication_test.go

@@ -27,6 +27,7 @@ import (
 
 	corerbd "github.com/ceph/ceph-csi/internal/rbd"
 	"github.com/ceph/ceph-csi/internal/rbd/types"
+	"github.com/ceph/ceph-csi/internal/util"
 
 	librbd "github.com/ceph/go-ceph/rbd"
 	"github.com/ceph/go-ceph/rbd/admin"
@@ -594,6 +595,16 @@ func TestGetGRPCError(t *testing.T) {
 			err:         nil,
 			expectedErr: status.Error(codes.OK, "ok string"),
 		},
+		{
+			name:        "ErrImageNotFound",
+			err:         corerbd.ErrImageNotFound,
+			expectedErr: status.Error(codes.NotFound, corerbd.ErrImageNotFound.Error()),
+		},
+		{
+			name:        "ErrPoolNotFound",
+			err:         util.ErrPoolNotFound,
+			expectedErr: status.Error(codes.NotFound, util.ErrPoolNotFound.Error()),
+		},
 	}
 
 	for _, tt := range tests {

internal/csi-addons/rbd/volumegroup.go

@@ -18,6 +18,7 @@ package rbd
 
 import (
 	"context"
+	"fmt"
 	"slices"
 
 	"github.com/ceph/ceph-csi/internal/rbd"
@@ -87,6 +88,11 @@ func (vs *VolumeGroupServer) CreateVolumeGroup(
 
 	// resolve all volumes
 	volumes := make([]types.Volume, len(req.GetVolumeIds()))
+	defer func() {
+		for _, vol := range volumes {
+			vol.Destroy(ctx)
+		}
+	}()
 	for i, id := range req.GetVolumeIds() {
 		vol, err := mgr.GetVolumeByID(ctx, id)
 		if err != nil {
@@ -97,9 +103,6 @@ func (vs *VolumeGroupServer) CreateVolumeGroup(
 				req.GetName(),
 				err.Error())
 		}
-
-		//nolint:gocritic // need to call .Destroy() for all volumes
-		defer vol.Destroy(ctx)
 		volumes[i] = vol
 	}
 
@@ -117,6 +120,21 @@ func (vs *VolumeGroupServer) CreateVolumeGroup(
 
 	log.DebugLog(ctx, "VolumeGroup %q has been created: %+v", req.GetName(), vg)
 
+	// extract the flatten mode
+	flattenMode, err := getFlattenMode(ctx, req.GetParameters())
+	if err != nil {
+		return nil, err
+	}
+	// Flatten the image if the flatten mode is set to FlattenModeForce
+	// before adding it to the volume group.
+	for _, v := range volumes {
+		err = v.HandleParentImageExistence(ctx, flattenMode)
+		if err != nil {
+			err = fmt.Errorf("failed to handle parent image for volume group %q: %w", vg, err)
+
+			return nil, getGRPCError(err)
+		}
+	}
 	// add each rbd-image to the RBDVolumeGroup
 	for _, vol := range volumes {
 		err = vg.AddVolume(ctx, vol)
@@ -328,17 +346,44 @@ func (vs *VolumeGroupServer) ModifyVolumeGroupMembership(
 		}
 	}
 
-	// add the new volumes to the group
-	for _, id := range toAdd {
-		vol, getErr := mgr.GetVolumeByID(ctx, id)
-		if getErr != nil {
+	// resolve all volumes
+	volumes := make([]types.Volume, len(toAdd))
+	defer func() {
+		for _, vol := range volumes {
+			vol.Destroy(ctx)
+		}
+	}()
+	for i, id := range toAdd {
+		var vol types.Volume
+		vol, err = mgr.GetVolumeByID(ctx, id)
+		if err != nil {
 			return nil, status.Errorf(
 				codes.NotFound,
 				"failed to find a volume with CSI ID %q: %v",
 				id,
 				err)
 		}
+		volumes[i] = vol
+	}
+
+	// extract the flatten mode
+	flattenMode, err := getFlattenMode(ctx, req.GetParameters())
+	if err != nil {
+		return nil, err
+	}
+	// Flatten the image if the flatten mode is set to FlattenModeForce
+	// before adding it to the volume group.
+	for _, vol := range volumes {
+		err = vol.HandleParentImageExistence(ctx, flattenMode)
+		if err != nil {
+			err = fmt.Errorf("failed to handle parent image for volume group %q: %w", vg, err)
+
+			return nil, getGRPCError(err)
+		}
+	}
 
+	// add the new volumes to the group
+	for _, vol := range volumes {
 		err = vg.AddVolume(ctx, vol)
 		if err != nil {
 			return nil, status.Errorf(

internal/csi-common/utils.go

@@ -128,6 +128,11 @@ func GetIDFromReplication(req interface{}) string {
 		if src != nil && src.GetVolume() != nil {
 			reqID = src.GetVolume().GetVolumeId()
 		}
+		if reqID == "" {
+			if src != nil && src.GetVolumegroup() != nil {
+				reqID = src.GetVolumegroup().GetVolumeGroupId()
+			}
+		}
 		if reqID == "" {
 			reqID = r.GetVolumeId() //nolint:nolintlint,staticcheck // req.VolumeId is deprecated
 		}