Repository: ceph/ceph-csi
Pull request #2656 (2021-11-18 17:47:21)

rbd: add a workaround to fix rbd snapshot scheduling
currently, we have a bug in the rbd mirror scheduling module. After doing failover and failback the scheduling is not getting updated and the mirroring snapshots are not getting created periodically as per the scheduling interval. This PR workaround this one by doing the below operations

- Create a dummy (csi-vol-dummy-ceph fsID) image per cluster and this image should be easily identified.

- During Promote operation on any image enables the mirroring on the dummy image. when we enable the mirroring on the dummy image the pool will get updated and the scheduling will be reconfigured.

- During Demote operation on any image disables the mirroring on the dummy image. the disabled need to be done to enable the mirroring again when we get the promote request to make the image as the primary

- When the DR is no more needed, this image needs to be manually cleanup for now as we don't want to add a check in the existing DeleteVolume code path for deleting dummy images as it impacts the performance of the DeleteVolume workflow.

Moved to add scheduling to the promote operation as scheduling need to be added when the image is promoted and this is the correct method of adding the scheduling to make the scheduling take place.

More details at https://bugzilla.redhat.com/show_bug.cgi?id=2019161
Signed-off-by: Madhu Rajanna madhupr007@gmail.com
Signed-off-by: Shyamsundar Ranganathan <srangana@redhat.com>

ShyamsundarR: Latest push was to add the "." to the end of the comment to address linter failure.
Madhu-1: @mergifyio rebase
Madhu-1: /retest all
Madhu-1: @mergifyio rebase
Madhu-1: >I1119 03:03:23.221198       1 connection.go:187] GRPC error: rpc error: code = InvalidArgument desc = invalid encryption kms configuration: failed connecting to Vault: failed to get the authentication token: Put "http://vault.cephcsi-e2e-ab3babb4a6ad.svc.cluster.local:8200/v1/auth/kubernetes/login": dial tcp: lookup vault.cephcsi-e2e-ab3babb4a6ad.svc.cluster.local: no such host

encryption tests are failing. not related to this PR.
Madhu-1: @humblec looks like cephcsi CI is broken due to a vault issue. am planning to merge this one manually. WDYT?
Madhu-1: > > I1119 03:03:23.221198       1 connection.go:187] GRPC error: rpc error: code = InvalidArgument desc = invalid encryption kms configuration: failed connecting to Vault: failed to get the authentication token: Put "http://vault.cephcsi-e2e-ab3babb4a6ad.svc.cluster.local:8200/v1/auth/kubernetes/login": dial tcp: lookup vault.cephcsi-e2e-ab3babb4a6ad.svc.cluster.local: no such host
> 
> encryption tests are failing. not related to this PR.

https://github.com/ceph/ceph-csi/issues/2657
Madhu-1: @humblec are you expecting a change for the above comments as part of this PR?
humblec: > @ShyamsundarR I tested your patch and it works as expected. So it is fine other than the legitimate comment from @Madhu-1. We should be able to merge it once that is fixed.

Thanks @BenamarMk for confirming. 

>@humblec are you expecting a change for the above comments as part of this PR?

@Madhu-1  yeah, the CI failure is on encryption, considering the urgency of this PR, I am fine to merge manually :+1: , I have a couple of minor comments though. 
Madhu-1: > > @ShyamsundarR I tested your patch and it works as expected. So it is fine other than the legitimate comment from @Madhu-1. We should be able to merge it once that is fixed.
> 
> Thanks @BenamarMk for confirming.
> 
> > @humblec are you expecting a change for the above comments as part of this PR?
> 
> @Madhu-1 yeah, the CI failure is on encryption, considering the urgency of this PR, I am fine to merge manually +1 , I have a couple of minor comments though.

