mirrors/ceph-csi
1
0
Fork 0
mirror of https://github.com/ceph/ceph-csi.git synced 2025-06-13 02:33:34 +00:00
Code Issues Packages Projects Releases Wiki Activity

rbd: get lastsyncbytes and lastsycduration for volrep

Browse Source 
This commit get more information from the description
like lastsyncbytes and lastsyncduration and send them
as a response of getvolumereplicationinfo request.

Signed-off-by: Yati Padia <ypadia@redhat.com>
This commit is contained in:
Yati Padia
2023-06-09 16:12:50 +05:30
committed by mergify[bot]
parent dbdb9086d8
commit 2e2e9044c0
6 changed files with 266 additions and 138 deletions
Download Patch File Download Diff File Expand all files Collapse all files

68
internal/csi-addons/rbd/replication.go
View File

@ -36,6 +36,7 @@ import (
"google.golang.org/grpc"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/status"
"google.golang.org/protobuf/types/known/durationpb"
"google.golang.org/protobuf/types/known/timestamppb"
)
@ -754,18 +755,14 @@ func (rs *ReplicationServer) GetVolumeReplicationInfo(ctx context.Context,
}
description := remoteStatus.Description
lastSyncTime, err := getLastSyncTime(description)
resp, err := getLastSyncInfo(description)
if err != nil {
if errors.Is(err, corerbd.ErrLastSyncTimeNotFound) {
return nil, status.Errorf(codes.NotFound, "failed to get last sync time: %v", err)
return nil, status.Errorf(codes.NotFound, "failed to get last sync info: %v", err)
}
log.ErrorLog(ctx, err.Error())
return nil, status.Errorf(codes.Internal, "failed to get last sync time: %v", err)
}
resp := &replication.GetVolumeReplicationInfoResponse{
LastSyncTime: lastSyncTime,
return nil, status.Errorf(codes.Internal, "failed to get last sync info: %v", err)
}
return resp, nil
@ -791,42 +788,71 @@ func RemoteStatus(gmis *librbd.GlobalMirrorImageStatus) (librbd.SiteMirrorImageS
return ss, err
}
// 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) {
// This function gets the local snapshot time, last sync snapshot seconds
// and last sync bytes from the description of localStatus and convert
// it into required types.
func getLastSyncInfo(description string) (*replication.GetVolumeReplicationInfoResponse, 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}"
// description = `replaying, {"bytes_per_second":0.0,"bytes_per_snapshot":81920.0,
// "last_snapshot_bytes":81920,"last_snapshot_sync_seconds":0,
// "local_snapshot_timestamp":1684675261,
// "remote_snapshot_timestamp":1684675261,"replay_state":"idle"}`
// In case there is no last snapshot bytes returns 0 as the
// LastSyncBytes is optional.
// In case there is no last snapshot sync seconds, it returns nil as the
// LastSyncDuration is optional.
// In case there is no local snapshot timestamp return an error as the
// LastSyncTime is required.
var response replication.GetVolumeReplicationInfoResponse
if description == "" {
return nil, fmt.Errorf("empty description: %w", corerbd.ErrLastSyncTimeNotFound)
}
splittedString := strings.SplitN(description, ",", 2)
if len(splittedString) == 1 {
return nil, fmt.Errorf("no local snapshot timestamp: %w", corerbd.ErrLastSyncTimeNotFound)
return nil, fmt.Errorf("no snapshot details: %w", corerbd.ErrLastSyncTimeNotFound)
}
type localStatus struct {
LocalSnapshotTime int64 `json:"local_snapshot_timestamp"`
LocalSnapshotTime int64 `json:"local_snapshot_timestamp"`
LastSnapshotBytes int64 `json:"last_snapshot_bytes"`
LastSnapshotDuration int64 `json:"last_snapshot_sync_seconds"`
}
var localSnapTime localStatus
err := json.Unmarshal([]byte(splittedString[1]), &localSnapTime)
var localSnapInfo localStatus
err := json.Unmarshal([]byte(splittedString[1]), &localSnapInfo)
if err != nil {
return nil, fmt.Errorf("failed to unmarshal description: %w", err)
return nil, fmt.Errorf("failed to unmarshal local snapshot info: %w", err)
}
// If the json unmarsal is successful but the local snapshot time is 0, we
// need to consider it as an error as the LastSyncTime is required.
if localSnapTime.LocalSnapshotTime == 0 {
if localSnapInfo.LocalSnapshotTime == 0 {
return nil, fmt.Errorf("empty local snapshot timestamp: %w", corerbd.ErrLastSyncTimeNotFound)
}
if localSnapInfo.LastSnapshotDuration == 0 {
response.LastSyncDuration = nil
} else {
// converts localSnapshotDuration of type int64 to string format with
// appended `s` seconds required for time.ParseDuration
lastDurationTime := fmt.Sprintf("%ds", localSnapInfo.LastSnapshotDuration)
// parse Duration from the lastDurationTime string
lastDuration, err := time.ParseDuration(lastDurationTime)
if err != nil {
return nil, fmt.Errorf("failed to parse last snapshot duration: %w", err)
}
// converts time.Duration to *durationpb.Duration
response.LastSyncDuration = durationpb.New(lastDuration)
}
lastUpdateTime := time.Unix(localSnapTime.LocalSnapshotTime, 0)
// converts localSnapshotTime of type int64 to time.Time
lastUpdateTime := time.Unix(localSnapInfo.LocalSnapshotTime, 0)
lastSyncTime := timestamppb.New(lastUpdateTime)
return lastSyncTime, nil
response.LastSyncTime = lastSyncTime
response.LastSyncBytes = localSnapInfo.LastSnapshotBytes
return &response, nil
}
func checkVolumeResyncStatus(localStatus librbd.SiteMirrorImageStatus) error {

117
internal/csi-addons/rbd/replication_test.go
View File

@ -20,6 +20,7 @@ import (
"context"
"errors"
"reflect"
"strconv"
"strings"
"testing"
"time"
@ -28,9 +29,11 @@ import (
librbd "github.com/ceph/go-ceph/rbd"
"github.com/ceph/go-ceph/rbd/admin"
"github.com/csi-addons/spec/lib/go/replication"
"github.com/stretchr/testify/assert"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/status"
"google.golang.org/protobuf/types/known/durationpb"
"google.golang.org/protobuf/types/known/timestamppb"
)
@ -442,58 +445,120 @@ func TestCheckRemoteSiteStatus(t *testing.T) {
}
}
func TestValidateLastSyncTime(t *testing.T) {
func TestValidateLastSyncInfo(t *testing.T) {
t.Parallel()
duration, err := time.ParseDuration(strconv.Itoa(int(56743)) + "s")
if err != nil {
t.Errorf("failed to parse duration)")
}
tests := []struct {
name string
description string
timestamp *timestamppb.Timestamp
info *replication.GetVolumeReplicationInfoResponse
expectedErr string
}{
{
"valid description",
name: "valid description",
//nolint:lll // sample output cannot be split into multiple lines.
`replaying,{"bytes_per_second":0.0,"bytes_per_snapshot":149504.0,"local_snapshot_timestamp":1662655501,"remote_snapshot_timestamp":1662655501}`,
timestamppb.New(time.Unix(1662655501, 0)),
"",
description: `replaying, {"bytes_per_second":0.0,"bytes_per_snapshot":81920.0,"last_snapshot_bytes":81920,"last_snapshot_sync_seconds":56743,"local_snapshot_timestamp":1684675261,"remote_snapshot_timestamp":1684675261,"replay_state":"idle"}`,
info: &replication.GetVolumeReplicationInfoResponse{
LastSyncTime: timestamppb.New(time.Unix(1684675261, 0)),
LastSyncDuration: durationpb.New(duration),
LastSyncBytes: 81920,
},
expectedErr: "",
},
{
"empty description",
"",
nil,
corerbd.ErrLastSyncTimeNotFound.Error(),
name: "empty description",
description: "",
info: &replication.GetVolumeReplicationInfoResponse{
LastSyncTime: nil,
LastSyncDuration: nil,
LastSyncBytes: 0,
},
expectedErr: corerbd.ErrLastSyncTimeNotFound.Error(),
},
{
"description without local_snapshot_timestamp",
`replaying,{"bytes_per_second":0.0,"bytes_per_snapshot":149504.0,"remote_snapshot_timestamp":1662655501}`,
nil,
"",
name: "description without last_snapshot_bytes",
//nolint:lll // sample output cannot be split into multiple lines.
description: `replaying, {"bytes_per_second":0.0,"last_snapshot_sync_seconds":56743,"local_snapshot_timestamp":1684675261,"remote_snapshot_timestamp":1684675261,"replay_state":"idle"}`,
info: &replication.GetVolumeReplicationInfoResponse{
LastSyncDuration: durationpb.New(duration),
LastSyncTime: timestamppb.New(time.Unix(1684675261, 0)),
LastSyncBytes: 0,
},
expectedErr: "",
},
{
"description with invalid JSON",
`replaying,{"bytes_per_second":0.0,"bytes_per_snapshot":149504.0","remote_snapshot_timestamp":1662655501`,
nil,
"failed to unmarshal",
name: "description without local_snapshot_time",
//nolint:lll // sample output cannot be split into multiple lines.
description: `replaying, {"bytes_per_second":0.0,"bytes_per_snapshot":81920.0,"last_snapshot_bytes":81920,"last_snapshot_sync_seconds":56743,"remote_snapshot_timestamp":1684675261,"replay_state":"idle"}`,
info: &replication.GetVolumeReplicationInfoResponse{
LastSyncDuration: nil,
LastSyncTime: nil,
LastSyncBytes: 0,
},
expectedErr: corerbd.ErrLastSyncTimeNotFound.Error(),
},
{
"description with no JSON",
`replaying`,
nil,
corerbd.ErrLastSyncTimeNotFound.Error(),
name: "description without last_snapshot_sync_seconds",
//nolint:lll // sample output cannot be split into multiple lines.
description: `replaying, {"bytes_per_second":0.0,"bytes_per_snapshot":81920.0,"last_snapshot_bytes":81920,"local_snapshot_timestamp":1684675261,"remote_snapshot_timestamp":1684675261,"replay_state":"idle"}`,
info: &replication.GetVolumeReplicationInfoResponse{
LastSyncDuration: nil,
LastSyncTime: timestamppb.New(time.Unix(1684675261, 0)),
LastSyncBytes: 81920,
},
expectedErr: "",
},
{
name: "description with invalid JSON",
//nolint:lll // sample output cannot be split into multiple lines.
description: `replaying,{"bytes_per_second":0.0,"last_snapshot_bytes":81920","bytes_per_snapshot":149504.0","remote_snapshot_timestamp":1662655501`,
info: &replication.GetVolumeReplicationInfoResponse{
LastSyncDuration: nil,
LastSyncTime: nil,
LastSyncBytes: 0,
},
expectedErr: "failed to unmarshal",
},
{
name: "description with no JSON",
description: `replaying`,
info: &replication.GetVolumeReplicationInfoResponse{
LastSyncDuration: nil,
LastSyncTime: nil,
LastSyncBytes: 0,
},
expectedErr: corerbd.ErrLastSyncTimeNotFound.Error(),
},
}
for _, tt := range tests {
tt := tt
t.Run(tt.name, func(t *testing.T) {
t.Parallel()
ts, err := getLastSyncTime(tt.description)
teststruct, err := getLastSyncInfo(tt.description)
if err != nil && !strings.Contains(err.Error(), tt.expectedErr) {
// returned error
t.Errorf("getLastSyncTime() returned error, expected: %v, got: %v",
t.Errorf("getLastSyncInfo() returned error, expected: %v, got: %v",
tt.expectedErr, err)
}
if !ts.AsTime().Equal(tt.timestamp.AsTime()) {
t.Errorf("getLastSyncTime() %v, expected %v", ts, tt.timestamp)
if teststruct != nil {
if teststruct.LastSyncTime.GetSeconds() != tt.info.LastSyncTime.GetSeconds() {
t.Errorf("name: %v, getLastSyncInfo() %v, expected %v", tt.name, teststruct.LastSyncTime, tt.info.LastSyncTime)
}
if tt.info.LastSyncDuration == nil && teststruct.LastSyncDuration != nil {
t.Errorf("name: %v, getLastSyncInfo() %v, expected %v", tt.name, teststruct.LastSyncDuration,
tt.info.LastSyncDuration)
}
if teststruct.LastSyncDuration.GetSeconds() != tt.info.LastSyncDuration.GetSeconds() {
t.Errorf("name: %v, getLastSyncInfo() %v, expected %v", tt.name, teststruct.LastSyncDuration,
tt.info.LastSyncDuration)
}
if teststruct.LastSyncBytes != tt.info.LastSyncBytes {
t.Errorf("name: %v, getLastSyncInfo() %v, expected %v", tt.name, teststruct.LastSyncBytes, tt.info.LastSyncBytes)
}
}
})
}