From c87759affef880d18155a4776dd90a1183912917 Mon Sep 17 00:00:00 2001 From: Christian Schwarz Date: Mon, 11 Mar 2019 13:46:36 +0100 Subject: [PATCH] replication/driver: automatic retries on connectivity-related errors --- Gopkg.lock | 11 + Gopkg.toml | 4 + client/status.go | 37 +- daemon/logging/build_logging.go | 2 + endpoint/endpoint.go | 31 ++ replication/driver/errorclass_enumer.go | 50 +++ replication/driver/replication_driver.go | 367 +++++++++++++++++- .../driver/replication_driver_debug.go | 29 ++ .../driver/replication_driver_logging.go | 25 ++ replication/driver/replication_driver_test.go | 15 +- replication/logic/pdu/pdu.pb.go | 251 ++++++++---- replication/logic/pdu/pdu.proto | 10 + replication/logic/replication_logic.go | 67 +++- replication/report/replication_report.go | 6 +- rpc/dataconn/dataconn_client.go | 20 + rpc/dataconn/dataconn_server.go | 9 + rpc/dataconn/dataconn_shared.go | 1 + rpc/dataconn/microbenchmark/microbenchmark.go | 6 + rpc/rpc_client.go | 66 ++++ util/envconst/envconst.go | 16 + 20 files changed, 933 insertions(+), 90 deletions(-) create mode 100644 replication/driver/errorclass_enumer.go create mode 100644 replication/driver/replication_driver_debug.go create mode 100644 replication/driver/replication_driver_logging.go diff --git a/Gopkg.lock b/Gopkg.lock index 9733251..7df6bf1 100644 --- a/Gopkg.lock +++ b/Gopkg.lock @@ -89,6 +89,14 @@ revision = "aa810b61a9c79d51363740d207bb46cf8e620ed5" version = "v1.2.0" +[[projects]] + digest = "1:ad92aa49f34cbc3546063c7eb2cabb55ee2278b72842eda80e2a20a8a06a8d73" + name = "github.com/google/uuid" + packages = ["."] + pruneopts = "" + revision = "0cd6bf5da1e1c83f8b45653022c74f71af0538a4" + version = "v1.1.1" + [[projects]] branch = "master" digest = "1:cb09475f771b9167fb9333629f5d6a7161572602ea040f1094602b0dc8709878" @@ -438,6 +446,7 @@ "github.com/go-logfmt/logfmt", "github.com/golang/protobuf/proto", "github.com/golang/protobuf/protoc-gen-go", + "github.com/google/uuid", "github.com/jinzhu/copier", "github.com/kr/pretty", "github.com/mattn/go-isatty", @@ -458,9 +467,11 @@ "golang.org/x/sys/unix", "golang.org/x/tools/cmd/stringer", "google.golang.org/grpc", + "google.golang.org/grpc/codes", "google.golang.org/grpc/credentials", "google.golang.org/grpc/keepalive", "google.golang.org/grpc/peer", + "google.golang.org/grpc/status", ] solver-name = "gps-cdcl" solver-version = 1 diff --git a/Gopkg.toml b/Gopkg.toml index 01e2aae..55c0a9b 100644 --- a/Gopkg.toml +++ b/Gopkg.toml @@ -59,3 +59,7 @@ required = [ [[constraint]] name = "google.golang.org/grpc" version = "1" + +[[constraint]] + version = "1.1.0" + name = "github.com/google/uuid" diff --git a/client/status.go b/client/status.go index e97db3a..ab06dca 100644 --- a/client/status.go +++ b/client/status.go @@ -347,10 +347,45 @@ func (t *tui) renderReplicationReport(rep *report.Report, history *bytesProgress return } + if rep.WaitReconnectError != nil { + t.printfDrawIndentedAndWrappedIfMultiline("Connectivity: %s", rep.WaitReconnectError) + t.newline() + } + if !rep.WaitReconnectSince.IsZero() { + delta := rep.WaitReconnectUntil.Sub(time.Now()).Round(time.Second) + if rep.WaitReconnectUntil.IsZero() || delta > 0 { + var until string + if rep.WaitReconnectUntil.IsZero() { + until = "waiting indefinitely" + } else { + until = fmt.Sprintf("hard fail in %s @ %s", delta, rep.WaitReconnectUntil) + } + t.printfDrawIndentedAndWrappedIfMultiline("Connectivity: reconnecting with exponential backoff (since %s) (%s)", + rep.WaitReconnectSince, until) + } else { + t.printfDrawIndentedAndWrappedIfMultiline("Connectivity: reconnects reached hard-fail timeout @ %s", rep.WaitReconnectUntil) + } + t.newline() + } + // TODO visualize more than the latest attempt by folding all attempts into one if len(rep.Attempts) == 0 { t.printf("no attempts made yet") return + } else { + t.printf("Attempt #%d", len(rep.Attempts)) + if len(rep.Attempts) > 1 { + t.printf(". Previous attempts failed with the follwing statuses:") + t.newline() + t.addIndent(1) + for i, a := range rep.Attempts[:len(rep.Attempts)-1] { + t.printfDrawIndentedAndWrappedIfMultiline("#%d: %s (failed at %s) (ran %s)", i + 1, a.State, a.FinishAt, a.FinishAt.Sub(a.StartAt)) + t.newline() + } + t.addIndent(-1) + } else { + t.newline() + } } latest := rep.Attempts[len(rep.Attempts)-1] @@ -369,8 +404,6 @@ func (t *tui) renderReplicationReport(rep *report.Report, history *bytesProgress t.newline() } - // TODO report sleep time between retry attempts once that is implemented - if latest.State != report.AttemptPlanning && latest.State != report.AttemptPlanningError { // Draw global progress bar // Progress: [---------------] diff --git a/daemon/logging/build_logging.go b/daemon/logging/build_logging.go index d54cdc1..8fb3718 100644 --- a/daemon/logging/build_logging.go +++ b/daemon/logging/build_logging.go @@ -14,6 +14,7 @@ import ( "github.com/zrepl/zrepl/daemon/snapper" "github.com/zrepl/zrepl/endpoint" "github.com/zrepl/zrepl/logger" + "github.com/zrepl/zrepl/replication/driver" "github.com/zrepl/zrepl/rpc" "github.com/zrepl/zrepl/rpc/transportmux" "github.com/zrepl/zrepl/tlsconf" @@ -77,6 +78,7 @@ const ( ) func WithSubsystemLoggers(ctx context.Context, log logger.Logger) context.Context { + ctx = driver.WithLogger(ctx, log.WithField(SubsysField, SubsysReplication)) ctx = endpoint.WithLogger(ctx, log.WithField(SubsysField, SubsyEndpoint)) ctx = pruner.WithLogger(ctx, log.WithField(SubsysField, SubsysPruning)) ctx = snapper.WithLogger(ctx, log.WithField(SubsysField, SubsysSnapshot)) diff --git a/endpoint/endpoint.go b/endpoint/endpoint.go index 325234e..3670069 100644 --- a/endpoint/endpoint.go +++ b/endpoint/endpoint.go @@ -107,6 +107,21 @@ func (p *Sender) DestroySnapshots(ctx context.Context, req *pdu.DestroySnapshots return doDestroySnapshots(ctx, dp, req.Snapshots) } +func (p *Sender) Ping(ctx context.Context, req *pdu.PingReq) (*pdu.PingRes, error) { + res := pdu.PingRes{ + Echo: req.GetMessage(), + } + return &res, nil +} + +func (p *Sender) PingDataconn(ctx context.Context, req *pdu.PingReq) (*pdu.PingRes, error) { + return p.Ping(ctx, req) +} + +func (p *Sender) WaitForConnectivity(ctx context.Context) (error) { + return nil +} + func (p *Sender) ReplicationCursor(ctx context.Context, req *pdu.ReplicationCursorReq) (*pdu.ReplicationCursorRes, error) { dp, err := p.filterCheckFS(req.Filesystem) if err != nil { @@ -278,6 +293,21 @@ func (s *Receiver) ListFilesystemVersions(ctx context.Context, req *pdu.ListFile return &pdu.ListFilesystemVersionsRes{Versions: rfsvs}, nil } +func (s *Receiver) Ping(ctx context.Context, req *pdu.PingReq) (*pdu.PingRes, error) { + res := pdu.PingRes{ + Echo: req.GetMessage(), + } + return &res, nil +} + +func (s *Receiver) PingDataconn(ctx context.Context, req *pdu.PingReq) (*pdu.PingRes, error) { + return s.Ping(ctx, req) +} + +func (s *Receiver) WaitForConnectivity(ctx context.Context) (error) { + return nil +} + func (s *Receiver) ReplicationCursor(context.Context, *pdu.ReplicationCursorReq) (*pdu.ReplicationCursorRes, error) { return nil, fmt.Errorf("ReplicationCursor not implemented for Receiver") } @@ -286,6 +316,7 @@ func (s *Receiver) Send(ctx context.Context, req *pdu.SendReq) (*pdu.SendRes, zf return nil, nil, fmt.Errorf("receiver does not implement Send()") } + func (s *Receiver) Receive(ctx context.Context, req *pdu.ReceiveReq, receive zfs.StreamCopier) (*pdu.ReceiveRes, error) { getLogger(ctx).Debug("incoming Receive") defer receive.Close() diff --git a/replication/driver/errorclass_enumer.go b/replication/driver/errorclass_enumer.go new file mode 100644 index 0000000..0a56c0e --- /dev/null +++ b/replication/driver/errorclass_enumer.go @@ -0,0 +1,50 @@ +// Code generated by "enumer -type=errorClass"; DO NOT EDIT. + +package driver + +import ( + "fmt" +) + +const _errorClassName = "errorClassUnknownerrorClassPermanenterrorClassTemporaryConnectivityRelated" + +var _errorClassIndex = [...]uint8{0, 17, 36, 74} + +func (i errorClass) String() string { + if i < 0 || i >= errorClass(len(_errorClassIndex)-1) { + return fmt.Sprintf("errorClass(%d)", i) + } + return _errorClassName[_errorClassIndex[i]:_errorClassIndex[i+1]] +} + +var _errorClassValues = []errorClass{0, 1, 2} + +var _errorClassNameToValueMap = map[string]errorClass{ + _errorClassName[0:17]: 0, + _errorClassName[17:36]: 1, + _errorClassName[36:74]: 2, +} + +// errorClassString retrieves an enum value from the enum constants string name. +// Throws an error if the param is not part of the enum. +func errorClassString(s string) (errorClass, error) { + if val, ok := _errorClassNameToValueMap[s]; ok { + return val, nil + } + return 0, fmt.Errorf("%s does not belong to errorClass values", s) +} + +// errorClassValues returns all values of the enum +func errorClassValues() []errorClass { + return _errorClassValues +} + +// IsAerrorClass returns "true" if the value is listed in the enum definition. "false" otherwise +func (i errorClass) IsAerrorClass() bool { + for _, v := range _errorClassValues { + if i == v { + return true + } + } + return false +} diff --git a/replication/driver/replication_driver.go b/replication/driver/replication_driver.go index 4d1a101..6a5a9ea 100644 --- a/replication/driver/replication_driver.go +++ b/replication/driver/replication_driver.go @@ -2,18 +2,69 @@ package driver import ( "context" + "errors" + "fmt" + "net" + "sort" + "strings" "sync" "time" "github.com/zrepl/zrepl/replication/report" "github.com/zrepl/zrepl/util/chainlock" + "github.com/zrepl/zrepl/util/envconst" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" ) +type interval struct { + begin time.Time + end time.Time +} + +func (w *interval) SetZero() { + w.begin = time.Time{} + w.end = time.Time{} +} + +// Duration of 0 means indefinite length +func (w *interval) Set(begin time.Time, duration time.Duration) { + if begin.IsZero() { + panic("zero begin time now allowed") + } + w.begin = begin + w.end = begin.Add(duration) +} + +// Returns the End of the interval if it has a defined length. +// For indefinite lengths, returns the zero value. +func (w *interval) End() time.Time { + return w.end +} + +// Return a context with a deadline at the interval's end. +// If the interval has indefinite length (duration 0 on Set), return ctx as is. +// The returned context.CancelFunc can be called either way. +func (w *interval) ContextWithDeadlineAtEnd(ctx context.Context) (context.Context, context.CancelFunc) { + if w.begin.IsZero() { + panic("must call Set before ContextWIthDeadlineAtEnd") + } + if w.end.IsZero() { + // indefinite length, just return context as is + return ctx, func() {} + } else { + return context.WithDeadline(ctx, w.end) + } +} + type run struct { l *chainlock.L startedAt, finishedAt time.Time + waitReconnect interval + waitReconnectError *timedError + // the attempts attempted so far: // All but the last in this slice must have finished with some errors. // The last attempt may not be finished and may not have errors. @@ -22,6 +73,7 @@ type run struct { type Planner interface { Plan(context.Context) ([]FS, error) + WaitForConnectivity(context.Context) error } // an attempt represents a single planning & execution of fs replications @@ -31,6 +83,7 @@ type attempt struct { l *chainlock.L startedAt, finishedAt time.Time + // after Planner.Plan was called, planErr and fss are mutually exclusive with regards to nil-ness // if both are nil, it must be assumed that Planner.Plan is active planErr *timedError @@ -60,11 +113,28 @@ func (e *timedError) IntoReportError() *report.TimedError { } type FS interface { + // Returns true if this FS and fs refer to the same filesystem returned + // by Planner.Plan in a previous attempt. + EqualToPreviousAttempt(fs FS) bool + // The returned steps are assumed to be dependent on exactly + // their direct predecessors in the returned list. PlanFS(context.Context) ([]Step, error) ReportInfo() *report.FilesystemInfo } type Step interface { + // Returns true iff the target snapshot is the same for this Step and other. + // We do not use TargetDate to avoid problems with wrong system time on + // snapshot creation. + // + // Implementations can assume that `other` is a step of the same filesystem, + // although maybe from a previous attempt. + // (`same` as defined by FS.EqualToPreviousAttempt) + // + // Note that TargetEquals should return true in a situation with one + // originally sent snapshot and a subsequent attempt's step that uses + // resumable send & recv. + TargetEquals(other Step) bool TargetDate() time.Time Step(context.Context) error ReportInfo() *report.StepInfo @@ -100,7 +170,11 @@ type step struct { type ReportFunc func() *report.Report type WaitFunc func(block bool) (done bool) +var maxAttempts = envconst.Int64("ZREPL_REPLICATION_MAX_ATTEMPTS", 3) +var reconnectHardFailTimeout = envconst.Duration("ZREPL_REPLICATION_RECONNECT_HARD_FAIL_TIMEOUT", 10*time.Minute) + func Do(ctx context.Context, planner Planner) (ReportFunc, WaitFunc) { + log := getLog(ctx) l := chainlock.New() run := &run{ l: l, @@ -111,15 +185,79 @@ func Do(ctx context.Context, planner Planner) (ReportFunc, WaitFunc) { go func() { defer close(done) - run.l.Lock() - a1 := &attempt{ - l: l, - startedAt: time.Now(), - planner: planner, + defer run.l.Lock().Unlock() + log.Debug("begin run") + defer log.Debug("run ended") + var prev *attempt + mainLog := log + for ano := 0; ano < int(maxAttempts) || maxAttempts == 0; ano++ { + log := mainLog.WithField("attempt_number", ano) + log.Debug("start attempt") + + run.waitReconnect.SetZero() + run.waitReconnectError = nil + + // do current attempt + cur := &attempt{ + l: l, + startedAt: time.Now(), + planner: planner, + } + run.attempts = append(run.attempts, cur) + run.l.DropWhile(func() { + cur.do(ctx, prev) + }) + prev = cur + if ctx.Err() != nil { + log.WithError(ctx.Err()).Info("context error") + return + } + + // error classification, bail out if done / permanent error + rep := cur.report() + log.WithField("attempt_state", rep.State).Debug("attempt state") + errRep := cur.errorReport() + + if rep.State == report.AttemptDone { + log.Debug("attempt completed successfully") + break + } + + mostRecentErr, mostRecentErrClass := errRep.MostRecent() + log.WithField("most_recent_err", mostRecentErr).WithField("most_recent_err_class", mostRecentErrClass).Debug("most recent error used for re-connect decision") + if mostRecentErr == nil { + // inconsistent reporting, let's bail out + log.Warn("attempt does not report done but error report does not report errors, aborting run") + break + } + log.WithError(mostRecentErr.Err).Error("most recent error in this attempt") + shouldReconnect := mostRecentErrClass == errorClassTemporaryConnectivityRelated + log.WithField("reconnect_decision", shouldReconnect).Debug("reconnect decision made") + if shouldReconnect { + run.waitReconnect.Set(time.Now(), reconnectHardFailTimeout) + log.WithField("deadline", run.waitReconnect.End()).Error("temporary connectivity-related error identified, start waiting for reconnect") + var connectErr error + var connectErrTime time.Time + run.l.DropWhile(func() { + ctx, cancel := run.waitReconnect.ContextWithDeadlineAtEnd(ctx) + defer cancel() + connectErr = planner.WaitForConnectivity(ctx) + connectErrTime = time.Now() + }) + if connectErr == nil { + log.Error("reconnect successful") // same level as 'begin with reconnect' message above + continue + } else { + run.waitReconnectError = newTimedError(connectErr, connectErrTime) + log.WithError(connectErr).Error("reconnecting failed, aborting run") + break + } + } else { + log.Error("most recent error cannot be solved by reconnecting, aborting run") + return + } + } - run.attempts = append(run.attempts, a1) - run.l.Unlock() - a1.do(ctx) }() @@ -141,7 +279,7 @@ func Do(ctx context.Context, planner Planner) (ReportFunc, WaitFunc) { return report, wait } -func (a *attempt) do(ctx context.Context) { +func (a *attempt) do(ctx context.Context, prev *attempt) { pfss, err := a.planner.Plan(ctx) errTime := time.Now() defer a.l.Lock().Unlock() @@ -160,14 +298,69 @@ func (a *attempt) do(ctx context.Context) { a.fss = append(a.fss, fs) } + prevs := make(map[*fs]*fs) + { + prevFSs := make(map[*fs][]*fs, len(pfss)) + if prev != nil { + debug("previous attempt has %d fss", len(a.fss)) + for _, fs := range a.fss { + for _, prevFS := range prev.fss { + if fs.fs.EqualToPreviousAttempt(prevFS.fs) { + l := prevFSs[fs] + l = append(l, prevFS) + prevFSs[fs] = l + } + } + } + } + type inconsistency struct { + cur *fs + prevs []*fs + } + var inconsistencies []inconsistency + for cur, fss := range prevFSs { + if len(fss) > 1 { + inconsistencies = append(inconsistencies, inconsistency{cur, fss}) + } + } + sort.SliceStable(inconsistencies, func(i, j int) bool { + return inconsistencies[i].cur.fs.ReportInfo().Name < inconsistencies[j].cur.fs.ReportInfo().Name + }) + if len(inconsistencies) > 0 { + var msg strings.Builder + msg.WriteString("cannot determine filesystem correspondences between different attempts:\n") + var inconsistencyLines []string + for _, i := range inconsistencies { + var prevNames []string + for _, prev := range i.prevs { + prevNames = append(prevNames, prev.fs.ReportInfo().Name) + } + l := fmt.Sprintf(" %s => %v", i.cur.fs.ReportInfo().Name, prevNames) + inconsistencyLines = append(inconsistencyLines, l) + } + fmt.Fprintf(&msg, strings.Join(inconsistencyLines, "\n")) + now := time.Now() + a.planErr = newTimedError(errors.New(msg.String()), now) + a.fss = nil + a.finishedAt = now + return + } + for cur, fss := range prevFSs { + if len(fss) > 0 { + prevs[cur] = fss[0] + } + } + } + // invariant: prevs contains an entry for each unambigious correspondence + stepQueue := newStepQueue() - defer stepQueue.Start(1)() + defer stepQueue.Start(1)() // TODO parallel replication var fssesDone sync.WaitGroup for _, f := range a.fss { fssesDone.Add(1) go func(f *fs) { defer fssesDone.Done() - f.do(ctx, stepQueue) + f.do(ctx, stepQueue, prevs[f]) }(f) } a.l.DropWhile(func() { @@ -176,10 +369,11 @@ func (a *attempt) do(ctx context.Context) { a.finishedAt = time.Now() } -func (fs *fs) do(ctx context.Context, pq *stepQueue) { +func (fs *fs) do(ctx context.Context, pq *stepQueue, prev *fs) { psteps, err := fs.fs.PlanFS(ctx) errTime := time.Now() defer fs.l.Lock().Unlock() + debug := debugPrefix("fs=%s", fs.fs.ReportInfo().Name) fs.planning.done = true if err != nil { fs.planning.err = newTimedError(err, errTime) @@ -192,6 +386,60 @@ func (fs *fs) do(ctx context.Context, pq *stepQueue) { } fs.planned.steps = append(fs.planned.steps, step) } + debug("iniital len(fs.planned.steps) = %d", len(fs.planned.steps)) + + // for not-first attempts, only allow fs.planned.steps + // up to including the originally planned target snapshot + if prev != nil && prev.planning.done && prev.planning.err == nil { + prevUncompleted := prev.planned.steps[prev.planned.step:] + if len(prevUncompleted) == 0 { + debug("prevUncompleted is empty") + return + } + if len(fs.planned.steps) == 0 { + debug("fs.planned.steps is empty") + return + } + prevFailed := prevUncompleted[0] + curFirst := fs.planned.steps[0] + // we assume that PlanFS retries prevFailed (using curFirst) + if !prevFailed.step.TargetEquals(curFirst.step) { + debug("Targets don't match") + // Two options: + // A: planning algorithm is broken + // B: manual user intervention inbetween + // Neither way will we make progress, so let's error out + stepFmt := func(step *step) string { + r := step.report() + s := r.Info + if r.IsIncremental() { + return fmt.Sprintf("%s=>%s", s.From, s.To) + } else { + return fmt.Sprintf("full=>%s", s.To) + } + } + msg := fmt.Sprintf("last attempt's uncompleted step %s does not correspond to this attempt's first planned step %s", + stepFmt(prevFailed), stepFmt(curFirst)) + fs.planned.stepErr = newTimedError(errors.New(msg), time.Now()) + return + } + // only allow until step targets diverge + min := len(prevUncompleted) + if min > len(fs.planned.steps) { + min = len(fs.planned.steps) + } + diverge := 0 + for ; diverge < min; diverge++ { + debug("diverge compare iteration %d", diverge) + if !fs.planned.steps[diverge].step.TargetEquals(prevUncompleted[diverge].step) { + break + } + } + debug("diverge is %d", diverge) + fs.planned.steps = fs.planned.steps[0:diverge] + } + debug("post-prev-merge len(fs.planned.steps) = %d", len(fs.planned.steps)) + for i, s := range fs.planned.steps { var ( err error @@ -215,9 +463,12 @@ func (fs *fs) do(ctx context.Context, pq *stepQueue) { // caller must hold lock l func (r *run) report() *report.Report { report := &report.Report{ - Attempts: make([]*report.AttemptReport, len(r.attempts)), - StartAt: r.startedAt, - FinishAt: r.finishedAt, + Attempts: make([]*report.AttemptReport, len(r.attempts)), + StartAt: r.startedAt, + FinishAt: r.finishedAt, + WaitReconnectSince: r.waitReconnect.begin, + WaitReconnectUntil: r.waitReconnect.end, + WaitReconnectError: r.waitReconnectError.IntoReportError(), } for i := range report.Attempts { report.Attempts[i] = r.attempts[i].report() @@ -299,3 +550,89 @@ func (s *step) report() *report.StepReport { } return r } + +type stepErrorReport struct { + err *timedError + step int +} + +//go:generate enumer -type=errorClass +type errorClass int + +const ( + errorClassUnknown errorClass = iota + errorClassPermanent + errorClassTemporaryConnectivityRelated +) + +type errorReport struct { + flattened []*timedError + // sorted DESCending by err time + byClass map[errorClass][]*timedError +} + +// caller must hold lock l +func (a *attempt) errorReport() *errorReport { + r := &errorReport{} + if a.planErr != nil { + r.flattened = append(r.flattened, a.planErr) + } + for _, fs := range a.fss { + if fs.planning.done && fs.planning.err != nil { + r.flattened = append(r.flattened, fs.planning.err) + } else if fs.planning.done && fs.planned.stepErr != nil { + r.flattened = append(r.flattened, fs.planned.stepErr) + } + } + + // build byClass + { + r.byClass = make(map[errorClass][]*timedError) + putClass := func(err *timedError, class errorClass) { + errs := r.byClass[class] + errs = append(errs, err) + r.byClass[class] = errs + } + for _, err := range r.flattened { + if neterr, ok := err.Err.(net.Error); ok && neterr.Temporary() { + putClass(err, errorClassTemporaryConnectivityRelated) + continue + } + if st, ok := status.FromError(err.Err); ok && st.Code() == codes.Unavailable { + // technically, codes.Unavailable could be returned by the gRPC endpoint, indicating overload, etc. + // for now, let's assume it only happens for connectivity issues, as specified in + // https://grpc.io/grpc/core/md_doc_statuscodes.html + putClass(err, errorClassTemporaryConnectivityRelated) + continue + } + putClass(err, errorClassPermanent) + } + for _, errs := range r.byClass { + sort.Slice(errs, func(i, j int) bool { + return errs[i].Time.After(errs[j].Time) // sort descendingly + }) + } + } + + return r +} + +func (r *errorReport) AnyError() *timedError { + for _, err := range r.flattened { + if err != nil { + return err + } + } + return nil +} + +func (r *errorReport) MostRecent() (err *timedError, errClass errorClass) { + for class, errs := range r.byClass { + // errs are sorted descendingly during construction + if len(errs) > 0 && (err == nil || errs[0].Time.After(err.Time)) { + err = errs[0] + errClass = class + } + } + return +} diff --git a/replication/driver/replication_driver_debug.go b/replication/driver/replication_driver_debug.go new file mode 100644 index 0000000..23f7ae7 --- /dev/null +++ b/replication/driver/replication_driver_debug.go @@ -0,0 +1,29 @@ +package driver + +import ( + "fmt" + "os" +) + +var debugEnabled bool = false + +func init() { + if os.Getenv("ZREPL_REPLICATION_DRIVER_DEBUG") != "" { + debugEnabled = true + } +} + +func debug(format string, args ...interface{}) { + if debugEnabled { + fmt.Fprintf(os.Stderr, "repl: driver: %s\n", fmt.Sprintf(format, args...)) + } +} + +type debugFunc func(format string, args ...interface{}) + +func debugPrefix(prefixFormat string, prefixFormatArgs ...interface{}) debugFunc { + prefix := fmt.Sprintf(prefixFormat, prefixFormatArgs...) + return func(format string, args ...interface{}) { + debug("%s: %s", prefix, fmt.Sprintf(format, args)) + } +} \ No newline at end of file diff --git a/replication/driver/replication_driver_logging.go b/replication/driver/replication_driver_logging.go new file mode 100644 index 0000000..ccaf7c8 --- /dev/null +++ b/replication/driver/replication_driver_logging.go @@ -0,0 +1,25 @@ +package driver + +import ( + "context" + + "github.com/zrepl/zrepl/logger" +) + +type Logger = logger.Logger + +type contexKey int + +const contexKeyLogger contexKey = iota + 1 + +func getLog(ctx context.Context) Logger { + l, ok := ctx.Value(contexKeyLogger).(Logger) + if !ok { + l = logger.NewNullLogger() + } + return l +} + +func WithLogger(ctx context.Context, log Logger) context.Context { + return context.WithValue(ctx, contexKeyLogger, log) +} diff --git a/replication/driver/replication_driver_test.go b/replication/driver/replication_driver_test.go index 6a5d62c..650434f 100644 --- a/replication/driver/replication_driver_test.go +++ b/replication/driver/replication_driver_test.go @@ -40,12 +40,20 @@ func (p *mockPlanner) Plan(ctx context.Context) ([]FS, error) { return p.fss, nil } +func (p *mockPlanner) WaitForConnectivity(context.Context) error { + return nil +} + type mockFS struct { globalStepCounter *uint32 name string steps []Step } +func (f *mockFS) EqualToPreviousAttempt(other FS) bool { + return f.name == other.(*mockFS).name +} + func (f *mockFS) PlanFS(ctx context.Context) ([]Step, error) { if f.steps != nil { panic("PlanFS used twice") @@ -118,6 +126,10 @@ func (f *mockStep) Step(ctx context.Context) error { return nil } +func (f *mockStep) TargetEquals(s Step) bool { + return f.ident == s.(*mockStep).ident +} + func (f *mockStep) TargetDate() time.Time { return f.targetDate } @@ -172,8 +184,7 @@ func TestReplication(t *testing.T) { diff, err := differ.Compare(prev, this) require.NoError(t, err) df := jsondiffformatter.NewDeltaFormatter() - res, err := df.Format(diff) - res = res + _, err = df.Format(diff) require.NoError(t, err) // uncomment the following line to get json diffs between each captured step // t.Logf("%s", res) diff --git a/replication/logic/pdu/pdu.pb.go b/replication/logic/pdu/pdu.pb.go index 6b7fd86..4424597 100644 --- a/replication/logic/pdu/pdu.pb.go +++ b/replication/logic/pdu/pdu.pb.go @@ -43,7 +43,7 @@ func (x FilesystemVersion_VersionType) String() string { return proto.EnumName(FilesystemVersion_VersionType_name, int32(x)) } func (FilesystemVersion_VersionType) EnumDescriptor() ([]byte, []int) { - return fileDescriptor_pdu_89315d819a6e0938, []int{5, 0} + return fileDescriptor_pdu_759e477f62b58e58, []int{5, 0} } type ListFilesystemReq struct { @@ -56,7 +56,7 @@ func (m *ListFilesystemReq) Reset() { *m = ListFilesystemReq{} } func (m *ListFilesystemReq) String() string { return proto.CompactTextString(m) } func (*ListFilesystemReq) ProtoMessage() {} func (*ListFilesystemReq) Descriptor() ([]byte, []int) { - return fileDescriptor_pdu_89315d819a6e0938, []int{0} + return fileDescriptor_pdu_759e477f62b58e58, []int{0} } func (m *ListFilesystemReq) XXX_Unmarshal(b []byte) error { return xxx_messageInfo_ListFilesystemReq.Unmarshal(m, b) @@ -88,7 +88,7 @@ func (m *ListFilesystemRes) Reset() { *m = ListFilesystemRes{} } func (m *ListFilesystemRes) String() string { return proto.CompactTextString(m) } func (*ListFilesystemRes) ProtoMessage() {} func (*ListFilesystemRes) Descriptor() ([]byte, []int) { - return fileDescriptor_pdu_89315d819a6e0938, []int{1} + return fileDescriptor_pdu_759e477f62b58e58, []int{1} } func (m *ListFilesystemRes) XXX_Unmarshal(b []byte) error { return xxx_messageInfo_ListFilesystemRes.Unmarshal(m, b) @@ -134,7 +134,7 @@ func (m *Filesystem) Reset() { *m = Filesystem{} } func (m *Filesystem) String() string { return proto.CompactTextString(m) } func (*Filesystem) ProtoMessage() {} func (*Filesystem) Descriptor() ([]byte, []int) { - return fileDescriptor_pdu_89315d819a6e0938, []int{2} + return fileDescriptor_pdu_759e477f62b58e58, []int{2} } func (m *Filesystem) XXX_Unmarshal(b []byte) error { return xxx_messageInfo_Filesystem.Unmarshal(m, b) @@ -179,7 +179,7 @@ func (m *ListFilesystemVersionsReq) Reset() { *m = ListFilesystemVersion func (m *ListFilesystemVersionsReq) String() string { return proto.CompactTextString(m) } func (*ListFilesystemVersionsReq) ProtoMessage() {} func (*ListFilesystemVersionsReq) Descriptor() ([]byte, []int) { - return fileDescriptor_pdu_89315d819a6e0938, []int{3} + return fileDescriptor_pdu_759e477f62b58e58, []int{3} } func (m *ListFilesystemVersionsReq) XXX_Unmarshal(b []byte) error { return xxx_messageInfo_ListFilesystemVersionsReq.Unmarshal(m, b) @@ -217,7 +217,7 @@ func (m *ListFilesystemVersionsRes) Reset() { *m = ListFilesystemVersion func (m *ListFilesystemVersionsRes) String() string { return proto.CompactTextString(m) } func (*ListFilesystemVersionsRes) ProtoMessage() {} func (*ListFilesystemVersionsRes) Descriptor() ([]byte, []int) { - return fileDescriptor_pdu_89315d819a6e0938, []int{4} + return fileDescriptor_pdu_759e477f62b58e58, []int{4} } func (m *ListFilesystemVersionsRes) XXX_Unmarshal(b []byte) error { return xxx_messageInfo_ListFilesystemVersionsRes.Unmarshal(m, b) @@ -259,7 +259,7 @@ func (m *FilesystemVersion) Reset() { *m = FilesystemVersion{} } func (m *FilesystemVersion) String() string { return proto.CompactTextString(m) } func (*FilesystemVersion) ProtoMessage() {} func (*FilesystemVersion) Descriptor() ([]byte, []int) { - return fileDescriptor_pdu_89315d819a6e0938, []int{5} + return fileDescriptor_pdu_759e477f62b58e58, []int{5} } func (m *FilesystemVersion) XXX_Unmarshal(b []byte) error { return xxx_messageInfo_FilesystemVersion.Unmarshal(m, b) @@ -339,7 +339,7 @@ func (m *SendReq) Reset() { *m = SendReq{} } func (m *SendReq) String() string { return proto.CompactTextString(m) } func (*SendReq) ProtoMessage() {} func (*SendReq) Descriptor() ([]byte, []int) { - return fileDescriptor_pdu_89315d819a6e0938, []int{6} + return fileDescriptor_pdu_759e477f62b58e58, []int{6} } func (m *SendReq) XXX_Unmarshal(b []byte) error { return xxx_messageInfo_SendReq.Unmarshal(m, b) @@ -420,7 +420,7 @@ func (m *Property) Reset() { *m = Property{} } func (m *Property) String() string { return proto.CompactTextString(m) } func (*Property) ProtoMessage() {} func (*Property) Descriptor() ([]byte, []int) { - return fileDescriptor_pdu_89315d819a6e0938, []int{7} + return fileDescriptor_pdu_759e477f62b58e58, []int{7} } func (m *Property) XXX_Unmarshal(b []byte) error { return xxx_messageInfo_Property.Unmarshal(m, b) @@ -470,7 +470,7 @@ func (m *SendRes) Reset() { *m = SendRes{} } func (m *SendRes) String() string { return proto.CompactTextString(m) } func (*SendRes) ProtoMessage() {} func (*SendRes) Descriptor() ([]byte, []int) { - return fileDescriptor_pdu_89315d819a6e0938, []int{8} + return fileDescriptor_pdu_759e477f62b58e58, []int{8} } func (m *SendRes) XXX_Unmarshal(b []byte) error { return xxx_messageInfo_SendRes.Unmarshal(m, b) @@ -524,7 +524,7 @@ func (m *ReceiveReq) Reset() { *m = ReceiveReq{} } func (m *ReceiveReq) String() string { return proto.CompactTextString(m) } func (*ReceiveReq) ProtoMessage() {} func (*ReceiveReq) Descriptor() ([]byte, []int) { - return fileDescriptor_pdu_89315d819a6e0938, []int{9} + return fileDescriptor_pdu_759e477f62b58e58, []int{9} } func (m *ReceiveReq) XXX_Unmarshal(b []byte) error { return xxx_messageInfo_ReceiveReq.Unmarshal(m, b) @@ -568,7 +568,7 @@ func (m *ReceiveRes) Reset() { *m = ReceiveRes{} } func (m *ReceiveRes) String() string { return proto.CompactTextString(m) } func (*ReceiveRes) ProtoMessage() {} func (*ReceiveRes) Descriptor() ([]byte, []int) { - return fileDescriptor_pdu_89315d819a6e0938, []int{10} + return fileDescriptor_pdu_759e477f62b58e58, []int{10} } func (m *ReceiveRes) XXX_Unmarshal(b []byte) error { return xxx_messageInfo_ReceiveRes.Unmarshal(m, b) @@ -601,7 +601,7 @@ func (m *DestroySnapshotsReq) Reset() { *m = DestroySnapshotsReq{} } func (m *DestroySnapshotsReq) String() string { return proto.CompactTextString(m) } func (*DestroySnapshotsReq) ProtoMessage() {} func (*DestroySnapshotsReq) Descriptor() ([]byte, []int) { - return fileDescriptor_pdu_89315d819a6e0938, []int{11} + return fileDescriptor_pdu_759e477f62b58e58, []int{11} } func (m *DestroySnapshotsReq) XXX_Unmarshal(b []byte) error { return xxx_messageInfo_DestroySnapshotsReq.Unmarshal(m, b) @@ -647,7 +647,7 @@ func (m *DestroySnapshotRes) Reset() { *m = DestroySnapshotRes{} } func (m *DestroySnapshotRes) String() string { return proto.CompactTextString(m) } func (*DestroySnapshotRes) ProtoMessage() {} func (*DestroySnapshotRes) Descriptor() ([]byte, []int) { - return fileDescriptor_pdu_89315d819a6e0938, []int{12} + return fileDescriptor_pdu_759e477f62b58e58, []int{12} } func (m *DestroySnapshotRes) XXX_Unmarshal(b []byte) error { return xxx_messageInfo_DestroySnapshotRes.Unmarshal(m, b) @@ -692,7 +692,7 @@ func (m *DestroySnapshotsRes) Reset() { *m = DestroySnapshotsRes{} } func (m *DestroySnapshotsRes) String() string { return proto.CompactTextString(m) } func (*DestroySnapshotsRes) ProtoMessage() {} func (*DestroySnapshotsRes) Descriptor() ([]byte, []int) { - return fileDescriptor_pdu_89315d819a6e0938, []int{13} + return fileDescriptor_pdu_759e477f62b58e58, []int{13} } func (m *DestroySnapshotsRes) XXX_Unmarshal(b []byte) error { return xxx_messageInfo_DestroySnapshotsRes.Unmarshal(m, b) @@ -734,7 +734,7 @@ func (m *ReplicationCursorReq) Reset() { *m = ReplicationCursorReq{} } func (m *ReplicationCursorReq) String() string { return proto.CompactTextString(m) } func (*ReplicationCursorReq) ProtoMessage() {} func (*ReplicationCursorReq) Descriptor() ([]byte, []int) { - return fileDescriptor_pdu_89315d819a6e0938, []int{14} + return fileDescriptor_pdu_759e477f62b58e58, []int{14} } func (m *ReplicationCursorReq) XXX_Unmarshal(b []byte) error { return xxx_messageInfo_ReplicationCursorReq.Unmarshal(m, b) @@ -882,7 +882,7 @@ func (m *ReplicationCursorReq_GetOp) Reset() { *m = ReplicationCursorReq func (m *ReplicationCursorReq_GetOp) String() string { return proto.CompactTextString(m) } func (*ReplicationCursorReq_GetOp) ProtoMessage() {} func (*ReplicationCursorReq_GetOp) Descriptor() ([]byte, []int) { - return fileDescriptor_pdu_89315d819a6e0938, []int{14, 0} + return fileDescriptor_pdu_759e477f62b58e58, []int{14, 0} } func (m *ReplicationCursorReq_GetOp) XXX_Unmarshal(b []byte) error { return xxx_messageInfo_ReplicationCursorReq_GetOp.Unmarshal(m, b) @@ -913,7 +913,7 @@ func (m *ReplicationCursorReq_SetOp) Reset() { *m = ReplicationCursorReq func (m *ReplicationCursorReq_SetOp) String() string { return proto.CompactTextString(m) } func (*ReplicationCursorReq_SetOp) ProtoMessage() {} func (*ReplicationCursorReq_SetOp) Descriptor() ([]byte, []int) { - return fileDescriptor_pdu_89315d819a6e0938, []int{14, 1} + return fileDescriptor_pdu_759e477f62b58e58, []int{14, 1} } func (m *ReplicationCursorReq_SetOp) XXX_Unmarshal(b []byte) error { return xxx_messageInfo_ReplicationCursorReq_SetOp.Unmarshal(m, b) @@ -954,7 +954,7 @@ func (m *ReplicationCursorRes) Reset() { *m = ReplicationCursorRes{} } func (m *ReplicationCursorRes) String() string { return proto.CompactTextString(m) } func (*ReplicationCursorRes) ProtoMessage() {} func (*ReplicationCursorRes) Descriptor() ([]byte, []int) { - return fileDescriptor_pdu_89315d819a6e0938, []int{15} + return fileDescriptor_pdu_759e477f62b58e58, []int{15} } func (m *ReplicationCursorRes) XXX_Unmarshal(b []byte) error { return xxx_messageInfo_ReplicationCursorRes.Unmarshal(m, b) @@ -1079,6 +1079,83 @@ func _ReplicationCursorRes_OneofSizer(msg proto.Message) (n int) { return n } +type PingReq struct { + Message string `protobuf:"bytes,1,opt,name=Message,proto3" json:"Message,omitempty"` + XXX_NoUnkeyedLiteral struct{} `json:"-"` + XXX_unrecognized []byte `json:"-"` + XXX_sizecache int32 `json:"-"` +} + +func (m *PingReq) Reset() { *m = PingReq{} } +func (m *PingReq) String() string { return proto.CompactTextString(m) } +func (*PingReq) ProtoMessage() {} +func (*PingReq) Descriptor() ([]byte, []int) { + return fileDescriptor_pdu_759e477f62b58e58, []int{16} +} +func (m *PingReq) XXX_Unmarshal(b []byte) error { + return xxx_messageInfo_PingReq.Unmarshal(m, b) +} +func (m *PingReq) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) { + return xxx_messageInfo_PingReq.Marshal(b, m, deterministic) +} +func (dst *PingReq) XXX_Merge(src proto.Message) { + xxx_messageInfo_PingReq.Merge(dst, src) +} +func (m *PingReq) XXX_Size() int { + return xxx_messageInfo_PingReq.Size(m) +} +func (m *PingReq) XXX_DiscardUnknown() { + xxx_messageInfo_PingReq.DiscardUnknown(m) +} + +var xxx_messageInfo_PingReq proto.InternalMessageInfo + +func (m *PingReq) GetMessage() string { + if m != nil { + return m.Message + } + return "" +} + +type PingRes struct { + // Echo must be PingReq.Message + Echo string `protobuf:"bytes,1,opt,name=Echo,proto3" json:"Echo,omitempty"` + XXX_NoUnkeyedLiteral struct{} `json:"-"` + XXX_unrecognized []byte `json:"-"` + XXX_sizecache int32 `json:"-"` +} + +func (m *PingRes) Reset() { *m = PingRes{} } +func (m *PingRes) String() string { return proto.CompactTextString(m) } +func (*PingRes) ProtoMessage() {} +func (*PingRes) Descriptor() ([]byte, []int) { + return fileDescriptor_pdu_759e477f62b58e58, []int{17} +} +func (m *PingRes) XXX_Unmarshal(b []byte) error { + return xxx_messageInfo_PingRes.Unmarshal(m, b) +} +func (m *PingRes) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) { + return xxx_messageInfo_PingRes.Marshal(b, m, deterministic) +} +func (dst *PingRes) XXX_Merge(src proto.Message) { + xxx_messageInfo_PingRes.Merge(dst, src) +} +func (m *PingRes) XXX_Size() int { + return xxx_messageInfo_PingRes.Size(m) +} +func (m *PingRes) XXX_DiscardUnknown() { + xxx_messageInfo_PingRes.DiscardUnknown(m) +} + +var xxx_messageInfo_PingRes proto.InternalMessageInfo + +func (m *PingRes) GetEcho() string { + if m != nil { + return m.Echo + } + return "" +} + func init() { proto.RegisterType((*ListFilesystemReq)(nil), "ListFilesystemReq") proto.RegisterType((*ListFilesystemRes)(nil), "ListFilesystemRes") @@ -1098,6 +1175,8 @@ func init() { proto.RegisterType((*ReplicationCursorReq_GetOp)(nil), "ReplicationCursorReq.GetOp") proto.RegisterType((*ReplicationCursorReq_SetOp)(nil), "ReplicationCursorReq.SetOp") proto.RegisterType((*ReplicationCursorRes)(nil), "ReplicationCursorRes") + proto.RegisterType((*PingReq)(nil), "PingReq") + proto.RegisterType((*PingRes)(nil), "PingRes") proto.RegisterEnum("FilesystemVersion_VersionType", FilesystemVersion_VersionType_name, FilesystemVersion_VersionType_value) } @@ -1113,6 +1192,7 @@ const _ = grpc.SupportPackageIsVersion4 // // For semantics around ctx use and closing/ending streaming RPCs, please refer to https://godoc.org/google.golang.org/grpc#ClientConn.NewStream. type ReplicationClient interface { + Ping(ctx context.Context, in *PingReq, opts ...grpc.CallOption) (*PingRes, error) ListFilesystems(ctx context.Context, in *ListFilesystemReq, opts ...grpc.CallOption) (*ListFilesystemRes, error) ListFilesystemVersions(ctx context.Context, in *ListFilesystemVersionsReq, opts ...grpc.CallOption) (*ListFilesystemVersionsRes, error) DestroySnapshots(ctx context.Context, in *DestroySnapshotsReq, opts ...grpc.CallOption) (*DestroySnapshotsRes, error) @@ -1127,6 +1207,15 @@ func NewReplicationClient(cc *grpc.ClientConn) ReplicationClient { return &replicationClient{cc} } +func (c *replicationClient) Ping(ctx context.Context, in *PingReq, opts ...grpc.CallOption) (*PingRes, error) { + out := new(PingRes) + err := c.cc.Invoke(ctx, "/Replication/Ping", in, out, opts...) + if err != nil { + return nil, err + } + return out, nil +} + func (c *replicationClient) ListFilesystems(ctx context.Context, in *ListFilesystemReq, opts ...grpc.CallOption) (*ListFilesystemRes, error) { out := new(ListFilesystemRes) err := c.cc.Invoke(ctx, "/Replication/ListFilesystems", in, out, opts...) @@ -1165,6 +1254,7 @@ func (c *replicationClient) ReplicationCursor(ctx context.Context, in *Replicati // ReplicationServer is the server API for Replication service. type ReplicationServer interface { + Ping(context.Context, *PingReq) (*PingRes, error) ListFilesystems(context.Context, *ListFilesystemReq) (*ListFilesystemRes, error) ListFilesystemVersions(context.Context, *ListFilesystemVersionsReq) (*ListFilesystemVersionsRes, error) DestroySnapshots(context.Context, *DestroySnapshotsReq) (*DestroySnapshotsRes, error) @@ -1175,6 +1265,24 @@ func RegisterReplicationServer(s *grpc.Server, srv ReplicationServer) { s.RegisterService(&_Replication_serviceDesc, srv) } +func _Replication_Ping_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { + in := new(PingReq) + if err := dec(in); err != nil { + return nil, err + } + if interceptor == nil { + return srv.(ReplicationServer).Ping(ctx, in) + } + info := &grpc.UnaryServerInfo{ + Server: srv, + FullMethod: "/Replication/Ping", + } + handler := func(ctx context.Context, req interface{}) (interface{}, error) { + return srv.(ReplicationServer).Ping(ctx, req.(*PingReq)) + } + return interceptor(ctx, in, info, handler) +} + func _Replication_ListFilesystems_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { in := new(ListFilesystemReq) if err := dec(in); err != nil { @@ -1251,6 +1359,10 @@ var _Replication_serviceDesc = grpc.ServiceDesc{ ServiceName: "Replication", HandlerType: (*ReplicationServer)(nil), Methods: []grpc.MethodDesc{ + { + MethodName: "Ping", + Handler: _Replication_Ping_Handler, + }, { MethodName: "ListFilesystems", Handler: _Replication_ListFilesystems_Handler, @@ -1272,54 +1384,57 @@ var _Replication_serviceDesc = grpc.ServiceDesc{ Metadata: "pdu.proto", } -func init() { proto.RegisterFile("pdu.proto", fileDescriptor_pdu_89315d819a6e0938) } +func init() { proto.RegisterFile("pdu.proto", fileDescriptor_pdu_759e477f62b58e58) } -var fileDescriptor_pdu_89315d819a6e0938 = []byte{ - // 735 bytes of a gzipped FileDescriptorProto - 0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x8c, 0x55, 0xdd, 0x6e, 0xda, 0x4a, - 0x10, 0xc6, 0x60, 0xc0, 0x0c, 0x51, 0x42, 0x36, 0x9c, 0xc8, 0xc7, 0xe7, 0x28, 0x42, 0xdb, 0x1b, - 0x52, 0xa9, 0x6e, 0x45, 0x7b, 0x53, 0x55, 0xaa, 0x54, 0x42, 0x7e, 0xa4, 0x56, 0x69, 0xb4, 0xd0, - 0x28, 0xca, 0x1d, 0x0d, 0xa3, 0xc4, 0x0a, 0xb0, 0xce, 0xee, 0xba, 0x0a, 0xbd, 0xec, 0x7b, 0xf4, - 0x41, 0xfa, 0x0e, 0xbd, 0xec, 0x03, 0x55, 0xbb, 0x60, 0xe3, 0x60, 0x23, 0x71, 0xe5, 0xfd, 0xbe, - 0x9d, 0x9d, 0x9d, 0xf9, 0x76, 0x66, 0x0c, 0xb5, 0x70, 0x14, 0xf9, 0xa1, 0xe0, 0x8a, 0xd3, 0x3d, - 0xd8, 0xfd, 0x14, 0x48, 0x75, 0x12, 0x8c, 0x51, 0xce, 0xa4, 0xc2, 0x09, 0xc3, 0x07, 0x7a, 0x95, - 0x25, 0x25, 0x79, 0x01, 0xf5, 0x25, 0x21, 0x5d, 0xab, 0x55, 0x6a, 0xd7, 0x3b, 0x75, 0x3f, 0x65, - 0x94, 0xde, 0x27, 0x4d, 0x28, 0x1f, 0x4f, 0x42, 0x35, 0x73, 0x8b, 0x2d, 0xab, 0xed, 0xb0, 0x39, - 0xa0, 0x5d, 0x80, 0xa5, 0x11, 0x21, 0x60, 0x5f, 0x0c, 0xd5, 0x9d, 0x6b, 0xb5, 0xac, 0x76, 0x8d, - 0x99, 0x35, 0x69, 0x41, 0x9d, 0xa1, 0x8c, 0x26, 0x38, 0xe0, 0xf7, 0x38, 0x35, 0xa7, 0x6b, 0x2c, - 0x4d, 0xd1, 0x77, 0xf0, 0xef, 0xd3, 0xe8, 0x2e, 0x51, 0xc8, 0x80, 0x4f, 0x25, 0xc3, 0x07, 0x72, - 0x90, 0xbe, 0x60, 0xe1, 0x38, 0xc5, 0xd0, 0x8f, 0xeb, 0x0f, 0x4b, 0xe2, 0x83, 0x13, 0xc3, 0x45, - 0x7e, 0xc4, 0xcf, 0x58, 0xb2, 0xc4, 0x86, 0xfe, 0xb1, 0x60, 0x37, 0xb3, 0x4f, 0x3a, 0x60, 0x0f, - 0x66, 0x21, 0x9a, 0xcb, 0xb7, 0x3b, 0x07, 0x59, 0x0f, 0xfe, 0xe2, 0xab, 0xad, 0x98, 0xb1, 0xd5, - 0x4a, 0x9c, 0x0f, 0x27, 0xb8, 0x48, 0xd7, 0xac, 0x35, 0x77, 0x1a, 0x05, 0x23, 0xb7, 0xd4, 0xb2, - 0xda, 0x36, 0x33, 0x6b, 0xf2, 0x3f, 0xd4, 0x8e, 0x04, 0x0e, 0x15, 0x0e, 0xae, 0x4e, 0x5d, 0xdb, - 0x6c, 0x2c, 0x09, 0xe2, 0x81, 0x63, 0x40, 0xc0, 0xa7, 0x6e, 0xd9, 0x78, 0x4a, 0x30, 0x3d, 0x84, - 0x7a, 0xea, 0x5a, 0xb2, 0x05, 0x4e, 0x7f, 0x3a, 0x0c, 0xe5, 0x1d, 0x57, 0x8d, 0x82, 0x46, 0x5d, - 0xce, 0xef, 0x27, 0x43, 0x71, 0xdf, 0xb0, 0xe8, 0x2f, 0x0b, 0xaa, 0x7d, 0x9c, 0x8e, 0x36, 0xd0, - 0x53, 0x07, 0x79, 0x22, 0xf8, 0x24, 0x0e, 0x5c, 0xaf, 0xc9, 0x36, 0x14, 0x07, 0xdc, 0x84, 0x5d, - 0x63, 0xc5, 0x01, 0x5f, 0x7d, 0x52, 0x3b, 0xf3, 0xa4, 0x26, 0x70, 0x3e, 0x09, 0x05, 0x4a, 0x69, - 0x02, 0x77, 0x58, 0x82, 0x75, 0x21, 0xf5, 0x70, 0x14, 0x85, 0x6e, 0x65, 0x5e, 0x48, 0x06, 0x90, - 0x7d, 0xa8, 0xf4, 0xc4, 0x8c, 0x45, 0x53, 0xb7, 0x6a, 0xe8, 0x05, 0xa2, 0x6f, 0xc0, 0xb9, 0x10, - 0x3c, 0x44, 0xa1, 0x66, 0x89, 0xa8, 0x56, 0x4a, 0xd4, 0x26, 0x94, 0x2f, 0x87, 0xe3, 0x28, 0x56, - 0x7a, 0x0e, 0xe8, 0x8f, 0x24, 0x63, 0x49, 0xda, 0xb0, 0xf3, 0x45, 0xe2, 0x68, 0xb5, 0x08, 0x1d, - 0xb6, 0x4a, 0x13, 0x0a, 0x5b, 0xc7, 0x8f, 0x21, 0xde, 0x28, 0x1c, 0xf5, 0x83, 0xef, 0x68, 0x32, - 0x2e, 0xb1, 0x27, 0x1c, 0x39, 0x04, 0x58, 0xc4, 0x13, 0xa0, 0x74, 0x6d, 0x53, 0x54, 0x35, 0x3f, - 0x0e, 0x91, 0xa5, 0x36, 0xe9, 0x15, 0x00, 0xc3, 0x1b, 0x0c, 0xbe, 0xe1, 0x26, 0xc2, 0x3f, 0x87, - 0xc6, 0xd1, 0x18, 0x87, 0x22, 0x1b, 0x67, 0x86, 0xa7, 0x5b, 0x29, 0xcf, 0x92, 0xde, 0xc2, 0x5e, - 0x0f, 0xa5, 0x12, 0x7c, 0x16, 0x57, 0xc0, 0x26, 0x9d, 0x43, 0x5e, 0x41, 0x2d, 0xb1, 0x77, 0x8b, - 0x6b, 0xbb, 0x63, 0x69, 0x44, 0xaf, 0x81, 0xac, 0x5c, 0xb4, 0x68, 0xb2, 0x18, 0x9a, 0x5b, 0xd6, - 0x34, 0x59, 0x6c, 0x63, 0x06, 0x89, 0x10, 0x5c, 0xc4, 0x2f, 0x66, 0x00, 0xed, 0xe5, 0x25, 0xa1, - 0x87, 0x54, 0x55, 0x27, 0x3e, 0x56, 0x71, 0x03, 0xef, 0xf9, 0xd9, 0x10, 0x58, 0x6c, 0x43, 0x7f, - 0x5b, 0xd0, 0x64, 0x18, 0x8e, 0x83, 0x1b, 0xd3, 0x24, 0x47, 0x91, 0x90, 0x5c, 0x6c, 0x22, 0xc6, - 0x4b, 0x28, 0xdd, 0xa2, 0x32, 0x21, 0xd5, 0x3b, 0xff, 0xf9, 0x79, 0x3e, 0xfc, 0x53, 0x54, 0x9f, - 0xc3, 0xb3, 0x02, 0xd3, 0x96, 0xfa, 0x80, 0x44, 0x65, 0x4a, 0x64, 0xed, 0x81, 0x7e, 0x7c, 0x40, - 0xa2, 0xf2, 0xaa, 0x50, 0x36, 0x0e, 0xbc, 0x67, 0x50, 0x36, 0x1b, 0xba, 0x49, 0x12, 0xe1, 0xe6, - 0x5a, 0x24, 0xb8, 0x6b, 0x43, 0x91, 0x87, 0x74, 0x90, 0x9b, 0x8d, 0x6e, 0xa1, 0xf9, 0x24, 0xd1, - 0x79, 0xd8, 0x67, 0x85, 0x64, 0x96, 0x38, 0xe7, 0x5c, 0xe1, 0x63, 0x20, 0xe7, 0xfe, 0x9c, 0xb3, - 0x02, 0x4b, 0x98, 0xae, 0x03, 0x95, 0xb9, 0x4a, 0x9d, 0x9f, 0x45, 0xdd, 0xbf, 0x89, 0x5b, 0xf2, - 0x16, 0x76, 0x9e, 0x8e, 0x50, 0x49, 0x88, 0x9f, 0xf9, 0x89, 0x78, 0x59, 0x4e, 0x92, 0x0b, 0xd8, - 0xcf, 0x9f, 0xbe, 0xc4, 0xf3, 0xd7, 0xce, 0x74, 0x6f, 0xfd, 0x9e, 0x24, 0xef, 0xa1, 0xb1, 0x5a, - 0x07, 0xa4, 0xe9, 0xe7, 0xd4, 0xb7, 0x97, 0xc7, 0x4a, 0xf2, 0x01, 0x76, 0x33, 0x92, 0x91, 0x7f, - 0x72, 0xdf, 0xc7, 0xcb, 0xa5, 0x65, 0xb7, 0x7c, 0x5d, 0x0a, 0x47, 0xd1, 0xd7, 0x8a, 0xf9, 0xa1, - 0xbe, 0xfe, 0x1b, 0x00, 0x00, 0xff, 0xff, 0xa3, 0xba, 0x8e, 0x63, 0x5d, 0x07, 0x00, 0x00, +var fileDescriptor_pdu_759e477f62b58e58 = []byte{ + // 779 bytes of a gzipped FileDescriptorProto + 0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x8c, 0x55, 0x51, 0x8f, 0xe2, 0x36, + 0x10, 0xde, 0x40, 0x80, 0x30, 0xac, 0xee, 0x58, 0x2f, 0x3d, 0xa5, 0x69, 0x7b, 0x42, 0xbe, 0x17, + 0xae, 0x52, 0xd3, 0x8a, 0xf6, 0xa5, 0xaa, 0x54, 0xa9, 0x2c, 0x7b, 0xbb, 0x52, 0xdb, 0x2b, 0x32, + 0xf4, 0xb4, 0xba, 0xb7, 0x14, 0x46, 0x6c, 0xb4, 0x80, 0x73, 0xb6, 0x53, 0x1d, 0x7d, 0xec, 0xbf, + 0xda, 0xff, 0xd0, 0xc7, 0xfe, 0xa0, 0x93, 0x4d, 0x1c, 0xb2, 0x24, 0x48, 0x3c, 0xc5, 0xdf, 0xe7, + 0xcf, 0xe3, 0x99, 0xf1, 0xcc, 0x04, 0xda, 0xc9, 0x22, 0x0d, 0x13, 0xc1, 0x15, 0xa7, 0x97, 0x70, + 0xf1, 0x5b, 0x2c, 0xd5, 0x9b, 0x78, 0x85, 0x72, 0x2b, 0x15, 0xae, 0x19, 0x7e, 0xa0, 0x77, 0x65, + 0x52, 0x92, 0x6f, 0xa0, 0xb3, 0x27, 0xa4, 0xef, 0xf4, 0xeb, 0x83, 0xce, 0xb0, 0x13, 0x16, 0x44, + 0xc5, 0x7d, 0xd2, 0x83, 0xc6, 0xf5, 0x3a, 0x51, 0x5b, 0xbf, 0xd6, 0x77, 0x06, 0x1e, 0xdb, 0x01, + 0x3a, 0x02, 0xd8, 0x8b, 0x08, 0x01, 0x77, 0x12, 0xa9, 0x7b, 0xdf, 0xe9, 0x3b, 0x83, 0x36, 0x33, + 0x6b, 0xd2, 0x87, 0x0e, 0x43, 0x99, 0xae, 0x71, 0xc6, 0x1f, 0x70, 0x63, 0x4e, 0xb7, 0x59, 0x91, + 0xa2, 0x3f, 0xc1, 0xe7, 0x4f, 0xbd, 0x7b, 0x87, 0x42, 0xc6, 0x7c, 0x23, 0x19, 0x7e, 0x20, 0x2f, + 0x8b, 0x17, 0x64, 0x86, 0x0b, 0x0c, 0xfd, 0xf5, 0xf8, 0x61, 0x49, 0x42, 0xf0, 0x2c, 0xcc, 0xe2, + 0x23, 0x61, 0x49, 0xc9, 0x72, 0x0d, 0xfd, 0xdf, 0x81, 0x8b, 0xd2, 0x3e, 0x19, 0x82, 0x3b, 0xdb, + 0x26, 0x68, 0x2e, 0x7f, 0x36, 0x7c, 0x59, 0xb6, 0x10, 0x66, 0x5f, 0xad, 0x62, 0x46, 0xab, 0x33, + 0xf1, 0x36, 0x5a, 0x63, 0x16, 0xae, 0x59, 0x6b, 0xee, 0x26, 0x8d, 0x17, 0x7e, 0xbd, 0xef, 0x0c, + 0x5c, 0x66, 0xd6, 0xe4, 0x4b, 0x68, 0x5f, 0x09, 0x8c, 0x14, 0xce, 0xee, 0x6e, 0x7c, 0xd7, 0x6c, + 0xec, 0x09, 0x12, 0x80, 0x67, 0x40, 0xcc, 0x37, 0x7e, 0xc3, 0x58, 0xca, 0x31, 0x7d, 0x0d, 0x9d, + 0xc2, 0xb5, 0xe4, 0x1c, 0xbc, 0xe9, 0x26, 0x4a, 0xe4, 0x3d, 0x57, 0xdd, 0x33, 0x8d, 0x46, 0x9c, + 0x3f, 0xac, 0x23, 0xf1, 0xd0, 0x75, 0xe8, 0xa3, 0x03, 0xad, 0x29, 0x6e, 0x16, 0x27, 0xe4, 0x53, + 0x3b, 0xf9, 0x46, 0xf0, 0xb5, 0x75, 0x5c, 0xaf, 0xc9, 0x33, 0xa8, 0xcd, 0xb8, 0x71, 0xbb, 0xcd, + 0x6a, 0x33, 0x7e, 0xf8, 0xa4, 0x6e, 0xe9, 0x49, 0x8d, 0xe3, 0x7c, 0x9d, 0x08, 0x94, 0xd2, 0x38, + 0xee, 0xb1, 0x1c, 0xeb, 0x42, 0x1a, 0xe3, 0x22, 0x4d, 0xfc, 0xe6, 0xae, 0x90, 0x0c, 0x20, 0x2f, + 0xa0, 0x39, 0x16, 0x5b, 0x96, 0x6e, 0xfc, 0x96, 0xa1, 0x33, 0x44, 0x7f, 0x00, 0x6f, 0x22, 0x78, + 0x82, 0x42, 0x6d, 0xf3, 0xa4, 0x3a, 0x85, 0xa4, 0xf6, 0xa0, 0xf1, 0x2e, 0x5a, 0xa5, 0x36, 0xd3, + 0x3b, 0x40, 0xff, 0xcd, 0x23, 0x96, 0x64, 0x00, 0xcf, 0xff, 0x94, 0xb8, 0x38, 0x2c, 0x42, 0x8f, + 0x1d, 0xd2, 0x84, 0xc2, 0xf9, 0xf5, 0xc7, 0x04, 0xe7, 0x0a, 0x17, 0xd3, 0xf8, 0x1f, 0x34, 0x11, + 0xd7, 0xd9, 0x13, 0x8e, 0xbc, 0x06, 0xc8, 0xfc, 0x89, 0x51, 0xfa, 0xae, 0x29, 0xaa, 0x76, 0x68, + 0x5d, 0x64, 0x85, 0x4d, 0x7a, 0x07, 0xc0, 0x70, 0x8e, 0xf1, 0xdf, 0x78, 0x4a, 0xe2, 0xbf, 0x86, + 0xee, 0xd5, 0x0a, 0x23, 0x51, 0xf6, 0xb3, 0xc4, 0xd3, 0xf3, 0x82, 0x65, 0x49, 0x97, 0x70, 0x39, + 0x46, 0xa9, 0x04, 0xdf, 0xda, 0x0a, 0x38, 0xa5, 0x73, 0xc8, 0x77, 0xd0, 0xce, 0xf5, 0x7e, 0xed, + 0x68, 0x77, 0xec, 0x45, 0xf4, 0x3d, 0x90, 0x83, 0x8b, 0xb2, 0x26, 0xb3, 0xd0, 0xdc, 0x72, 0xa4, + 0xc9, 0xac, 0xc6, 0x0c, 0x12, 0x21, 0xb8, 0xb0, 0x2f, 0x66, 0x00, 0x1d, 0x57, 0x05, 0xa1, 0x87, + 0x54, 0x4b, 0x07, 0xbe, 0x52, 0xb6, 0x81, 0x2f, 0xc3, 0xb2, 0x0b, 0xcc, 0x6a, 0xe8, 0x7f, 0x0e, + 0xf4, 0x18, 0x26, 0xab, 0x78, 0x6e, 0x9a, 0xe4, 0x2a, 0x15, 0x92, 0x8b, 0x53, 0x92, 0xf1, 0x2d, + 0xd4, 0x97, 0xa8, 0x8c, 0x4b, 0x9d, 0xe1, 0x17, 0x61, 0x95, 0x8d, 0xf0, 0x06, 0xd5, 0x1f, 0xc9, + 0xed, 0x19, 0xd3, 0x4a, 0x7d, 0x40, 0xa2, 0x32, 0x25, 0x72, 0xf4, 0xc0, 0xd4, 0x1e, 0x90, 0xa8, + 0x82, 0x16, 0x34, 0x8c, 0x81, 0xe0, 0x15, 0x34, 0xcc, 0x86, 0x6e, 0x92, 0x3c, 0x71, 0xbb, 0x5c, + 0xe4, 0x78, 0xe4, 0x42, 0x8d, 0x27, 0x74, 0x56, 0x19, 0x8d, 0x6e, 0xa1, 0xdd, 0x24, 0xd1, 0x71, + 0xb8, 0xb7, 0x67, 0xf9, 0x2c, 0xf1, 0xde, 0x72, 0x85, 0x1f, 0x63, 0xb9, 0xb3, 0xe7, 0xdd, 0x9e, + 0xb1, 0x9c, 0x19, 0x79, 0xd0, 0xdc, 0x65, 0x89, 0xbe, 0x82, 0xd6, 0x24, 0xde, 0x2c, 0x75, 0x5a, + 0x7c, 0x68, 0xfd, 0x8e, 0x52, 0x46, 0x4b, 0xdb, 0x54, 0x16, 0xd2, 0xaf, 0xac, 0x48, 0xea, 0xb6, + 0xbb, 0x9e, 0xdf, 0x73, 0xdb, 0x76, 0x7a, 0x3d, 0x7c, 0xac, 0xe9, 0x19, 0x90, 0xbb, 0x46, 0x02, + 0x70, 0xb5, 0x9c, 0x78, 0x61, 0x66, 0x3a, 0xb0, 0x2b, 0x49, 0x7e, 0x84, 0xe7, 0x4f, 0x47, 0xb4, + 0x24, 0x24, 0x2c, 0xfd, 0xa4, 0x82, 0x32, 0x27, 0xc9, 0x04, 0x5e, 0x54, 0x4f, 0x77, 0x12, 0x84, + 0x47, 0xff, 0x19, 0xc1, 0xf1, 0x3d, 0x49, 0x7e, 0x86, 0xee, 0x61, 0x9d, 0x91, 0x5e, 0x58, 0xd1, + 0x3f, 0x41, 0x15, 0x2b, 0xc9, 0x2f, 0x70, 0x51, 0x7a, 0x12, 0xf2, 0x59, 0xe5, 0xfb, 0x07, 0x95, + 0xb4, 0x1c, 0x35, 0xde, 0xd7, 0x93, 0x45, 0xfa, 0x57, 0xd3, 0xfc, 0xb0, 0xbf, 0xff, 0x14, 0x00, + 0x00, 0xff, 0xff, 0xaf, 0xa9, 0xda, 0x41, 0xbd, 0x07, 0x00, 0x00, } diff --git a/replication/logic/pdu/pdu.proto b/replication/logic/pdu/pdu.proto index 1b66916..15be58e 100644 --- a/replication/logic/pdu/pdu.proto +++ b/replication/logic/pdu/pdu.proto @@ -2,6 +2,7 @@ syntax = "proto3"; option go_package = "pdu"; service Replication { + rpc Ping (PingReq) returns (PingRes); rpc ListFilesystems (ListFilesystemReq) returns (ListFilesystemRes); rpc ListFilesystemVersions (ListFilesystemVersionsReq) returns (ListFilesystemVersionsRes); rpc DestroySnapshots (DestroySnapshotsReq) returns (DestroySnapshotsRes); @@ -120,3 +121,12 @@ message ReplicationCursorRes { bool Notexist = 2; } } + +message PingReq { + string Message = 1; +} + +message PingRes { + // Echo must be PingReq.Message + string Echo = 1; +} \ No newline at end of file diff --git a/replication/logic/replication_logic.go b/replication/logic/replication_logic.go index 1ad6aa0..cf29596 100644 --- a/replication/logic/replication_logic.go +++ b/replication/logic/replication_logic.go @@ -26,6 +26,7 @@ type Endpoint interface { 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) + WaitForConnectivity(ctx context.Context) (error) } type Sender interface { @@ -64,6 +65,44 @@ func (p *Planner) Plan(ctx context.Context) ([]driver.FS, error) { return dfss, nil } +func (p *Planner) WaitForConnectivity(ctx context.Context) error { + var wg sync.WaitGroup + doPing := func(endpoint Endpoint, errOut *error) { + defer wg.Done() + err := endpoint.WaitForConnectivity(ctx) + if err != nil { + *errOut = err + } else { + *errOut = nil + } + } + wg.Add(2) + var senderErr, receiverErr error + go doPing(p.sender, &senderErr) + go doPing(p.receiver, &receiverErr) + wg.Wait() + if senderErr == nil && receiverErr == nil { + return nil + } else if senderErr != nil && receiverErr != nil { + if senderErr.Error() == receiverErr.Error() { + return fmt.Errorf("sender and receiver are not reachable: %s", senderErr.Error()) + } else { + return fmt.Errorf("sender and receiver are not reachable:\n sender: %s\n receiver: %s", senderErr, receiverErr) + } + } else { + var side string + var err *error + if senderErr != nil { + side = "sender" + err = &senderErr + } else { + side = "receiver" + err = &receiverErr + } + return fmt.Errorf("%s is not reachable: %s", side, *err) + } +} + type Filesystem struct { sender Sender receiver Receiver @@ -73,6 +112,15 @@ type Filesystem struct { promBytesReplicated prometheus.Counter // compat } +func (f *Filesystem) EqualToPreviousAttempt(other driver.FS) bool { + g, ok := other.(*Filesystem) + if !ok { + return false + } + // TODO: use GUIDs (issued by zrepl, not those from ZFS) + return f.Path == g.Path +} + func (f *Filesystem) PlanFS(ctx context.Context) ([]driver.Step, error) { steps, err := f.doPlanning(ctx) if err != nil { @@ -99,6 +147,18 @@ type Step struct { expectedSize int64 // 0 means no size estimate present / possible } +func (s *Step) TargetEquals(other driver.Step) bool { + t, ok := other.(*Step) + if !ok { + return false + } + if !s.parent.EqualToPreviousAttempt(t.parent) { + panic("Step interface promise broken: parent filesystems must be same") + } + return s.from.GetGuid() == t.from.GetGuid() && + s.to.GetGuid() == t.to.GetGuid() +} + func (s *Step) TargetDate() time.Time { return s.to.SnapshotTime() // FIXME compat name } @@ -112,8 +172,13 @@ func (s *Step) ReportInfo() *report.StepInfo { if s.byteCounter != nil { byteCounter = s.byteCounter.Count() } + // FIXME stick to zfs convention of from and to + from := "" + if s.from != nil { + from = s.from.RelName() + } return &report.StepInfo{ - From: s.from.RelName(), + From: from, To: s.to.RelName(), BytesExpected: s.expectedSize, BytesReplicated: byteCounter, diff --git a/replication/report/replication_report.go b/replication/report/replication_report.go index bce4a23..7f8f605 100644 --- a/replication/report/replication_report.go +++ b/replication/report/replication_report.go @@ -6,8 +6,10 @@ import ( ) type Report struct { - StartAt, FinishAt time.Time - Attempts []*AttemptReport + StartAt, FinishAt time.Time + WaitReconnectSince, WaitReconnectUntil time.Time + WaitReconnectError *TimedError + Attempts []*AttemptReport } var _, _ = json.Marshal(&Report{}) diff --git a/rpc/dataconn/dataconn_client.go b/rpc/dataconn/dataconn_client.go index 324dc2c..8473d97 100644 --- a/rpc/dataconn/dataconn_client.go +++ b/rpc/dataconn/dataconn_client.go @@ -213,3 +213,23 @@ func (c *Client) ReqRecv(ctx context.Context, req *pdu.ReceiveReq, streamCopier return res.res, cause } + + +func (c *Client) ReqPing(ctx context.Context, req *pdu.PingReq) (*pdu.PingRes, error) { + conn, err := c.getWire(ctx) + if err != nil { + return nil, err + } + defer c.putWire(conn) + + if err := c.send(ctx, conn, EndpointPing, req, nil); err != nil { + return nil, err + } + + var res pdu.PingRes + if err := c.recv(ctx, conn, &res); err != nil { + return nil, err + } + + return &res, nil +} \ No newline at end of file diff --git a/rpc/dataconn/dataconn_server.go b/rpc/dataconn/dataconn_server.go index 8c0e670..646545f 100644 --- a/rpc/dataconn/dataconn_server.go +++ b/rpc/dataconn/dataconn_server.go @@ -25,6 +25,8 @@ type Handler interface { // It is guaranteed that Server calls Receive with a stream that holds the IdleConnTimeout // configured in ServerConfig.Shared.IdleConnTimeout. Receive(ctx context.Context, r *pdu.ReceiveReq, receive zfs.StreamCopier) (*pdu.ReceiveRes, error) + // PingDataconn handles a PingReq + PingDataconn(ctx context.Context, r *pdu.PingReq) (*pdu.PingRes, error) } type Logger = logger.Logger @@ -125,6 +127,13 @@ func (s *Server) serveConn(nc *transport.AuthConn) { return } res, handlerErr = s.h.Receive(ctx, &req, &streamCopier{streamConn: c, closeStreamOnClose: false}) // SHADOWING + case EndpointPing: + var req pdu.PingReq + if err := proto.Unmarshal(reqStructured, &req); err != nil { + s.log.WithError(err).Error("cannot unmarshal ping request") + return + } + res, handlerErr = s.h.PingDataconn(ctx, &req) // SHADOWING default: s.log.WithField("endpoint", endpoint).Error("unknown endpoint") handlerErr = fmt.Errorf("requested endpoint does not exist") diff --git a/rpc/dataconn/dataconn_shared.go b/rpc/dataconn/dataconn_shared.go index 0ea5a34..43ccf92 100644 --- a/rpc/dataconn/dataconn_shared.go +++ b/rpc/dataconn/dataconn_shared.go @@ -10,6 +10,7 @@ import ( ) const ( + EndpointPing string = "/v1/ping" EndpointSend string = "/v1/send" EndpointRecv string = "/v1/recv" ) diff --git a/rpc/dataconn/microbenchmark/microbenchmark.go b/rpc/dataconn/microbenchmark/microbenchmark.go index a39c53d..0e11a8e 100644 --- a/rpc/dataconn/microbenchmark/microbenchmark.go +++ b/rpc/dataconn/microbenchmark/microbenchmark.go @@ -77,6 +77,12 @@ func (devNullHandler) Receive(ctx context.Context, r *pdu.ReceiveReq, stream zfs return &res, err } +func (devNullHandler) PingDataconn(ctx context.Context, r *pdu.PingReq) (*pdu.PingRes, error) { + return &pdu.PingRes{ + Echo: r.GetMessage(), + }, nil +} + type tcpConnecter struct { addr string } diff --git a/rpc/rpc_client.go b/rpc/rpc_client.go index ba55134..c34dc3b 100644 --- a/rpc/rpc_client.go +++ b/rpc/rpc_client.go @@ -2,11 +2,16 @@ package rpc import ( "context" + "errors" + "fmt" "net" + "sync" + "sync/atomic" "time" "google.golang.org/grpc" + "github.com/google/uuid" "github.com/zrepl/zrepl/replication/logic" "github.com/zrepl/zrepl/replication/logic/pdu" "github.com/zrepl/zrepl/rpc/dataconn" @@ -101,6 +106,67 @@ func (c *Client) ReplicationCursor(ctx context.Context, in *pdu.ReplicationCurso return c.controlClient.ReplicationCursor(ctx, in) } +func (c *Client) WaitForConnectivity(ctx context.Context) error { + ctx, cancel := context.WithCancel(ctx) + defer cancel() + msg := uuid.New().String() + req := pdu.PingReq{Message: msg} + var ctrlOk, dataOk int32 + loggers := GetLoggersOrPanic(ctx) + var wg sync.WaitGroup + wg.Add(2) + checkRes := func(res *pdu.PingRes, err error, logger Logger, okVar *int32) { + if err == nil && res.GetEcho() != req.GetMessage() { + err = errors.New("pilot message not echoed correctly") + } + if err == context.Canceled { + err = nil + } + if err != nil { + logger.WithError(err).Error("ping failed") + atomic.StoreInt32(okVar, 0) + cancel() + } else { + atomic.StoreInt32(okVar, 1) + } + } + go func() { + defer wg.Done() + ctrl, ctrlErr := c.controlClient.Ping(ctx, &req, grpc.FailFast(false)) + checkRes(ctrl, ctrlErr, loggers.Control, &ctrlOk) + }() + go func() { + defer wg.Done() + for ctx.Err() == nil { + data, dataErr := c.dataClient.ReqPing(ctx, &req) + // dataClient uses transport.Connecter, which doesn't expose FailFast(false) + // => we need to mask dial timeouts + if err, ok := dataErr.(interface{ Temporary() bool }); ok && err.Temporary() { + continue + } + // it's not a dial timeout, + checkRes(data, dataErr, loggers.Data, &dataOk) + return + } + }() + wg.Wait() + var what string + if ctrlOk == 1 && dataOk == 1 { + return nil + } + if ctrlOk == 0 { + what += "control" + } + if dataOk == 0 { + if len(what) > 0 { + what += " and data" + } else { + what += "data" + } + } + return fmt.Errorf("%s rpc failed to respond to ping rpcs", what) +} + func (c *Client) ResetConnectBackoff() { c.controlConn.ResetConnectBackoff() } diff --git a/util/envconst/envconst.go b/util/envconst/envconst.go index 8c13190..44bc9b8 100644 --- a/util/envconst/envconst.go +++ b/util/envconst/envconst.go @@ -40,3 +40,19 @@ func Int64(varname string, def int64) int64 { cache.Store(varname, d) return d } + +func Bool(varname string, def bool) bool { + if v, ok := cache.Load(varname); ok { + return v.(bool) + } + e := os.Getenv(varname) + if e == "" { + return def + } + d, err := strconv.ParseBool(e) + if err != nil { + panic(err) + } + cache.Store(varname, d) + return d +}