WIP: draft snapmgmt

README.md

@@ -1,9 +1,8 @@
 [![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 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)
+[![Donate via Liberapay](https://img.shields.io/liberapay/receives/zrepl.svg?logo=liberapay)](https://liberapay.com/zrepl/donate)
 [![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
@@ -24,7 +23,6 @@ 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.
@@ -48,8 +46,11 @@ 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`.
-`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`.
+
+### 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.
 
 ### Project Structure
 
@@ -61,15 +62,14 @@ If you just want to install the Go dependencies, run `./lazy.sh godep`.
 │   └── 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
-├── dist                    # supplemental material for users & package maintainers
 ├── docs                    # sphinx-based documentation
+├── dist                    # supplemental material for users & package maintainers
 │   ├── **/*.rst            # documentation in reStructuredText
 │   ├── sphinxconf
 │   │   └── conf.py         # sphinx config (see commit 445a280 why its not in docs/)
@@ -78,7 +78,6 @@ If you just want to install the Go dependencies, run `./lazy.sh godep`.
 │   └── 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
@@ -93,13 +92,14 @@ If you just want to install the Go dependencies, run `./lazy.sh godep`.
 │   ├── 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               # transport implementations
+├── transport               # transports implementation
 │   ├── fromconfig
 │   ├── local
 │   ├── ssh
 │   ├── tcp
 │   └── tls
 ├── util
+├── vendor                  # managed by dep
 ├── version                 # abstraction for versions (filled during build by Makefile)
 └── zfs                     # zfs(8) wrappers
 ```
@@ -134,15 +134,3 @@ 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,8 +60,6 @@ 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)
 	}

daemon/control.go

@@ -147,8 +147,8 @@ func (j *controlJob) Run(ctx context.Context) {
 	server := http.Server{
 		Handler: mux,
 		// control socket is local, 1s timeout should be more than sufficient, even on a loaded system
-		WriteTimeout: envconst.Duration("ZREPL_DAEMON_CONTROL_WRITE_TIMEOUT", 1*time.Second),
-		ReadTimeout:  envconst.Duration("ZREPL_DAEMON_CONTROL_READ_TIMEOUT", 1*time.Second),
+		WriteTimeout: 1 * time.Second,
+		ReadTimeout:  1 * time.Second,
 	}
 
 outer:

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

@@ -0,0 +1,129 @@
+# 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.go

@@ -198,11 +198,7 @@ func (s *Sender) Send(ctx context.Context, r *pdu.SendReq) (*pdu.SendRes, zfs.St
 	if err != nil {
 		return nil, nil, err
 	}
-	defer func(guardp **semaphore.AcquireGuard) {
-		if *guardp != nil {
-			(*guardp).Release()
-		}
-	}(&guard)
+	defer guard.Release()
 
 	si, err := zfs.ZFSSendDry(ctx, sendArgs)
 	if err != nil {
@@ -259,13 +255,6 @@ func (s *Sender) Send(ctx context.Context, r *pdu.SendReq) (*pdu.SendRes, zfs.St
 	if err != nil {
 		return nil, nil, errors.Wrap(err, "zfs send failed")
 	}
-
-	// defer releasing guard until streamCopier is closed
-	streamCopier.SetPostCloseCallback(func(_ error) {
-		guard.Release()
-	})
-	guard = nil
-
 	return res, streamCopier, nil
 }
 

replication/logic/replication_logic.go

@@ -262,6 +262,7 @@ 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-band messaging mechanisms.
+	// will likely use out-of-bounds 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 cert/key pair")
+		return nil, errors.Wrap(err, "cannot parse cer/key pair")
 	}
 
 	clientCNs := make(map[string]struct{}, len(in.ClientCNs))

zfs/zfs.go

@@ -350,8 +350,7 @@ func (a ZFSSendArgs) buildCommonSendArgs() ([]string, error) {
 }
 
 type ReadCloserCopier struct {
-	recorder          readErrRecorder
-	postCloseCallback func(closeErr error)
+	recorder readErrRecorder
 }
 
 type readErrRecorder struct {
@@ -403,17 +402,8 @@ func (c *ReadCloserCopier) Read(p []byte) (n int, err error) {
 	return c.recorder.Read(p)
 }
 
-// caller must ensure that this function is not executing concurrently to Close
-func (c *ReadCloserCopier) SetPostCloseCallback(callback func(closeErr error)) {
-	c.postCloseCallback = callback
-}
-
 func (c *ReadCloserCopier) Close() error {
-	err := c.recorder.Close()
-	if c.postCloseCallback != nil {
-		c.postCloseCallback(err)
-	}
-	return err
+	return c.recorder.ReadCloser.Close()
 }
 
 func pipeWithCapacityHint(capacity int) (r, w *os.File, err error) {