replication: refactor driving logic (no more explicit state machine)
This commit is contained in:
Christian Schwarz
@@ -0,0 +1,114 @@
|
||||
package mainfsm
|
||||
|
||||
import (
|
||||
"sort"
|
||||
|
||||
. "github.com/zrepl/zrepl/replication/logic/pdu"
|
||||
)
|
||||
|
||||
type ConflictNoCommonAncestor struct {
|
||||
SortedSenderVersions, SortedReceiverVersions []*FilesystemVersion
|
||||
}
|
||||
|
||||
func (c *ConflictNoCommonAncestor) Error() string {
|
||||
return "no common snapshot or suitable bookmark between sender and receiver"
|
||||
}
|
||||
|
||||
type ConflictDiverged struct {
|
||||
SortedSenderVersions, SortedReceiverVersions []*FilesystemVersion
|
||||
CommonAncestor *FilesystemVersion
|
||||
SenderOnly, ReceiverOnly []*FilesystemVersion
|
||||
}
|
||||
|
||||
func (c *ConflictDiverged) Error() string {
|
||||
return "the receiver's latest snapshot is not present on sender"
|
||||
}
|
||||
|
||||
func SortVersionListByCreateTXGThenBookmarkLTSnapshot(fsvslice []*FilesystemVersion) []*FilesystemVersion {
|
||||
lesser := func(s []*FilesystemVersion) func(i, j int) bool {
|
||||
return func(i, j int) bool {
|
||||
if s[i].CreateTXG < s[j].CreateTXG {
|
||||
return true
|
||||
}
|
||||
if s[i].CreateTXG == s[j].CreateTXG {
|
||||
// Bookmark < Snapshot
|
||||
return s[i].Type == FilesystemVersion_Bookmark && s[j].Type == FilesystemVersion_Snapshot
|
||||
}
|
||||
return false
|
||||
}
|
||||
}
|
||||
if sort.SliceIsSorted(fsvslice, lesser(fsvslice)) {
|
||||
return fsvslice
|
||||
}
|
||||
sorted := make([]*FilesystemVersion, len(fsvslice))
|
||||
copy(sorted, fsvslice)
|
||||
sort.Slice(sorted, lesser(sorted))
|
||||
return sorted
|
||||
}
|
||||
|
||||
// conflict may be a *ConflictDiverged or a *ConflictNoCommonAncestor
|
||||
func IncrementalPath(receiver, sender []*FilesystemVersion) (incPath []*FilesystemVersion, conflict error) {
|
||||
|
||||
receiver = SortVersionListByCreateTXGThenBookmarkLTSnapshot(receiver)
|
||||
sender = SortVersionListByCreateTXGThenBookmarkLTSnapshot(sender)
|
||||
|
||||
// Find most recent common ancestor by name, preferring snapshots over bookmarks
|
||||
|
||||
mrcaRcv := len(receiver) - 1
|
||||
mrcaSnd := len(sender) - 1
|
||||
|
||||
for mrcaRcv >= 0 && mrcaSnd >= 0 {
|
||||
if receiver[mrcaRcv].Guid == sender[mrcaSnd].Guid {
|
||||
// Since we arrive from the end of the array, and because we defined bookmark < snapshot,
|
||||
// this condition will match snapshot first, which is what we want because it gives us
|
||||
// size estimation
|
||||
break
|
||||
}
|
||||
receiverCreation, err := receiver[mrcaRcv].CreationAsTime()
|
||||
if err != nil {
|
||||
panic(err) // FIXME move this to a sorting phase before
|
||||
}
|
||||
senderCreation, err := sender[mrcaSnd].CreationAsTime()
|
||||
if err != nil {
|
||||
panic(err) // FIXME move this to the sorting phase before
|
||||
}
|
||||
|
||||
if receiverCreation.Before(senderCreation) {
|
||||
mrcaSnd--
|
||||
} else {
|
||||
mrcaRcv--
|
||||
}
|
||||
}
|
||||
|
||||
if mrcaRcv == -1 || mrcaSnd == -1 {
|
||||
return nil, &ConflictNoCommonAncestor{
|
||||
SortedSenderVersions: sender,
|
||||
SortedReceiverVersions: receiver,
|
||||
}
|
||||
}
|
||||
|
||||
if mrcaRcv != len(receiver)-1 {
|
||||
return nil, &ConflictDiverged{
|
||||
SortedSenderVersions: sender,
|
||||
SortedReceiverVersions: receiver,
|
||||
CommonAncestor: sender[mrcaSnd],
|
||||
SenderOnly: sender[mrcaSnd+1:],
|
||||
ReceiverOnly: receiver[mrcaRcv+1:],
|
||||
}
|
||||
}
|
||||
|
||||
// incPath must not contain bookmarks except initial one,
|
||||
incPath = make([]*FilesystemVersion, 0, len(sender))
|
||||
incPath = append(incPath, sender[mrcaSnd])
|
||||
// it's ok if incPath[0] is a bookmark, but not the subsequent ones in the incPath
|
||||
for i := mrcaSnd + 1; i < len(sender); i++ {
|
||||
if sender[i].Type == FilesystemVersion_Snapshot && incPath[len(incPath)-1].Guid != sender[i].Guid {
|
||||
incPath = append(incPath, sender[i])
|
||||
}
|
||||
}
|
||||
if len(incPath) == 1 {
|
||||
// nothing to do
|
||||
incPath = incPath[1:]
|
||||
}
|
||||
return incPath, nil
|
||||
}
|
||||
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,122 @@
|
||||
syntax = "proto3";
|
||||
option go_package = "pdu";
|
||||
|
||||
service Replication {
|
||||
rpc ListFilesystems (ListFilesystemReq) returns (ListFilesystemRes);
|
||||
rpc ListFilesystemVersions (ListFilesystemVersionsReq) returns (ListFilesystemVersionsRes);
|
||||
rpc DestroySnapshots (DestroySnapshotsReq) returns (DestroySnapshotsRes);
|
||||
rpc ReplicationCursor (ReplicationCursorReq) returns (ReplicationCursorRes);
|
||||
// for Send and Recv, see package rpc
|
||||
}
|
||||
|
||||
message ListFilesystemReq {}
|
||||
|
||||
message ListFilesystemRes {
|
||||
repeated Filesystem Filesystems = 1;
|
||||
bool Empty = 2;
|
||||
}
|
||||
|
||||
message Filesystem {
|
||||
string Path = 1;
|
||||
string ResumeToken = 2;
|
||||
}
|
||||
|
||||
message ListFilesystemVersionsReq {
|
||||
string Filesystem = 1;
|
||||
}
|
||||
|
||||
message ListFilesystemVersionsRes {
|
||||
repeated FilesystemVersion Versions = 1;
|
||||
}
|
||||
|
||||
message FilesystemVersion {
|
||||
enum VersionType {
|
||||
Snapshot = 0;
|
||||
Bookmark = 1;
|
||||
}
|
||||
VersionType Type = 1;
|
||||
string Name = 2;
|
||||
uint64 Guid = 3;
|
||||
uint64 CreateTXG = 4;
|
||||
string Creation = 5; // RFC 3339
|
||||
}
|
||||
|
||||
|
||||
message SendReq {
|
||||
string Filesystem = 1;
|
||||
string From = 2;
|
||||
// May be empty / null to request a full transfer of From
|
||||
string To = 3;
|
||||
|
||||
// If ResumeToken is not empty, the resume token that CAN be tried for 'zfs send' by the sender.
|
||||
// The sender MUST indicate in SendRes.UsedResumeToken
|
||||
// If it does not work, the sender SHOULD clear the resume token on their side
|
||||
// and use From and To instead
|
||||
// If ResumeToken is not empty, the GUIDs of From and To
|
||||
// MUST correspond to those encoded in the ResumeToken.
|
||||
// Otherwise, the Sender MUST return an error.
|
||||
string ResumeToken = 4;
|
||||
bool Compress = 5;
|
||||
bool Dedup = 6;
|
||||
|
||||
bool DryRun = 7;
|
||||
}
|
||||
|
||||
message Property {
|
||||
string Name = 1;
|
||||
string Value = 2;
|
||||
}
|
||||
|
||||
message SendRes {
|
||||
// Whether the resume token provided in the request has been used or not.
|
||||
bool UsedResumeToken = 2;
|
||||
|
||||
// Expected stream size determined by dry run, not exact.
|
||||
// 0 indicates that for the given SendReq, no size estimate could be made.
|
||||
int64 ExpectedSize = 3;
|
||||
|
||||
repeated Property Properties = 4;
|
||||
}
|
||||
|
||||
message ReceiveReq {
|
||||
string Filesystem = 1; // FIXME should be snapshot name, we can enforce that on recv
|
||||
|
||||
// If true, the receiver should clear the resume token before perfoming the zfs recv of the stream in the request
|
||||
bool ClearResumeToken = 2;
|
||||
}
|
||||
|
||||
message ReceiveRes {}
|
||||
|
||||
message DestroySnapshotsReq {
|
||||
string Filesystem = 1;
|
||||
// Path to filesystem, snapshot or bookmark to be destroyed
|
||||
repeated FilesystemVersion Snapshots = 2;
|
||||
}
|
||||
|
||||
message DestroySnapshotRes {
|
||||
FilesystemVersion Snapshot = 1;
|
||||
string Error = 2;
|
||||
}
|
||||
|
||||
message DestroySnapshotsRes {
|
||||
repeated DestroySnapshotRes Results = 1;
|
||||
}
|
||||
|
||||
message ReplicationCursorReq {
|
||||
string Filesystem = 1;
|
||||
message GetOp {}
|
||||
message SetOp {
|
||||
string Snapshot = 2;
|
||||
}
|
||||
oneof op {
|
||||
GetOp get = 2;
|
||||
SetOp set = 3;
|
||||
}
|
||||
}
|
||||
|
||||
message ReplicationCursorRes {
|
||||
oneof Result {
|
||||
uint64 Guid = 1;
|
||||
bool Notexist = 2;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,76 @@
|
||||
package pdu
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"github.com/zrepl/zrepl/zfs"
|
||||
"time"
|
||||
)
|
||||
|
||||
func (v *FilesystemVersion) RelName() string {
|
||||
zv, err := v.ZFSFilesystemVersion()
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
return zv.String()
|
||||
}
|
||||
|
||||
func (v FilesystemVersion_VersionType) ZFSVersionType() zfs.VersionType {
|
||||
switch v {
|
||||
case FilesystemVersion_Snapshot:
|
||||
return zfs.Snapshot
|
||||
case FilesystemVersion_Bookmark:
|
||||
return zfs.Bookmark
|
||||
default:
|
||||
panic(fmt.Sprintf("unexpected v.Type %#v", v))
|
||||
}
|
||||
}
|
||||
|
||||
func FilesystemVersionFromZFS(fsv *zfs.FilesystemVersion) *FilesystemVersion {
|
||||
var t FilesystemVersion_VersionType
|
||||
switch fsv.Type {
|
||||
case zfs.Bookmark:
|
||||
t = FilesystemVersion_Bookmark
|
||||
case zfs.Snapshot:
|
||||
t = FilesystemVersion_Snapshot
|
||||
default:
|
||||
panic("unknown fsv.Type: " + fsv.Type)
|
||||
}
|
||||
return &FilesystemVersion{
|
||||
Type: t,
|
||||
Name: fsv.Name,
|
||||
Guid: fsv.Guid,
|
||||
CreateTXG: fsv.CreateTXG,
|
||||
Creation: fsv.Creation.Format(time.RFC3339),
|
||||
}
|
||||
}
|
||||
|
||||
func FilesystemVersionCreation(t time.Time) string {
|
||||
return t.Format(time.RFC3339)
|
||||
}
|
||||
|
||||
func (v *FilesystemVersion) CreationAsTime() (time.Time, error) {
|
||||
return time.Parse(time.RFC3339, v.Creation)
|
||||
}
|
||||
|
||||
// implement fsfsm.FilesystemVersion
|
||||
func (v *FilesystemVersion) SnapshotTime() time.Time {
|
||||
t, err := v.CreationAsTime()
|
||||
if err != nil {
|
||||
panic(err) // FIXME
|
||||
}
|
||||
return t
|
||||
}
|
||||
|
||||
func (v *FilesystemVersion) ZFSFilesystemVersion() (*zfs.FilesystemVersion, error) {
|
||||
ct, err := v.CreationAsTime()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &zfs.FilesystemVersion{
|
||||
Type: v.Type.ZFSVersionType(),
|
||||
Name: v.Name,
|
||||
Guid: v.Guid,
|
||||
CreateTXG: v.CreateTXG,
|
||||
Creation: ct,
|
||||
}, nil
|
||||
}
|
||||
@@ -0,0 +1,68 @@
|
||||
package pdu
|
||||
|
||||
import (
|
||||
"github.com/stretchr/testify/assert"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
func TestFilesystemVersion_RelName(t *testing.T) {
|
||||
|
||||
type TestCase struct {
|
||||
In FilesystemVersion
|
||||
Out string
|
||||
Panic bool
|
||||
}
|
||||
|
||||
creat := FilesystemVersionCreation(time.Now())
|
||||
tcs := []TestCase{
|
||||
{
|
||||
In: FilesystemVersion{
|
||||
Type: FilesystemVersion_Snapshot,
|
||||
Name: "foobar",
|
||||
Creation: creat,
|
||||
},
|
||||
Out: "@foobar",
|
||||
},
|
||||
{
|
||||
In: FilesystemVersion{
|
||||
Type: FilesystemVersion_Bookmark,
|
||||
Name: "foobar",
|
||||
Creation: creat,
|
||||
},
|
||||
Out: "#foobar",
|
||||
},
|
||||
{
|
||||
In: FilesystemVersion{
|
||||
Type: 2342,
|
||||
Name: "foobar",
|
||||
Creation: creat,
|
||||
},
|
||||
Panic: true,
|
||||
},
|
||||
}
|
||||
|
||||
for _, tc := range tcs {
|
||||
if tc.Panic {
|
||||
assert.Panics(t, func() {
|
||||
tc.In.RelName()
|
||||
})
|
||||
} else {
|
||||
o := tc.In.RelName()
|
||||
assert.Equal(t, tc.Out, o)
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
func TestFilesystemVersion_ZFSFilesystemVersion(t *testing.T) {
|
||||
|
||||
empty := &FilesystemVersion{}
|
||||
_, err:= empty.ZFSFilesystemVersion()
|
||||
assert.Error(t, err)
|
||||
|
||||
dateInvalid := &FilesystemVersion{Creation: "foobar"}
|
||||
_, err = dateInvalid.ZFSFilesystemVersion()
|
||||
assert.Error(t, err)
|
||||
|
||||
}
|
||||
@@ -0,0 +1,430 @@
|
||||
package logic
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/prometheus/client_golang/prometheus"
|
||||
|
||||
"github.com/zrepl/zrepl/replication/driver"
|
||||
. "github.com/zrepl/zrepl/replication/logic/diff"
|
||||
"github.com/zrepl/zrepl/replication/logic/pdu"
|
||||
"github.com/zrepl/zrepl/replication/report"
|
||||
"github.com/zrepl/zrepl/util/bytecounter"
|
||||
"github.com/zrepl/zrepl/zfs"
|
||||
)
|
||||
|
||||
// Endpoint represents one side of the replication.
|
||||
//
|
||||
// An endpoint is either in Sender or Receiver mode, represented by the correspondingly
|
||||
// named interfaces defined in this package.
|
||||
type Endpoint interface {
|
||||
// Does not include placeholder filesystems
|
||||
ListFilesystems(ctx context.Context, req *pdu.ListFilesystemReq) (*pdu.ListFilesystemRes, error)
|
||||
ListFilesystemVersions(ctx context.Context, req *pdu.ListFilesystemVersionsReq) (*pdu.ListFilesystemVersionsRes, error)
|
||||
DestroySnapshots(ctx context.Context, req *pdu.DestroySnapshotsReq) (*pdu.DestroySnapshotsRes, error)
|
||||
}
|
||||
|
||||
type Sender interface {
|
||||
Endpoint
|
||||
// If a non-nil io.ReadCloser is returned, it is guaranteed to be closed before
|
||||
// any next call to the parent github.com/zrepl/zrepl/replication.Endpoint.
|
||||
// If the send request is for dry run the io.ReadCloser will be nil
|
||||
Send(ctx context.Context, r *pdu.SendReq) (*pdu.SendRes, zfs.StreamCopier, error)
|
||||
ReplicationCursor(ctx context.Context, req *pdu.ReplicationCursorReq) (*pdu.ReplicationCursorRes, error)
|
||||
}
|
||||
|
||||
type Receiver interface {
|
||||
Endpoint
|
||||
// Receive sends r and sendStream (the latter containing a ZFS send stream)
|
||||
// to the parent github.com/zrepl/zrepl/replication.Endpoint.
|
||||
Receive(ctx context.Context, req *pdu.ReceiveReq, receive zfs.StreamCopier) (*pdu.ReceiveRes, error)
|
||||
}
|
||||
|
||||
type Planner struct {
|
||||
sender Sender
|
||||
receiver Receiver
|
||||
|
||||
promSecsPerState *prometheus.HistogramVec // labels: state
|
||||
promBytesReplicated *prometheus.CounterVec // labels: filesystem
|
||||
}
|
||||
|
||||
func (p *Planner) Plan(ctx context.Context) ([]driver.FS, error) {
|
||||
fss, err := p.doPlanning(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
dfss := make([]driver.FS, len(fss))
|
||||
for i := range dfss {
|
||||
dfss[i] = fss[i]
|
||||
}
|
||||
return dfss, nil
|
||||
}
|
||||
|
||||
type Filesystem struct {
|
||||
sender Sender
|
||||
receiver Receiver
|
||||
|
||||
Path string // compat
|
||||
receiverFSExists bool // compat
|
||||
promBytesReplicated prometheus.Counter // compat
|
||||
}
|
||||
|
||||
func (f *Filesystem) PlanFS(ctx context.Context) ([]driver.Step, error) {
|
||||
steps, err := f.doPlanning(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
dsteps := make([]driver.Step, len(steps))
|
||||
for i := range dsteps {
|
||||
dsteps[i] = steps[i]
|
||||
}
|
||||
return dsteps, nil
|
||||
}
|
||||
func (f *Filesystem) ReportInfo() *report.FilesystemInfo {
|
||||
return &report.FilesystemInfo{Name: f.Path} // FIXME compat name
|
||||
}
|
||||
|
||||
type Step struct {
|
||||
sender Sender
|
||||
receiver Receiver
|
||||
|
||||
parent *Filesystem
|
||||
from, to *pdu.FilesystemVersion // compat
|
||||
|
||||
byteCounter bytecounter.StreamCopier
|
||||
expectedSize int64 // 0 means no size estimate present / possible
|
||||
}
|
||||
|
||||
func (s *Step) TargetDate() time.Time {
|
||||
return s.to.SnapshotTime() // FIXME compat name
|
||||
}
|
||||
|
||||
func (s *Step) Step(ctx context.Context) error {
|
||||
return s.doReplication(ctx)
|
||||
}
|
||||
|
||||
func (s *Step) ReportInfo() *report.StepInfo {
|
||||
var byteCounter int64
|
||||
if s.byteCounter != nil {
|
||||
byteCounter = s.byteCounter.Count()
|
||||
}
|
||||
return &report.StepInfo{
|
||||
From: s.from.RelName(),
|
||||
To: s.to.RelName(),
|
||||
BytesExpected: s.expectedSize,
|
||||
BytesReplicated: byteCounter,
|
||||
}
|
||||
}
|
||||
|
||||
func NewPlanner(secsPerState *prometheus.HistogramVec, bytesReplicated *prometheus.CounterVec, sender Sender, receiver Receiver) *Planner {
|
||||
return &Planner{
|
||||
sender: sender,
|
||||
receiver: receiver,
|
||||
promSecsPerState: secsPerState,
|
||||
promBytesReplicated: bytesReplicated,
|
||||
}
|
||||
}
|
||||
func resolveConflict(conflict error) (path []*pdu.FilesystemVersion, msg string) {
|
||||
if noCommonAncestor, ok := conflict.(*ConflictNoCommonAncestor); ok {
|
||||
if len(noCommonAncestor.SortedReceiverVersions) == 0 {
|
||||
// TODO this is hard-coded replication policy: most recent snapshot as source
|
||||
var mostRecentSnap *pdu.FilesystemVersion
|
||||
for n := len(noCommonAncestor.SortedSenderVersions) - 1; n >= 0; n-- {
|
||||
if noCommonAncestor.SortedSenderVersions[n].Type == pdu.FilesystemVersion_Snapshot {
|
||||
mostRecentSnap = noCommonAncestor.SortedSenderVersions[n]
|
||||
break
|
||||
}
|
||||
}
|
||||
if mostRecentSnap == nil {
|
||||
return nil, "no snapshots available on sender side"
|
||||
}
|
||||
return []*pdu.FilesystemVersion{mostRecentSnap}, fmt.Sprintf("start replication at most recent snapshot %s", mostRecentSnap.RelName())
|
||||
}
|
||||
}
|
||||
return nil, "no automated way to handle conflict type"
|
||||
}
|
||||
|
||||
func (p *Planner) doPlanning(ctx context.Context) ([]*Filesystem, error) {
|
||||
|
||||
log := getLogger(ctx)
|
||||
|
||||
log.Info("start planning")
|
||||
|
||||
slfssres, err := p.sender.ListFilesystems(ctx, &pdu.ListFilesystemReq{})
|
||||
if err != nil {
|
||||
log.WithError(err).WithField("errType", fmt.Sprintf("%T", err)).Error("error listing sender filesystems")
|
||||
return nil, err
|
||||
}
|
||||
sfss := slfssres.GetFilesystems()
|
||||
// no progress here since we could run in a live-lock on connectivity issues
|
||||
|
||||
rlfssres, err := p.receiver.ListFilesystems(ctx, &pdu.ListFilesystemReq{})
|
||||
if err != nil {
|
||||
log.WithError(err).WithField("errType", fmt.Sprintf("%T", err)).Error("error listing receiver filesystems")
|
||||
return nil, err
|
||||
}
|
||||
rfss := rlfssres.GetFilesystems()
|
||||
|
||||
q := make([]*Filesystem, 0, len(sfss))
|
||||
for _, fs := range sfss {
|
||||
|
||||
receiverFSExists := false
|
||||
for _, rfs := range rfss {
|
||||
if rfs.Path == fs.Path {
|
||||
receiverFSExists = true
|
||||
}
|
||||
}
|
||||
|
||||
ctr := p.promBytesReplicated.WithLabelValues(fs.Path)
|
||||
|
||||
q = append(q, &Filesystem{
|
||||
sender: p.sender,
|
||||
receiver: p.receiver,
|
||||
Path: fs.Path,
|
||||
receiverFSExists: receiverFSExists,
|
||||
promBytesReplicated: ctr,
|
||||
})
|
||||
}
|
||||
|
||||
return q, nil
|
||||
}
|
||||
|
||||
func (fs *Filesystem) doPlanning(ctx context.Context) ([]*Step, error) {
|
||||
|
||||
log := getLogger(ctx).WithField("filesystem", fs.Path)
|
||||
|
||||
log.Debug("assessing filesystem")
|
||||
|
||||
sfsvsres, err := fs.sender.ListFilesystemVersions(ctx, &pdu.ListFilesystemVersionsReq{Filesystem: fs.Path})
|
||||
if err != nil {
|
||||
log.WithError(err).Error("cannot get remote filesystem versions")
|
||||
return nil, err
|
||||
}
|
||||
sfsvs := sfsvsres.GetVersions()
|
||||
|
||||
if len(sfsvs) < 1 {
|
||||
err := errors.New("sender does not have any versions")
|
||||
log.Error(err.Error())
|
||||
return nil, err
|
||||
}
|
||||
|
||||
var rfsvs []*pdu.FilesystemVersion
|
||||
if fs.receiverFSExists {
|
||||
rfsvsres, err := fs.receiver.ListFilesystemVersions(ctx, &pdu.ListFilesystemVersionsReq{Filesystem: fs.Path})
|
||||
if err != nil {
|
||||
log.WithError(err).Error("receiver error")
|
||||
return nil, err
|
||||
}
|
||||
rfsvs = rfsvsres.GetVersions()
|
||||
} else {
|
||||
rfsvs = []*pdu.FilesystemVersion{}
|
||||
}
|
||||
|
||||
path, conflict := IncrementalPath(rfsvs, sfsvs)
|
||||
if conflict != nil {
|
||||
var msg string
|
||||
path, msg = resolveConflict(conflict) // no shadowing allowed!
|
||||
if path != nil {
|
||||
log.WithField("conflict", conflict).Info("conflict")
|
||||
log.WithField("resolution", msg).Info("automatically resolved")
|
||||
} else {
|
||||
log.WithField("conflict", conflict).Error("conflict")
|
||||
log.WithField("problem", msg).Error("cannot resolve conflict")
|
||||
}
|
||||
}
|
||||
if path == nil {
|
||||
return nil, conflict
|
||||
}
|
||||
|
||||
steps := make([]*Step, 0, len(path))
|
||||
// FIXME unify struct declarations => initializer?
|
||||
if len(path) == 1 {
|
||||
steps = append(steps, &Step{
|
||||
parent: fs,
|
||||
sender: fs.sender,
|
||||
receiver: fs.receiver,
|
||||
from: nil,
|
||||
to: path[0],
|
||||
})
|
||||
} else {
|
||||
for i := 0; i < len(path)-1; i++ {
|
||||
steps = append(steps, &Step{
|
||||
parent: fs,
|
||||
sender: fs.sender,
|
||||
receiver: fs.receiver,
|
||||
from: path[i],
|
||||
to: path[i+1],
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
log.Debug("compute send size estimate")
|
||||
errs := make(chan error, len(steps))
|
||||
var wg sync.WaitGroup
|
||||
fanOutCtx, fanOutCancel := context.WithCancel(ctx)
|
||||
defer fanOutCancel()
|
||||
for _, step := range steps {
|
||||
wg.Add(1)
|
||||
go func(step *Step) {
|
||||
defer wg.Done()
|
||||
err := step.updateSizeEstimate(fanOutCtx)
|
||||
if err != nil {
|
||||
log.WithError(err).WithField("step", step).Error("error computing size estimate")
|
||||
fanOutCancel()
|
||||
}
|
||||
errs <- err
|
||||
}(step)
|
||||
}
|
||||
wg.Wait()
|
||||
close(errs)
|
||||
var significantErr error = nil
|
||||
for err := range errs {
|
||||
if err != nil {
|
||||
if significantErr == nil || significantErr == context.Canceled {
|
||||
significantErr = err
|
||||
}
|
||||
}
|
||||
}
|
||||
if significantErr != nil {
|
||||
return nil, significantErr
|
||||
}
|
||||
|
||||
log.Debug("filesystem planning finished")
|
||||
return steps, nil
|
||||
}
|
||||
|
||||
// type FilesystemsReplicationFailedError struct {
|
||||
// FilesystemsWithError []*fsrep.Replication
|
||||
// }
|
||||
|
||||
// func (e FilesystemsReplicationFailedError) Error() string {
|
||||
// allSame := true
|
||||
// lastErr := e.FilesystemsWithError[0].Err().Error()
|
||||
// for _, fs := range e.FilesystemsWithError {
|
||||
// fsErr := fs.Err().Error()
|
||||
// allSame = allSame && lastErr == fsErr
|
||||
// }
|
||||
|
||||
// fsstr := "multiple filesystems"
|
||||
// if len(e.FilesystemsWithError) == 1 {
|
||||
// fsstr = fmt.Sprintf("filesystem %s", e.FilesystemsWithError[0].FS())
|
||||
// }
|
||||
// errorStr := lastErr
|
||||
// if !allSame {
|
||||
// errorStr = "multiple different errors"
|
||||
// }
|
||||
// return fmt.Sprintf("%s could not be replicated: %s", fsstr, errorStr)
|
||||
// }
|
||||
|
||||
func (s *Step) updateSizeEstimate(ctx context.Context) error {
|
||||
|
||||
log := getLogger(ctx)
|
||||
|
||||
sr := s.buildSendRequest(true)
|
||||
|
||||
log.Debug("initiate dry run send request")
|
||||
sres, _, err := s.sender.Send(ctx, sr)
|
||||
if err != nil {
|
||||
log.WithError(err).Error("dry run send request failed")
|
||||
return err
|
||||
}
|
||||
s.expectedSize = sres.ExpectedSize
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *Step) buildSendRequest(dryRun bool) (sr *pdu.SendReq) {
|
||||
fs := s.parent.Path
|
||||
if s.from == nil {
|
||||
sr = &pdu.SendReq{
|
||||
Filesystem: fs,
|
||||
To: s.to.RelName(),
|
||||
DryRun: dryRun,
|
||||
}
|
||||
} else {
|
||||
sr = &pdu.SendReq{
|
||||
Filesystem: fs,
|
||||
From: s.from.RelName(),
|
||||
To: s.to.RelName(),
|
||||
DryRun: dryRun,
|
||||
}
|
||||
}
|
||||
return sr
|
||||
}
|
||||
|
||||
func (s *Step) doReplication(ctx context.Context) error {
|
||||
|
||||
fs := s.parent.Path
|
||||
|
||||
log := getLogger(ctx)
|
||||
sr := s.buildSendRequest(false)
|
||||
|
||||
log.Debug("initiate send request")
|
||||
sres, sstreamCopier, err := s.sender.Send(ctx, sr)
|
||||
if err != nil {
|
||||
log.WithError(err).Error("send request failed")
|
||||
return err
|
||||
}
|
||||
if sstreamCopier == nil {
|
||||
err := errors.New("send request did not return a stream, broken endpoint implementation")
|
||||
return err
|
||||
}
|
||||
defer sstreamCopier.Close()
|
||||
|
||||
// Install a byte counter to track progress + for status report
|
||||
s.byteCounter = bytecounter.NewStreamCopier(sstreamCopier)
|
||||
defer func() {
|
||||
s.parent.promBytesReplicated.Add(float64(s.byteCounter.Count()))
|
||||
}()
|
||||
|
||||
rr := &pdu.ReceiveReq{
|
||||
Filesystem: fs,
|
||||
ClearResumeToken: !sres.UsedResumeToken,
|
||||
}
|
||||
log.Debug("initiate receive request")
|
||||
_, err = s.receiver.Receive(ctx, rr, s.byteCounter)
|
||||
if err != nil {
|
||||
log.
|
||||
WithError(err).
|
||||
WithField("errType", fmt.Sprintf("%T", err)).
|
||||
Error("receive request failed (might also be error on sender)")
|
||||
// This failure could be due to
|
||||
// - an unexpected exit of ZFS on the sending side
|
||||
// - an unexpected exit of ZFS on the receiving side
|
||||
// - a connectivity issue
|
||||
return err
|
||||
}
|
||||
log.Debug("receive finished")
|
||||
|
||||
log.Debug("advance replication cursor")
|
||||
req := &pdu.ReplicationCursorReq{
|
||||
Filesystem: fs,
|
||||
Op: &pdu.ReplicationCursorReq_Set{
|
||||
Set: &pdu.ReplicationCursorReq_SetOp{
|
||||
Snapshot: s.to.GetName(),
|
||||
},
|
||||
},
|
||||
}
|
||||
_, err = s.sender.ReplicationCursor(ctx, req)
|
||||
if err != nil {
|
||||
log.WithError(err).Error("error advancing replication cursor")
|
||||
// If this fails and replication planning restarts, the diff algorithm will find
|
||||
// that cursor out of place. This is not a problem because then, it would just use another FS
|
||||
// However, we FIXME have no means to just update the cursor in a
|
||||
// second replication attempt right after this one where we don't have new snaps yet
|
||||
return err
|
||||
}
|
||||
|
||||
return err
|
||||
}
|
||||
|
||||
func (s *Step) String() string {
|
||||
if s.from == nil { // FIXME: ZFS semantics are that to is nil on non-incremental send
|
||||
return fmt.Sprintf("%s%s (full)", s.parent.Path, s.to.RelName())
|
||||
} else {
|
||||
return fmt.Sprintf("%s(%s => %s)", s.parent.Path, s.from.RelName(), s.to.RelName())
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,28 @@
|
||||
package logic
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
"github.com/zrepl/zrepl/logger"
|
||||
)
|
||||
|
||||
type contextKey int
|
||||
|
||||
const (
|
||||
contextKeyLog contextKey = iota
|
||||
)
|
||||
|
||||
type Logger = logger.Logger
|
||||
|
||||
func WithLogger(ctx context.Context, l Logger) context.Context {
|
||||
ctx = context.WithValue(ctx, contextKeyLog, l)
|
||||
return ctx
|
||||
}
|
||||
|
||||
func getLogger(ctx context.Context) Logger {
|
||||
l, ok := ctx.Value(contextKeyLog).(Logger)
|
||||
if !ok {
|
||||
l = logger.NewNullLogger()
|
||||
}
|
||||
return l
|
||||
}
|
||||
Reference in New Issue
Block a user