Compare commits

..

33 Commits

Author SHA1 Message Date
Christian Schwarz 515ddb8307 endpoing: better log message for v1 replication cursors 2020-04-06 12:51:06 +02:00
Christian Schwarz f5f94219fd endpoint: fix ListStale for step-* abstractions by using replication cursors and falling back to step holds for initial replication 2020-04-06 12:50:11 +02:00
Christian Schwarz ac2eb9f86b zfs: fix half-nil error reporting of dataset-does-not-exist for ZFSListChan and ZFSBookmark 2020-04-06 12:48:35 +02:00
Christian Schwarz 7061282e08 endpoint: rename abstraction super-"classes" 2020-04-06 01:06:52 +02:00
Christian Schwarz 4fd369b67f consider replication cursor when determining stale step-holds and bookmarks 2020-04-06 01:04:58 +02:00
Christian Schwarz da8b168573 zfs changes fixup 2020-04-05 20:57:49 +02:00
Christian Schwarz 2c15db9cfe range bounds fixup 2020-04-05 20:57:34 +02:00
Christian Schwarz 1fc4e62128 Merge branch 'problame/zfs-command-logging-and-status' into problame/holds-release-and-hold-leak-fix-v2 2020-04-05 20:51:35 +02:00
Christian Schwarz a02b4c19e9 endpoint: step hold cleanup based on replication cursor bookmark 2020-04-05 20:15:49 +02:00
Christian Schwarz 4f255cef09 zfs changes platformtest 2020-04-05 20:14:38 +02:00
Christian Schwarz 05852bde7a fixup e82aea5862: gofmt 2020-04-05 20:14:11 +02:00
Christian Schwarz 4e0574e7d4 envconst.Var 2020-04-05 20:12:32 +02:00
Christian Schwarz 8cab6e95ad endpoint abstr: use userrefs to only issue zfs holds if there is at least one hold 2020-04-05 20:10:37 +02:00
Christian Schwarz 568a112519 fixup zfs changes 2020-04-05 20:01:30 +02:00
Christian Schwarz f70393378f endpoint: zfs abstraction: use new api (fixup) 2020-04-05 19:58:59 +02:00
Christian Schwarz b16a9ede46 zfs: userrefs, platformtests for ListFilesystemVersions and ListMapping (likely needs fixup from next commit) 2020-04-05 19:19:56 +02:00
Christian Schwarz e82aea5862 range-based createtxg queries 2020-04-05 14:55:11 +02:00
Christian Schwarz 86698c2789 FIXUP typo in environment variable 2020-03-30 12:19:40 +02:00
Christian Schwarz 8c27e57edf endpoint: environment variable for disabling cleanup of stale step holds on HintMostRecentCommonAncestor 2020-03-30 11:39:37 +02:00
Christian Schwarz 4ecbb32044 remove outdated FIXME 2020-03-30 00:57:20 +02:00
Christian Schwarz d09ca2447a Merge branch 'problame/zfs-command-logging-and-status' into problame/holds-release-and-hold-leak-fix-v2 2020-03-29 19:07:18 +02:00
Christian Schwarz bab424021a zfs: introduce pkg zfs/zfscmd for command logging, status, prometheus metrics
refs #196
2020-03-29 19:06:25 +02:00
Christian Schwarz bc291e622f Merge branch 'problame/zfs-command-logging-and-status' into problame/holds-release-and-hold-leak-fix-v2 2020-03-29 19:04:21 +02:00
Christian Schwarz deeca76aa0 zfs: introduce pkg zfs/zfscmd for command logging, status, prometheus metrics
refs #196
2020-03-29 19:02:41 +02:00
Christian Schwarz 8bfaba1665 WIP SQUASH MERGE Merge branch 'problame/zfs-command-logging-and-status' into problame/holds-release-and-hold-leak-fix-v2 2020-03-29 14:08:44 +02:00
Christian Schwarz 815b432063 zfs: introduce pkg zfs/zfscmd for command logging, status, prometheus metrics
refs #196
2020-03-29 14:01:05 +02:00
Christian Schwarz 0b44e25c8b endpoint: concurrent queries 2020-03-27 20:17:48 +01:00
Christian Schwarz 8755847a40 fix move replication cursor 2020-03-27 17:36:38 +01:00
Christian Schwarz 586a4ffac8 SQUASH THIS MERGE branch 'problame/zfs-command-logging-and-status' into problame/holds-release-and-hold-leak-fix-v2 2020-03-27 17:16:29 +01:00
Christian Schwarz 4e13ea46de zfs: introduce pkg zfs/zfscmd for command logging, status, prometheus metrics
refs #196
2020-03-27 17:00:03 +01:00
Christian Schwarz 47d7bba192 build: go1.14 + address tlsconf deprecation notice
fixes #286
2020-03-27 00:29:16 +01:00
Christian Schwarz 4a4ec4d9b4 endpoint: add filesystem path validation for HintMostRecentCommonAncestor and SendCompleted RPCs 2020-03-27 00:12:36 +01:00
Christian Schwarz f3734ed0d4 endpoint: refactor, fix stale holds on initial replication, holds release subcmds
- endpoint abstractions now share an interface `Abstraction`
- pkg 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
- the `zrepl holds list` command consumes endpoint.ListAbstractions
- the new `zrepl holds release-{all,stale}` commands can be used
  to remove abstractions of package endpoint

