WIP fix racy holds

fixes #280

README.md

@@ -1,8 +1,9 @@
 [![GitHub license](https://img.shields.io/github/license/zrepl/zrepl.svg)](https://github.com/zrepl/zrepl/blob/master/LICENSE)
 [![Language: Go](https://img.shields.io/badge/language-Go-6ad7e5.svg)](https://golang.org/)
 [![User Docs](https://img.shields.io/badge/docs-web-blue.svg)](https://zrepl.github.io)
-[![Donate via PayPal](https://img.shields.io/badge/donate-paypal-yellow.svg)](https://www.paypal.com/cgi-bin/webscr?cmd=_s-xclick&hosted_button_id=R5QSXJVYHGX96)
+[![Donate via Patreon](https://img.shields.io/endpoint.svg?url=https%3A%2F%2Fshieldsio-patreon.herokuapp.com%2Fzrepl%2Fpledges&style=flat&color=yellow)](https://www.patreon.com/zrepl)
 [![Donate via Liberapay](https://img.shields.io/liberapay/receives/zrepl.svg?logo=liberapay)](https://liberapay.com/zrepl/donate)
+[![Donate via PayPal](https://img.shields.io/badge/donate-paypal-yellow.svg)](https://www.paypal.com/cgi-bin/webscr?cmd=_s-xclick&hosted_button_id=R5QSXJVYHGX96)
 [![Twitter](https://img.shields.io/twitter/url/https/github.com/zrepl/zrepl.svg?style=social)](https://twitter.com/intent/tweet?text=Wow:&url=https%3A%2F%2Fgithub.com%2Fzrepl%2Fzrepl)
 
 # zrepl
@@ -23,6 +24,7 @@ zrepl is a one-stop ZFS backup & replication solution.
    If so, think of an expressive configuration example.
 2. Think of at least one use case that generalizes from your concrete application.
 3. Open an issue on GitHub with example conf & use case attached.
+4. **Optional**: [Post a bounty](https://www.bountysource.com/teams/zrepl) on the issue, or [contact Christian Schwarz](https://cschwarz.com) for contract work.
 
 The above does not apply if you already implemented everything.
 Check out the *Coding Workflow* section below for details.
@@ -46,11 +48,8 @@ zrepl is written in [Go](https://golang.org) and uses [Go modules](https://githu
 The documentation is written in [ReStructured Text](http://docutils.sourceforge.net/rst.html) using the [Sphinx](https://www.sphinx-doc.org) framework.
 
 To get started, run `./lazy.sh devsetup` to easily install build dependencies and read `docs/installation.rst -> Compiling from Source`.
-
-### Overall Architecture
-
-The application architecture is documented as part of the user docs in the *Implementation* section (`docs/content/impl`).
-Make sure to develop an understanding how zrepl is typically used by studying the user docs first.
+`lazy.sh` uses `python3-pip` to fetch the build dependencies for the docs - you might want to use a [venv](https://docs.python.org/3/library/venv.html).
+If you just want to install the Go dependencies, run `./lazy.sh godep`.
 
 ### Project Structure
 
@@ -62,14 +61,15 @@ Make sure to develop an understanding how zrepl is typically used by studying th
 │   └── samples
 ├── daemon                  # the implementation of `zrepl daemon` subcommand
 │   ├── filters
+│   ├── hooks               # snapshot hooks
 │   ├── job                 # job implementations
 │   ├── logging             # logging outlets + formatters
 │   ├── nethelpers
 │   ├── prometheus
 │   ├── pruner              # pruner implementation
 │   ├── snapper             # snapshotter implementation
-├── docs                    # sphinx-based documentation
 ├── dist                    # supplemental material for users & package maintainers
+├── docs                    # sphinx-based documentation
 │   ├── **/*.rst            # documentation in reStructuredText
 │   ├── sphinxconf
 │   │   └── conf.py         # sphinx config (see commit 445a280 why its not in docs/)
@@ -78,6 +78,7 @@ Make sure to develop an understanding how zrepl is typically used by studying th
 │   └── public_git          # checkout of zrepl.github.io managed by above shell script
 ├── endpoint                # implementation of replication endpoints (=> package replication)
 ├── logger                  # our own logger package
+├── platformtest            # test suite for our zfs abstractions (error classification, etc)
 ├── pruning                 # pruning rules (the logic, not the actual execution)
 │   └── retentiongrid
 ├── replication
@@ -92,14 +93,13 @@ Make sure to develop an understanding how zrepl is typically used by studying th
 │   ├── transportmux        # TCP connecter and listener used to split control & data traffic
 │   └── versionhandshake    # replication protocol version handshake perfomed on newly established connections
 ├── tlsconf                 # abstraction for Go TLS server + client config
-├── transport               # transports implementation
+├── transport               # transport implementations
 │   ├── fromconfig
 │   ├── local
 │   ├── ssh
 │   ├── tcp
 │   └── tls
 ├── util
-├── vendor                  # managed by dep
 ├── version                 # abstraction for versions (filled during build by Makefile)
 └── zfs                     # zfs(8) wrappers
 ```
@@ -134,3 +134,15 @@ There will not be a big refactoring (an attempt was made, but it's destroying to
 
 However, new contributions & patches should fix naming without further notice in the commit message.
 
+### RPC debugging
+
+Optionally, there are various RPC-related environment varibles, that if set to something != `""` will produce additional debug output on stderr:
+
+https://github.com/zrepl/zrepl/blob/master/rpc/rpc_debug.go#L11
+
+https://github.com/zrepl/zrepl/blob/master/rpc/dataconn/dataconn_debug.go#L11
+
+https://github.com/zrepl/zrepl/blob/master/rpc/dataconn/stream/stream_debug.go#L11
+
+https://github.com/zrepl/zrepl/blob/master/rpc/dataconn/heartbeatconn/heartbeatconn_debug.go#L11
+

client/testcmd.go

@@ -60,6 +60,8 @@ func runTestFilterCmd(subcommand *cli.Subcommand, args []string) error {
 		confFilter = j.Filesystems
 	case *config.PushJob:
 		confFilter = j.Filesystems
+	case *config.SnapJob:
+		confFilter = j.Filesystems
 	default:
 		return fmt.Errorf("job type %T does not have filesystems filter", j)
 	}

docs/installation.rst

@@ -101,9 +101,9 @@ and serves as a reference for build dependencies and procedure:
 
 ::
 
-    git clone https://github.com/zrepl/zrepl.git
-    cd zrepl
-    sudo docker build -t zrepl_build -f build.Dockerfile .
+    git clone https://github.com/zrepl/zrepl.git && \
+    cd zrepl && \
+    sudo docker build -t zrepl_build -f build.Dockerfile . && \
     sudo docker run -it --rm \
         -v "${PWD}:/src" \
         --user "$(id -u):$(id -g)" \

draft_snapmgmt.md

@@ -1,129 +0,0 @@
-# Tiered Snapshotting, Pruning & Replication
-
-## Use Case
-
-* Differently paced snapshotting & replication
-    * 20 x 10min-spaced snapshots for rollback in case of adminstrative mistakes.
-        * Should never be replicated
-    * Daily Snapshots for off-site replication
-        * at 3:00 AM to avoid impacting production workload
-
-* Sometimes, out-of-routine snapshot for off-site replication
-    * e.g. before system update
-
-
-## Config Draft
-
-```yaml
-- type: push
-  connect:
-    type: tcp
-    address: "backup-server.foo.bar:8888"
-  filesystems: {
-    "<": true,
-    "tmp": false
-  }
-  send:
-    snap_filters:
-    - type: snapmgmt # auto-add this snap filter if snapmgmt != {} 
-      # every snapmgmt method produces a list of snaps_filters which this snap_filter type represents
-    - type: regex
-      regex: "manual_.*" # admin may create snapshots named like `manual_pre_upgrade` and have those replicated as well
-      # TODO: how does a manual `zfs snapshot` trigger replication? `zrepl signal wakeup ...` good enough for now
-  snapmgmt:
-    type: tiered
-    prefix: zrepl_
-    tiers:
-    - name: local
-      cron: every 10 minutes
-      keep: { local: 10, remote: 0 }
-    - name: daily
-      cron: every day at 3 AM
-      keep: { local: 14, remote: 30 }
-    - name: weekly
-      cron: every sunday at 3 AM
-      keep: { local: 0, remote: 20 }
-    - name: monthly
-      cron: every last sunday of the month at 3 AM
-      keep: { local: 0, remote: 12 }
-```
-
-TODO pull - likely breaks existing split of responsibilities about pruning between `pull` and `source`
-
-## Implementation
-
-A `snapmgmt` type implements both snapshotting and *local* pruning in one service (`RunSnapshotterAndLocalPruner()`)
-The `snapmgmt` type `tiered` works as follows:
-
-* Snapshots are created per tier:
-  * snapshot name = `${PREFIX}_${JOBNAME}_TIERED_${thistier.name}_${NOW}`
-  * property `zrepl:JOBNAME:tier=${thistier.name}`
-  * property `zrepl:JOBNAME:destroy_at=${thistier.destroy_at(NOW)}`
-* Sender-side pruning by
-  * filtering all snaps by userprop `zrepl:JOBNAME:tier=${thistier.name}`
-  * destroying all snaps in the filtered list with `(zrepl:JOBNAME:destroy_at).Before(NOW)`
-  * sleeping until MIN of
-    * `MIN(filtered_list.zrepl:JOBNAME:destroy_at)`
-    * `(<-newSnapshots).DestroyAt)`
-
-The replication enigne remains orthogonal to `snapmgmt`, but the protocol allows a sender to present a filtered view of snapshots in the `ListFilesystemVersions` RPC response.
-The filters are specified in a list, and have OR semantics, i.e., if one filter matches, the snapshot is presented tothe replication engine.
-The filter type supporting the feature proposed in this issue is the `tiered` filter type:
-it includes snapshots that are
-* named by the snapshotter above: `${PREFIX}_${JOBNAME}_${snapmgmt.tiers|.name}_${NOW}`
-* and the respective tier has a non-zero `keep.remote`
-* TODO: deriving the tier from the snapshot name is unclean.
-  We could instead allow that a replication filter is not pure (i.e. may use zfs.ZFSGet).
-  Would work for both `push` and `source` jobs.
-
-Receiver-side pruning is still unsolved (TODO):
-
-* We could replicate the `destroy_at` property, and have independent pruning on the receive-side.
-    * While there are cases where this is desirable (use case "guarantee that replicated data is destroyed after X months")
-    * it might be undesirable in cases where the receive-side is a backup (which should not self-destroy)
-* We could require each `snapmgmt` method to have a `PruneReceiver(receiver)` method, and require that `receiver` provides an RPC endpoint to get the `destroy_at` infromation required to evaluate the prune policy (e.g. a flag in the request, like `GetDestroyAt=true`)
-
-## Other future `snapmgmt` methods
-
-```yaml
-snapmgmt:
-  # zrepl signal ondemand JOBNAME SNAPNAME EXPIRE_AT
-  # => creates snapshot @SNAPNAME for all filesystems matched by `$JOB.filesystems` with expiration date `EXPIRE_AT`
-  # => snapshot has user prop `zrepl:JOBNAME:ondemand=on`
-  # replication filter `snapmgmt` filters by `ondemand` property
-  # pruning filter uses expieration data encoded in property
-  type: ondemand
-
-snapmgmt:
-  # no snapshots, never-matching filter
-  type: manual
-```
-
-## Existing Use-Cases
-
-### Existing Configurations & Migration to New Config
-
-`snapmgmt` replaces the previous `snapshotting` and `pruning` fields.
-Proposed migration for existing configs:
-
-```yaml
-snapmgmt:
-  type: pre-snapmgmt
-  snapshotting:
-    type: periodic
-    interval: 10m
-    prefix: zrepl_
-  pruning:
-    keep_sender:
-      - type: not_replicated
-      - type: last_n
-        count: 10
-      - type: grid
-        grid: 1x1h(keep=all) | 24x1h | 14x1d
-        regex: "^zrepl_.*"
-    keep_receiver:
-      - type: grid
-        grid: 1x1h(keep=all) | 24x1h | 35x1d | 6x30d
-        regex: "^zrepl_.*"
-```
-

endpoint/endpoint_zfs.go

@@ -411,6 +411,7 @@ func ListZFSHoldsAndBookmarks(ctx context.Context, fsfilter zfs.DatasetFilter) (
 	for _, fs := range fss {
 		err := listZFSHoldsAndBookmarksImplFS(ctx, out, fs)
 		if err != nil {
+			// FIXME if _, ok := err.(*zfs.DatasetDoesNotExist); ok { noop }
 			return nil, errors.Wrapf(err, "list holds and bookmarks on %q", fs.ToString())
 		}
 	}
@@ -420,7 +421,8 @@ func ListZFSHoldsAndBookmarks(ctx context.Context, fsfilter zfs.DatasetFilter) (
 func listZFSHoldsAndBookmarksImplFS(ctx context.Context, out *ListHoldsAndBookmarksOutput, fs *zfs.DatasetPath) error {
 	fsvs, err := zfs.ZFSListFilesystemVersions(fs, nil)
 	if err != nil {
-		return errors.Wrapf(err, "list filesystem versions of %q", fs)
+		// FIXME if _, ok := err.(*zfs.DatasetDoesNotExist); ok { noop }
+		return errors.Wrapf(err, "list filesystem versions of %q", fs.ToString())
 	}
 	for _, v := range fsvs {
 		switch v.Type {
@@ -429,7 +431,12 @@ func listZFSHoldsAndBookmarksImplFS(ctx context.Context, out *ListHoldsAndBookma
 		case zfs.Snapshot:
 			holds, err := zfs.ZFSHolds(ctx, fs.ToString(), v.Name)
 			if err != nil {
-				return errors.Wrapf(err, "get holds of %q", v.ToAbsPath(fs))
+				if _, ok := err.(*zfs.DatasetDoesNotExist); ok {
+					holds  = []string{}
+					// fallthrough
+				} else {
+					return errors.Wrapf(err, "get holds of %q", v.ToAbsPath(fs))
+				}
 			}
 			for _, tag := range holds {
 				listZFSHoldsAndBookmarksImplSnapshotTryParseHold(ctx, out, fs, v, tag)

platformtest/tests/idempotentHold.go

@@ -3,6 +3,7 @@ package tests
 import (
 	"fmt"
 
+	"github.com/stretchr/testify/require"
 	"github.com/zrepl/zrepl/platformtest"
 	"github.com/zrepl/zrepl/zfs"
 )
@@ -13,35 +14,28 @@ func IdempotentHold(ctx *platformtest.Context) {
 		DESTROYROOT
 		CREATEROOT
 		+  "foo bar"
-		+  "foo bar@1"
-	`)
-	defer platformtest.Run(ctx, platformtest.PanicErr, ctx.RootDataset, `
-		R zfs release zrepl_platformtest "${ROOTDS}/foo bar@1"
-		-  "foo bar@1"
-		-  "foo bar"
+		+  "foo bar@1 2"
 	`)
 
 	fs := fmt.Sprintf("%s/foo bar", ctx.RootDataset)
-	v1 := sendArgVersion(fs, "@1")
+	snap := sendArgVersion(fs, "@1 2")
 
 	tag := "zrepl_platformtest"
-	err := zfs.ZFSHold(ctx, fs, v1, tag)
+	err := zfs.ZFSHold(ctx, fs, snap, tag)
 	if err != nil {
 		panic(err)
 	}
 
-	err = zfs.ZFSHold(ctx, fs, v1, tag)
-	if err != nil {
-		panic(err)
-	}
-
-	vnonexistent := zfs.ZFSSendArgVersion{
-		RelName: "@nonexistent",
-		GUID:    0xbadf00d,
-	}
-	err = zfs.ZFSHold(ctx, fs, vnonexistent, tag)
-	if err == nil {
-		panic("still expecting error for nonexistent snapshot")
-	}
+	// existing holds
+	holds, err := zfs.ZFSHolds(ctx, fs, "1 2")
+	require.NoError(ctx, err)
+	require.Equal(ctx, []string{tag}, holds)
 
+	holds, err = zfs.ZFSHolds(ctx, fs, "non existent")
+	ctx.Logf("holds=%v", holds)
+	ctx.Logf("errT=%T", err)
+	ctx.Logf("err=%s", err)
+	notExist, ok := err.(*zfs.DatasetDoesNotExist)
+	require.True(ctx, ok)
+	require.Equal(ctx, fs+"@non existent", notExist.Path)
 }

platformtest/tests/listHolds.go

@@ -0,0 +1,47 @@
+package tests
+
+import (
+	"fmt"
+
+	"github.com/zrepl/zrepl/platformtest"
+	"github.com/zrepl/zrepl/zfs"
+)
+
+func ListHolds(ctx *platformtest.Context) {
+
+	platformtest.Run(ctx, platformtest.PanicErr, ctx.RootDataset, `
+		DESTROYROOT
+		CREATEROOT
+		+  "foo bar"
+		+  "foo bar@1"
+	`)
+	defer platformtest.Run(ctx, platformtest.PanicErr, ctx.RootDataset, `
+		R zfs release zrepl_platformtest "${ROOTDS}/foo bar@1"
+		-  "foo bar@1"
+		-  "foo bar"
+	`)
+
+	fs := fmt.Sprintf("%s/foo bar", ctx.RootDataset)
+	v1 := sendArgVersion(fs, "@1")
+
+	tag := "zrepl_platformtest"
+	err := zfs.ZFSHold(ctx, fs, v1, tag)
+	if err != nil {
+		panic(err)
+	}
+
+	err = zfs.ZFSHold(ctx, fs, v1, tag)
+	if err != nil {
+		panic(err)
+	}
+
+	vnonexistent := zfs.ZFSSendArgVersion{
+		RelName: "@nonexistent",
+		GUID:    0xbadf00d,
+	}
+	err = zfs.ZFSHold(ctx, fs, vnonexistent, tag)
+	if err == nil {
+		panic("still expecting error for nonexistent snapshot")
+	}
+
+}

platformtest/tests/tests.go

@@ -31,4 +31,5 @@ var Cases = []Case{
 	SendArgsValidationEncryptedSendOfUnencryptedDatasetForbidden,
 	SendArgsValidationResumeTokenEncryptionMismatchForbidden,
 	SendArgsValidationResumeTokenDifferentFilesystemForbidden,
+	ListHolds,
 }

replication/logic/replication_logic.go

@@ -262,7 +262,6 @@ func (p *Planner) doPlanning(ctx context.Context) ([]*Filesystem, error) {
 		return nil, err
 	}
 	sfss := slfssres.GetFilesystems()
-	// no progress here since we could run in a live-lock on connectivity issues
 
 	rlfssres, err := p.receiver.ListFilesystems(ctx, &pdu.ListFilesystemReq{})
 	if err != nil {

rpc/dataconn/timeoutconn/timeoutconn.go

@@ -49,7 +49,7 @@ type Wire interface {
 	// No data that could otherwise be Read is lost as a consequence of this call.
 	// The use case for this API is abortive connection shutdown.
 	// To provide any value over draining Wire using io.Read, an implementation
-	// will likely use out-of-bounds messaging mechanisms.
+	// will likely use out-of-band messaging mechanisms.
 	// TODO WaitForPeerClose() (supported bool, err error)
 }
 

transport/tls/serve_tls.go

@@ -32,7 +32,7 @@ func TLSListenerFactoryFromConfig(c *config.Global, in *config.TLSServe) (transp
 
 	serverCert, err := tls.LoadX509KeyPair(in.Cert, in.Key)
 	if err != nil {
-		return nil, errors.Wrap(err, "cannot parse cer/key pair")
+		return nil, errors.Wrap(err, "cannot parse cert/key pair")
 	}
 
 	clientCNs := make(map[string]struct{}, len(in.ClientCNs))

zfs/holds.go

@@ -59,6 +59,7 @@ success:
 	return nil
 }
 
+// If the snapshot does not exist, the returned error is of type *DatasetDoesNotExist
 func ZFSHolds(ctx context.Context, fs, snap string) ([]string, error) {
 	if err := validateZFSFilesystem(fs); err != nil {
 		return nil, errors.Wrap(err, "`fs` is not a valid filesystem path")