Pruner: parallel pruning

refs #62

Gopkg.lock

@@ -83,7 +83,7 @@
   branch = "master"
   name = "github.com/problame/go-netssh"
   packages = ["."]
-  revision = "53a2e445f8ace7ec678f2d8cdd9c1428dfef4562"
+  revision = "ffa145d2506e222977205e7666a9722d6b9959ac"
 
 [[projects]]
   branch = "master"

build.Dockerfile

@@ -11,7 +11,6 @@ RUN /tmp/lazy.sh devsetup
 # prepare volume mount of git checkout to /zrepl
 RUN mkdir -p /go/src/github.com/zrepl/zrepl
 RUN chmod -R 0777 /go
-RUN mkdir -p /.cache && chmod -R 0777 /.cache
 WORKDIR /go/src/github.com/zrepl/zrepl
 
 

cmd/autosnap.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"github.com/zrepl/zrepl/zfs"
 	"sort"
+	"sync"
 	"time"
 )
 
@@ -165,33 +166,46 @@ func (a *IntervalAutosnap) doSnapshots(didSnaps chan struct{}) {
 
 	// don't cache the result from previous run in case the user added
 	// a new dataset in the meantime
-	ds, stop := a.filterFilesystems()
+	fss, stop := a.filterFilesystems()
 	if stop {
 		return
 	}
 
+	a.task.Log().Info("beginning parallel snapshots")
+
 	// TODO channel programs -> allow a little jitter?
-	for _, d := range ds {
+	var wg sync.WaitGroup
-		suffix := time.Now().In(time.UTC).Format("20060102_150405_000")
+	for fsi := range fss {
-		snapname := fmt.Sprintf("%s%s", a.Prefix, suffix)
+		wg.Add(1)
+		go func(fs *zfs.DatasetPath) {
+			defer wg.Done()
 
-		l := a.task.Log().WithField(logFSField, d.ToString()).
+			suffix := time.Now().In(time.UTC).Format("20060102_150405_000")
-			WithField("snapname", snapname)
+			snapname := fmt.Sprintf("%s%s", a.Prefix, suffix)
 
-		l.Info("create snapshot")
+			l := a.task.Log().WithField(logFSField, fs.ToString()).
-		err := zfs.ZFSSnapshot(d, snapname, false)
+				WithField("snapname", snapname)
-		if err != nil {
-			a.task.Log().WithError(err).Error("cannot create snapshot")
-		}
 
-		l.Info("create corresponding bookmark")
+			l.Info("create snapshot")
-		err = zfs.ZFSBookmark(d, snapname, snapname)
+			err := zfs.ZFSSnapshot(fs, snapname, false)
-		if err != nil {
+			if err != nil {
-			a.task.Log().WithError(err).Error("cannot create bookmark")
+				l.WithError(err).Error("cannot create snapshot")
-		}
+				return
+			}
 
+			l.Info("create corresponding bookmark")
+			err = zfs.ZFSBookmark(fs, snapname, snapname)
+			if err != nil {
+				l.WithError(err).Error("cannot create bookmark")
+			}
+
+		}(fss[fsi])
 	}
 
+	a.task.Log().Info("waiting for parallel snapshots to finish")
+	wg.Wait()
+	a.task.Log().Info("snapshots finished")
+
 	select {
 	case didSnaps <- struct{}{}:
 	default:

cmd/config_connect.go

@@ -9,7 +9,6 @@ import (
 	"github.com/mitchellh/mapstructure"
 	"github.com/pkg/errors"
 	"github.com/problame/go-netssh"
-	"time"
 )
 
 type SSHStdinserverConnecter struct {
@@ -20,8 +19,6 @@ type SSHStdinserverConnecter struct {
 	TransportOpenCommand []string `mapstructure:"transport_open_command"`
 	SSHCommand           string   `mapstructure:"ssh_command"`
 	Options              []string
-	DialTimeout          string `mapstructure:"dial_timeout"`
-	dialTimeout          time.Duration
 }
 
 func parseSSHStdinserverConnecter(i map[string]interface{}) (c *SSHStdinserverConnecter, err error) {
@@ -32,15 +29,6 @@ func parseSSHStdinserverConnecter(i map[string]interface{}) (c *SSHStdinserverCo
 		return nil, err
 	}
 
-	if c.DialTimeout != "" {
-		c.dialTimeout, err = time.ParseDuration(c.DialTimeout)
-		if err != nil {
-			return nil, errors.Wrap(err, "cannot parse dial_timeout")
-		}
-	} else {
-		c.dialTimeout = 10 * time.Second
-	}
-
 	// TODO assert fields are filled
 	return
 
@@ -50,15 +38,9 @@ func (c *SSHStdinserverConnecter) Connect() (rwc io.ReadWriteCloser, err error)
 
 	var endpoint netssh.Endpoint
 	if err = copier.Copy(&endpoint, c); err != nil {
-		return nil, errors.WithStack(err)
+		return
 	}
-	var dialCtx context.Context
+	if rwc, err = netssh.Dial(context.TODO(), endpoint); err != nil {
-	dialCtx, dialCancel := context.WithTimeout(context.TODO(), c.dialTimeout) // context.TODO tied to error handling below
-	defer dialCancel()
-	if rwc, err = netssh.Dial(dialCtx, endpoint); err != nil {
-		if err == context.DeadlineExceeded {
-			err = errors.Errorf("dial_timeout of %s exceeded", c.dialTimeout)
-		}
 		err = errors.WithStack(err)
 		return
 	}

cmd/prune.go

@@ -3,7 +3,9 @@ package cmd
 import (
 	"context"
 	"fmt"
+	"github.com/zrepl/zrepl/util"
 	"github.com/zrepl/zrepl/zfs"
+	"sync"
 	"time"
 )
 
@@ -23,9 +25,8 @@ type PruneResult struct {
 	Remove     []zfs.FilesystemVersion
 }
 
+// FIXME must not call p.task.Enter because it runs in parallel
 func (p *Pruner) filterFilesystems() (filesystems []*zfs.DatasetPath, stop bool) {
-	p.task.Enter("filter_fs")
-	defer p.task.Finish()
 	filesystems, err := zfs.ZFSListMapping(p.DatasetFilter)
 	if err != nil {
 		p.task.Log().WithError(err).Error("error applying filesystem filter")
@@ -38,9 +39,8 @@ func (p *Pruner) filterFilesystems() (filesystems []*zfs.DatasetPath, stop bool)
 	return filesystems, false
 }
 
+// FIXME must not call p.task.Enter because it runs in parallel
 func (p *Pruner) filterVersions(fs *zfs.DatasetPath) (fsversions []zfs.FilesystemVersion, stop bool) {
-	p.task.Enter("filter_versions")
-	defer p.task.Finish()
 	log := p.task.Log().WithField(logFSField, fs.ToString())
 
 	filter := NewPrefixFilter(p.SnapshotPrefix)
@@ -56,9 +56,8 @@ func (p *Pruner) filterVersions(fs *zfs.DatasetPath) (fsversions []zfs.Filesyste
 	return fsversions, false
 }
 
-func (p *Pruner) pruneFilesystem(fs *zfs.DatasetPath) (r PruneResult, valid bool) {
+// FIXME must not call p.task.Enter because it runs in parallel
-	p.task.Enter("prune_fs")
+func (p *Pruner) pruneFilesystem(fs *zfs.DatasetPath, destroySemaphore util.Semaphore) (r PruneResult, valid bool) {
-	defer p.task.Finish()
 	log := p.task.Log().WithField(logFSField, fs.ToString())
 
 	fsversions, stop := p.filterVersions(fs)
@@ -66,9 +65,7 @@ func (p *Pruner) pruneFilesystem(fs *zfs.DatasetPath) (r PruneResult, valid bool
 		return
 	}
 
-	p.task.Enter("prune_policy")
 	keep, remove, err := p.PrunePolicy.Prune(fs, fsversions)
-	p.task.Finish()
 	if err != nil {
 		log.WithError(err).Error("error evaluating prune policy")
 		return
@@ -81,33 +78,37 @@ func (p *Pruner) pruneFilesystem(fs *zfs.DatasetPath) (r PruneResult, valid bool
 
 	r = PruneResult{fs, fsversions, keep, remove}
 
-	makeFields := func(v zfs.FilesystemVersion) (fields map[string]interface{}) {
+	var wg sync.WaitGroup
-		fields = make(map[string]interface{})
+	for v := range remove {
-		fields["version"] = v.ToAbsPath(fs)
+		wg.Add(1)
-		timeSince := v.Creation.Sub(p.Now)
+		go func(v zfs.FilesystemVersion) {
-		fields["age_ns"] = timeSince
+			defer wg.Done()
-		const day time.Duration = 24 * time.Hour
+			// log fields
-		days := timeSince / day
+			fields := make(map[string]interface{})
-		remainder := timeSince % day
+			fields["version"] = v.ToAbsPath(fs)
-		fields["age_str"] = fmt.Sprintf("%dd%s", days, remainder)
+			timeSince := v.Creation.Sub(p.Now)
-		return
+			fields["age_ns"] = timeSince
-	}
+			const day time.Duration = 24 * time.Hour
+			days := timeSince / day
+			remainder := timeSince % day
+			fields["age_str"] = fmt.Sprintf("%dd%s", days, remainder)
 
-	for _, v := range remove {
+			log.WithFields(fields).Info("destroying version")
-		fields := makeFields(v)
+			// echo what we'll do and exec zfs destroy if not dry run
-		log.WithFields(fields).Info("destroying version")
+			// TODO special handling for EBUSY (zfs hold)
-		// echo what we'll do and exec zfs destroy if not dry run
+			// TODO error handling for clones? just echo to cli, skip over, and exit with non-zero status code (we're idempotent)
-		// TODO special handling for EBUSY (zfs hold)
+			if !p.DryRun {
-		// TODO error handling for clones? just echo to cli, skip over, and exit with non-zero status code (we're idempotent)
+				destroySemaphore.Down()
-		if !p.DryRun {
+				err := zfs.ZFSDestroyFilesystemVersion(fs, v)
-			p.task.Enter("destroy")
+				destroySemaphore.Up()
-			err := zfs.ZFSDestroyFilesystemVersion(fs, v)
+				if err != nil {
-			p.task.Finish()
+					log.WithFields(fields).WithError(err).Error("error destroying version")
-			if err != nil {
+				}
-				log.WithFields(fields).WithError(err).Error("error destroying version")
 			}
-		}
+		}(remove[v])
 	}
+	wg.Wait()
+
 	return r, true
 }
 
@@ -124,13 +125,31 @@ func (p *Pruner) Run(ctx context.Context) (r []PruneResult, err error) {
 		return
 	}
 
-	r = make([]PruneResult, 0, len(filesystems))
+	maxConcurrentDestroy := len(filesystems)
+	p.task.Log().WithField("max_concurrent_destroy", maxConcurrentDestroy).Info("begin concurrent destroy")
+	destroySem := util.NewSemaphore(maxConcurrentDestroy)
 
+	resChan := make(chan PruneResult, len(filesystems))
+	var wg sync.WaitGroup
 	for _, fs := range filesystems {
-		res, ok := p.pruneFilesystem(fs)
+		wg.Add(1)
-		if ok {
+		go func(fs *zfs.DatasetPath) {
-			r = append(r, res)
+			defer wg.Done()
-		}
+			res, ok := p.pruneFilesystem(fs, destroySem)
+			if ok {
+				resChan <- res
+			}
+
+		}(fs)
+	}
+	wg.Wait()
+	close(resChan)
+
+	p.task.Log().Info("destroys done")
+
+	r = make([]PruneResult, 0, len(filesystems))
+	for res := range resChan {
+		r = append(r, res)
 	}
 
 	return

docs/changelog.rst

@@ -1,7 +1,6 @@
 .. |break_config| replace:: **[BREAK]**
 .. |break| replace:: **[BREAK]**
 .. |bugfix| replace:: [BUG]
-.. |docs| replace:: [DOCS]
 .. |feature| replace:: [FEATURE]
 
 Changelog
@@ -20,13 +19,7 @@ Developers should consult the git commit log or GitHub issue tracker.
   * Make sure to understand the meaning bookmarks have for :ref:`maximum replication downtime <replication-downtime>`.
   * Example: :sampleconf:`pullbackup/productionhost.yml`
 
-* |break| :commit:`ccd062e`: ``ssh+stdinserver`` transport: changed protocol requires daemon restart on both sides
+* |break| :commit:`ccd062e`: both sides of a replication setup must be updated and restarted. Otherwise the connecting side will hang and not time out.
-
-  * The delicate procedure of talking to the serving-side zrepl daemon via the stdinserver proxy command now has better error handling.
-  * This includes handshakes between client+proxy and client + remote daemo, which is not implemented in previous versions of zrepl.
-  * The connecting side will therefore time out, with the message ``dial_timeout of 10s exceeded``.
-  * Both sides of a replication setup must be updated and restarted. Otherwise the connecting side will hang and not time out.
-
 * |break_config| :commit:`2bfcfa5`: first outlet in ``global.logging`` is now used for logging meta-errors, for example problems encountered when writing to other outlets.
 * |feature| :issue:`10`: ``zrepl control status`` subcommand
 
@@ -41,7 +34,6 @@ Developers should consult the git commit log or GitHub issue tracker.
 * |bugfix| :issue:`8` and :issue:`56`: ``ssh+stdinserver`` transport properly reaps SSH child processes
 * |bugfix| :commit:`cef63ac`: ``human`` format now prints non-string values correctly
 * |bugfix| :issue:`26`: slow TCP outlets no longer block the daemon
-* |docs| :issue:`64`: tutorial: document ``known_host`` file entry
 
 0.0.2
 -----

docs/configuration/transports.rst

@@ -85,9 +85,8 @@ Connect Mode
         user: root
         port: 22
         identity_file: /etc/zrepl/ssh/identity
-        options: # optional, default [], `-o` arguments passed to ssh
+        options: # optional
         - "Compression=on"
-        dial_timeout: # optional, default 10s, max time.Duration until initial handshake is completed
 
 The connecting zrepl daemon
 
@@ -102,13 +101,10 @@ The connecting zrepl daemon
       #. The remote user, host and port correspond to those configured.
       #. Further options can be specified using the ``options`` field, which appends each entry in the list to the command line using ``-o $entry``.
 
-#. Wraps the pipe ends in an ``io.ReadWriteCloser`` and uses it for RPC.
+1. Wraps the pipe ends in an ``io.ReadWriteCloser`` and uses it for RPC.
-
 
 As discussed in the section above, the connecting zrepl daemon expects that ``zrepl stdinserver $client_identity`` is  executed automatically via an ``authorized_keys`` file entry.
 
-The ``known_hosts`` file used by the ssh command must contain an entry for the serving host, e.g., ``app-srv.example.com`` in the example above.
-
 .. NOTE::
 
     The environment variables of the underlying SSH process are cleared. ``$SSH_AUTH_SOCK`` will not be available.

docs/index.rst

@@ -16,7 +16,7 @@ zrepl - ZFS replication
 Getting started
 ~~~~~~~~~~~~~~~
 
-The :ref:`10 minutes tutorial setup <tutorial>` gives you a first impression.
+The :ref:`5 minute tutorial setup <tutorial>` gives you a first impression.
 
 Main Features
 ~~~~~~~~~~~~~

docs/tutorial.rst

@@ -60,7 +60,7 @@ Follow the :ref:`OS-specific installation instructions <installation>` and come
 Configure ``backup-srv``
 ------------------------
 
-We define a **pull job** named ``pull_app-srv`` in the |mainconfig| on host ``backup-srv``: ::
+We define a **pull job** named ``pull_app-srv`` in the |mainconfig|: ::
 
     jobs:
     - name: pull_app-srv
@@ -94,13 +94,6 @@ It uses the private key specified at ``connect.identity_file`` which we still ne
 Note that most use cases do not benefit from separate keypairs per remote endpoint.
 Thus, it is sufficient to create one keypair and use it for all ``connect`` directives on one host.
 
-zrepl uses ssh's default ``known_hosts`` file, which must contain a host identification entry for ``app-srv.example.com``.
-If that entry does not already exist, we need to generate it.
-Run the following command, compare the host fingerprints, and confirm with yes if they match.
-You will not be able to get a shell with the identity file we just generated, which is fine. ::
-
-    ssh -i /etc/zrepl/ssh/identity root@app-srv.example.com
-
 Learn more about :ref:`transport-ssh+stdinserver` transport and the :ref:`pull job <job-pull>` format.
 
 .. _tutorial-configure-app-srv:
@@ -108,7 +101,7 @@ Learn more about :ref:`transport-ssh+stdinserver` transport and the :ref:`pull j
 Configure ``app-srv``
 ---------------------
 
-We define a corresponding **source job** named ``pull_backup`` in the |mainconfig| on host ``app-srv``:  ::
+We define a corresponding **source job** named ``pull_backup`` in the |mainconfig|: ::
 
     jobs:
     - name: pull_backup

rpc/datatype_string.go

@@ -2,7 +2,7 @@
 
 package rpc
 
-import "strconv"
+import "fmt"
 
 const _DataType_name = "DataTypeNoneDataTypeControlDataTypeMarshaledJSONDataTypeOctets"
 
@@ -11,7 +11,7 @@ var _DataType_index = [...]uint8{0, 12, 27, 48, 62}
 func (i DataType) String() string {
 	i -= 1
 	if i >= DataType(len(_DataType_index)-1) {
-		return "DataType(" + strconv.FormatInt(int64(i+1), 10) + ")"
+		return fmt.Sprintf("DataType(%d)", i+1)
 	}
 	return _DataType_name[_DataType_index[i]:_DataType_index[i+1]]
 }

rpc/frametype_string.go

@@ -2,7 +2,7 @@
 
 package rpc
 
-import "strconv"
+import "fmt"
 
 const (
 	_FrameType_name_0 = "FrameTypeHeaderFrameTypeDataFrameTypeTrailer"
@@ -11,6 +11,7 @@ const (
 
 var (
 	_FrameType_index_0 = [...]uint8{0, 15, 28, 44}
+	_FrameType_index_1 = [...]uint8{0, 12}
 )
 
 func (i FrameType) String() string {
@@ -21,6 +22,6 @@ func (i FrameType) String() string {
 	case i == 255:
 		return _FrameType_name_1
 	default:
-		return "FrameType(" + strconv.FormatInt(int64(i), 10) + ")"
+		return fmt.Sprintf("FrameType(%d)", i)
 	}
 }

rpc/status_string.go

@@ -2,7 +2,7 @@
 
 package rpc
 
-import "strconv"
+import "fmt"
 
 const _Status_name = "StatusOKStatusRequestErrorStatusServerErrorStatusError"
 
@@ -11,7 +11,7 @@ var _Status_index = [...]uint8{0, 8, 26, 43, 54}
 func (i Status) String() string {
 	i -= 1
 	if i >= Status(len(_Status_index)-1) {
-		return "Status(" + strconv.FormatInt(int64(i+1), 10) + ")"
+		return fmt.Sprintf("Status(%d)", i+1)
 	}
 	return _Status_name[_Status_index[i]:_Status_index[i+1]]
 }

util/semaphore.go

@@ -0,0 +1,17 @@
+package util
+
+type Semaphore struct {
+	c chan struct{}
+}
+
+func NewSemaphore(cap int) Semaphore {
+	return Semaphore{make(chan struct{}, cap)}
+}
+
+func (s Semaphore) Down() {
+	s.c <- struct{}{}
+}
+
+func (s Semaphore) Up() {
+	<-s.c
+}

zfs/conflict_string.go

@@ -2,7 +2,7 @@
 
 package zfs
 
-import "strconv"
+import "fmt"
 
 const _Conflict_name = "ConflictIncrementalConflictAllRightConflictNoCommonAncestorConflictDiverged"
 
@@ -10,7 +10,7 @@ var _Conflict_index = [...]uint8{0, 19, 35, 59, 75}
 
 func (i Conflict) String() string {
 	if i < 0 || i >= Conflict(len(_Conflict_index)-1) {
-		return "Conflict(" + strconv.FormatInt(int64(i), 10) + ")"
+		return fmt.Sprintf("Conflict(%d)", i)
 	}
 	return _Conflict_name[_Conflict_index[i]:_Conflict_index[i+1]]
 }