2021-02-11 10:01:38 +00:00
|
|
|
/*
|
|
|
|
Copyright 2021 The Ceph-CSI Authors.
|
|
|
|
|
|
|
|
Licensed under the Apache License, Version 2.0 (the "License");
|
|
|
|
you may not use this file except in compliance with the License.
|
|
|
|
You may obtain a copy of the License at
|
|
|
|
|
|
|
|
http://www.apache.org/licenses/LICENSE-2.0
|
|
|
|
|
|
|
|
Unless required by applicable law or agreed to in writing, software
|
|
|
|
distributed under the License is distributed on an "AS IS" BASIS,
|
|
|
|
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
|
|
See the License for the specific language governing permissions and
|
|
|
|
limitations under the License.
|
|
|
|
*/
|
|
|
|
|
|
|
|
package rbd
|
|
|
|
|
|
|
|
import (
|
|
|
|
"context"
|
2022-09-09 11:26:41 +00:00
|
|
|
"encoding/json"
|
2021-02-11 10:01:38 +00:00
|
|
|
"errors"
|
2021-06-23 13:14:48 +00:00
|
|
|
"fmt"
|
2021-06-28 11:38:42 +00:00
|
|
|
"regexp"
|
2021-02-11 10:01:38 +00:00
|
|
|
"strconv"
|
2021-03-18 07:16:36 +00:00
|
|
|
"strings"
|
2021-08-05 09:11:28 +00:00
|
|
|
"time"
|
2021-02-11 10:01:38 +00:00
|
|
|
|
|
|
|
"github.com/ceph/ceph-csi/internal/util"
|
2021-08-24 15:03:25 +00:00
|
|
|
"github.com/ceph/ceph-csi/internal/util/log"
|
2021-02-11 10:01:38 +00:00
|
|
|
|
|
|
|
librbd "github.com/ceph/go-ceph/rbd"
|
2021-06-28 11:38:42 +00:00
|
|
|
"github.com/ceph/go-ceph/rbd/admin"
|
2021-04-06 06:46:35 +00:00
|
|
|
"github.com/csi-addons/spec/lib/go/replication"
|
2022-08-15 05:23:50 +00:00
|
|
|
"google.golang.org/grpc"
|
2021-02-11 10:01:38 +00:00
|
|
|
"google.golang.org/grpc/codes"
|
|
|
|
"google.golang.org/grpc/status"
|
2022-09-09 11:26:41 +00:00
|
|
|
"google.golang.org/protobuf/types/known/timestamppb"
|
2021-02-11 10:01:38 +00:00
|
|
|
)
|
|
|
|
|
|
|
|
// imageMirroringMode is used to indicate the mirroring mode for an RBD image.
|
|
|
|
type imageMirroringMode string
|
|
|
|
|
|
|
|
const (
|
|
|
|
// imageMirrorModeSnapshot uses snapshots to propagate RBD images between
|
|
|
|
// ceph clusters.
|
|
|
|
imageMirrorModeSnapshot imageMirroringMode = "snapshot"
|
2021-11-26 06:38:21 +00:00
|
|
|
// imageMirrorModeJournal uses journaling to propagate RBD images between
|
|
|
|
// ceph clusters.
|
|
|
|
imageMirrorModeJournal imageMirroringMode = "journal"
|
2021-02-11 10:01:38 +00:00
|
|
|
)
|
|
|
|
|
|
|
|
const (
|
|
|
|
// mirroringMode + key to get the imageMirroringMode from parameters.
|
|
|
|
imageMirroringKey = "mirroringMode"
|
|
|
|
// forceKey + key to get the force option from parameters.
|
|
|
|
forceKey = "force"
|
2021-06-28 11:38:42 +00:00
|
|
|
|
|
|
|
// schedulingIntervalKey to get the schedulingInterval from the
|
|
|
|
// parameters.
|
|
|
|
// Interval of time between scheduled snapshots. Typically in the form
|
|
|
|
// <num><m,h,d>.
|
|
|
|
schedulingIntervalKey = "schedulingInterval"
|
|
|
|
|
|
|
|
// schedulingStartTimeKey to get the schedulingStartTime from the
|
|
|
|
// parameters.
|
|
|
|
// (optional) StartTime is the time the snapshot schedule
|
|
|
|
// begins, can be specified using the ISO 8601 time format.
|
|
|
|
schedulingStartTimeKey = "schedulingStartTime"
|
2021-02-11 10:01:38 +00:00
|
|
|
)
|
|
|
|
|
2021-03-17 07:27:04 +00:00
|
|
|
// ReplicationServer struct of rbd CSI driver with supported methods of Replication
|
|
|
|
// controller server spec.
|
|
|
|
type ReplicationServer struct {
|
|
|
|
// added UnimplementedControllerServer as a member of
|
|
|
|
// ControllerServer. if replication spec add more RPC services in the proto
|
|
|
|
// file, then we don't need to add all RPC methods leading to forward
|
|
|
|
// compatibility.
|
|
|
|
*replication.UnimplementedControllerServer
|
|
|
|
// Embed ControllerServer as it implements helper functions
|
|
|
|
*ControllerServer
|
2021-02-11 10:01:38 +00:00
|
|
|
}
|
|
|
|
|
2022-08-15 05:23:50 +00:00
|
|
|
func (rs *ReplicationServer) RegisterService(server grpc.ServiceRegistrar) {
|
|
|
|
replication.RegisterControllerServer(server, rs)
|
|
|
|
}
|
|
|
|
|
2021-02-11 10:01:38 +00:00
|
|
|
// getForceOption extracts the force option from the GRPC request parameters.
|
|
|
|
// If not set, the default will be set to false.
|
|
|
|
func getForceOption(ctx context.Context, parameters map[string]string) (bool, error) {
|
|
|
|
val, ok := parameters[forceKey]
|
|
|
|
if !ok {
|
2021-08-24 15:03:25 +00:00
|
|
|
log.WarningLog(ctx, "%s is not set in parameters, setting to default (%v)", forceKey, false)
|
2021-07-22 05:45:17 +00:00
|
|
|
|
2021-02-11 10:01:38 +00:00
|
|
|
return false, nil
|
|
|
|
}
|
|
|
|
force, err := strconv.ParseBool(val)
|
|
|
|
if err != nil {
|
|
|
|
return false, status.Errorf(codes.Internal, err.Error())
|
|
|
|
}
|
2021-07-22 05:45:17 +00:00
|
|
|
|
2021-02-11 10:01:38 +00:00
|
|
|
return force, nil
|
|
|
|
}
|
|
|
|
|
|
|
|
// getMirroringMode gets the mirroring mode from the input GRPC request parameters.
|
|
|
|
// mirroringMode is the key to check the mode in the parameters.
|
|
|
|
func getMirroringMode(ctx context.Context, parameters map[string]string) (librbd.ImageMirrorMode, error) {
|
|
|
|
val, ok := parameters[imageMirroringKey]
|
|
|
|
if !ok {
|
2021-08-24 15:03:25 +00:00
|
|
|
log.WarningLog(
|
2021-06-25 11:52:34 +00:00
|
|
|
ctx,
|
|
|
|
"%s is not set in parameters, setting to mirroringMode to default (%s)",
|
|
|
|
imageMirroringKey,
|
|
|
|
imageMirrorModeSnapshot)
|
2021-07-22 05:45:17 +00:00
|
|
|
|
2021-02-11 10:01:38 +00:00
|
|
|
return librbd.ImageMirrorModeSnapshot, nil
|
|
|
|
}
|
|
|
|
|
|
|
|
var mirroringMode librbd.ImageMirrorMode
|
|
|
|
switch imageMirroringMode(val) {
|
|
|
|
case imageMirrorModeSnapshot:
|
|
|
|
mirroringMode = librbd.ImageMirrorModeSnapshot
|
2021-11-26 06:38:21 +00:00
|
|
|
case imageMirrorModeJournal:
|
|
|
|
mirroringMode = librbd.ImageMirrorModeJournal
|
2021-02-11 10:01:38 +00:00
|
|
|
default:
|
|
|
|
return mirroringMode, status.Errorf(codes.InvalidArgument, "%s %s not supported", imageMirroringKey, val)
|
|
|
|
}
|
2021-07-22 05:45:17 +00:00
|
|
|
|
2021-02-11 10:01:38 +00:00
|
|
|
return mirroringMode, nil
|
|
|
|
}
|
|
|
|
|
2021-11-15 06:00:13 +00:00
|
|
|
// validateSchedulingDetails gets the mirroring mode and scheduling details from the
|
2021-06-28 11:38:42 +00:00
|
|
|
// input GRPC request parameters and validates the scheduling is only supported
|
2021-08-09 04:48:32 +00:00
|
|
|
// for snapshot mirroring mode.
|
2021-11-26 06:38:21 +00:00
|
|
|
func validateSchedulingDetails(ctx context.Context, parameters map[string]string) error {
|
2021-06-28 11:38:42 +00:00
|
|
|
var err error
|
|
|
|
|
|
|
|
val := parameters[imageMirroringKey]
|
|
|
|
|
|
|
|
switch imageMirroringMode(val) {
|
2021-11-26 06:38:21 +00:00
|
|
|
case imageMirrorModeJournal:
|
|
|
|
// journal mirror mode does not require scheduling parameters
|
|
|
|
if _, ok := parameters[schedulingIntervalKey]; ok {
|
|
|
|
log.WarningLog(ctx, "%s parameter cannot be used with %s mirror mode, ignoring it",
|
|
|
|
schedulingIntervalKey, string(imageMirrorModeJournal))
|
|
|
|
}
|
|
|
|
if _, ok := parameters[schedulingStartTimeKey]; ok {
|
|
|
|
log.WarningLog(ctx, "%s parameter cannot be used with %s mirror mode, ignoring it",
|
|
|
|
schedulingStartTimeKey, string(imageMirrorModeJournal))
|
|
|
|
}
|
|
|
|
|
|
|
|
return nil
|
2021-06-28 11:38:42 +00:00
|
|
|
case imageMirrorModeSnapshot:
|
2021-08-09 04:48:32 +00:00
|
|
|
// If mirroring mode is not set in parameters, we are defaulting mirroring
|
|
|
|
// mode to snapshot. Discard empty mirroring mode from validation as it is
|
|
|
|
// an optional parameter.
|
|
|
|
case "":
|
2021-06-28 11:38:42 +00:00
|
|
|
default:
|
2021-11-15 06:00:13 +00:00
|
|
|
return status.Error(codes.InvalidArgument, "scheduling is only supported for snapshot mode")
|
2021-06-28 11:38:42 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
// validate mandatory interval field
|
|
|
|
interval, ok := parameters[schedulingIntervalKey]
|
|
|
|
if ok && interval == "" {
|
2021-11-15 06:00:13 +00:00
|
|
|
return status.Error(codes.InvalidArgument, "scheduling interval cannot be empty")
|
2021-06-28 11:38:42 +00:00
|
|
|
}
|
2021-11-15 06:00:13 +00:00
|
|
|
adminStartTime := admin.StartTime(parameters[schedulingStartTimeKey])
|
2021-06-28 11:38:42 +00:00
|
|
|
if !ok {
|
|
|
|
// startTime is alone not supported it has to be present with interval
|
|
|
|
if adminStartTime != "" {
|
2021-11-15 06:00:13 +00:00
|
|
|
return status.Errorf(codes.InvalidArgument,
|
2021-06-28 11:38:42 +00:00
|
|
|
"%q parameter is supported only with %q",
|
|
|
|
schedulingStartTimeKey,
|
|
|
|
schedulingIntervalKey)
|
|
|
|
}
|
|
|
|
}
|
|
|
|
if interval != "" {
|
2021-11-15 06:00:13 +00:00
|
|
|
err = validateSchedulingInterval(interval)
|
2021-06-28 11:38:42 +00:00
|
|
|
if err != nil {
|
2021-11-15 06:00:13 +00:00
|
|
|
return status.Error(codes.InvalidArgument, err.Error())
|
2021-06-28 11:38:42 +00:00
|
|
|
}
|
|
|
|
}
|
2021-07-22 05:45:17 +00:00
|
|
|
|
2021-11-15 06:00:13 +00:00
|
|
|
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])
|
2021-06-28 11:38:42 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
// validateSchedulingInterval return the interval as it is if its ending with
|
|
|
|
// `m|h|d` or else it will return error.
|
2021-11-15 06:00:13 +00:00
|
|
|
func validateSchedulingInterval(interval string) error {
|
2021-07-13 12:21:05 +00:00
|
|
|
re := regexp.MustCompile(`^\d+[mhd]$`)
|
2021-06-28 11:38:42 +00:00
|
|
|
if re.MatchString(interval) {
|
2021-11-15 06:00:13 +00:00
|
|
|
return nil
|
2021-06-28 11:38:42 +00:00
|
|
|
}
|
2021-07-22 05:45:17 +00:00
|
|
|
|
2021-11-15 06:00:13 +00:00
|
|
|
return errors.New("interval specified without d, h, m suffix")
|
2021-06-28 11:38:42 +00:00
|
|
|
}
|
|
|
|
|
2021-02-11 10:01:38 +00:00
|
|
|
// EnableVolumeReplication extracts the RBD volume information from the
|
|
|
|
// volumeID, If the image is present it will enable the mirroring based on the
|
|
|
|
// user provided information.
|
2021-03-17 07:27:04 +00:00
|
|
|
func (rs *ReplicationServer) EnableVolumeReplication(ctx context.Context,
|
2021-02-11 10:01:38 +00:00
|
|
|
req *replication.EnableVolumeReplicationRequest,
|
|
|
|
) (*replication.EnableVolumeReplicationResponse, error) {
|
2021-03-17 07:27:04 +00:00
|
|
|
volumeID := req.GetVolumeId()
|
|
|
|
if volumeID == "" {
|
|
|
|
return nil, status.Error(codes.InvalidArgument, "empty volume ID in request")
|
|
|
|
}
|
|
|
|
cr, err := util.NewUserCredentials(req.GetSecrets())
|
2021-02-11 10:01:38 +00:00
|
|
|
if err != nil {
|
2021-03-17 07:27:04 +00:00
|
|
|
return nil, status.Error(codes.Internal, err.Error())
|
|
|
|
}
|
|
|
|
defer cr.DeleteCredentials()
|
|
|
|
|
2021-11-26 06:38:21 +00:00
|
|
|
err = validateSchedulingDetails(ctx, req.GetParameters())
|
2021-06-28 11:38:42 +00:00
|
|
|
if err != nil {
|
|
|
|
return nil, err
|
|
|
|
}
|
|
|
|
|
2021-03-17 07:27:04 +00:00
|
|
|
if acquired := rs.VolumeLocks.TryAcquire(volumeID); !acquired {
|
2021-08-24 15:03:25 +00:00
|
|
|
log.ErrorLog(ctx, util.VolumeOperationAlreadyExistsFmt, volumeID)
|
2021-07-22 05:45:17 +00:00
|
|
|
|
2021-03-17 07:27:04 +00:00
|
|
|
return nil, status.Errorf(codes.Aborted, util.VolumeOperationAlreadyExistsFmt, volumeID)
|
|
|
|
}
|
|
|
|
defer rs.VolumeLocks.Release(volumeID)
|
|
|
|
|
2021-12-09 08:00:52 +00:00
|
|
|
rbdVol, err := GenVolFromVolID(ctx, volumeID, cr, req.GetSecrets())
|
2021-03-17 07:27:04 +00:00
|
|
|
defer rbdVol.Destroy()
|
|
|
|
if err != nil {
|
|
|
|
switch {
|
|
|
|
case errors.Is(err, 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.Pool, volumeID)
|
|
|
|
default:
|
|
|
|
err = status.Errorf(codes.Internal, err.Error())
|
|
|
|
}
|
2021-07-22 05:45:17 +00:00
|
|
|
|
2021-02-11 10:01:38 +00:00
|
|
|
return nil, err
|
|
|
|
}
|
|
|
|
// extract the mirroring mode
|
|
|
|
mirroringMode, err := getMirroringMode(ctx, req.GetParameters())
|
|
|
|
if err != nil {
|
|
|
|
return nil, err
|
|
|
|
}
|
|
|
|
|
|
|
|
mirroringInfo, err := rbdVol.getImageMirroringInfo()
|
|
|
|
if err != nil {
|
2021-08-24 15:03:25 +00:00
|
|
|
log.ErrorLog(ctx, err.Error())
|
2021-07-22 05:45:17 +00:00
|
|
|
|
2021-02-11 10:01:38 +00:00
|
|
|
return nil, status.Error(codes.Internal, err.Error())
|
|
|
|
}
|
|
|
|
|
|
|
|
if mirroringInfo.State != librbd.MirrorImageEnabled {
|
|
|
|
err = rbdVol.enableImageMirroring(mirroringMode)
|
|
|
|
if err != nil {
|
2021-08-24 15:03:25 +00:00
|
|
|
log.ErrorLog(ctx, err.Error())
|
2021-07-22 05:45:17 +00:00
|
|
|
|
2021-02-11 10:01:38 +00:00
|
|
|
return nil, status.Error(codes.Internal, err.Error())
|
|
|
|
}
|
|
|
|
}
|
2021-06-28 11:38:42 +00:00
|
|
|
|
2021-02-11 10:01:38 +00:00
|
|
|
return &replication.EnableVolumeReplicationResponse{}, nil
|
|
|
|
}
|
|
|
|
|
|
|
|
// DisableVolumeReplication extracts the RBD volume information from the
|
|
|
|
// volumeID, If the image is present and the mirroring is enabled on the RBD
|
|
|
|
// image it will disable the mirroring.
|
2021-03-17 07:27:04 +00:00
|
|
|
func (rs *ReplicationServer) DisableVolumeReplication(ctx context.Context,
|
2021-02-11 10:01:38 +00:00
|
|
|
req *replication.DisableVolumeReplicationRequest,
|
|
|
|
) (*replication.DisableVolumeReplicationResponse, error) {
|
2021-03-17 07:27:04 +00:00
|
|
|
volumeID := req.GetVolumeId()
|
|
|
|
if volumeID == "" {
|
|
|
|
return nil, status.Error(codes.InvalidArgument, "empty volume ID in request")
|
|
|
|
}
|
|
|
|
cr, err := util.NewUserCredentials(req.GetSecrets())
|
2021-02-11 10:01:38 +00:00
|
|
|
if err != nil {
|
2021-03-17 07:27:04 +00:00
|
|
|
return nil, status.Error(codes.Internal, err.Error())
|
2021-02-11 10:01:38 +00:00
|
|
|
}
|
2021-03-17 07:27:04 +00:00
|
|
|
defer cr.DeleteCredentials()
|
2021-02-11 10:01:38 +00:00
|
|
|
|
2021-03-17 07:27:04 +00:00
|
|
|
if acquired := rs.VolumeLocks.TryAcquire(volumeID); !acquired {
|
2021-08-24 15:03:25 +00:00
|
|
|
log.ErrorLog(ctx, util.VolumeOperationAlreadyExistsFmt, volumeID)
|
2021-07-22 05:45:17 +00:00
|
|
|
|
2021-03-17 07:27:04 +00:00
|
|
|
return nil, status.Errorf(codes.Aborted, util.VolumeOperationAlreadyExistsFmt, volumeID)
|
|
|
|
}
|
|
|
|
defer rs.VolumeLocks.Release(volumeID)
|
|
|
|
|
2021-12-09 08:00:52 +00:00
|
|
|
rbdVol, err := GenVolFromVolID(ctx, volumeID, cr, req.GetSecrets())
|
2021-03-17 07:27:04 +00:00
|
|
|
defer rbdVol.Destroy()
|
|
|
|
if err != nil {
|
|
|
|
switch {
|
|
|
|
case errors.Is(err, 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.Pool, volumeID)
|
|
|
|
default:
|
|
|
|
err = status.Errorf(codes.Internal, err.Error())
|
|
|
|
}
|
2021-07-22 05:45:17 +00:00
|
|
|
|
2021-03-17 07:27:04 +00:00
|
|
|
return nil, err
|
|
|
|
}
|
2021-02-11 10:01:38 +00:00
|
|
|
// extract the force option
|
|
|
|
force, err := getForceOption(ctx, req.GetParameters())
|
|
|
|
if err != nil {
|
|
|
|
return nil, err
|
|
|
|
}
|
|
|
|
|
|
|
|
mirroringInfo, err := rbdVol.getImageMirroringInfo()
|
|
|
|
if err != nil {
|
2021-08-24 15:03:25 +00:00
|
|
|
log.ErrorLog(ctx, err.Error())
|
2021-07-22 05:45:17 +00:00
|
|
|
|
2021-02-11 10:01:38 +00:00
|
|
|
return nil, status.Error(codes.Internal, err.Error())
|
|
|
|
}
|
|
|
|
|
|
|
|
switch mirroringInfo.State {
|
2021-03-30 04:44:32 +00:00
|
|
|
// image is already in disabled state
|
|
|
|
case librbd.MirrorImageDisabled:
|
|
|
|
// image mirroring is still disabling
|
|
|
|
case librbd.MirrorImageDisabling:
|
|
|
|
return nil, status.Errorf(codes.Aborted, "%s is in disabling state", volumeID)
|
2021-02-11 10:01:38 +00:00
|
|
|
case librbd.MirrorImageEnabled:
|
2021-08-12 07:05:56 +00:00
|
|
|
return disableVolumeReplication(rbdVol, mirroringInfo, force)
|
|
|
|
default:
|
2021-09-02 16:01:17 +00:00
|
|
|
return nil, status.Errorf(codes.InvalidArgument, "image is in %s Mode", mirroringInfo.State)
|
2021-08-12 07:05:56 +00:00
|
|
|
}
|
2021-07-22 05:45:17 +00:00
|
|
|
|
2021-08-12 07:05:56 +00:00
|
|
|
return &replication.DisableVolumeReplicationResponse{}, nil
|
|
|
|
}
|
2021-07-22 05:45:17 +00:00
|
|
|
|
2021-08-12 07:05:56 +00:00
|
|
|
func disableVolumeReplication(rbdVol *rbdVolume,
|
|
|
|
mirroringInfo *librbd.MirrorImageInfo,
|
2022-06-01 10:17:19 +00:00
|
|
|
force bool,
|
|
|
|
) (*replication.DisableVolumeReplicationResponse, error) {
|
2021-08-12 07:05:56 +00:00
|
|
|
if !mirroringInfo.Primary {
|
|
|
|
// Return success if the below condition is met
|
|
|
|
// Local image is secondary
|
|
|
|
// Local image is in up+replaying state
|
|
|
|
|
|
|
|
// If the image is in a secondary and its state is up+replaying means
|
2022-04-14 11:59:47 +00:00
|
|
|
// its a healthy secondary and the image is primary somewhere in the
|
2021-08-12 07:05:56 +00:00
|
|
|
// remote cluster and the local image is getting replayed. Return
|
|
|
|
// success for the Disabling mirroring as we cannot disable mirroring
|
|
|
|
// on the secondary image, when the image on the primary site gets
|
|
|
|
// disabled the image on all the remote (secondary) clusters will get
|
|
|
|
// auto-deleted. This helps in garbage collecting the volume
|
|
|
|
// replication Kubernetes artifacts after failback operation.
|
|
|
|
localStatus, rErr := rbdVol.getLocalState()
|
|
|
|
if rErr != nil {
|
|
|
|
return nil, status.Error(codes.Internal, rErr.Error())
|
2021-03-17 07:50:30 +00:00
|
|
|
}
|
2021-08-12 07:05:56 +00:00
|
|
|
if localStatus.Up && localStatus.State == librbd.MirrorImageStatusStateReplaying {
|
|
|
|
return &replication.DisableVolumeReplicationResponse{}, nil
|
2021-03-17 07:50:30 +00:00
|
|
|
}
|
2021-07-22 05:45:17 +00:00
|
|
|
|
2021-08-12 07:05:56 +00:00
|
|
|
return nil, status.Errorf(codes.InvalidArgument,
|
|
|
|
"secondary image status is up=%t and state=%s",
|
|
|
|
localStatus.Up,
|
|
|
|
localStatus.State)
|
|
|
|
}
|
|
|
|
err := rbdVol.disableImageMirroring(force)
|
|
|
|
if err != nil {
|
|
|
|
return nil, status.Error(codes.Internal, err.Error())
|
|
|
|
}
|
|
|
|
// the image state can be still disabling once we disable the mirroring
|
|
|
|
// check the mirroring is disabled or not
|
|
|
|
mirroringInfo, err = rbdVol.getImageMirroringInfo()
|
|
|
|
if err != nil {
|
|
|
|
return nil, status.Error(codes.Internal, err.Error())
|
|
|
|
}
|
|
|
|
if mirroringInfo.State == librbd.MirrorImageDisabling {
|
|
|
|
return nil, status.Errorf(codes.Aborted, "%s is in disabling state", rbdVol.VolID)
|
2021-02-11 10:01:38 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
return &replication.DisableVolumeReplicationResponse{}, nil
|
|
|
|
}
|
|
|
|
|
|
|
|
// PromoteVolume extracts the RBD volume information from the volumeID, If the
|
|
|
|
// image is present, mirroring is enabled and the image is in demoted state it
|
|
|
|
// will promote the volume as primary.
|
|
|
|
// If the image is already primary it will return success.
|
2021-03-17 07:27:04 +00:00
|
|
|
func (rs *ReplicationServer) PromoteVolume(ctx context.Context,
|
2021-02-11 10:01:38 +00:00
|
|
|
req *replication.PromoteVolumeRequest,
|
|
|
|
) (*replication.PromoteVolumeResponse, error) {
|
2021-03-17 07:27:04 +00:00
|
|
|
volumeID := req.GetVolumeId()
|
|
|
|
if volumeID == "" {
|
|
|
|
return nil, status.Error(codes.InvalidArgument, "empty volume ID in request")
|
|
|
|
}
|
|
|
|
cr, err := util.NewUserCredentials(req.GetSecrets())
|
2021-02-11 10:01:38 +00:00
|
|
|
if err != nil {
|
2021-03-17 07:27:04 +00:00
|
|
|
return nil, status.Error(codes.Internal, err.Error())
|
2021-02-11 10:01:38 +00:00
|
|
|
}
|
2021-03-17 07:27:04 +00:00
|
|
|
defer cr.DeleteCredentials()
|
2021-02-11 10:01:38 +00:00
|
|
|
|
2021-03-17 07:27:04 +00:00
|
|
|
if acquired := rs.VolumeLocks.TryAcquire(volumeID); !acquired {
|
2021-08-24 15:03:25 +00:00
|
|
|
log.ErrorLog(ctx, util.VolumeOperationAlreadyExistsFmt, volumeID)
|
2021-07-22 05:45:17 +00:00
|
|
|
|
2021-03-17 07:27:04 +00:00
|
|
|
return nil, status.Errorf(codes.Aborted, util.VolumeOperationAlreadyExistsFmt, volumeID)
|
|
|
|
}
|
|
|
|
defer rs.VolumeLocks.Release(volumeID)
|
|
|
|
|
2021-12-09 08:00:52 +00:00
|
|
|
rbdVol, err := GenVolFromVolID(ctx, volumeID, cr, req.GetSecrets())
|
2021-03-17 07:27:04 +00:00
|
|
|
defer rbdVol.Destroy()
|
|
|
|
if err != nil {
|
|
|
|
switch {
|
|
|
|
case errors.Is(err, 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.Pool, volumeID)
|
|
|
|
default:
|
|
|
|
err = status.Errorf(codes.Internal, err.Error())
|
|
|
|
}
|
2021-07-22 05:45:17 +00:00
|
|
|
|
2021-03-17 07:27:04 +00:00
|
|
|
return nil, err
|
|
|
|
}
|
2021-02-11 10:01:38 +00:00
|
|
|
|
|
|
|
mirroringInfo, err := rbdVol.getImageMirroringInfo()
|
|
|
|
if err != nil {
|
2021-08-24 15:03:25 +00:00
|
|
|
log.ErrorLog(ctx, err.Error())
|
2021-07-22 05:45:17 +00:00
|
|
|
|
2021-02-11 10:01:38 +00:00
|
|
|
return nil, status.Error(codes.Internal, err.Error())
|
|
|
|
}
|
|
|
|
|
|
|
|
if mirroringInfo.State != librbd.MirrorImageEnabled {
|
2021-06-25 11:52:34 +00:00
|
|
|
return nil, status.Errorf(
|
|
|
|
codes.InvalidArgument,
|
|
|
|
"mirroring is not enabled on %s, image is in %d Mode",
|
|
|
|
rbdVol.VolID,
|
|
|
|
mirroringInfo.State)
|
2021-02-11 10:01:38 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
// promote secondary to primary
|
|
|
|
if !mirroringInfo.Primary {
|
2021-12-22 03:36:37 +00:00
|
|
|
if req.GetForce() {
|
|
|
|
// workaround for https://github.com/ceph/ceph-csi/issues/2736
|
|
|
|
// TODO: remove this workaround when the issue is fixed
|
|
|
|
err = rbdVol.forcePromoteImage(cr)
|
|
|
|
} else {
|
|
|
|
err = rbdVol.promoteImage(req.GetForce())
|
|
|
|
}
|
2021-02-11 10:01:38 +00:00
|
|
|
if err != nil {
|
2021-08-24 15:03:25 +00:00
|
|
|
log.ErrorLog(ctx, err.Error())
|
2021-04-06 05:48:01 +00:00
|
|
|
// In case of the DR the image on the primary site cannot be
|
|
|
|
// demoted as the cluster is down, during failover the image need
|
|
|
|
// to be force promoted. RBD returns `Device or resource busy`
|
|
|
|
// error message if the image cannot be promoted for above reason.
|
|
|
|
// Return FailedPrecondition so that replication operator can send
|
|
|
|
// request to force promote the image.
|
|
|
|
if strings.Contains(err.Error(), "Device or resource busy") {
|
|
|
|
return nil, status.Error(codes.FailedPrecondition, err.Error())
|
|
|
|
}
|
2021-07-22 05:45:17 +00:00
|
|
|
|
2021-02-11 10:01:38 +00:00
|
|
|
return nil, status.Error(codes.Internal, err.Error())
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2021-11-15 06:00:13 +00:00
|
|
|
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)
|
|
|
|
}
|
|
|
|
|
2021-02-11 10:01:38 +00:00
|
|
|
return &replication.PromoteVolumeResponse{}, 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.
|
|
|
|
// If the image is already secondary it will return success.
|
2021-03-17 07:27:04 +00:00
|
|
|
func (rs *ReplicationServer) DemoteVolume(ctx context.Context,
|
2021-02-11 10:01:38 +00:00
|
|
|
req *replication.DemoteVolumeRequest,
|
|
|
|
) (*replication.DemoteVolumeResponse, error) {
|
2021-03-17 07:27:04 +00:00
|
|
|
volumeID := req.GetVolumeId()
|
|
|
|
if volumeID == "" {
|
|
|
|
return nil, status.Error(codes.InvalidArgument, "empty volume ID in request")
|
|
|
|
}
|
|
|
|
cr, err := util.NewUserCredentials(req.GetSecrets())
|
2021-02-11 10:01:38 +00:00
|
|
|
if err != nil {
|
2021-03-17 07:27:04 +00:00
|
|
|
return nil, status.Error(codes.Internal, err.Error())
|
|
|
|
}
|
|
|
|
defer cr.DeleteCredentials()
|
|
|
|
|
|
|
|
if acquired := rs.VolumeLocks.TryAcquire(volumeID); !acquired {
|
2021-08-24 15:03:25 +00:00
|
|
|
log.ErrorLog(ctx, util.VolumeOperationAlreadyExistsFmt, volumeID)
|
2021-07-22 05:45:17 +00:00
|
|
|
|
2021-03-17 07:27:04 +00:00
|
|
|
return nil, status.Errorf(codes.Aborted, util.VolumeOperationAlreadyExistsFmt, volumeID)
|
2021-02-11 10:01:38 +00:00
|
|
|
}
|
2021-03-17 07:27:04 +00:00
|
|
|
defer rs.VolumeLocks.Release(volumeID)
|
2021-02-11 10:01:38 +00:00
|
|
|
|
2021-12-09 08:00:52 +00:00
|
|
|
rbdVol, err := GenVolFromVolID(ctx, volumeID, cr, req.GetSecrets())
|
2021-03-17 07:27:04 +00:00
|
|
|
defer rbdVol.Destroy()
|
|
|
|
if err != nil {
|
|
|
|
switch {
|
|
|
|
case errors.Is(err, 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.Pool, volumeID)
|
|
|
|
default:
|
|
|
|
err = status.Errorf(codes.Internal, err.Error())
|
|
|
|
}
|
2021-07-22 05:45:17 +00:00
|
|
|
|
2021-03-17 07:27:04 +00:00
|
|
|
return nil, err
|
|
|
|
}
|
2021-02-11 10:01:38 +00:00
|
|
|
mirroringInfo, err := rbdVol.getImageMirroringInfo()
|
|
|
|
if err != nil {
|
2021-08-24 15:03:25 +00:00
|
|
|
log.ErrorLog(ctx, err.Error())
|
2021-07-22 05:45:17 +00:00
|
|
|
|
2021-02-11 10:01:38 +00:00
|
|
|
return nil, status.Error(codes.Internal, err.Error())
|
|
|
|
}
|
|
|
|
|
|
|
|
if mirroringInfo.State != librbd.MirrorImageEnabled {
|
2021-06-25 11:52:34 +00:00
|
|
|
return nil, status.Errorf(
|
|
|
|
codes.InvalidArgument,
|
|
|
|
"mirroring is not enabled on %s, image is in %d Mode",
|
|
|
|
rbdVol.VolID,
|
|
|
|
mirroringInfo.State)
|
2021-02-11 10:01:38 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
// demote image to secondary
|
|
|
|
if mirroringInfo.Primary {
|
|
|
|
err = rbdVol.demoteImage()
|
|
|
|
if err != nil {
|
2021-08-24 15:03:25 +00:00
|
|
|
log.ErrorLog(ctx, err.Error())
|
2021-07-22 05:45:17 +00:00
|
|
|
|
2021-02-11 10:01:38 +00:00
|
|
|
return nil, status.Error(codes.Internal, err.Error())
|
|
|
|
}
|
|
|
|
}
|
2021-07-22 05:45:17 +00:00
|
|
|
|
2021-02-11 10:01:38 +00:00
|
|
|
return &replication.DemoteVolumeResponse{}, nil
|
|
|
|
}
|
|
|
|
|
2021-06-23 13:14:48 +00:00
|
|
|
// checkRemoteSiteStatus checks the state of the remote cluster.
|
|
|
|
// It returns true if the state of the remote cluster is up and unknown.
|
|
|
|
func checkRemoteSiteStatus(ctx context.Context, mirrorStatus *librbd.GlobalMirrorImageStatus) bool {
|
|
|
|
ready := true
|
2022-08-08 17:53:41 +00:00
|
|
|
found := false
|
2021-06-23 13:14:48 +00:00
|
|
|
for _, s := range mirrorStatus.SiteStatuses {
|
2021-10-25 10:28:37 +00:00
|
|
|
log.UsefulLog(
|
|
|
|
ctx,
|
2021-10-25 11:27:23 +00:00
|
|
|
"peer site mirrorUUID=%q, daemon up=%t, mirroring state=%q, description=%q and lastUpdate=%d",
|
2021-10-25 10:28:37 +00:00
|
|
|
s.MirrorUUID,
|
2021-10-25 11:27:23 +00:00
|
|
|
s.Up,
|
|
|
|
s.State,
|
2021-10-25 10:28:37 +00:00
|
|
|
s.Description,
|
|
|
|
s.LastUpdate)
|
2021-06-23 13:14:48 +00:00
|
|
|
if s.MirrorUUID != "" {
|
2022-08-08 17:53:41 +00:00
|
|
|
found = true
|
|
|
|
// If ready is already "false" do not flip it based on another remote peer status
|
|
|
|
if ready && (s.State != librbd.MirrorImageStatusStateUnknown || !s.Up) {
|
2021-06-23 13:14:48 +00:00
|
|
|
ready = false
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2022-08-08 17:53:41 +00:00
|
|
|
// Return readiness only if at least one remote peer status was processed
|
|
|
|
return found && ready
|
2021-06-23 13:14:48 +00:00
|
|
|
}
|
|
|
|
|
2021-02-11 10:01:38 +00:00
|
|
|
// ResyncVolume extracts the RBD volume information from the volumeID, If the
|
|
|
|
// image is present, mirroring is enabled and the image is in demoted state.
|
|
|
|
// If yes it will resync the image to correct the split-brain.
|
2021-03-17 07:27:04 +00:00
|
|
|
func (rs *ReplicationServer) ResyncVolume(ctx context.Context,
|
2021-02-11 10:01:38 +00:00
|
|
|
req *replication.ResyncVolumeRequest,
|
|
|
|
) (*replication.ResyncVolumeResponse, error) {
|
2021-03-17 07:27:04 +00:00
|
|
|
volumeID := req.GetVolumeId()
|
|
|
|
if volumeID == "" {
|
|
|
|
return nil, status.Error(codes.InvalidArgument, "empty volume ID in request")
|
|
|
|
}
|
|
|
|
cr, err := util.NewUserCredentials(req.GetSecrets())
|
2021-02-11 10:01:38 +00:00
|
|
|
if err != nil {
|
2021-03-17 07:27:04 +00:00
|
|
|
return nil, status.Error(codes.Internal, err.Error())
|
|
|
|
}
|
|
|
|
defer cr.DeleteCredentials()
|
|
|
|
|
|
|
|
if acquired := rs.VolumeLocks.TryAcquire(volumeID); !acquired {
|
2021-08-24 15:03:25 +00:00
|
|
|
log.ErrorLog(ctx, util.VolumeOperationAlreadyExistsFmt, volumeID)
|
2021-07-22 05:45:17 +00:00
|
|
|
|
2021-03-17 07:27:04 +00:00
|
|
|
return nil, status.Errorf(codes.Aborted, util.VolumeOperationAlreadyExistsFmt, volumeID)
|
|
|
|
}
|
|
|
|
defer rs.VolumeLocks.Release(volumeID)
|
2021-12-09 08:00:52 +00:00
|
|
|
rbdVol, err := GenVolFromVolID(ctx, volumeID, cr, req.GetSecrets())
|
2021-03-17 07:27:04 +00:00
|
|
|
defer rbdVol.Destroy()
|
|
|
|
if err != nil {
|
|
|
|
switch {
|
|
|
|
case errors.Is(err, 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.Pool, volumeID)
|
|
|
|
default:
|
|
|
|
err = status.Errorf(codes.Internal, err.Error())
|
|
|
|
}
|
2021-07-22 05:45:17 +00:00
|
|
|
|
2021-02-11 10:01:38 +00:00
|
|
|
return nil, err
|
|
|
|
}
|
|
|
|
|
|
|
|
mirroringInfo, err := rbdVol.getImageMirroringInfo()
|
|
|
|
if err != nil {
|
|
|
|
// in case of Resync the image will get deleted and gets recreated and
|
2021-03-18 07:30:57 +00:00
|
|
|
// it takes time for this operation.
|
2021-08-24 15:03:25 +00:00
|
|
|
log.ErrorLog(ctx, err.Error())
|
2021-07-22 05:45:17 +00:00
|
|
|
|
2021-03-18 07:30:57 +00:00
|
|
|
return nil, status.Error(codes.Aborted, err.Error())
|
2021-02-11 10:01:38 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
if mirroringInfo.State != librbd.MirrorImageEnabled {
|
|
|
|
return nil, status.Error(codes.InvalidArgument, "image mirroring is not enabled")
|
|
|
|
}
|
|
|
|
|
|
|
|
// return error if the image is still primary
|
|
|
|
if mirroringInfo.Primary {
|
|
|
|
return nil, status.Error(codes.InvalidArgument, "image is in primary state")
|
|
|
|
}
|
|
|
|
|
2021-07-06 12:19:55 +00:00
|
|
|
mirrorStatus, err := rbdVol.getImageMirroringStatus()
|
2021-02-11 10:01:38 +00:00
|
|
|
if err != nil {
|
2021-09-14 15:29:25 +00:00
|
|
|
// the image gets recreated after issuing resync
|
2021-02-11 10:01:38 +00:00
|
|
|
if errors.Is(err, ErrImageNotFound) {
|
2021-09-14 15:29:25 +00:00
|
|
|
// caller retries till RBD syncs an initial version of the image to
|
|
|
|
// report its status in the resync call. Ideally, this line will not
|
|
|
|
// be executed as the error would get returned due to getImageMirroringInfo
|
|
|
|
// failing to find an image above.
|
|
|
|
return nil, status.Error(codes.Aborted, err.Error())
|
2021-02-11 10:01:38 +00:00
|
|
|
}
|
2021-08-24 15:03:25 +00:00
|
|
|
log.ErrorLog(ctx, err.Error())
|
2021-07-22 05:45:17 +00:00
|
|
|
|
2021-02-11 10:01:38 +00:00
|
|
|
return nil, status.Error(codes.Internal, err.Error())
|
|
|
|
}
|
|
|
|
ready := false
|
2021-06-23 13:14:48 +00:00
|
|
|
|
|
|
|
localStatus, err := mirrorStatus.LocalStatus()
|
|
|
|
if err != nil {
|
2021-08-24 15:03:25 +00:00
|
|
|
log.ErrorLog(ctx, err.Error())
|
2021-06-23 13:14:48 +00:00
|
|
|
|
|
|
|
return nil, fmt.Errorf("failed to get local status: %w", err)
|
|
|
|
}
|
2021-10-25 11:22:14 +00:00
|
|
|
|
|
|
|
// convert the last update time to UTC
|
|
|
|
lastUpdateTime := time.Unix(localStatus.LastUpdate, 0).UTC()
|
|
|
|
log.UsefulLog(
|
|
|
|
ctx,
|
2021-10-25 11:27:23 +00:00
|
|
|
"local status: daemon up=%t, image mirroring state=%q, description=%q and lastUpdate=%s",
|
|
|
|
localStatus.Up,
|
|
|
|
localStatus.State,
|
2021-10-25 11:22:14 +00:00
|
|
|
localStatus.Description,
|
|
|
|
lastUpdateTime)
|
|
|
|
|
2021-04-28 07:46:00 +00:00
|
|
|
// To recover from split brain (up+error) state the image need to be
|
|
|
|
// demoted and requested for resync on site-a and then the image on site-b
|
|
|
|
// should be demoted. The volume should be marked to ready=true when the
|
|
|
|
// image state on both the clusters are up+unknown because during the last
|
|
|
|
// snapshot syncing the data gets copied first and then image state on the
|
|
|
|
// site-a changes to up+unknown.
|
|
|
|
|
|
|
|
// If the image state on both the sites are up+unknown consider that
|
|
|
|
// complete data is synced as the last snapshot
|
|
|
|
// gets exchanged between the clusters.
|
2021-10-25 11:22:14 +00:00
|
|
|
if localStatus.State == librbd.MirrorImageStatusStateUnknown && localStatus.Up {
|
2021-06-23 13:14:48 +00:00
|
|
|
ready = checkRemoteSiteStatus(ctx, mirrorStatus)
|
2021-02-11 10:01:38 +00:00
|
|
|
}
|
2021-03-18 07:16:36 +00:00
|
|
|
|
2022-06-22 05:48:10 +00:00
|
|
|
err = resyncVolume(localStatus, rbdVol, req.Force)
|
|
|
|
if err != nil {
|
|
|
|
log.ErrorLog(ctx, err.Error())
|
2021-09-14 15:29:25 +00:00
|
|
|
|
2022-06-22 05:48:10 +00:00
|
|
|
return nil, err
|
2021-03-18 07:16:36 +00:00
|
|
|
}
|
|
|
|
|
2021-10-25 10:24:12 +00:00
|
|
|
err = checkVolumeResyncStatus(localStatus)
|
|
|
|
if err != nil {
|
|
|
|
return nil, status.Error(codes.Internal, err.Error())
|
|
|
|
}
|
2021-08-05 09:11:28 +00:00
|
|
|
|
2021-11-18 10:59:50 +00:00
|
|
|
err = repairResyncedImageID(ctx, rbdVol, ready)
|
|
|
|
if err != nil {
|
|
|
|
return nil, status.Errorf(codes.Internal, "failed to resync Image ID: %s", err.Error())
|
|
|
|
}
|
|
|
|
|
2021-02-11 10:01:38 +00:00
|
|
|
resp := &replication.ResyncVolumeResponse{
|
|
|
|
Ready: ready,
|
|
|
|
}
|
2021-07-22 05:45:17 +00:00
|
|
|
|
2021-02-11 10:01:38 +00:00
|
|
|
return resp, nil
|
|
|
|
}
|
2021-10-14 11:34:18 +00:00
|
|
|
|
2022-09-09 11:26:41 +00:00
|
|
|
// GetVolumeReplicationInfo extracts the RBD volume information from the volumeID, If the
|
|
|
|
// image is present, mirroring is enabled and the image is in primary state.
|
|
|
|
func (rs *ReplicationServer) GetVolumeReplicationInfo(ctx context.Context,
|
|
|
|
req *replication.GetVolumeReplicationInfoRequest,
|
|
|
|
) (*replication.GetVolumeReplicationInfoResponse, error) {
|
|
|
|
volumeID := req.GetVolumeId()
|
|
|
|
if volumeID == "" {
|
|
|
|
return nil, status.Error(codes.InvalidArgument, "empty volume ID in request")
|
|
|
|
}
|
|
|
|
cr, err := util.NewUserCredentials(req.GetSecrets())
|
|
|
|
if err != nil {
|
|
|
|
return nil, status.Error(codes.Internal, err.Error())
|
|
|
|
}
|
|
|
|
defer cr.DeleteCredentials()
|
|
|
|
|
|
|
|
if acquired := rs.VolumeLocks.TryAcquire(volumeID); !acquired {
|
|
|
|
log.ErrorLog(ctx, util.VolumeOperationAlreadyExistsFmt, volumeID)
|
|
|
|
|
|
|
|
return nil, status.Errorf(codes.Aborted, util.VolumeOperationAlreadyExistsFmt, volumeID)
|
|
|
|
}
|
|
|
|
defer rs.VolumeLocks.Release(volumeID)
|
|
|
|
rbdVol, err := GenVolFromVolID(ctx, volumeID, cr, req.GetSecrets())
|
|
|
|
defer rbdVol.Destroy()
|
|
|
|
if err != nil {
|
|
|
|
switch {
|
|
|
|
case errors.Is(err, 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.Pool, volumeID)
|
|
|
|
default:
|
|
|
|
err = status.Errorf(codes.Internal, err.Error())
|
|
|
|
}
|
|
|
|
|
|
|
|
return nil, err
|
|
|
|
}
|
|
|
|
|
|
|
|
mirroringInfo, err := rbdVol.getImageMirroringInfo()
|
|
|
|
if err != nil {
|
|
|
|
log.ErrorLog(ctx, err.Error())
|
|
|
|
|
|
|
|
return nil, status.Error(codes.Aborted, err.Error())
|
|
|
|
}
|
|
|
|
|
|
|
|
if mirroringInfo.State != librbd.MirrorImageEnabled {
|
|
|
|
return nil, status.Error(codes.InvalidArgument, "image mirroring is not enabled")
|
|
|
|
}
|
|
|
|
|
|
|
|
// return error if the image is not in primary state
|
|
|
|
if !mirroringInfo.Primary {
|
|
|
|
return nil, status.Error(codes.InvalidArgument, "image is not in primary state")
|
|
|
|
}
|
|
|
|
|
|
|
|
mirrorStatus, err := rbdVol.getImageMirroringStatus()
|
|
|
|
if err != nil {
|
|
|
|
if errors.Is(err, ErrImageNotFound) {
|
|
|
|
return nil, status.Error(codes.Aborted, err.Error())
|
|
|
|
}
|
|
|
|
log.ErrorLog(ctx, err.Error())
|
|
|
|
|
|
|
|
return nil, status.Error(codes.Internal, err.Error())
|
|
|
|
}
|
|
|
|
|
2022-09-14 11:09:35 +00:00
|
|
|
remoteStatus, err := RemoteStatus(mirrorStatus)
|
2022-09-09 11:26:41 +00:00
|
|
|
if err != nil {
|
|
|
|
log.ErrorLog(ctx, err.Error())
|
|
|
|
|
2022-09-14 11:09:35 +00:00
|
|
|
return nil, fmt.Errorf("failed to get remote status: %w", err)
|
2022-09-09 11:26:41 +00:00
|
|
|
}
|
|
|
|
|
2022-09-14 11:09:35 +00:00
|
|
|
description := remoteStatus.Description
|
2022-09-09 11:26:41 +00:00
|
|
|
lastSyncTime, err := getLastSyncTime(description)
|
|
|
|
if err != nil {
|
|
|
|
return nil, fmt.Errorf("failed to get last sync time: %w", err)
|
|
|
|
}
|
|
|
|
|
|
|
|
resp := &replication.GetVolumeReplicationInfoResponse{
|
|
|
|
LastSyncTime: lastSyncTime,
|
|
|
|
}
|
|
|
|
|
|
|
|
return resp, nil
|
|
|
|
}
|
|
|
|
|
2022-09-14 11:09:35 +00:00
|
|
|
// RemoteStatus returns one SiteMirrorImageStatus item from the SiteStatuses
|
|
|
|
// slice that corresponds to the remote site's status. If the remote status
|
|
|
|
// is not found than the error ErrNotExist will be returned.
|
|
|
|
func RemoteStatus(gmis *librbd.GlobalMirrorImageStatus) (librbd.SiteMirrorImageStatus, error) {
|
|
|
|
var (
|
|
|
|
ss librbd.SiteMirrorImageStatus
|
|
|
|
err error = librbd.ErrNotExist
|
|
|
|
)
|
|
|
|
for i := range gmis.SiteStatuses {
|
|
|
|
if gmis.SiteStatuses[i].MirrorUUID != "" {
|
|
|
|
ss = gmis.SiteStatuses[i]
|
|
|
|
err = nil
|
|
|
|
|
|
|
|
break
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
return ss, err
|
|
|
|
}
|
|
|
|
|
2022-09-09 11:26:41 +00:00
|
|
|
// This function gets the local snapshot time from the description
|
|
|
|
// of localStatus and converts it into required type.
|
|
|
|
func getLastSyncTime(description string) (*timestamppb.Timestamp, error) {
|
|
|
|
// Format of the description will be as followed:
|
|
|
|
// description = "replaying,{"bytes_per_second":0.0,
|
|
|
|
// "bytes_per_snapshot":149504.0,"local_snapshot_timestamp":1662655501
|
|
|
|
// ,"remote_snapshot_timestamp":1662655501}"
|
|
|
|
// In case there is no local snapshot timestamp we can pass the default value
|
|
|
|
if description == "" {
|
|
|
|
return nil, nil
|
|
|
|
}
|
|
|
|
splittedString := strings.SplitN(description, ",", 2)
|
2022-09-14 11:09:35 +00:00
|
|
|
if len(splittedString) == 1 {
|
|
|
|
return nil, nil
|
|
|
|
}
|
2022-09-09 11:26:41 +00:00
|
|
|
type localStatus struct {
|
|
|
|
LocalSnapshotTime int64 `json:"local_snapshot_timestamp"`
|
|
|
|
}
|
|
|
|
|
|
|
|
var localSnapTime localStatus
|
|
|
|
err := json.Unmarshal([]byte(splittedString[1]), &localSnapTime)
|
|
|
|
if err != nil {
|
|
|
|
return nil, fmt.Errorf("failed to unmarshal description: %w", err)
|
|
|
|
}
|
|
|
|
|
|
|
|
lastUpdateTime := time.Unix(localSnapTime.LocalSnapshotTime, 0)
|
|
|
|
lastSyncTime := timestamppb.New(lastUpdateTime)
|
|
|
|
|
|
|
|
return lastSyncTime, nil
|
|
|
|
}
|
|
|
|
|
2022-06-22 05:48:10 +00:00
|
|
|
func resyncVolume(localStatus librbd.SiteMirrorImageStatus, rbdVol *rbdVolume, force bool) error {
|
|
|
|
if resyncRequired(localStatus) {
|
|
|
|
// If the force option is not set return the error message to retry
|
|
|
|
// with Force option.
|
|
|
|
if !force {
|
|
|
|
return status.Errorf(codes.FailedPrecondition,
|
|
|
|
"image is in %q state, description (%s). Force resync to recover volume",
|
|
|
|
localStatus.State, localStatus.Description)
|
|
|
|
}
|
|
|
|
err := rbdVol.resyncImage()
|
|
|
|
if err != nil {
|
|
|
|
return status.Error(codes.Internal, err.Error())
|
|
|
|
}
|
|
|
|
|
|
|
|
// If we issued a resync, return a non-final error as image needs to be recreated
|
|
|
|
// locally. Caller retries till RBD syncs an initial version of the image to
|
|
|
|
// report its status in the resync request.
|
|
|
|
return status.Error(codes.Unavailable, "awaiting initial resync due to split brain")
|
|
|
|
}
|
|
|
|
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
|
2021-10-25 10:24:12 +00:00
|
|
|
func checkVolumeResyncStatus(localStatus librbd.SiteMirrorImageStatus) error {
|
|
|
|
// we are considering 2 states to check resync started and resync completed
|
|
|
|
// as below. all other states will be considered as an error state so that
|
|
|
|
// cephCSI can return error message and volume replication operator can
|
|
|
|
// mark the VolumeReplication status as not resyncing for the volume.
|
|
|
|
|
|
|
|
// If the state is Replaying means the resync is going on.
|
|
|
|
// Once the volume on remote cluster is demoted and resync
|
2022-08-08 11:53:35 +00:00
|
|
|
// is completed the image state will be moved to UNKNOWN.
|
|
|
|
// RBD mirror daemon should be always running on the primary cluster.
|
|
|
|
if !localStatus.Up || (localStatus.State != librbd.MirrorImageStatusStateReplaying &&
|
|
|
|
localStatus.State != librbd.MirrorImageStatusStateUnknown) {
|
2021-10-25 10:24:12 +00:00
|
|
|
return fmt.Errorf(
|
2022-08-08 11:53:35 +00:00
|
|
|
"not resyncing. Local status: daemon up=%t image is in %q state",
|
|
|
|
localStatus.Up, localStatus.State)
|
2021-10-25 10:24:12 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
|
2021-10-14 11:34:18 +00:00
|
|
|
// resyncRequired returns true if local image is in split-brain state and image
|
|
|
|
// needs resync.
|
|
|
|
func resyncRequired(localStatus librbd.SiteMirrorImageStatus) bool {
|
|
|
|
// resync is required if the image is in error state or the description
|
|
|
|
// contains split-brain message.
|
|
|
|
// In some corner cases like `re-player shutdown` the local image will not
|
|
|
|
// be in an error state. It would be also worth considering the `description`
|
|
|
|
// field to make sure about split-brain.
|
2021-10-25 11:22:14 +00:00
|
|
|
if localStatus.State == librbd.MirrorImageStatusStateError ||
|
2022-06-01 10:17:19 +00:00
|
|
|
strings.Contains(localStatus.Description, "split-brain") {
|
2021-10-14 11:34:18 +00:00
|
|
|
return true
|
|
|
|
}
|
|
|
|
|
|
|
|
return false
|
|
|
|
}
|
2021-11-18 10:59:50 +00:00
|
|
|
|
|
|
|
// repairResyncedImageID updates the existing image ID with new one.
|
|
|
|
func repairResyncedImageID(ctx context.Context, rv *rbdVolume, ready bool) error {
|
|
|
|
// During resync operation the local image will get deleted and a new
|
|
|
|
// image is recreated by the rbd mirroring. The new image will have a
|
|
|
|
// new image ID. Once resync is completed update the image ID in the OMAP
|
|
|
|
// to get the image removed from the trash during DeleteVolume.
|
|
|
|
|
|
|
|
// if the image is not completely resynced skip repairing image ID.
|
|
|
|
if !ready {
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
j, err := volJournal.Connect(rv.Monitors, rv.RadosNamespace, rv.conn.Creds)
|
|
|
|
if err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
defer j.Destroy()
|
|
|
|
// reset the image ID which is stored in the existing OMAP
|
|
|
|
return rv.repairImageID(ctx, j, true)
|
|
|
|
}
|