@humblec we can address that in a follow-up PR as it's not a blocker for this PR?
humblec: @Madhu-1 @ShyamsundarR compared to previous PR, I couldnt get the corner case this PR fixed though. Appreciated if we can have a quick mention some where in this PR.  It looks to me that, now the dummy image scheduling has been set to short/1minute period for other images to rearrange the reschedule  ( may be more iterations for them to reschedule) compared to previous version, I could be wrong. 
humblec: > > > @ShyamsundarR I tested your patch and it works as expected. So it is fine other than the legitimate comment from @Madhu-1. We should be able to merge it once that is fixed.
> > 
> > 
> > Thanks @BenamarMk for confirming.
> > > @humblec are you expecting a change for the above comments as part of this PR?
> > 
> > 
> > @Madhu-1 yeah, the CI failure is on encryption, considering the urgency of this PR, I am fine to merge manually +1 , I have a couple of minor comments though.
> 
> @humblec we can address that in a follow-up PR as it's not a blocker for this PR?

sure, nw  go ahead 
Madhu-1: > @Madhu-1 @ShyamsundarR compared to previous PR, I couldnt get the corner case this PR fixed though. Appreciated if we can have a quick mention some where in this PR. It looks to me that, now the dummy image scheduling has been set to short/1minute period for other images to rearrange the reschedule ( may be more iterations for them to reschedule) compared to previous version, I could be wrong.

the previous PR was not resetting the scheduling for all the rbd images the enable and disable was done only on one PromoteVolume. To reset the scheduling the enable and disable of mirroring need to be done on all the images. more detail are in this PR description

Patch:
diff --git a/internal/rbd/rbd_util.go b/internal/rbd/rbd_util.go
index 3dff543fd..1ac823be9 100644
--- a/internal/rbd/rbd_util.go
+++ b/internal/rbd/rbd_util.go
@@ -2050,28 +2050,6 @@ func (ri *rbdImage) addSnapshotScheduling(
 		return err
 	}
 	adminConn := ra.MirrorSnashotSchedule()
