From 4fd369b67fdbdf5fb9dee298806742169251d6b7 Mon Sep 17 00:00:00 2001 From: Christian Schwarz Date: Mon, 6 Apr 2020 01:04:58 +0200 Subject: [PATCH] consider replication cursor when determining stale step-holds and bookmarks --- client/hold_create_step_hold.go | 59 +++++++++++ client/holds.go | 1 + client/holds_create.go | 14 +++ endpoint/endpoint.go | 4 +- endpoint/endpoint_zfs_abstraction.go | 116 +++++++++++++++------- endpoint/endpoint_zfs_abstraction_step.go | 27 +++-- 6 files changed, 173 insertions(+), 48 deletions(-) create mode 100644 client/hold_create_step_hold.go create mode 100644 client/holds_create.go diff --git a/client/hold_create_step_hold.go b/client/hold_create_step_hold.go new file mode 100644 index 0000000..8052c1a --- /dev/null +++ b/client/hold_create_step_hold.go @@ -0,0 +1,59 @@ +package client + +import ( + "context" + "fmt" + + "github.com/pkg/errors" + "github.com/spf13/pflag" + "github.com/zrepl/zrepl/cli" + "github.com/zrepl/zrepl/endpoint" + "github.com/zrepl/zrepl/zfs" +) + +var holdsCreateStepHoldFlags struct { + target string + jobid JobIDFlag +} + +var holdsCmdCreateStepHold = &cli.Subcommand{ + Use: "step", + Run: doHoldsCreateStep, + NoRequireConfig: true, + Short: `create a step hold or bookmark`, + SetupFlags: func(f *pflag.FlagSet) { + f.StringVarP(&holdsCreateStepHoldFlags.target, "target", "t", "", "snapshot to be held / bookmark to be held") + f.VarP(&holdsCreateStepHoldFlags.jobid, "jobid", "j", "jobid for which the hold is installed") + }, +} + +func doHoldsCreateStep(sc *cli.Subcommand, args []string) error { + if len(args) > 0 { + return errors.New("subcommand takes no arguments") + } + + f := &holdsCreateStepHoldFlags + + fs, _, _, err := zfs.DecomposeVersionString(f.target) + if err != nil { + return errors.Wrapf(err, "%q invalid target", f.target) + } + + if f.jobid.FlagValue() == nil { + return errors.Errorf("jobid must be set") + } + + ctx := context.Background() + + v, err := zfs.ZFSGetFilesystemVersion(ctx, f.target) + if err != nil { + return errors.Wrapf(err, "get info about target %q", f.target) + } + + step, err := endpoint.HoldStep(ctx, fs, v, *f.jobid.FlagValue()) + if err != nil { + return errors.Wrap(err, "create step hold") + } + fmt.Println(step.String()) + return nil +} diff --git a/client/holds.go b/client/holds.go index 548b11d..e0d182d 100644 --- a/client/holds.go +++ b/client/holds.go @@ -22,6 +22,7 @@ var ( holdsCmdList, holdsCmdReleaseAll, holdsCmdReleaseStale, + holdsCmdCreate, } }, } diff --git a/client/holds_create.go b/client/holds_create.go new file mode 100644 index 0000000..0bc619c --- /dev/null +++ b/client/holds_create.go @@ -0,0 +1,14 @@ +package client + +import "github.com/zrepl/zrepl/cli" + +var holdsCmdCreate = &cli.Subcommand{ + Use: "create", + NoRequireConfig: true, + Short: `create zrepl-managed holds and boomkmarks (for debugging & development only!)`, + SetupSubcommands: func() []*cli.Subcommand { + return []*cli.Subcommand{ + holdsCmdCreateStepHold, + } + }, +} diff --git a/endpoint/endpoint.go b/endpoint/endpoint.go index 414051e..62cd9e8 100644 --- a/endpoint/endpoint.go +++ b/endpoint/endpoint.go @@ -323,7 +323,7 @@ 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.FromVersion, 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 @@ -332,7 +332,7 @@ func (s *Sender) Send(ctx context.Context, r *pdu.SendReq) (*pdu.SendRes, zfs.St } } // make sure `To` doesn't go away in order to make this step resumable - err = HoldStep(ctx, sendArgs.FS, sendArgs.ToVersion, 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.ToVersion) } diff --git a/endpoint/endpoint_zfs_abstraction.go b/endpoint/endpoint_zfs_abstraction.go index d29a0d6..4809344 100644 --- a/endpoint/endpoint_zfs_abstraction.go +++ b/endpoint/endpoint_zfs_abstraction.go @@ -77,25 +77,6 @@ func (t AbstractionType) MustValidate() error { return nil } -// Number of instances of this abstraction type that are live (not stale) -// per (FS,JobID). -1 for infinity. -func (t AbstractionType) NumLivePerFsAndJob() int { - switch t { - case AbstractionStepBookmark: - return 2 - case AbstractionStepHold: - return 2 - case AbstractionLastReceivedHold: - return 1 - case AbstractionReplicationCursorBookmarkV1: - return -1 - case AbstractionReplicationCursorBookmarkV2: - return 1 - default: - panic(t) - } -} - type AbstractionJSON struct{ Abstraction } var _ json.Marshaler = (*AbstractionJSON)(nil) @@ -138,6 +119,15 @@ func AbstractionTypeSetFromStrings(sts []string) (AbstractionTypeSet, error) { return ats, nil } +func (s AbstractionTypeSet) Contains(q AbstractionTypeSet) bool { + for k := range q { + if _, ok := s[k]; !ok { + return false + } + } + return true +} + func (s AbstractionTypeSet) String() string { sts := make([]string, 0, len(s)) for i := range s { @@ -652,18 +642,22 @@ func BatchDestroy(ctx context.Context, abs []Abstraction) <-chan BatchDestroyRes type StalenessInfo struct { ConstructedWithQuery ListZFSHoldsAndBookmarksQuery - All []Abstraction Live []Abstraction Stale []Abstraction } +type fsAndJobId struct { + fs string + jobId JobID +} + func ListStale(ctx context.Context, q ListZFSHoldsAndBookmarksQuery) (*StalenessInfo, error) { if !q.CreateTXG.IsUnbounded() { // we must determine the most recent step per FS, can't allow that return nil, errors.New("ListStale cannot have Until != nil set on query") } - abs, absErr, err := ListAbstractions(ctx, q) + qAbs, absErr, err := ListAbstractions(ctx, q) if err != nil { return nil, err } @@ -671,22 +665,53 @@ func ListStale(ctx context.Context, q ListZFSHoldsAndBookmarksQuery) (*Staleness // can't go on here because we can't determine the most recent step return nil, ListAbstractionsErrors(absErr) } - si := listStaleFiltering(abs) + + buildMostRecentFrom := qAbs + if !q.What[AbstractionReplicationCursorBookmarkV2] { + cq := q + cq.What = AbstractionTypeSet{AbstractionReplicationCursorBookmarkV2: true} + buildMostRecentFrom, absErr, err = ListAbstractions(ctx, cq) // no shadow + if err != nil { + return nil, err + } + if len(absErr) > 0 { + // can't go on here because we can't determine the most recent step + return nil, ListAbstractionsErrors(absErr) + } + } + mostRecentCursorByFS := make(map[fsAndJobId]*Abstraction) // empty map => will always return nil + for _, a := range buildMostRecentFrom { + if a.GetType() != AbstractionReplicationCursorBookmarkV2 { + continue + } + key := fsAndJobId{a.GetFS(), *a.GetJobID()} + mra := mostRecentCursorByFS[key] + if mra == nil || (*mra).GetCreateTXG() < a.GetCreateTXG() { + a := a + mra = &a + } + mostRecentCursorByFS[key] = mra + } + + si := listStaleFiltering(qAbs, mostRecentCursorByFS) si.ConstructedWithQuery = q return si, nil } -// The last AbstractionType.NumLive() step holds per (FS,Job,AbstractionType) are live -// others are stale. +type fsAjobAtype struct { + fsAndJobId + Type AbstractionType +} + +// For step holds and bookmarks, only those older than the most recent replication cursor +// of their (filesystem,job) is considered because younger ones cannot be stale by definition +// (if we destroy them, we might actually lose the hold on the `To` for an ongoing incremental replication) +// +// For replication cursors and last-received-holds, only the most recent one is kept. // // the returned StalenessInfo.ConstructedWithQuery is not set -func listStaleFiltering(abs []Abstraction) *StalenessInfo { +func listStaleFiltering(abs []Abstraction, mostRecentCursorByFS map[fsAndJobId]*Abstraction) *StalenessInfo { - type fsAjobAtype struct { - FS string - Job JobID - Type AbstractionType - } var noJobId []Abstraction by := make(map[fsAjobAtype][]Abstraction) for _, a := range abs { @@ -694,31 +719,46 @@ func listStaleFiltering(abs []Abstraction) *StalenessInfo { noJobId = append(noJobId, a) continue } - faj := fsAjobAtype{a.GetFS(), *a.GetJobID(), a.GetType()} + faj := fsAjobAtype{fsAndJobId{a.GetFS(), *a.GetJobID()}, a.GetType()} l := by[faj] l = append(l, a) by[faj] = l } ret := &StalenessInfo{ - All: abs, Live: noJobId, Stale: []Abstraction{}, } - // sort descending (highest createtxg first), then cut off for k := range by { l := by[k] + // sort descending (highest createtxg first), then cut off sort.Slice(l, func(i, j int) bool { return l[i].GetCreateTXG() > l[j].GetCreateTXG() }) - cutoff := k.Type.NumLivePerFsAndJob() - if cutoff == -1 || len(l) <= cutoff { - ret.Live = append(ret.Live, l...) + if k.Type == AbstractionStepHold || k.Type == AbstractionStepBookmark { + // all older than the most recent cursor are stale, others are always live + mostRecent := mostRecentCursorByFS[k.fsAndJobId] + if mostRecent == nil { + ret.Live = append(ret.Live, l...) + } else { + for _, a := range l { + if a.GetCreateTXG() > (*mostRecent).GetCreateTXG() { + ret.Live = append(ret.Live, a) + } else { + ret.Stale = append(ret.Stale, a) + } + } + } + } else if k.Type == AbstractionReplicationCursorBookmarkV2 || k.Type == AbstractionLastReceivedHold { + // all but the most recent are stale by definition (we always _move_ them) + if len(l) > 0 { + ret.Live = append(ret.Live, l[0]) + ret.Stale = append(ret.Stale, l[1:]...) + } } else { - ret.Live = append(ret.Live, l[0:cutoff]...) - ret.Stale = append(ret.Stale, l[cutoff:]...) + ret.Live = append(ret.Live, l...) } } diff --git a/endpoint/endpoint_zfs_abstraction_step.go b/endpoint/endpoint_zfs_abstraction_step.go index caafab1..61e57dd 100644 --- a/endpoint/endpoint_zfs_abstraction_step.go +++ b/endpoint/endpoint_zfs_abstraction_step.go @@ -63,19 +63,25 @@ func ParseStepBookmarkName(fullname string) (guid uint64, jobID JobID, err error // idempotently hold / step-bookmark `version` // // returns ErrBookmarkCloningNotSupported if version is a bookmark and bookmarking bookmarks is not supported by ZFS -func HoldStep(ctx context.Context, fs string, v zfs.FilesystemVersion, jobID JobID) error { +func HoldStep(ctx context.Context, fs string, v zfs.FilesystemVersion, jobID JobID) (Abstraction, error) { if v.IsSnapshot() { tag, err := StepHoldTag(jobID) if err != nil { - return errors.Wrap(err, "step hold tag") + return nil, errors.Wrap(err, "step hold tag") } if err := zfs.ZFSHold(ctx, fs, v, tag); err != nil { - return errors.Wrap(err, "step hold: zfs") + return nil, errors.Wrap(err, "step hold: zfs") } - return nil + return &ListHoldsAndBookmarksOutputHold{ + Type: AbstractionStepHold, + FS: fs, + Tag: tag, + JobID: jobID, + FilesystemVersion: v, + }, nil } if !v.IsBookmark() { @@ -84,7 +90,7 @@ func HoldStep(ctx context.Context, fs string, v zfs.FilesystemVersion, jobID Job bmname, err := StepBookmarkName(fs, v.Guid, jobID) if err != nil { - return errors.Wrap(err, "create step bookmark: determine bookmark name") + return nil, errors.Wrap(err, "create step bookmark: determine bookmark name") } // idempotently create bookmark err = zfs.ZFSBookmark(ctx, fs, v.ToSendArgVersion(), bmname) @@ -95,11 +101,16 @@ func HoldStep(ctx context.Context, fs string, v zfs.FilesystemVersion, jobID Job // is most likely not going to be successful. Also, there's the possibility that // the caller might want to filter what snapshots are eligibile, and this would // complicate things even further. - return err // TODO go1.13 use wrapping + return nil, err // TODO go1.13 use wrapping } - return errors.Wrap(err, "create step bookmark: zfs") + return nil, errors.Wrap(err, "create step bookmark: zfs") } - return nil + return &ListHoldsAndBookmarksOutputBookmark{ + Type: AbstractionStepBookmark, + FS: fs, + FilesystemVersion: v, + JobID: jobID, + }, nil } // idempotently release the step-hold on v if v is a snapshot