Co-authored-by: InsanePrawn <insane.prawny@gmail.com>

supersedes #282
fixes #280
fixes #278
2020-03-27 00:12:29 +01:00
18 changed files with 129 additions and 248 deletions
+3 -8
View File
@@ -65,7 +65,6 @@ wrapup-and-checksum:
-acf $(NOARCH_TARBALL) \
$(ARTIFACTDIR)/docs/html \
$(ARTIFACTDIR)/bash_completion \
$(ARTIFACTDIR)/_zrepl.zsh_completion \
$(ARTIFACTDIR)/go_env.txt \
dist \
config/samples
@@ -160,7 +159,7 @@ platformtest: # do not track dependency on platformtest-bin to allow build of pl
$(ZREPL_PLATFORMTEST_ARGS)
##################### NOARCH #####################
.PHONY: noarch $(ARTIFACTDIR)/bash_completion $(ARTIFACTDIR)/_zrepl.zsh_completion $(ARTIFACTDIR)/go_env.txt docs docs-clean
.PHONY: noarch $(ARTIFACTDIR)/bash_completion $(ARTIFACTDIR)/go_env.txt docs docs-clean
$(ARTIFACTDIR):
@@ -168,16 +167,12 @@ $(ARTIFACTDIR):
$(ARTIFACTDIR)/docs: $(ARTIFACTDIR)
mkdir -p "$@"
noarch: $(ARTIFACTDIR)/bash_completion $(ARTIFACTDIR)/_zrepl.zsh_completion $(ARTIFACTDIR)/go_env.txt docs
noarch: $(ARTIFACTDIR)/bash_completion $(ARTIFACTDIR)/go_env.txt docs
# pass
$(ARTIFACTDIR)/bash_completion:
$(MAKE) zrepl-bin GOOS=$(GOHOSTOS) GOARCH=$(GOHOSTARCH)
artifacts/zrepl-$(GOHOSTOS)-$(GOHOSTARCH) gencompletion bash "$@"
$(ARTIFACTDIR)/_zrepl.zsh_completion:
$(MAKE) zrepl-bin GOOS=$(GOHOSTOS) GOARCH=$(GOHOSTARCH)
artifacts/zrepl-$(GOHOSTOS)-$(GOHOSTARCH) gencompletion zsh "$@"
artifacts/zrepl-$(GOHOSTOS)-$(GOHOSTARCH) bashcomp "$@"
$(ARTIFACTDIR)/go_env.txt:
$(GO_ENV_VARS) $(GO) env > $@
+11 -37
View File
@@ -19,38 +19,9 @@ var rootCmd = &cobra.Command{
Short: "One-stop ZFS replication solution",
}
func init() {
rootCmd.PersistentFlags().StringVar(&rootArgs.configPath, "config", "", "config file path")
}
var genCompletionCmd = &cobra.Command{
Use: "gencompletion",
Short: "generate shell auto-completions",
}
type completionCmdInfo struct {
genFunc func(outpath string) error
help string
}
var completionCmdMap = map[string]completionCmdInfo{
"zsh": {
rootCmd.GenZshCompletionFile,
" save to file `_zrepl` in your zsh's $fpath",
},
"bash": {
rootCmd.GenBashCompletionFile,
" save to a path and source that path in your .bashrc",
},
}
func init() {
for sh, info := range completionCmdMap {
sh, info := sh, info
genCompletionCmd.AddCommand(&cobra.Command{
Use: fmt.Sprintf("%s path/to/out/file", sh),
Short: fmt.Sprintf("generate %s completions", sh),
Example: info.help,
var bashcompCmd = &cobra.Command{
Use: "bashcomp path/to/out/file",
Short: "generate bash completions",
Run: func(cmd *cobra.Command, args []string) {
if len(args) != 1 {
fmt.Fprintf(os.Stderr, "specify exactly one positional agument\n")
@@ -60,14 +31,17 @@ func init() {
}
os.Exit(1)
}
if err := info.genFunc(args[0]); err != nil {
fmt.Fprintf(os.Stderr, "error generating %s completion: %s", sh, err)
if err := rootCmd.GenBashCompletionFile(args[0]); err != nil {
fmt.Fprintf(os.Stderr, "error generating bash completion: %s", err)
os.Exit(1)
}
},
})
}
rootCmd.AddCommand(genCompletionCmd)
Hidden: true,
}
func init() {
rootCmd.PersistentFlags().StringVar(&rootArgs.configPath, "config", "", "config file path")
rootCmd.AddCommand(bashcompCmd)
}
type Subcommand struct {
@@ -12,28 +12,28 @@ import (
"github.com/zrepl/zrepl/zfs"
)
var zabsCreateStepHoldFlags struct {
var holdsCreateStepHoldFlags struct {
target string
jobid JobIDFlag
}
var zabsCmdCreateStepHold = &cli.Subcommand{
var holdsCmdCreateStepHold = &cli.Subcommand{
Use: "step",
Run: doZabsCreateStep,
Run: doHoldsCreateStep,
NoRequireConfig: true,
Short: `create a step hold or bookmark`,
SetupFlags: func(f *pflag.FlagSet) {
f.StringVarP(&zabsCreateStepHoldFlags.target, "target", "t", "", "snapshot to be held / bookmark to be held")
f.VarP(&zabsCreateStepHoldFlags.jobid, "jobid", "j", "jobid for which the hold is installed")
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 doZabsCreateStep(sc *cli.Subcommand, args []string) error {
func doHoldsCreateStep(sc *cli.Subcommand, args []string) error {
if len(args) > 0 {
return errors.New("subcommand takes no arguments")
}
f := &zabsCreateStepHoldFlags
f := &holdsCreateStepHoldFlags
fs, _, _, err := zfs.DecomposeVersionString(f.target)
if err != nil {
+10 -10
View File
@@ -14,15 +14,15 @@ import (
)
var (
ZFSAbstractionsCmd = &cli.Subcommand{
Use: "zfs-abstraction",
Short: "manage abstractions that zrepl builds on top of ZFS",
HoldsCmd = &cli.Subcommand{
Use: "holds",
Short: "manage holds & step bookmarks",
SetupSubcommands: func() []*cli.Subcommand {
return []*cli.Subcommand{
zabsCmdList,
zabsCmdReleaseAll,
zabsCmdReleaseStale,
zabsCmdCreate,
holdsCmdList,
holdsCmdReleaseAll,
holdsCmdReleaseStale,
holdsCmdCreate,
}
},
}
@@ -30,7 +30,7 @@ var (
// a common set of CLI flags that map to the fields of an
// endpoint.ListZFSHoldsAndBookmarksQuery
type zabsFilterFlags struct {
type holdsFilterFlags struct {
Filesystems FilesystemsFilterFlag
Job JobIDFlag
Types AbstractionTypesFlag
@@ -38,7 +38,7 @@ type zabsFilterFlags struct {
}
// produce a query from the CLI flags
func (f zabsFilterFlags) Query() (endpoint.ListZFSHoldsAndBookmarksQuery, error) {
func (f holdsFilterFlags) Query() (endpoint.ListZFSHoldsAndBookmarksQuery, error) {
q := endpoint.ListZFSHoldsAndBookmarksQuery{
FS: f.Filesystems.FlagValue(),
What: f.Types.FlagValue(),
@@ -48,7 +48,7 @@ func (f zabsFilterFlags) Query() (endpoint.ListZFSHoldsAndBookmarksQuery, error)
return q, q.Validate()
}
func (f *zabsFilterFlags) registerZabsFilterFlags(s *pflag.FlagSet, verb string) {
func (f *holdsFilterFlags) registerHoldsFilterFlags(s *pflag.FlagSet, verb string) {
// Note: the default value is defined in the .FlagValue methods
s.Var(&f.Filesystems, "fs", fmt.Sprintf("only %s holds on the specified filesystem [default: all filesystems] [comma-separated list of <dataset-pattern>:<ok|!> pairs]", verb))
s.Var(&f.Job, "job", fmt.Sprintf("only %s holds created by the specified job [default: any job]", verb))
+14
View File
@@ -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,
}
},
}
@@ -16,23 +16,23 @@ import (
"github.com/zrepl/zrepl/util/chainlock"
)
var zabsListFlags struct {
Filter zabsFilterFlags
var holdsListFlags struct {
Filter holdsFilterFlags
Json bool
}
var zabsCmdList = &cli.Subcommand{
var holdsCmdList = &cli.Subcommand{
Use: "list",
Short: `list zrepl ZFS abstractions`,
Run: doZabsList,
Run: doHoldsList,
NoRequireConfig: true,
Short: "list holds and bookmarks",
SetupFlags: func(f *pflag.FlagSet) {
zabsListFlags.Filter.registerZabsFilterFlags(f, "list")
f.BoolVar(&zabsListFlags.Json, "json", false, "emit JSON")
holdsListFlags.Filter.registerHoldsFilterFlags(f, "list")
f.BoolVar(&holdsListFlags.Json, "json", false, "emit JSON")
},
}
func doZabsList(sc *cli.Subcommand, args []string) error {
func doHoldsList(sc *cli.Subcommand, args []string) error {
var err error
ctx := context.Background()
@@ -40,7 +40,7 @@ func doZabsList(sc *cli.Subcommand, args []string) error {
return errors.New("this subcommand takes no positional arguments")
}
q, err := zabsListFlags.Filter.Query()
q, err := holdsListFlags.Filter.Query()
if err != nil {
return errors.Wrap(err, "invalid filter specification on command line")
}
@@ -62,7 +62,7 @@ func doZabsList(sc *cli.Subcommand, args []string) error {
for a := range abstractions {
func() {
defer line.Lock().Unlock()
if zabsListFlags.Json {
if holdsListFlags.Json {
enc.SetIndent("", " ")
if err := enc.Encode(abstractions); err != nil {
panic(err)
@@ -15,35 +15,35 @@ import (
)
// shared between release-all and release-step
var zabsReleaseFlags struct {
Filter zabsFilterFlags
var holdsReleaseFlags struct {
Filter holdsFilterFlags
Json bool
DryRun bool
}
func registerZabsReleaseFlags(s *pflag.FlagSet) {
zabsReleaseFlags.Filter.registerZabsFilterFlags(s, "release")
s.BoolVar(&zabsReleaseFlags.Json, "json", false, "emit json instead of pretty-printed")
s.BoolVar(&zabsReleaseFlags.DryRun, "dry-run", false, "do a dry-run")
func registerHoldsReleaseFlags(s *pflag.FlagSet) {
holdsReleaseFlags.Filter.registerHoldsFilterFlags(s, "release")
s.BoolVar(&holdsReleaseFlags.Json, "json", false, "emit json instead of pretty-printed")
s.BoolVar(&holdsReleaseFlags.DryRun, "dry-run", false, "do a dry-run")
}
var zabsCmdReleaseAll = &cli.Subcommand{
var holdsCmdReleaseAll = &cli.Subcommand{
Use: "release-all",
Run: doZabsReleaseAll,
Run: doHoldsReleaseAll,
NoRequireConfig: true,
Short: `(DANGEROUS) release ALL zrepl ZFS abstractions (mostly useful for uninstalling zrepl completely or for "de-zrepl-ing" a filesystem)`,
SetupFlags: registerZabsReleaseFlags,
Short: `(DANGEROUS) release all zrepl-managed holds and bookmarks, mostly useful for uninstalling zrepl`,
SetupFlags: registerHoldsReleaseFlags,
}
var zabsCmdReleaseStale = &cli.Subcommand{
var holdsCmdReleaseStale = &cli.Subcommand{
Use: "release-stale",
Run: doZabsReleaseStale,
Run: doHoldsReleaseStale,
NoRequireConfig: true,
Short: `release stale zrepl ZFS abstractions (useful if zrepl has a bug and does not do it by itself)`,
SetupFlags: registerZabsReleaseFlags,
Short: `release stale zrepl-managed holds and boomkarks (useful if zrepl has a bug and doesn't do it by itself)`,
SetupFlags: registerHoldsReleaseFlags,
}
func doZabsReleaseAll(sc *cli.Subcommand, args []string) error {
func doHoldsReleaseAll(sc *cli.Subcommand, args []string) error {
var err error
ctx := context.Background()
@@ -51,7 +51,7 @@ func doZabsReleaseAll(sc *cli.Subcommand, args []string) error {
return errors.New("this subcommand takes no positional arguments")
}
q, err := zabsReleaseFlags.Filter.Query()
q, err := holdsReleaseFlags.Filter.Query()
if err != nil {
return errors.Wrap(err, "invalid filter specification on command line")
}
@@ -65,10 +65,10 @@ func doZabsReleaseAll(sc *cli.Subcommand, args []string) error {
// proceed anyways with rest of abstractions
}
return doZabsRelease_Common(ctx, abstractions)
return doHoldsRelease_Common(ctx, abstractions)
}
func doZabsReleaseStale(sc *cli.Subcommand, args []string) error {
func doHoldsReleaseStale(sc *cli.Subcommand, args []string) error {
var err error
ctx := context.Background()
@@ -77,7 +77,7 @@ func doZabsReleaseStale(sc *cli.Subcommand, args []string) error {
return errors.New("this subcommand takes no positional arguments")
}
q, err := zabsReleaseFlags.Filter.Query()
q, err := holdsReleaseFlags.Filter.Query()
if err != nil {
return errors.Wrap(err, "invalid filter specification on command line")
}
@@ -87,13 +87,13 @@ func doZabsReleaseStale(sc *cli.Subcommand, args []string) error {
return err // context clear by invocation of command
}
return doZabsRelease_Common(ctx, stalenessInfo.Stale)
return doHoldsRelease_Common(ctx, stalenessInfo.Stale)
}
func doZabsRelease_Common(ctx context.Context, destroy []endpoint.Abstraction) error {
func doHoldsRelease_Common(ctx context.Context, destroy []endpoint.Abstraction) error {
if zabsReleaseFlags.DryRun {
if zabsReleaseFlags.Json {
if holdsReleaseFlags.DryRun {
if holdsReleaseFlags.Json {
m, err := json.MarshalIndent(destroy, "", " ")
if err != nil {
panic(err)
@@ -121,7 +121,7 @@ func doZabsRelease_Common(ctx context.Context, destroy []endpoint.Abstraction) e
for res := range outcome {
hadErr = hadErr || res.DestroyErr != nil
if zabsReleaseFlags.Json {
if holdsReleaseFlags.Json {
err := enc.Encode(res)
if err != nil {
colorErr.Fprintf(os.Stderr, "cannot marshal there were errors in destroying the abstractions")
-14
View File
@@ -1,14 +0,0 @@
package client
import "github.com/zrepl/zrepl/cli"
var zabsCmdCreate = &cli.Subcommand{
Use: "create",
NoRequireConfig: true,
Short: `create zrepl ZFS abstractions (mostly useful for debugging & development, users should not need to use this command)`,
SetupSubcommands: func() []*cli.Subcommand {
return []*cli.Subcommand{
zabsCmdCreateStepHold,
}
},
}
-2
View File
@@ -62,9 +62,7 @@ Actual changelog:
* |bugfix| |docs| snapshotting: clarify sync-up behavior and warn about filesystems
that will not be snapshotted until the sync-up phase is over
* |docs| Document new replication features in the :ref:`config overview <overview-how-replication-works>` and :repomasterlink:`replication/design.md`.
* |feature| documented subcommand to generate ``bash`` and ``zsh`` completions
* **[MAINTAINER NOTICE]** New platform tests in this version, please make sure you run them for your distro!
* **[MAINTAINER NOTICE]** Please add the shell completions to the zrepl packages.
0.2.1
-----
+2 -2
View File
@@ -38,8 +38,8 @@ CLI Overview
* - ``zrepl migrate``
- | perform on-disk state / ZFS property migrations
| (see :ref:`changelog <changelog>` for details)
* - ``zrepl zfs-abstractions``
- list and remove zrepl's abstractions on top of ZFS, e.g. holds and step bookmarks (see :ref:`overview <replication-cursor-and-last-received-hold>` )
* - ``zrepl holds``
- list and remove holds and step bookmarks created by zrepl (see :ref:`overview <replication-cursor-and-last-received-hold>` )
.. _usage-zrepl-daemon:
+4 -4
View File
@@ -150,11 +150,11 @@ func (p *Sender) HintMostRecentCommonAncestor(ctx context.Context, r *pdu.HintMo
}
// take care of stale step holds
log.WithField("step-holds-cleanup-mode", senderHintMostRecentCommonAncestorStepCleanupMode).
log.WithField("step-holds-cleanup-mode", hintMostRecentCommonAncestorStepCleanupMode).
Debug("taking care of possibly stale step holds")
doStepCleanup := false
var stepCleanupSince *CreateTXGRangeBound
switch senderHintMostRecentCommonAncestorStepCleanupMode {
switch hintMostRecentCommonAncestorStepCleanupMode {
case StepCleanupNoCleanup:
doStepCleanup = false
case StepCleanupRangeSinceUnbounded:
@@ -180,7 +180,7 @@ func (p *Sender) HintMostRecentCommonAncestor(ctx context.Context, r *pdu.HintMo
}
}
default:
panic(senderHintMostRecentCommonAncestorStepCleanupMode)
panic(hintMostRecentCommonAncestorStepCleanupMode)
}
if !doStepCleanup {
log.Info("skipping cleanup of prior invocations' step holds due to environment variable setting")
@@ -218,7 +218,7 @@ func (m *HintMostRecentCommonAncestorStepCleanupMode) Set(s string) error {
return nil
}
var senderHintMostRecentCommonAncestorStepCleanupMode = *envconst.Var("ZREPL_ENDPOINT_SENDER_HINT_MOST_RECENT_STEP_HOLD_CLEANUP_MODE", &StepCleanupRangeSinceReplicationCursor).(*HintMostRecentCommonAncestorStepCleanupMode)
var hintMostRecentCommonAncestorStepCleanupMode = *envconst.Var("ZREPL_ENDPOINT_HINT_MOST_RECENT_STEP_HOLD_CLEANUP_MODE", &StepCleanupRangeSinceReplicationCursor).(*HintMostRecentCommonAncestorStepCleanupMode)
var maxConcurrentZFSSendSemaphore = semaphore.New(envconst.Int64("ZREPL_ENDPOINT_MAX_CONCURRENT_SEND", 10))
+1 -1
View File
@@ -17,7 +17,7 @@ func init() {
cli.AddSubcommand(client.PprofCmd)
cli.AddSubcommand(client.TestCmd)
cli.AddSubcommand(client.MigrateCmd)
cli.AddSubcommand(client.ZFSAbstractionsCmd)
cli.AddSubcommand(client.HoldsCmd)
}
func main() {
+1 -26
View File
@@ -18,32 +18,7 @@ Hence, when trying to map algorithm to implementation, use the code in package `
* the recv-side fs doesn't get newer snapshots than `to` in the meantime
* guaranteed because the zrepl model of the receiver assumes ownership of the filesystems it receives into
* if that assumption is broken, future replication attempts will fail with a conflict
* The [Algorithm for Planning and Executing Replication of an Filesystems](#zrepl-algo-filesystem) is a design draft and not used.
However, there were some noteworthy lessons learned when implementing the algorithm for a single step:
* In order to avoid leaking `step-hold`s and `step-bookmarks`, if the replication planner is invoked a second time after a replication step (either initial or incremental) has been attempted but failed to completed, the replication planner must
* A) either guarantee that it will resume that replication step, and continue as if nothing happened or
* B) release the step holds and bookmarks and clear the partially received state on the sending side.
* Option A is what we want to do: we use the step algorithm to achieve resumability in the first place!
* Option B is not done by zrepl except if the sending side doesn't support resuming.
In that case however, we need not release any holds since the behavior is to re-start the send
from the beginning.
* However, there is one **edge-case to Option A for initial replication**:
* If initial replication "`full a`" fails without leaving resumable state, the step holds on the sending side are still present, which makes sense because
* a) we want resumability and
* b) **the sending side cannot immediately be informed post-failure whether the initial replication left any state that would mandate keeping the step hold**, because the network connection might have failed.
* Thus, the sending side must keep the step hold for "`full a`" until it knows more.
* In the current implementation, it knows more when the next replication attempt is made, the planner is invoked, the diffing algorithm run, and the `HintMostRecentCommonAncestor` RPC is sent by the active side, communicating the most recent common version shared betwen sender and receiver.
* At this point, the sender can safely throw away any step holds with CreateTXG's older than that version.
* **The `step-hold`, `step-bookmark`, `last-received-hold` and `replication-cursor` abstractions are currently local concepts of package `endpoint` and not part of the replication protocol**
* This is not necessarilty the best design decision and should be revisited some point:
* The (quite expensive) `HintMostRecentCommonAncestor` RPC impl on the sender would not be necessary if step holds were part of the replication protocol:
* We only need the `HintMostRecentCommonAncestor` info for the aforementioned edge-case during initial replication, where the receive is aborted without any partial received state being stored on the receiver (due to network failure, wrong zfs invocation, bad permissions, etc):
* **The replication planner does not know about the step holds, thus it cannot deterministically pick up where it left of (right at the start of the last failing initial replication).**
* Instead, it will seem like no prior invocation happened at all, and it will apply its policy for initial replication to pick a new `full b != full a`, **thereby leaking the step holds of `full a`**.
* In contrast, if the replication planner created the step holds and knew about them, it could use the step holds as an indicator where it left off and re-start from there (of course asserting that the thereby inferred step is compatible with the state of the receiving side).
* (What we do in zrepl right now is to hard-code the initial replication policy, and hard-code that assumption in `endpoint.ListStale` as well.)
* The cummulative cleanup done in `HintMostRecentCommonAncestor` provides a nice self-healing aspect, though.
* The [Algorithm for Planning and Executing Replication of an Filesystems](#zrepl-algo-filesystem) is a design draft and not used
* We also have [Notes on Planning and Executing Replication of Multiple Filesystems](#zrepl-algo-multiple-filesystems-notes)
---
+27 -36
View File
@@ -46,7 +46,7 @@ func (x Tri) String() string {
return proto.EnumName(Tri_name, int32(x))
}
func (Tri) EnumDescriptor() ([]byte, []int) {
return fileDescriptor_pdu_e59763dc61674a79, []int{0}
return fileDescriptor_pdu_0f43b713cd3bf056, []int{0}
}
type FilesystemVersion_VersionType int32
@@ -69,7 +69,7 @@ func (x FilesystemVersion_VersionType) String() string {
return proto.EnumName(FilesystemVersion_VersionType_name, int32(x))
}
func (FilesystemVersion_VersionType) EnumDescriptor() ([]byte, []int) {
return fileDescriptor_pdu_e59763dc61674a79, []int{5, 0}
return fileDescriptor_pdu_0f43b713cd3bf056, []int{5, 0}
}
type ListFilesystemReq struct {
@@ -82,7 +82,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_e59763dc61674a79, []int{0}
return fileDescriptor_pdu_0f43b713cd3bf056, []int{0}
}
func (m *ListFilesystemReq) XXX_Unmarshal(b []byte) error {
return xxx_messageInfo_ListFilesystemReq.Unmarshal(m, b)
@@ -113,7 +113,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_e59763dc61674a79, []int{1}
return fileDescriptor_pdu_0f43b713cd3bf056, []int{1}
}
func (m *ListFilesystemRes) XXX_Unmarshal(b []byte) error {
return xxx_messageInfo_ListFilesystemRes.Unmarshal(m, b)
@@ -154,7 +154,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_e59763dc61674a79, []int{2}
return fileDescriptor_pdu_0f43b713cd3bf056, []int{2}
}
func (m *Filesystem) XXX_Unmarshal(b []byte) error {
return xxx_messageInfo_Filesystem.Unmarshal(m, b)
@@ -213,7 +213,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_e59763dc61674a79, []int{3}
return fileDescriptor_pdu_0f43b713cd3bf056, []int{3}
}
func (m *ListFilesystemVersionsReq) XXX_Unmarshal(b []byte) error {
return xxx_messageInfo_ListFilesystemVersionsReq.Unmarshal(m, b)
@@ -251,7 +251,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_e59763dc61674a79, []int{4}
return fileDescriptor_pdu_0f43b713cd3bf056, []int{4}
}
func (m *ListFilesystemVersionsRes) XXX_Unmarshal(b []byte) error {
return xxx_messageInfo_ListFilesystemVersionsRes.Unmarshal(m, b)
@@ -293,7 +293,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_e59763dc61674a79, []int{5}
return fileDescriptor_pdu_0f43b713cd3bf056, []int{5}
}
func (m *FilesystemVersion) XXX_Unmarshal(b []byte) error {
return xxx_messageInfo_FilesystemVersion.Unmarshal(m, b)
@@ -371,7 +371,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_e59763dc61674a79, []int{6}
return fileDescriptor_pdu_0f43b713cd3bf056, []int{6}
}
func (m *SendReq) XXX_Unmarshal(b []byte) error {
return xxx_messageInfo_SendReq.Unmarshal(m, b)
@@ -445,7 +445,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_e59763dc61674a79, []int{7}
return fileDescriptor_pdu_0f43b713cd3bf056, []int{7}
}
func (m *Property) XXX_Unmarshal(b []byte) error {
return xxx_messageInfo_Property.Unmarshal(m, b)
@@ -496,7 +496,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_e59763dc61674a79, []int{8}
return fileDescriptor_pdu_0f43b713cd3bf056, []int{8}
}
func (m *SendRes) XXX_Unmarshal(b []byte) error {
return xxx_messageInfo_SendRes.Unmarshal(m, b)
@@ -548,7 +548,7 @@ func (m *SendCompletedReq) Reset() { *m = SendCompletedReq{} }
func (m *SendCompletedReq) String() string { return proto.CompactTextString(m) }
func (*SendCompletedReq) ProtoMessage() {}
func (*SendCompletedReq) Descriptor() ([]byte, []int) {
return fileDescriptor_pdu_e59763dc61674a79, []int{9}
return fileDescriptor_pdu_0f43b713cd3bf056, []int{9}
}
func (m *SendCompletedReq) XXX_Unmarshal(b []byte) error {
return xxx_messageInfo_SendCompletedReq.Unmarshal(m, b)
@@ -585,7 +585,7 @@ func (m *SendCompletedRes) Reset() { *m = SendCompletedRes{} }
func (m *SendCompletedRes) String() string { return proto.CompactTextString(m) }
func (*SendCompletedRes) ProtoMessage() {}
func (*SendCompletedRes) Descriptor() ([]byte, []int) {
return fileDescriptor_pdu_e59763dc61674a79, []int{10}
return fileDescriptor_pdu_0f43b713cd3bf056, []int{10}
}
func (m *SendCompletedRes) XXX_Unmarshal(b []byte) error {
return xxx_messageInfo_SendCompletedRes.Unmarshal(m, b)
@@ -608,7 +608,7 @@ var xxx_messageInfo_SendCompletedRes proto.InternalMessageInfo
type ReceiveReq struct {
Filesystem string `protobuf:"bytes,1,opt,name=Filesystem,proto3" json:"Filesystem,omitempty"`
To *FilesystemVersion `protobuf:"bytes,2,opt,name=To,proto3" json:"To,omitempty"`
// If true, the receiver should clear the resume token before performing the
// If true, the receiver should clear the resume token before perfoming the
// zfs recv of the stream in the request
ClearResumeToken bool `protobuf:"varint,3,opt,name=ClearResumeToken,proto3" json:"ClearResumeToken,omitempty"`
XXX_NoUnkeyedLiteral struct{} `json:"-"`
@@ -620,7 +620,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_e59763dc61674a79, []int{11}
return fileDescriptor_pdu_0f43b713cd3bf056, []int{11}
}
func (m *ReceiveReq) XXX_Unmarshal(b []byte) error {
return xxx_messageInfo_ReceiveReq.Unmarshal(m, b)
@@ -671,7 +671,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_e59763dc61674a79, []int{12}
return fileDescriptor_pdu_0f43b713cd3bf056, []int{12}
}
func (m *ReceiveRes) XXX_Unmarshal(b []byte) error {
return xxx_messageInfo_ReceiveRes.Unmarshal(m, b)
@@ -704,7 +704,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_e59763dc61674a79, []int{13}
return fileDescriptor_pdu_0f43b713cd3bf056, []int{13}
}
func (m *DestroySnapshotsReq) XXX_Unmarshal(b []byte) error {
return xxx_messageInfo_DestroySnapshotsReq.Unmarshal(m, b)
@@ -750,7 +750,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_e59763dc61674a79, []int{14}
return fileDescriptor_pdu_0f43b713cd3bf056, []int{14}
}
func (m *DestroySnapshotRes) XXX_Unmarshal(b []byte) error {
return xxx_messageInfo_DestroySnapshotRes.Unmarshal(m, b)
@@ -795,7 +795,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_e59763dc61674a79, []int{15}
return fileDescriptor_pdu_0f43b713cd3bf056, []int{15}
}
func (m *DestroySnapshotsRes) XXX_Unmarshal(b []byte) error {
return xxx_messageInfo_DestroySnapshotsRes.Unmarshal(m, b)
@@ -833,7 +833,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_e59763dc61674a79, []int{16}
return fileDescriptor_pdu_0f43b713cd3bf056, []int{16}
}
func (m *ReplicationCursorReq) XXX_Unmarshal(b []byte) error {
return xxx_messageInfo_ReplicationCursorReq.Unmarshal(m, b)
@@ -874,7 +874,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_e59763dc61674a79, []int{17}
return fileDescriptor_pdu_0f43b713cd3bf056, []int{17}
}
func (m *ReplicationCursorRes) XXX_Unmarshal(b []byte) error {
return xxx_messageInfo_ReplicationCursorRes.Unmarshal(m, b)
@@ -1010,7 +1010,7 @@ 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_e59763dc61674a79, []int{18}
return fileDescriptor_pdu_0f43b713cd3bf056, []int{18}
}
func (m *PingReq) XXX_Unmarshal(b []byte) error {
return xxx_messageInfo_PingReq.Unmarshal(m, b)
@@ -1049,7 +1049,7 @@ 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_e59763dc61674a79, []int{19}
return fileDescriptor_pdu_0f43b713cd3bf056, []int{19}
}
func (m *PingRes) XXX_Unmarshal(b []byte) error {
return xxx_messageInfo_PingRes.Unmarshal(m, b)
@@ -1078,15 +1078,6 @@ func (m *PingRes) GetEcho() string {
type HintMostRecentCommonAncestorReq struct {
Filesystem string `protobuf:"bytes,1,opt,name=Filesystem,proto3" json:"Filesystem,omitempty"`
// A copy of the FilesystemVersion on the sending side that the replication
// algorithm identified as a shared most recent common version between sending
// and receiving side.
//
// If nil, this is an indication that the replication algorithm could not
// find a common ancestor between the two sides.
// NOTE: nilness does not mean that replication never happened - there could
// as well be a replication conflict. thus, dont' jump to conclusions too
// rapidly here.
SenderVersion *FilesystemVersion `protobuf:"bytes,2,opt,name=SenderVersion,proto3" json:"SenderVersion,omitempty"`
XXX_NoUnkeyedLiteral struct{} `json:"-"`
XXX_unrecognized []byte `json:"-"`
@@ -1097,7 +1088,7 @@ func (m *HintMostRecentCommonAncestorReq) Reset() { *m = HintMostRecentC
func (m *HintMostRecentCommonAncestorReq) String() string { return proto.CompactTextString(m) }
func (*HintMostRecentCommonAncestorReq) ProtoMessage() {}
func (*HintMostRecentCommonAncestorReq) Descriptor() ([]byte, []int) {
return fileDescriptor_pdu_e59763dc61674a79, []int{20}
return fileDescriptor_pdu_0f43b713cd3bf056, []int{20}
}
func (m *HintMostRecentCommonAncestorReq) XXX_Unmarshal(b []byte) error {
return xxx_messageInfo_HintMostRecentCommonAncestorReq.Unmarshal(m, b)
@@ -1141,7 +1132,7 @@ func (m *HintMostRecentCommonAncestorRes) Reset() { *m = HintMostRecentC
func (m *HintMostRecentCommonAncestorRes) String() string { return proto.CompactTextString(m) }
func (*HintMostRecentCommonAncestorRes) ProtoMessage() {}
func (*HintMostRecentCommonAncestorRes) Descriptor() ([]byte, []int) {
return fileDescriptor_pdu_e59763dc61674a79, []int{21}
return fileDescriptor_pdu_0f43b713cd3bf056, []int{21}
}
func (m *HintMostRecentCommonAncestorRes) XXX_Unmarshal(b []byte) error {
return xxx_messageInfo_HintMostRecentCommonAncestorRes.Unmarshal(m, b)
@@ -1458,9 +1449,9 @@ var _Replication_serviceDesc = grpc.ServiceDesc{
Metadata: "pdu.proto",
}
func init() { proto.RegisterFile("pdu.proto", fileDescriptor_pdu_e59763dc61674a79) }
func init() { proto.RegisterFile("pdu.proto", fileDescriptor_pdu_0f43b713cd3bf056) }
var fileDescriptor_pdu_e59763dc61674a79 = []byte{
var fileDescriptor_pdu_0f43b713cd3bf056 = []byte{
// 892 bytes of a gzipped FileDescriptorProto
0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x8c, 0x56, 0xdf, 0x6f, 0xdb, 0x36,
0x10, 0x8e, 0x6c, 0x39, 0x91, 0xcf, 0xe9, 0xea, 0x5c, 0xb2, 0x42, 0x13, 0xba, 0xce, 0xe3, 0x86,
-10
View File
@@ -129,16 +129,6 @@ message PingRes {
message HintMostRecentCommonAncestorReq {
string Filesystem = 1;
// A copy of the FilesystemVersion on the sending side that the replication
// algorithm identified as a shared most recent common version between sending
// and receiving side.
//
// If nil, this is an indication that the replication algorithm could not
// find a common ancestor between the two sides.
// NOTE: nilness does not mean that replication never happened - there could
// as well be a replication conflict. thus, dont' jump to conclusions too
// rapidly here.
FilesystemVersion SenderVersion = 2;
}
message HintMostRecentCommonAncestorRes {}
+1 -1
View File
@@ -152,7 +152,7 @@ func (m *HandshakeMessage) DecodeReader(r io.Reader, maxLen int) error {
func DoHandshakeCurrentVersion(conn net.Conn, deadline time.Time) *HandshakeError {
// current protocol version is hardcoded here
return DoHandshakeVersion(conn, deadline, 3)
return DoHandshakeVersion(conn, deadline, 2)
}
const HandshakeMessageMaxLen = 16 * 4096
+3 -16
View File
@@ -1,7 +1,6 @@
package zfscmd
import (
"os/exec"
"time"
)
@@ -29,22 +28,10 @@ func waitPreLogging(c *Cmd, now time.Time) {
}
func waitPostLogging(c *Cmd, err error, now time.Time) {
var total, system, user float64
total = c.Runtime().Seconds()
if ee, ok := err.(*exec.ExitError); ok {
system = ee.ProcessState.SystemTime().Seconds()
user = ee.ProcessState.UserTime().Seconds()
} else {
system = -1
user = -1
}
log := c.log().
WithField("total_time_s", total).
WithField("systemtime_s", system).
WithField("usertime_s", user)
WithField("total_time_s", c.Runtime().Seconds()).
WithField("systemtime_s", c.cmd.ProcessState.SystemTime().Seconds()).
WithField("usertime_s", c.cmd.ProcessState.UserTime().Seconds())
if err == nil {
log.Info("command exited without error")
-29
View File
@@ -1,9 +1,7 @@
package zfscmd
import (
"bufio"
"bytes"
"context"
"io"
"os/exec"
"testing"
@@ -60,30 +58,3 @@ func TestCmdStderrBehaviorStdoutPipe(t *testing.T) {
require.True(t, ok)
require.Empty(t, ee.Stderr) // !!!!! probably not what one would expect if we only redirect stdout
}
func TestCmdProcessState(t *testing.T) {
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
cmd := exec.CommandContext(ctx, "bash", "-c", "echo running; sleep 3600")
stdout, err := cmd.StdoutPipe()
require.NoError(t, err)
err = cmd.Start()
require.NoError(t, err)
r := bufio.NewReader(stdout)
line, err := r.ReadString('\n')
require.NoError(t, err)
require.Equal(t, "running\n", line)
// we know it's running and sleeping
cancel()
err = cmd.Wait()
t.Logf("wait err %T\n%s", err, err)
require.Error(t, err)
ee, ok := err.(*exec.ExitError)
require.True(t, ok)
require.NotNil(t, ee.ProcessState)
require.Contains(t, ee.Error(), "killed")
}