endpoint: refactor, fix stale holds on initial replication failure, holds release subcmds, more efficient ZFS queries
The motivation for this recatoring are based on two independent issues: - @JMoVS found that the changes merged as part of #259 slowed his OS X based installation down significantly. Analysis of the zfs command logging introduced in #296 showed that `zfs holds` took most of the execution time, and they pointed out that not all of those `zfs holds` invocations were actually necessary. I.e.: zrepl was inefficient about retrieving information from ZFS. - @InsanePrawn found that failures on initial replication would lead to step holds accumulating on the sending side, i.e. they would never be cleaned up in the HintMostRecentCommonAncestor RPC handler. That was because we only sent that RPC if there was a most recent common ancestor detected during replication planning. @InsanePrawn prototyped an implementation of a `zrepl holds release` command to mitigate the situation. As part of that development work and back-and-forth with @problame, it became evident that the abstractions that #259 built on top of zfs in package endpoint (step holds, replication cursor, last-received-hold), were not well-represented for re-use in the `zrepl holds release` subocommand. This commit refactors package endpoint to address both of these issues: - endpoint abstractions now share an interface `Abstraction` that, among other things, provides a uniform `Destroy()` method. However, that method should not be destroyed directly but instead the package-level `BatchDestroy` function should be used in order to allow for a migration to zfs channel programs in the future. - endpoint now has a query facitilty (`ListAbstractions`) which is used to find on-disk - step holds and bookmarks - replication cursors (v1, v2) - last-received-holds By describing the query in a struct, we can centralized the retrieval of information via the ZFS CLI and only have to be clever once. We are "clever" in the following ways: - When asking for hold-based abstractions, we only run `zfs holds` on snapshot that have `userrefs` > 0 - To support this functionality, add field `UserRefs` to zfs.FilesystemVersion and retrieve it anywhere we retrieve zfs.FilesystemVersion from ZFS. - When asking only for bookmark-based abstractions, we only run `zfs list -t bookmark`, not with snapshots. - Currently unused (except for CLI) per-filesystem concurrent lookup - Option to only include abstractions with CreateTXG in a specified range - refactor `endpoint`'s various ZFS info retrieval methods to use `ListAbstractions` - change the `zrepl holds list` command to consume endpoint.ListAbstractions - Add a `ListStale` method which, given a query template, lists stale holds and bookmarks. - it uses replication cursor has different modes - the new `zrepl holds release-{all,stale}` commands can be used to remove abstractions of package endpoint - Adjust HintMostRecentCommonAncestor RPC for stale-holds cleanup: - send it also if no most recent common ancestor exists between sender and receiver - have the sender clean up its abstractions when it receives the RPC with no most recent common ancestor, using `ListStale` - Due to changed semantics, bump the protocol version. - Adjust HintMostRecentCommonAncestor RPC for performance problems encountered by @JMoVS - by default, per (job,fs)-combination, only consider cleaning step holds in the createtxg range `[last replication cursor,conservatively-estimated-receive-side-version)` - this behavior ensures resumability at cost proportional to the time that replication was donw - however, as explained in a comment, we might leak holds if the zrepl daemon stops running - that trade-off is acceptable because in the presumably rare this might happen the user has two tools at their hand: - Tool 1: run `zrepl holds release-stale` - Tool 2: use env var `ZREPL_ENDPOINT_SENDER_HINT_MOST_RECENT_STEP_HOLD_CLEANUP_MODE` to adjust the lower bound of the createtxg range (search for it in the code). The env var can also be used to disable hold-cleanup on the send-side entirely. supersedes closes #293 supersedes closes #282 fixes #280 fixes #278 Additionaly, we fixed a couple of bugs: - zfs: fix half-nil error reporting of dataset-does-not-exist for ZFSListChan and ZFSBookmark - endpoint: Sender's `HintMostRecentCommonAncestor` handler would not check whether access to the specified filesystem was allowed.
This commit is contained in:
Christian Schwarz
+133
-45
@@ -96,7 +96,7 @@ func (s *Sender) ListFilesystemVersions(ctx context.Context, r *pdu.ListFilesyst
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
fsvs, err := zfs.ZFSListFilesystemVersions(lp, nil)
|
||||
fsvs, err := zfs.ZFSListFilesystemVersions(lp, zfs.ListFilesystemVersionsOptions{})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
@@ -110,34 +110,116 @@ func (s *Sender) ListFilesystemVersions(ctx context.Context, r *pdu.ListFilesyst
|
||||
}
|
||||
|
||||
func (p *Sender) HintMostRecentCommonAncestor(ctx context.Context, r *pdu.HintMostRecentCommonAncestorReq) (*pdu.HintMostRecentCommonAncestorRes, error) {
|
||||
var err error
|
||||
|
||||
fs := r.GetFilesystem()
|
||||
mostRecent, err := sendArgsFromPDUAndValidateExists(ctx, fs, r.GetSenderVersion())
|
||||
fsp, err := p.filterCheckFS(r.GetFilesystem())
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
fs := fsp.ToString()
|
||||
|
||||
log := getLogger(ctx).WithField("fs", fs).WithField("hinted_most_recent", fmt.Sprintf("%#v", r.GetSenderVersion()))
|
||||
|
||||
log.WithField("full_hint", r).Debug("full hint")
|
||||
|
||||
if r.GetSenderVersion() == nil {
|
||||
// no common ancestor found, likely due to failed prior replication attempt
|
||||
// => release stale step holds to prevent them from accumulating
|
||||
// (they can accumulate on initial replication because each inital replication step might hold a different `to`)
|
||||
// => replication cursors cannot accumulate because we always _move_ the replication cursor
|
||||
log.Debug("releasing all step holds on the filesystem")
|
||||
TryReleaseStepStaleFS(ctx, fs, p.jobId)
|
||||
return &pdu.HintMostRecentCommonAncestorRes{}, nil
|
||||
}
|
||||
|
||||
// we were hinted a specific common ancestor
|
||||
|
||||
mostRecentVersion, err := sendArgsFromPDUAndValidateExistsAndGetVersion(ctx, fs, r.GetSenderVersion())
|
||||
if err != nil {
|
||||
msg := "HintMostRecentCommonAncestor rpc with nonexistent most recent version"
|
||||
getLogger(ctx).WithField("fs", fs).WithField("hinted_most_recent", fmt.Sprintf("%#v", mostRecent)).
|
||||
Warn(msg)
|
||||
log.Warn(msg)
|
||||
return nil, errors.Wrap(err, msg)
|
||||
}
|
||||
|
||||
// move replication cursor to this position
|
||||
_, err = MoveReplicationCursor(ctx, fs, mostRecent, p.jobId)
|
||||
destroyedCursors, err := MoveReplicationCursor(ctx, fs, mostRecentVersion, p.jobId)
|
||||
if err == zfs.ErrBookmarkCloningNotSupported {
|
||||
getLogger(ctx).Debug("not creating replication cursor from bookmark because ZFS does not support it")
|
||||
log.Debug("not creating replication cursor from bookmark because ZFS does not support it")
|
||||
// fallthrough
|
||||
} else if err != nil {
|
||||
return nil, errors.Wrap(err, "cannot set replication cursor to hinted version")
|
||||
}
|
||||
|
||||
// cleanup previous steps
|
||||
if err := ReleaseStepAll(ctx, fs, mostRecent, p.jobId); err != nil {
|
||||
return nil, errors.Wrap(err, "cannot cleanup prior invocation's step holds and bookmarks")
|
||||
// take care of stale step holds
|
||||
log.WithField("step-holds-cleanup-mode", senderHintMostRecentCommonAncestorStepCleanupMode).
|
||||
Debug("taking care of possibly stale step holds")
|
||||
doStepCleanup := false
|
||||
var stepCleanupSince *CreateTXGRangeBound
|
||||
switch senderHintMostRecentCommonAncestorStepCleanupMode {
|
||||
case StepCleanupNoCleanup:
|
||||
doStepCleanup = false
|
||||
case StepCleanupRangeSinceUnbounded:
|
||||
doStepCleanup = true
|
||||
stepCleanupSince = nil
|
||||
case StepCleanupRangeSinceReplicationCursor:
|
||||
doStepCleanup = true
|
||||
// Use the destroyed replication cursors as indicator how far the previous replication got.
|
||||
// To be precise: We limit the amount of visisted snapshots to exactly those snapshots
|
||||
// created since the last successful replication cursor movement (i.e. last successful replication step)
|
||||
//
|
||||
// If we crash now, we'll leak the step we are about to release, but the performance gain
|
||||
// of limiting the amount of snapshots we visit makes up for that.
|
||||
// Users have the `zrepl holds release-stale` command to cleanup leaked step holds.
|
||||
for _, destroyed := range destroyedCursors {
|
||||
if stepCleanupSince == nil {
|
||||
stepCleanupSince = &CreateTXGRangeBound{
|
||||
CreateTXG: destroyed.GetCreateTXG(),
|
||||
Inclusive: &zfs.NilBool{B: true},
|
||||
}
|
||||
} else if destroyed.GetCreateTXG() < stepCleanupSince.CreateTXG {
|
||||
stepCleanupSince.CreateTXG = destroyed.GetCreateTXG()
|
||||
}
|
||||
}
|
||||
default:
|
||||
panic(senderHintMostRecentCommonAncestorStepCleanupMode)
|
||||
}
|
||||
if !doStepCleanup {
|
||||
log.Info("skipping cleanup of prior invocations' step holds due to environment variable setting")
|
||||
} else {
|
||||
if err := ReleaseStepCummulativeInclusive(ctx, fs, stepCleanupSince, mostRecentVersion, p.jobId); err != nil {
|
||||
return nil, errors.Wrap(err, "cannot cleanup prior invocation's step holds and bookmarks")
|
||||
} else {
|
||||
log.Info("step hold cleanup done")
|
||||
}
|
||||
}
|
||||
|
||||
return &pdu.HintMostRecentCommonAncestorRes{}, nil
|
||||
}
|
||||
|
||||
type HintMostRecentCommonAncestorStepCleanupMode struct{ string }
|
||||
|
||||
var (
|
||||
StepCleanupRangeSinceReplicationCursor = HintMostRecentCommonAncestorStepCleanupMode{"range-since-replication-cursor"}
|
||||
StepCleanupRangeSinceUnbounded = HintMostRecentCommonAncestorStepCleanupMode{"range-since-unbounded"}
|
||||
StepCleanupNoCleanup = HintMostRecentCommonAncestorStepCleanupMode{"no-cleanup"}
|
||||
)
|
||||
|
||||
func (m HintMostRecentCommonAncestorStepCleanupMode) String() string { return string(m.string) }
|
||||
func (m *HintMostRecentCommonAncestorStepCleanupMode) Set(s string) error {
|
||||
switch s {
|
||||
case StepCleanupRangeSinceReplicationCursor.String():
|
||||
*m = StepCleanupRangeSinceReplicationCursor
|
||||
case StepCleanupRangeSinceUnbounded.String():
|
||||
*m = StepCleanupRangeSinceUnbounded
|
||||
case StepCleanupNoCleanup.String():
|
||||
*m = StepCleanupNoCleanup
|
||||
default:
|
||||
return fmt.Errorf("unknown step cleanup mode %q", s)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
var senderHintMostRecentCommonAncestorStepCleanupMode = *envconst.Var("ZREPL_ENDPOINT_SENDER_HINT_MOST_RECENT_STEP_HOLD_CLEANUP_MODE", &StepCleanupRangeSinceReplicationCursor).(*HintMostRecentCommonAncestorStepCleanupMode)
|
||||
|
||||
var maxConcurrentZFSSendSemaphore = semaphore.New(envconst.Int64("ZREPL_ENDPOINT_MAX_CONCURRENT_SEND", 10))
|
||||
|
||||
func uncheckedSendArgsFromPDU(fsv *pdu.FilesystemVersion) *zfs.ZFSSendArgVersion {
|
||||
@@ -147,15 +229,16 @@ func uncheckedSendArgsFromPDU(fsv *pdu.FilesystemVersion) *zfs.ZFSSendArgVersion
|
||||
return &zfs.ZFSSendArgVersion{RelName: fsv.GetRelName(), GUID: fsv.Guid}
|
||||
}
|
||||
|
||||
func sendArgsFromPDUAndValidateExists(ctx context.Context, fs string, fsv *pdu.FilesystemVersion) (*zfs.ZFSSendArgVersion, error) {
|
||||
v := uncheckedSendArgsFromPDU(fsv)
|
||||
if v == nil {
|
||||
return nil, errors.New("must not be nil")
|
||||
func sendArgsFromPDUAndValidateExistsAndGetVersion(ctx context.Context, fs string, fsv *pdu.FilesystemVersion) (v zfs.FilesystemVersion, err error) {
|
||||
sendArgs := uncheckedSendArgsFromPDU(fsv)
|
||||
if sendArgs == nil {
|
||||
return v, errors.New("must not be nil")
|
||||
}
|
||||
if err := v.ValidateExists(ctx, fs); err != nil {
|
||||
return nil, err
|
||||
version, err := sendArgs.ValidateExistsAndGetVersion(ctx, fs)
|
||||
if err != nil {
|
||||
return v, err
|
||||
}
|
||||
return v, nil
|
||||
return version, nil
|
||||
}
|
||||
|
||||
func (s *Sender) Send(ctx context.Context, r *pdu.SendReq) (*pdu.SendRes, zfs.StreamCopier, error) {
|
||||
@@ -182,7 +265,7 @@ func (s *Sender) Send(ctx context.Context, r *pdu.SendReq) (*pdu.SendRes, zfs.St
|
||||
return nil, nil, fmt.Errorf("unknown pdu.Tri variant %q", r.Encrypted)
|
||||
}
|
||||
|
||||
sendArgs := zfs.ZFSSendArgs{
|
||||
sendArgsUnvalidated := zfs.ZFSSendArgsUnvalidated{
|
||||
FS: r.Filesystem,
|
||||
From: uncheckedSendArgsFromPDU(r.GetFrom()), // validated by zfs.ZFSSendDry / zfs.ZFSSend
|
||||
To: uncheckedSendArgsFromPDU(r.GetTo()), // validated by zfs.ZFSSendDry / zfs.ZFSSend
|
||||
@@ -190,6 +273,11 @@ func (s *Sender) Send(ctx context.Context, r *pdu.SendReq) (*pdu.SendRes, zfs.St
|
||||
ResumeToken: r.ResumeToken, // nil or not nil, depending on decoding success
|
||||
}
|
||||
|
||||
sendArgs, err := sendArgsUnvalidated.Validate(ctx)
|
||||
if err != nil {
|
||||
return nil, nil, errors.Wrap(err, "validate send arguments")
|
||||
}
|
||||
|
||||
getLogger(ctx).Debug("acquire concurrent send semaphore")
|
||||
// TODO use try-acquire and fail with resource-exhaustion rpc status
|
||||
// => would require handling on the client-side
|
||||
@@ -224,7 +312,7 @@ func (s *Sender) Send(ctx context.Context, r *pdu.SendReq) (*pdu.SendRes, zfs.St
|
||||
// update replication cursor
|
||||
if sendArgs.From != nil {
|
||||
// For all but the first replication, this should always be a no-op because SendCompleted already moved the cursor
|
||||
_, err = MoveReplicationCursor(ctx, sendArgs.FS, sendArgs.From, s.jobId)
|
||||
_, err = MoveReplicationCursor(ctx, sendArgs.FS, sendArgs.FromVersion, s.jobId)
|
||||
if err == zfs.ErrBookmarkCloningNotSupported {
|
||||
getLogger(ctx).Debug("not creating replication cursor from bookmark because ZFS does not support it")
|
||||
// fallthrough
|
||||
@@ -235,18 +323,18 @@ func (s *Sender) Send(ctx context.Context, r *pdu.SendReq) (*pdu.SendRes, zfs.St
|
||||
|
||||
// make sure `From` doesn't go away in order to make this step resumable
|
||||
if sendArgs.From != nil {
|
||||
err := HoldStep(ctx, sendArgs.FS, sendArgs.From, s.jobId)
|
||||
_, err := HoldStep(ctx, sendArgs.FS, *sendArgs.FromVersion, s.jobId)
|
||||
if err == zfs.ErrBookmarkCloningNotSupported {
|
||||
getLogger(ctx).Debug("not creating step bookmark because ZFS does not support it")
|
||||
// fallthrough
|
||||
} else if err != nil {
|
||||
return nil, nil, errors.Wrap(err, "cannot create step bookmark")
|
||||
return nil, nil, errors.Wrapf(err, "cannot hold `from` version %q before starting send", *sendArgs.FromVersion)
|
||||
}
|
||||
}
|
||||
// make sure `To` doesn't go away in order to make this step resumable
|
||||
err = HoldStep(ctx, sendArgs.FS, sendArgs.To, s.jobId)
|
||||
_, err = HoldStep(ctx, sendArgs.FS, sendArgs.ToVersion, s.jobId)
|
||||
if err != nil {
|
||||
return nil, nil, errors.Wrapf(err, "cannot hold `to` version %q before starting send", sendArgs.To.RelName)
|
||||
return nil, nil, errors.Wrapf(err, "cannot hold `to` version %q before starting send", sendArgs.ToVersion)
|
||||
}
|
||||
|
||||
// step holds & replication cursor released / moved forward in s.SendCompleted => s.moveCursorAndReleaseSendHolds
|
||||
@@ -259,27 +347,32 @@ func (s *Sender) Send(ctx context.Context, r *pdu.SendReq) (*pdu.SendRes, zfs.St
|
||||
}
|
||||
|
||||
func (p *Sender) SendCompleted(ctx context.Context, r *pdu.SendCompletedReq) (*pdu.SendCompletedRes, error) {
|
||||
orig := r.GetOriginalReq() // may be nil, always use proto getters
|
||||
fs := orig.GetFilesystem()
|
||||
|
||||
var err error
|
||||
var from *zfs.ZFSSendArgVersion
|
||||
orig := r.GetOriginalReq() // may be nil, always use proto getters
|
||||
fsp, err := p.filterCheckFS(orig.GetFilesystem())
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
fs := fsp.ToString()
|
||||
|
||||
var from *zfs.FilesystemVersion
|
||||
if orig.GetFrom() != nil {
|
||||
from, err = sendArgsFromPDUAndValidateExists(ctx, fs, orig.GetFrom()) // no shadow
|
||||
f, err := sendArgsFromPDUAndValidateExistsAndGetVersion(ctx, fs, orig.GetFrom()) // no shadow
|
||||
if err != nil {
|
||||
return nil, errors.Wrap(err, "validate `from` exists")
|
||||
}
|
||||
from = &f
|
||||
}
|
||||
to, err := sendArgsFromPDUAndValidateExists(ctx, fs, orig.GetTo())
|
||||
to, err := sendArgsFromPDUAndValidateExistsAndGetVersion(ctx, fs, orig.GetTo())
|
||||
if err != nil {
|
||||
return nil, errors.Wrap(err, "validate `to` exists")
|
||||
}
|
||||
|
||||
log := getLogger(ctx).WithField("to_guid", to.GUID).
|
||||
log := getLogger(ctx).WithField("to_guid", to.Guid).
|
||||
WithField("fs", fs).
|
||||
WithField("to", to.RelName)
|
||||
if from != nil {
|
||||
log = log.WithField("from", from.RelName).WithField("from_guid", from.GUID)
|
||||
log = log.WithField("from", from.RelName).WithField("from_guid", from.Guid)
|
||||
}
|
||||
|
||||
log.Debug("move replication cursor to most recent common version")
|
||||
@@ -320,18 +413,14 @@ func (p *Sender) SendCompleted(ctx context.Context, r *pdu.SendCompletedReq) (*p
|
||||
return
|
||||
}
|
||||
log.Debug("release step-hold of or step-bookmark on `from`")
|
||||
err := ReleaseStep(ctx, fs, from, p.jobId)
|
||||
err := ReleaseStep(ctx, fs, *from, p.jobId)
|
||||
if err != nil {
|
||||
if dne, ok := err.(*zfs.DatasetDoesNotExist); ok {
|
||||
// If bookmark cloning is not supported, `from` might be the old replication cursor
|
||||
// and thus have already been destroyed by MoveReplicationCursor above
|
||||
// In that case, nonexistence of `from` is not an error, otherwise it is.
|
||||
fsp, err := zfs.NewDatasetPath(fs)
|
||||
if err != nil {
|
||||
panic(err) // fs has been validated multiple times above
|
||||
}
|
||||
for _, fsv := range destroyedCursors {
|
||||
if fsv.ToAbsPath(fsp) == dne.Path {
|
||||
for _, c := range destroyedCursors {
|
||||
if c.GetFullPath() == dne.Path {
|
||||
log.Info("`from` was a replication cursor and has already been destroyed")
|
||||
return
|
||||
}
|
||||
@@ -557,8 +646,9 @@ func (s *Receiver) ListFilesystemVersions(ctx context.Context, req *pdu.ListFile
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
// TODO share following code with sender
|
||||
|
||||
fsvs, err := zfs.ZFSListFilesystemVersions(lp, nil)
|
||||
fsvs, err := zfs.ZFSListFilesystemVersions(lp, zfs.ListFilesystemVersionsOptions{})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
@@ -610,9 +700,6 @@ func (s *Receiver) Receive(ctx context.Context, req *pdu.ReceiveReq, receive zfs
|
||||
if to == nil {
|
||||
return nil, errors.New("`To` must not be nil")
|
||||
}
|
||||
if err := to.ValidateInMemory(lp.ToString()); err != nil {
|
||||
return nil, errors.Wrap(err, "`To` invalid")
|
||||
}
|
||||
if !to.IsSnapshot() {
|
||||
return nil, errors.New("`To` must be a snapshot")
|
||||
}
|
||||
@@ -725,7 +812,8 @@ func (s *Receiver) Receive(ctx context.Context, req *pdu.ReceiveReq, receive zfs
|
||||
}
|
||||
|
||||
// validate that we actually received what the sender claimed
|
||||
if err := to.ValidateExists(ctx, lp.ToString()); err != nil {
|
||||
toRecvd, err := to.ValidateExistsAndGetVersion(ctx, lp.ToString())
|
||||
if err != nil {
|
||||
msg := "receive request's `To` version does not match what we received in the stream"
|
||||
getLogger(ctx).WithError(err).WithField("snap", snapFullPath).Error(msg)
|
||||
getLogger(ctx).Error("aborting recv request, but keeping received snapshot for inspection")
|
||||
@@ -734,7 +822,7 @@ func (s *Receiver) Receive(ctx context.Context, req *pdu.ReceiveReq, receive zfs
|
||||
|
||||
if s.conf.UpdateLastReceivedHold {
|
||||
getLogger(ctx).Debug("move last-received-hold")
|
||||
if err := MoveLastReceivedHold(ctx, lp.ToString(), *to, s.conf.JobID); err != nil {
|
||||
if err := MoveLastReceivedHold(ctx, lp.ToString(), toRecvd, s.conf.JobID); err != nil {
|
||||
return nil, errors.Wrap(err, "cannot move last-received-hold")
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user