rbd: check local image state during promote operation

rbd mirroring CLI calls are async and it doesn't wait
for the operation to be completed. ex:- `rbd mirror image enable`
it will enable the mirroring on the image but it doesn't
ensure that the image is mirroring enabled and healthy
primary. The same goes for the promote volume also.
This commits adds a check-in PromoteVolume to make sure
the image in a healthy state i.e `up+stopped`.

note:- not considering any intermediate states to make
sure the image is completely healthy before responding
success to the RPC call.

Signed-off-by: Madhu Rajanna <[email protected]>

internal/rbd/errors.go

@@ -42,4 +42,6 @@ var (
 	ErrMissingImageNameInVolID = errors.New("rbd image name information can not be empty in volID")
 	// ErrDecodeClusterIDFromMonsInVolID is returned when mons hash decoding on migration volID.
 	ErrDecodeClusterIDFromMonsInVolID = errors.New("failed to get clusterID from monitors hash in volID")
+	// ErrUnHealthyImage is returned when image is not healthy.
+	ErrUnHealthyImage = errors.New("image is not healthy")
 )

internal/rbd/replicationcontrollerserver.go

@@ -517,6 +517,11 @@ func (rs *ReplicationServer) PromoteVolume(ctx context.Context,
 		}
 	}
 
+	err = checkHealthyPrimary(ctx, rbdVol)
+	if err != nil {
+		return nil, status.Error(codes.Internal, err.Error())
+	}
+
 	var mode librbd.ImageMirrorMode
 	mode, err = getMirroringMode(ctx, req.GetParameters())
 	if err != nil {
@@ -546,6 +551,35 @@ func (rs *ReplicationServer) PromoteVolume(ctx context.Context,
 	return &replication.PromoteVolumeResponse{}, nil
 }
 
+// checkHealthyPrimary checks if the is a healhty primary or not.
+// healthy primary image will be in up+stopped state, for states other
+// than this it returns an error message.
+func checkHealthyPrimary(ctx context.Context, rbdVol *rbdVolume) error {
+	mirrorStatus, err := rbdVol.getImageMirroringStatus()
+	if err != nil {
+		return err
+	}
+	localStatus, err := mirrorStatus.LocalStatus()
+	if err != nil {
+		// LocalStatus can fail if the local site status is not found in
+		// mirroring status. Log complete sites status to debug why getting
+		// local status failed
+		log.ErrorLog(ctx, "mirroring status is %+v", mirrorStatus)
+
+		return fmt.Errorf("failed to get local status: %w", err)
+	}
+
+	if !localStatus.Up && localStatus.State != librbd.MirrorImageStatusStateStopped {
+		return fmt.Errorf("%s %w. State is up=%t, state=%q",
+			rbdVol,
+			ErrUnHealthyImage,
+			localStatus.Up,
+			localStatus.State)
+	}
+
+	return nil
+}
+
 // DemoteVolume extracts the RBD volume information from the
 // volumeID, If the image is present, mirroring is enabled and the
 // image is in promoted state it will demote the volume as secondary.