-	// list all the snapshot scheduling and check at least one image scheduling
-	// exists with specified interval.
-	ssList, err := adminConn.List(ls)
-	if err != nil {
-		return err
-	}
-
-	for _, ss := range ssList {
-		// make sure we are matching image level scheduling. The
-		// `adminConn.List` lists the global level scheduling also.
-		if ss.Name == ri.String() {
-			for _, s := range ss.Schedule {
-				// TODO: Add support to check start time also.
-				// The start time is currently stored with different format
-				// in ceph. Comparison is not possible unless we know in
-				// which format ceph is storing it.
-				if s.Interval == interval {
-					return err
-				}
-			}
-		}
-	}
 	err = adminConn.Add(ls, interval, startTime)
 	if err != nil {
 		return err
diff --git a/internal/rbd/replicationcontrollerserver.go b/internal/rbd/replicationcontrollerserver.go
index e00d1b7a5..f1729e2d1 100644
--- a/internal/rbd/replicationcontrollerserver.go
+++ b/internal/rbd/replicationcontrollerserver.go
@@ -23,6 +23,7 @@ import (
 	"regexp"
 	"strconv"
 	"strings"
+	"sync"
 	"time"
 
 	"github.com/ceph/ceph-csi/internal/util"
@@ -63,6 +64,18 @@ const (
 	schedulingStartTimeKey = "schedulingStartTime"
 )
 
+type operation string
+
+var (
+	// pool+"/"+key to check dummy image is created.
+	dummyImageCreated operation = "dummyImageCreated"
+	// Read write lock to ensure that only one operation is happening at a time.
+	operationLock = sync.Map{}
+
+	// Lock to serialize operations on the dummy image to tickle RBD snapshot schedule.
+	dummyImageOpsLock sync.Mutex
+)
+
 // ReplicationServer struct of rbd CSI driver with supported methods of Replication
 // controller server spec.
 type ReplicationServer struct {
@@ -117,12 +130,10 @@ func getMirroringMode(ctx context.Context, parameters map[string]string) (librbd
 	return mirroringMode, nil
 }
 
-// getSchedulingDetails gets the mirroring mode and scheduling details from the
+// validateSchedulingDetails gets the mirroring mode and scheduling details from the
 // input GRPC request parameters and validates the scheduling is only supported
 // for snapshot mirroring mode.
-func getSchedulingDetails(parameters map[string]string) (admin.Interval, admin.StartTime, error) {
-	admInt := admin.NoInterval
-	adminStartTime := admin.NoStartTime
+func validateSchedulingDetails(parameters map[string]string) error {
 	var err error
 
 	val := parameters[imageMirroringKey]
@@ -134,43 +145,49 @@ func getSchedulingDetails(parameters map[string]string) (admin.Interval, admin.S
 	// an optional parameter.
 	case "":
 	default:
-		return admInt, adminStartTime, status.Error(codes.InvalidArgument, "scheduling is only supported for snapshot mode")
+		return status.Error(codes.InvalidArgument, "scheduling is only supported for snapshot mode")
 	}
 
 	// validate mandatory interval field
 	interval, ok := parameters[schedulingIntervalKey]
 	if ok && interval == "" {
-		return admInt, adminStartTime, status.Error(codes.InvalidArgument, "scheduling interval cannot be empty")
+		return status.Error(codes.InvalidArgument, "scheduling interval cannot be empty")
 	}
-	adminStartTime = admin.StartTime(parameters[schedulingStartTimeKey])
+	adminStartTime := admin.StartTime(parameters[schedulingStartTimeKey])
 	if !ok {
 		// startTime is alone not supported it has to be present with interval
 		if adminStartTime != "" {
-			return admInt, admin.NoStartTime, status.Errorf(codes.InvalidArgument,
+			return status.Errorf(codes.InvalidArgument,
 				"%q parameter is supported only with %q",
 				schedulingStartTimeKey,
 				schedulingIntervalKey)
 		}
 	}
 	if interval != "" {
-		admInt, err = validateSchedulingInterval(interval)
+		err = validateSchedulingInterval(interval)
 		if err != nil {
-			return admInt, admin.NoStartTime, status.Error(codes.InvalidArgument, err.Error())
+			return status.Error(codes.InvalidArgument, err.Error())
 		}
 	}
 
-	return admInt, adminStartTime, nil
+	return nil
+}
+
+// getSchedulingDetails returns scheduling interval and scheduling startTime.
+func getSchedulingDetails(parameters map[string]string) (admin.Interval, admin.StartTime) {
+	return admin.Interval(parameters[schedulingIntervalKey]),
+		admin.StartTime(parameters[schedulingStartTimeKey])
 }
 
 // validateSchedulingInterval return the interval as it is if its ending with
 // `m|h|d` or else it will return error.
-func validateSchedulingInterval(interval string) (admin.Interval, error) {
+func validateSchedulingInterval(interval string) error {
 	re := regexp.MustCompile(`^\d+[mhd]$`)
 	if re.MatchString(interval) {
-		return admin.Interval(interval), nil
+		return nil
 	}
 
-	return "", errors.New("interval specified without d, h, m suffix")
+	return errors.New("interval specified without d, h, m suffix")
 }
 
 // EnableVolumeReplication extracts the RBD volume information from the
@@ -189,7 +206,7 @@ func (rs *ReplicationServer) EnableVolumeReplication(ctx context.Context,
 	}
 	defer cr.DeleteCredentials()
 
-	interval, startTime, err := getSchedulingDetails(req.GetParameters())
+	err = validateSchedulingDetails(req.GetParameters())
 	if err != nil {
 		return nil, err
 	}
@@ -228,6 +245,11 @@ func (rs *ReplicationServer) EnableVolumeReplication(ctx context.Context,
 		return nil, status.Error(codes.Internal, err.Error())
 	}
 
+	err = createDummyImage(ctx, rbdVol)
+	if err != nil {
+		return nil, status.Errorf(codes.Internal, "failed to create dummy image %s", err.Error())
+	}
+
 	if mirroringInfo.State != librbd.MirrorImageEnabled {
 		err = rbdVol.enableImageMirroring(mirroringMode)
 		if err != nil {
@@ -237,20 +259,78 @@ func (rs *ReplicationServer) EnableVolumeReplication(ctx context.Context,
 		}
 	}
 
-	if interval != "" {
-		err = rbdVol.addSnapshotScheduling(interval, startTime)
+	return &replication.EnableVolumeReplicationResponse{}, nil
+}
+
+// getDummyImageName returns the csi-vol-dummy+cluster FSID as the image name.
+// each cluster should have a unique dummy image created. choosing the cluster
+// FSID for the same reason.
+func getDummyImageName(conn *util.ClusterConnection) (string, error) {
+	id, err := conn.GetFSID()
+	if err != nil {
+		return "", err
+	}
+
+	return fmt.Sprintf("csi-vol-dummy-%s", id), nil
+}
+
+// getOperationName returns the operation name for the given operation type
+// combined with the pool name.
+func getOperationName(poolName string, optName operation) string {
+	return fmt.Sprintf("%s/%s", poolName, optName)
+}
+
+// createDummyImage creates a dummy image as a workaround for the rbd
+// scheduling problem.
+func createDummyImage(ctx context.Context, rbdVol *rbdVolume) error {
+	optName := getOperationName(rbdVol.Pool, dummyImageCreated)
+	if _, ok := operationLock.Load(optName); !ok {
+		// create a dummy image
+		imgName, err := getDummyImageName(rbdVol.conn)
 		if err != nil {
-			return nil, err
+			return err
 		}
-		log.DebugLog(
-			ctx,
-			"Added scheduling at interval %s, start time %s for volume %s",
-			interval,
-			startTime,
-			rbdVol)
+		dummyVol := rbdVol
+		dummyVol.RbdImageName = imgName
+		err = createImage(ctx, dummyVol, dummyVol.conn.Creds)
+		if err != nil && !strings.Contains(err.Error(), "File exists") {
+			return err
+		}
+		operationLock.Store(optName, true)
 	}
 
-	return &replication.EnableVolumeReplicationResponse{}, nil
+	return nil
+}
+
+// tickleMirroringOnDummyImage disables and reenables mirroring on the dummy image, and sets a
+// schedule of a minute for the dummy image, to force a schedule refresh for other mirrored images
+// within a minute.
+func tickleMirroringOnDummyImage(rbdVol *rbdVolume, mirroringMode librbd.ImageMirrorMode) error {
+	imgName, err := getDummyImageName(rbdVol.conn)
+	if err != nil {
+		return err
+	}
+	dummyVol := rbdVol
+	dummyVol.RbdImageName = imgName
+
+	dummyImageOpsLock.Lock()
+	defer dummyImageOpsLock.Unlock()
+	err = dummyVol.disableImageMirroring(false)
+	if err != nil {
+		return err
+	}
+
+	err = dummyVol.enableImageMirroring(mirroringMode)
+	if err != nil {
+		return err
+	}
+
+	err = dummyVol.addSnapshotScheduling(admin.Interval("1m"), admin.NoStartTime)
+	if err != nil {
+		return err
+	}
+
+	return nil
 }
 
 // DisableVolumeReplication extracts the RBD volume information from the
@@ -437,6 +517,32 @@ func (rs *ReplicationServer) PromoteVolume(ctx context.Context,
 		}
 	}
 
+	var mode librbd.ImageMirrorMode
+	mode, err = getMirroringMode(ctx, req.GetParameters())
+	if err != nil {
+		return nil, status.Errorf(codes.Internal, "failed to get mirroring mode %s", err.Error())
+	}
+
+	log.DebugLog(ctx, "Attempting to tickle dummy image for restarting RBD schedules")
+	err = tickleMirroringOnDummyImage(rbdVol, mode)
+	if err != nil {
+		return nil, status.Errorf(codes.Internal, "failed to enable mirroring on dummy image %s", err.Error())
+	}
+
+	interval, startTime := getSchedulingDetails(req.GetParameters())
+	if interval != admin.NoInterval {
+		err = rbdVol.addSnapshotScheduling(interval, startTime)
+		if err != nil {
+			return nil, err
+		}
+		log.DebugLog(
+			ctx,
+			"Added scheduling at interval %s, start time %s for volume %s",
+			interval,
+			startTime,
+			rbdVol)
+	}
+
 	return &replication.PromoteVolumeResponse{}, nil
 }
 
diff --git a/internal/util/connection.go b/internal/util/connection.go
index 9625d4e30..0aef6848b 100644
--- a/internal/util/connection.go
+++ b/internal/util/connection.go
@@ -114,6 +114,14 @@ func (cc *ClusterConnection) GetFSAdmin() (*ca.FSAdmin, error) {
 	return ca.NewFromConn(cc.conn), nil
 }
 
+func (cc *ClusterConnection) GetFSID() (string, error) {
+	if cc.conn == nil {
+		return "", errors.New("cluster is not connected yet")
+	}
+
+	return cc.conn.GetFSID()
+}
+
 // GetRBDAdmin get RBDAdmin to administrate rbd volumes.
 func (cc *ClusterConnection) GetRBDAdmin() (*ra.RBDAdmin, error) {
 	if cc.conn == nil {


Test patch:
diff --git a/internal/rbd/replicationcontrollerserver_test.go b/internal/rbd/replicationcontrollerserver_test.go
index 1d49e8954..cf306710a 100644
--- a/internal/rbd/replicationcontrollerserver_test.go
+++ b/internal/rbd/replicationcontrollerserver_test.go
@@ -29,37 +29,31 @@ func TestValidateSchedulingInterval(t *testing.T) {
 	tests := []struct {
 		name     string
 		interval string
-		want     admin.Interval
 		wantErr  bool
 	}{
 		{
 			"valid interval in minutes",
 			"3m",
-			admin.Interval("3m"),
 			false,
 		},
 		{
 			"valid interval in hour",
 			"22h",
-			admin.Interval("22h"),
 			false,
 		},
 		{
 			"valid interval in days",
 			"13d",
-			admin.Interval("13d"),
 			false,
 		},
 		{
 			"invalid interval without number",
 			"d",
-			admin.Interval(""),
 			true,
 		},
 		{
 			"invalid interval without (m|h|d) suffix",
 			"12",
-			admin.Interval(""),
 			true,
 		},
 	}
@@ -67,27 +61,22 @@ func TestValidateSchedulingInterval(t *testing.T) {
 		tt := tt
 		t.Run(tt.name, func(t *testing.T) {
 			t.Parallel()
-			got, err := validateSchedulingInterval(tt.interval)
+			err := validateSchedulingInterval(tt.interval)
 			if (err != nil) != tt.wantErr {
 				t.Errorf("validateSchedulingInterval() error = %v, wantErr %v", err, tt.wantErr)
 
 				return
 			}
-			if !reflect.DeepEqual(got, tt.want) {
-				t.Errorf("validateSchedulingInterval() = %v, want %v", got, tt.want)
-			}
 		})
 	}
 }
 
-func TestGetSchedulingDetails(t *testing.T) {
+func TestValidateSchedulingDetails(t *testing.T) {
 	t.Parallel()
 	tests := []struct {
-		name          string
-		parameters    map[string]string
-		wantInterval  admin.Interval
-		wantStartTime admin.StartTime
-		wantErr       bool
+		name       string
+		parameters map[string]string
+		wantErr    bool
 	}{
 		{
 			"valid parameters",
@@ -96,8 +85,6 @@ func TestGetSchedulingDetails(t *testing.T) {
 				schedulingIntervalKey:  "1h",
 				schedulingStartTimeKey: "14:00:00-05:00",
 			},
-			admin.Interval("1h"),
-			admin.StartTime("14:00:00-05:00"),
 			false,
 		},
 		{
@@ -106,8 +93,6 @@ func TestGetSchedulingDetails(t *testing.T) {
 				imageMirroringKey:     string(imageMirrorModeSnapshot),
 				schedulingIntervalKey: "1h",
 			},
-			admin.Interval("1h"),
-			admin.NoStartTime,
 			false,
 		},
 		{
@@ -116,8 +101,6 @@ func TestGetSchedulingDetails(t *testing.T) {
 				imageMirroringKey:     "journal",
 				schedulingIntervalKey: "1h",
 			},
-			admin.NoInterval,
-			admin.NoStartTime,
 			true,
 		},
 		{
@@ -126,8 +109,6 @@ func TestGetSchedulingDetails(t *testing.T) {
 				imageMirroringKey:      string(imageMirrorModeSnapshot),
 				schedulingStartTimeKey: "14:00:00-05:00",
 			},
-			admin.NoInterval,
-			admin.NoStartTime,
 			true,
 		},
 		{
@@ -135,15 +116,11 @@ func TestGetSchedulingDetails(t *testing.T) {
 			map[string]string{
 				imageMirroringKey: string(imageMirrorModeSnapshot),
 			},
-			admin.NoInterval,
-			admin.NoStartTime,
 			false,
 		},
 		{
 			"when no parameters and scheduling details are specified",
 			map[string]string{},
-			admin.NoInterval,
-			admin.NoStartTime,
 			false,
 		},
 		{
@@ -152,8 +129,6 @@ func TestGetSchedulingDetails(t *testing.T) {
 				schedulingIntervalKey:  "1h",
 				schedulingStartTimeKey: "14:00:00-05:00",
 			},
-			admin.Interval("1h"),
-			admin.StartTime("14:00:00-05:00"),
 			false,
 		},
 	}
@@ -161,12 +136,63 @@ func TestGetSchedulingDetails(t *testing.T) {
 		tt := tt
 		t.Run(tt.name, func(t *testing.T) {
 			t.Parallel()
-			interval, startTime, err := getSchedulingDetails(tt.parameters)
+			err := validateSchedulingDetails(tt.parameters)
 			if (err != nil) != tt.wantErr {
 				t.Errorf("getSchedulingDetails() error = %v, wantErr %v", err, tt.wantErr)
 
 				return
 			}
+		})
+	}
+}
+
+func TestGetSchedulingDetails(t *testing.T) {
+	t.Parallel()
+	tests := []struct {
+		name          string
+		parameters    map[string]string
+		wantInterval  admin.Interval
+		wantStartTime admin.StartTime
+	}{
+		{
+			"valid parameters",
+			map[string]string{
+				schedulingIntervalKey:  "1h",
+				schedulingStartTimeKey: "14:00:00-05:00",
+			},
+			admin.Interval("1h"),
+			admin.StartTime("14:00:00-05:00"),
+		},
+		{
+			"valid parameters when optional startTime is missing",
+			map[string]string{
+				imageMirroringKey:     string(imageMirrorModeSnapshot),
+				schedulingIntervalKey: "1h",
+			},
+			admin.Interval("1h"),
+			admin.NoStartTime,
+		},
+		{
+			"when startTime is specified without interval",
+			map[string]string{
+				imageMirroringKey:      string(imageMirrorModeSnapshot),
+				schedulingStartTimeKey: "14:00:00-05:00",
+			},
+			admin.NoInterval,
+			admin.StartTime("14:00:00-05:00"),
+		},
+		{
+			"when no parameters and scheduling details are specified",
+			map[string]string{},
+			admin.NoInterval,
+			admin.NoStartTime,
+		},
+	}
+	for _, tt := range tests {
+		tt := tt
+		t.Run(tt.name, func(t *testing.T) {
+			t.Parallel()
+			interval, startTime := getSchedulingDetails(tt.parameters)
 			if !reflect.DeepEqual(interval, tt.wantInterval) {
 				t.Errorf("getSchedulingDetails() interval = %v, want %v", interval, tt.wantInterval)
 			}