fixups

cc @asomers

.circleci/config.yml

@@ -358,7 +358,7 @@ jobs:
           subcommand: docdep
       - run:
           command: |
-            git config --global user.email "me@cschwarz.com"
+            git config --global user.email "zreplbot@cschwarz.com"
             git config --global user.name "zrepl-github-io-ci"
 
       # https://circleci.com/docs/2.0/add-ssh-key/#adding-multiple-keys-with-blank-hostnames

client/status/viewmodel/bytesprogresshistory.go

@@ -11,7 +11,6 @@ type bytesProgressHistory struct {
 	last        *byteProgressMeasurement // pointer as poor man's optional
 	changeCount int
 	lastChange  time.Time
-	bpsAvg      float64
 }
 
 func (p *bytesProgressHistory) Update(currentVal int64) (bytesPerSecondAvg int64, changeCount int) {
@@ -38,11 +37,8 @@ func (p *bytesProgressHistory) Update(currentVal int64) (bytesPerSecondAvg int64
 	deltaT := time.Since(p.last.time)
 	rate := float64(deltaV) / deltaT.Seconds()
 
-	factor := 0.3
-	p.bpsAvg = (1-factor)*p.bpsAvg + factor*rate
-
 	p.last.time = time.Now()
 	p.last.val = currentVal
 
-	return int64(p.bpsAvg), p.changeCount
+	return int64(rate), p.changeCount
 }

client/status/viewmodel/render.go

@@ -356,9 +356,17 @@ func renderReplicationReport(t *stringbuilder.B, rep *report.Report, history *by
 		// Progress: [---------------]
 		expected, replicated, containsInvalidSizeEstimates := latest.BytesSum()
 		rate, changeCount := history.Update(replicated)
+		eta := time.Duration(0)
+		if rate > 0 {
+			eta = time.Duration((expected-replicated)/rate) * time.Second
+		}
+
 		t.Write("Progress: ")
 		t.DrawBar(50, replicated, expected, changeCount)
 		t.Write(fmt.Sprintf(" %s / %s @ %s/s", ByteCountBinary(replicated), ByteCountBinary(expected), ByteCountBinary(rate)))
+		if eta != 0 {
+			t.Write(fmt.Sprintf(" (%s remaining)", humanizeDuration(eta)))
+		}
 		t.Newline()
 		if containsInvalidSizeEstimates {
 			t.Write("NOTE: not all steps could be size-estimated, total estimate is likely imprecise!")
@@ -383,6 +391,30 @@ func renderReplicationReport(t *stringbuilder.B, rep *report.Report, history *by
 	}
 }
 
+func humanizeDuration(duration time.Duration) string {
+	days := int64(duration.Hours() / 24)
+	hours := int64(math.Mod(duration.Hours(), 24))
+	minutes := int64(math.Mod(duration.Minutes(), 60))
+	seconds := int64(math.Mod(duration.Seconds(), 60))
+
+	var parts []string
+
+	force := false
+	chunks := []int64{days, hours, minutes, seconds}
+	for i, chunk := range chunks {
+		if force || chunk > 0 {
+			padding := 0
+			if force {
+				padding = 2
+			}
+			parts = append(parts, fmt.Sprintf("%*d%c", padding, chunk, "dhms"[i]))
+			force = true
+		}
+	}
+
+	return strings.Join(parts, " ")
+}
+
 func renderPrunerReport(t *stringbuilder.B, r *pruner.Report, fsfilter FilterFunc) {
 	if r == nil {
 		t.Printf("...\n")

config/config.go

@@ -13,6 +13,7 @@ import (
 	"github.com/pkg/errors"
 	"github.com/zrepl/yaml-config"
 
+	"github.com/zrepl/zrepl/util/datasizeunit"
 	zfsprop "github.com/zrepl/zrepl/zfs/property"
 )
 
@@ -87,6 +88,8 @@ type SendOptions struct {
 	Compressed       bool `yaml:"compressed,optional,default=false"`
 	EmbeddedData     bool `yaml:"embbeded_data,optional,default=false"`
 	Saved            bool `yaml:"saved,optional,default=false"`
+
+	BandwidthLimit *BandwidthLimit `yaml:"bandwidth_limit,optional,fromdefaults"`
 }
 
 type RecvOptions struct {
@@ -96,6 +99,15 @@ type RecvOptions struct {
 	// Reencrypt bool `yaml:"reencrypt"`
 
 	Properties *PropertyRecvOptions `yaml:"properties,fromdefaults"`
+
+	BandwidthLimit *BandwidthLimit `yaml:"bandwidth_limit,optional,fromdefaults"`
+}
+
+var _ yaml.Unmarshaler = &datasizeunit.Bits{}
+
+type BandwidthLimit struct {
+	Max            datasizeunit.Bits `yaml:"max,default=-1 B"`
+	BucketCapacity datasizeunit.Bits `yaml:"bucket_capacity,default=128 KiB"`
 }
 
 type Replication struct {
@@ -113,10 +125,6 @@ type ReplicationOptionsConcurrency struct {
 	SizeEstimates int `yaml:"size_estimates,optional,default=4"`
 }
 
-func (l *RecvOptions) SetDefault() {
-	*l = RecvOptions{Properties: &PropertyRecvOptions{}}
-}
-
 type PropertyRecvOptions struct {
 	Inherit  []zfsprop.Property          `yaml:"inherit,optional"`
 	Override map[zfsprop.Property]string `yaml:"override,optional"`

config/samples/bandwidth_limit.yml

@@ -0,0 +1,41 @@
+
+jobs:
+  - type: sink
+    name: "limited_sink"
+    root_fs: "fs0"
+    recv:
+      bandwidth_limit:
+        max: 12345 B
+    serve:
+      type: local
+      listener_name: localsink
+
+  - type: push
+    name: "limited_push"
+    connect:
+      type: local
+      listener_name: localsink
+      client_identity: local_backup
+    filesystems: {
+      "root<": true,
+    }
+    send:
+      bandwidth_limit:
+        max: 54321 B
+        bucket_capacity: 1024 B
+    snapshotting:
+      type: manual
+    pruning:
+      keep_sender:
+        - type: last_n
+          count: 1
+      keep_receiver:
+        - type: last_n
+          count: 1
+
+  - type: sink
+    name: "nolimit_sink"
+    root_fs: "fs1"
+    serve:
+      type: local
+      listener_name: localsink

config/samples/quickstart_backup_to_external_disk.yml

@@ -5,7 +5,7 @@
 # quick start section which inlines this example.
 #
 # CUSTOMIZATIONS YOU WILL LIKELY WANT TO APPLY:
-# - adjust the name of the production pool `system` in the `filesystems` filter of jobs `snapjob` and `push_to_derive`
+# - adjust the name of the production pool `system` in the `filesystems` filter of jobs `snapjob` and `push_to_drive`
 # - adjust the name of the backup pool `backuppool` in the `backuppool_sink` job
 # - adjust the occurences of `myhostname` to the name of the system you are backing up (cannot be easily changed once you start replicating)
 # - make sure the `zrepl_` prefix is not being used by any other zfs tools you might have installed (it likely isn't)

daemon/job/build_jobs.go

@@ -8,6 +8,7 @@ import (
 	"github.com/pkg/errors"
 
 	"github.com/zrepl/zrepl/config"
+	"github.com/zrepl/zrepl/util/bandwidthlimit"
 )
 
 func JobsFromConfig(c *config.Config) ([]Job, error) {
@@ -107,3 +108,13 @@ func validateReceivingSidesDoNotOverlap(receivingRootFSs []string) error {
 	}
 	return nil
 }
+
+func buildBandwidthLimitConfig(in *config.BandwidthLimit) (c bandwidthlimit.Config, _ error) {
+	if in.Max.ToBytes() > 0 && int64(in.Max.ToBytes()) == 0 {
+		return c, fmt.Errorf("bandwidth limit `max` is too small, must at least specify one byte")
+	}
+	return bandwidthlimit.Config{
+		Max:            int64(in.Max.ToBytes()),
+		BucketCapacity: int64(in.BucketCapacity.ToBytes()),
+	}, nil
+}

daemon/job/build_jobs_sendrecvoptions.go

@@ -22,7 +22,12 @@ func buildSenderConfig(in SendingJobConfig, jobID endpoint.JobID) (*endpoint.Sen
 		return nil, errors.Wrap(err, "cannot build filesystem filter")
 	}
 	sendOpts := in.GetSendOptions()
-	return &endpoint.SenderConfig{
+	bwlim, err := buildBandwidthLimitConfig(sendOpts.BandwidthLimit)
+	if err != nil {
+		return nil, errors.Wrap(err, "cannot build bandwith limit config")
+	}
+
+	sc := &endpoint.SenderConfig{
 		FSF:   fsf,
 		JobID: jobID,
 
@@ -34,7 +39,15 @@ func buildSenderConfig(in SendingJobConfig, jobID endpoint.JobID) (*endpoint.Sen
 		SendCompressed:       sendOpts.Compressed,
 		SendEmbeddedData:     sendOpts.EmbeddedData,
 		SendSaved:            sendOpts.Saved,
-	}, nil
+
+		BandwidthLimit: bwlim,
+	}
+
+	if err := sc.Validate(); err != nil {
+		return nil, errors.Wrap(err, "cannot build sender config")
+	}
+
+	return sc, nil
 }
 
 type ReceivingJobConfig interface {
@@ -53,6 +66,12 @@ func buildReceiverConfig(in ReceivingJobConfig, jobID endpoint.JobID) (rc endpoi
 	}
 
 	recvOpts := in.GetRecvOptions()
+
+	bwlim, err := buildBandwidthLimitConfig(recvOpts.BandwidthLimit)
+	if err != nil {
+		return rc, errors.Wrap(err, "cannot build bandwith limit config")
+	}
+
 	rc = endpoint.ReceiverConfig{
 		JobID:                      jobID,
 		RootWithoutClientComponent: rootFs,
@@ -60,6 +79,8 @@ func buildReceiverConfig(in ReceivingJobConfig, jobID endpoint.JobID) (rc endpoi
 
 		InheritProperties:  recvOpts.Properties.Inherit,
 		OverrideProperties: recvOpts.Properties.Override,
+
+		BandwidthLimit: bwlim,
 	}
 	if err := rc.Validate(); err != nil {
 		return rc, errors.Wrap(err, "cannot build receiver config")

daemon/job/build_jobs_test.go

@@ -119,6 +119,14 @@ func TestSampleConfigsAreBuiltWithoutErrors(t *testing.T) {
 		t.Errorf("glob failed: %+v", err)
 	}
 
+	type additionalCheck struct {
+		state int
+		test  func(t *testing.T, jobs []Job)
+	}
+	additionalChecks := map[string]*additionalCheck{
+		"bandwidth_limit.yml": {test: testSampleConfig_BandwidthLimit},
+	}
+
 	for _, p := range paths {
 
 		if path.Ext(p) != ".yml" {
@@ -126,10 +134,20 @@ func TestSampleConfigsAreBuiltWithoutErrors(t *testing.T) {
 			continue
 		}
 
+		filename := path.Base(p)
+		t.Logf("checking for presence additonal checks for file %q", filename)
+		additionalCheck := additionalChecks[filename]
+		if additionalCheck == nil {
+			t.Logf("no additional checks")
+		} else {
+			t.Logf("additional check present")
+			additionalCheck.state = 1
+		}
+
 		t.Run(p, func(t *testing.T) {
 			c, err := config.ParseConfig(p)
 			if err != nil {
-				t.Errorf("error parsing %s:\n%+v", p, err)
+				t.Fatalf("error parsing %s:\n%+v", p, err)
 			}
 
 			t.Logf("file: %s", p)
@@ -138,11 +156,57 @@ func TestSampleConfigsAreBuiltWithoutErrors(t *testing.T) {
 			tls.FakeCertificateLoading(t)
 			jobs, err := JobsFromConfig(c)
 			t.Logf("jobs: %#v", jobs)
-			assert.NoError(t, err)
+			require.NoError(t, err)
+
+			if additionalCheck != nil {
+				additionalCheck.test(t, jobs)
+				additionalCheck.state = 2
+			}
 		})
 
 	}
 
+	for basename, c := range additionalChecks {
+		if c.state == 0 {
+			panic("univisited additional check " + basename)
+		}
+	}
+
+}
+
+func testSampleConfig_BandwidthLimit(t *testing.T, jobs []Job) {
+	require.Len(t, jobs, 3)
+
+	{
+		limitedSink, ok := jobs[0].(*PassiveSide)
+		require.True(t, ok, "%T", jobs[0])
+		limitedSinkMode, ok := limitedSink.mode.(*modeSink)
+		require.True(t, ok, "%T", limitedSink)
+
+		assert.Equal(t, int64(12345), limitedSinkMode.receiverConfig.BandwidthLimit.Max)
+		assert.Equal(t, int64(1<<17), limitedSinkMode.receiverConfig.BandwidthLimit.BucketCapacity)
+	}
+
+	{
+		limitedPush, ok := jobs[1].(*ActiveSide)
+		require.True(t, ok, "%T", jobs[1])
+		limitedPushMode, ok := limitedPush.mode.(*modePush)
+		require.True(t, ok, "%T", limitedPush)
+
+		assert.Equal(t, int64(54321), limitedPushMode.senderConfig.BandwidthLimit.Max)
+		assert.Equal(t, int64(1024), limitedPushMode.senderConfig.BandwidthLimit.BucketCapacity)
+	}
+
+	{
+		unlimitedSink, ok := jobs[2].(*PassiveSide)
+		require.True(t, ok, "%T", jobs[2])
+		unlimitedSinkMode, ok := unlimitedSink.mode.(*modeSink)
+		require.True(t, ok, "%T", unlimitedSink)
+
+		max := unlimitedSinkMode.receiverConfig.BandwidthLimit.Max
+		assert.Less(t, max, int64(0), max, "unlimited mode <=> negative value for .Max, see bandwidthlimit.Config")
+	}
+
 }
 
 func TestReplicationOptions(t *testing.T) {

docs/configuration/prune.rst

@@ -124,43 +124,51 @@ The syntax to describe the bucket list is as follows:
 
 ::
 
-          This grid spec produces the following list of adjacent buckets. For the sake of simplicity,
-          we subject all snapshots to the grid pruning policy by settings `regex: .*`.
+   Assume the following grid spec:
 
-          `
-           grid: 1x1h(keep=all) | 2x2h | 1x3h
-           regex: .*
-          `
+      grid: 1x1h(keep=all) | 2x2h | 1x3h
 
-          0h        1h        2h        3h        4h        5h        6h        7h        8h        9h
-          |         |         |         |         |         |         |         |         |         |
-          |-Bucket1-|-----Bucket 2------|------Bucket 3-----|-----------Bucket 4----------|
-          | keep=all|      keep=1       |       keep=1      |            keep=1           |
+   This grid spec produces the following constellation of buckets:
+
+   0h        1h        2h        3h        4h        5h        6h        7h        8h        9h
+   |         |         |         |         |         |         |         |         |         |
+   |-Bucket1-|-----Bucket2-------|------Bucket3------|-----------Bucket4-----------|
+   | keep=all|      keep=1       |       keep=1      |            keep=1           |
 
 
 
-          Let us consider the following set of snapshots @a-zA-C:
+   Now assume that we have a set of snapshots @a, @b, ..., @D.
+   Snapshot @a is the most recent snapshot.
+   Snapshot @D is the oldest snapshot, it is almost 9 hours older than snapshot @a.
+   We place the snapshots on the same timeline as the buckets:
 
 
-          |  a  b  c  d  e  f  g  h  i  j  k  l  m  n  o  p  q  r  s  t  u  v  w  x  y  z  A  B  C  D        |
+   0h        1h        2h        3h        4h        5h        6h        7h        8h        9h
+   |         |         |         |         |         |         |         |         |         |
+   |-Bucket1-|-----Bucket2-------|------Bucket3------|-----------Bucket4-----------|
+   | keep=all|      keep=1       |       keep=1      |            keep=1           |
+   |         |                   |                   |                             |
+   |  a  b  c| d  e  f  g  h  i  j  k  l  m  n  o  p |q  r  s  t  u  v  w  x  y  z |A  B  C  D
 
-          The `grid` algorithm maps them to their respective buckets:
+   The result is the following mapping of snapshots to buckets:
 
-          Bucket 1: a, b, c
-          Bucket 2: d,e,f,g,h,i,j
-          Bucket 3: k,l,m,n,o,p
-          Bucket 4: q,r, q,r,s,t,u,v,w,x,y,z
-          None:     A,B,C,D
+   Bucket1:   a, b, c
+   Bucket2:   d,e,f,g,h,i,j
+   Bucket3:   k,l,m,n,o,p
+   Bucket4:   q,r,s,t,u,v,w,x,y,z
+   No bucket: A,B,C,D
 
-          It then applies the per-bucket pruning logic described above which resulting in the
-          following list of remaining snapshots.
+   For each bucket, we now prune snapshots until it only contains `keep` snapshots.
+   Newer snapshots are destroyed first.
+   Snapshots that do not fall into a bucket are always destroyed.
 
-          |  a  b  c                    j                 p                             z                    |
-
-          Note that it only makes sense to grow (not shorten) the interval duration for buckets
-          further in the past since each bucket acts like a low-pass filter for incoming snapshots
-          and adding a less-low-pass-filter after a low-pass one has no effect.
+   Result after pruning:
 
+   0h        1h        2h        3h        4h        5h        6h        7h        8h        9h
+   |         |         |         |         |         |         |         |         |         |
+   |-Bucket1-|-----Bucket2-------|------Bucket3------|-----------Bucket4-----------|
+   |         |                   |                   |                             |
+   |  a  b  c|                   j                 p |                           z  |
 
 .. _prune-keep-last-n:
 

docs/configuration/sendrecvoptions.rst

@@ -36,6 +36,9 @@ See the `upstream man page <https://openzfs.github.io/openzfs-docs/man/8/zfs-sen
     * - ``encrypted``
       -
       - Specific to zrepl, :ref:`see below <job-send-options-encrypted>`.
+    * - ``bandwidth_limit``
+      -
+      - Specific to zrepl, :ref:`see below <job-send-recv-options-bandwidth-limit>`.
     * - ``raw``
       - ``-w``
       - Use ``encrypted`` to only allow encrypted sends.
@@ -138,6 +141,7 @@ Recv Options
          override: {
            "org.openzfs.systemd:ignore": "on"
          }
+       bandwidth_limit: ... # see below
      ...
 
 .. _job-recv-options--inherit-and-override:
@@ -212,3 +216,25 @@ and property replication is enabled, the receiver must :ref:`inherit the followi
 * ``keylocation``
 * ``keyformat``
 * ``encryption``
+
+Common Options
+~~~~~~~~~~~~~~
+
+.. _job-send-recv-options-bandwidth-limit:
+
+Bandwidth Limit (send & recv)
+-----------------------------
+
+::
+
+   bandwidth_limit:
+     max: 23.5 MiB # -1 is the default and disabled rate limiting
+     bucket_capacity: # token bucket capacity in bytes; defaults to 128KiB
+
+Both ``send`` and ``recv`` can be limited to a maximum bandwidth through ``bandwidth_limit``.
+For most users, it should be sufficient to just set ``bandwidth_limit.max``.
+The ``bandwidth_limit.bucket_capacity`` refers to the `token bucket size <https://github.com/juju/ratelimit>`_.
+
+The bandwidth limit only applies to the payload data, i.e., the ZFS send stream.
+It does not account for transport protocol overheads.
+The scope is the job level, i.e., all :ref:`concurrent <replication-option-concurrency>` sends or incoming receives of a job share the bandwidth limit.

docs/publish.sh

@@ -1,6 +1,5 @@
 #!/bin/bash
-set -eo pipefail
-
+set -euo pipefail
 
 NON_INTERACTIVE=false
 DO_CLONE=false
@@ -13,7 +12,7 @@ while getopts "ca" arg; do
             DO_CLONE=true
             ;;
         *)
-            echo invalid option
+            echo "invalid option '-$arg'"
             exit 1
             ;;
     esac
@@ -27,11 +26,6 @@ checkout_repo_msg() {
     echo "clone ${GHPAGESREPO} to ${PUBLICDIR}:"
 }
 
-exit_msg() {
-    echo "error, exiting..."
-}
-trap exit_msg EXIT
-
 if ! type sphinx-versioning >/dev/null; then
     echo "install sphinx-versioning and come back"
     exit 1
@@ -71,6 +65,9 @@ git reset --hard origin/master
 
 echo "cleaning GitHub pages repo"
 git rm -rf .
+cat > .gitignore <<EOF
+**/.doctrees
+EOF
 
 popd
 
@@ -96,7 +93,13 @@ COMMIT_MSG="sphinx-versioning render from publish.sh - $(date -u) - ${CURRENT_CO
 pushd "$PUBLICDIR"
 
 echo "adding and commiting all changes in GitHub pages repo"
+git add .gitignore
 git add -A
-git commit -m "$COMMIT_MSG"
+if [ "$(git status --porcelain)" != "" ]; then
+    git commit -m "$COMMIT_MSG"
+else
+    echo "nothing to commit"
+fi
+echo "pushing to GitHub pages repo"
 git push origin master
 

docs/supporters.rst

@@ -32,6 +32,8 @@ We would like to thank the following people and organizations for supporting zre
 
    <div class="fa fa-code" style="width: 1em;"></div>
 
+* |supporter-gold| Prominic.NET, Inc.
+* |supporter-std| Torsten Blum
 * |supporter-gold| Cyberiada GmbH
 * |supporter-std| `Gordon Schulz <https://github.com/azmodude>`_
 * |supporter-std| `@jwittlincohen <https://github.com/jwittlincohen>`_

endpoint/endpoint.go

@@ -14,6 +14,7 @@ import (
 	"github.com/zrepl/zrepl/daemon/logging/trace"
 
 	"github.com/zrepl/zrepl/replication/logic/pdu"
+	"github.com/zrepl/zrepl/util/bandwidthlimit"
 	"github.com/zrepl/zrepl/util/chainedio"
 	"github.com/zrepl/zrepl/util/chainlock"
 	"github.com/zrepl/zrepl/util/envconst"
@@ -34,6 +35,8 @@ type SenderConfig struct {
 	SendCompressed       bool
 	SendEmbeddedData     bool
 	SendSaved            bool
+
+	BandwidthLimit bandwidthlimit.Config
 }
 
 func (c *SenderConfig) Validate() error {
@@ -44,6 +47,9 @@ func (c *SenderConfig) Validate() error {
 	if _, err := StepHoldTag(c.JobID); err != nil {
 		return fmt.Errorf("JobID cannot be used for hold tag: %s", err)
 	}
+	if err := bandwidthlimit.ValidateConfig(c.BandwidthLimit); err != nil {
+		return errors.Wrap(err, "`BandwidthLimit` field invalid")
+	}
 	return nil
 }
 
@@ -54,16 +60,21 @@ type Sender struct {
 	FSFilter zfs.DatasetFilter
 	jobId    JobID
 	config   SenderConfig
+	bwLimit  bandwidthlimit.Wrapper
 }
 
 func NewSender(conf SenderConfig) *Sender {
 	if err := conf.Validate(); err != nil {
 		panic("invalid config" + err.Error())
 	}
+
+	ratelimiter := bandwidthlimit.WrapperFromConfig(conf.BandwidthLimit)
+
 	return &Sender{
 		FSFilter: conf.FSF,
 		jobId:    conf.JobID,
 		config:   conf,
+		bwLimit:  ratelimiter,
 	}
 }
 
@@ -301,12 +312,16 @@ func (s *Sender) Send(ctx context.Context, r *pdu.SendReq) (*pdu.SendRes, io.Rea
 		abstractionsCacheSingleton.TryBatchDestroy(ctx, s.jobId, sendArgs.FS, destroyTypes, keep, check)
 	}()
 
-	sendStream, err := zfs.ZFSSend(ctx, sendArgs)
+	var sendStream io.ReadCloser
+	sendStream, err = zfs.ZFSSend(ctx, sendArgs)
 	if err != nil {
 		// it's ok to not destroy the abstractions we just created here, a new send attempt will take care of it
 		return nil, nil, errors.Wrap(err, "zfs send failed")
 	}
 
+	// apply rate limit
+	sendStream = s.bwLimit.WrapReadCloser(sendStream)
+
 	return res, sendStream, nil
 }
 
@@ -439,6 +454,8 @@ type ReceiverConfig struct {
 
 	InheritProperties  []zfsprop.Property
 	OverrideProperties map[zfsprop.Property]string
+
+	BandwidthLimit bandwidthlimit.Config
 }
 
 func (c *ReceiverConfig) copyIn() {
@@ -475,6 +492,11 @@ func (c *ReceiverConfig) Validate() error {
 	if c.RootWithoutClientComponent.Length() <= 0 {
 		return errors.New("RootWithoutClientComponent must not be an empty dataset path")
 	}
+
+	if err := bandwidthlimit.ValidateConfig(c.BandwidthLimit); err != nil {
+		return errors.Wrap(err, "`BandwidthLimit` field invalid")
+	}
+
 	return nil
 }
 
@@ -484,6 +506,8 @@ type Receiver struct {
 
 	conf ReceiverConfig // validated
 
+	bwLimit bandwidthlimit.Wrapper
+
 	recvParentCreationMtx *chainlock.L
 }
 
@@ -495,6 +519,7 @@ func NewReceiver(config ReceiverConfig) *Receiver {
 	return &Receiver{
 		conf:                  config,
 		recvParentCreationMtx: chainlock.New(),
+		bwLimit:               bandwidthlimit.WrapperFromConfig(config.BandwidthLimit),
 	}
 }
 
@@ -787,6 +812,9 @@ func (s *Receiver) Receive(ctx context.Context, req *pdu.ReceiveReq, receive io.
 		return nil, errors.Wrap(err, "cannot determine whether we can use resumable send & recv")
 	}
 
+	// apply rate limit
+	receive = s.bwLimit.WrapReadCloser(receive)
+
 	var peek bytes.Buffer
 	var MaxPeek = envconst.Int64("ZREPL_ENDPOINT_RECV_PEEK_SIZE", 1<<20)
 	log.WithField("max_peek_bytes", MaxPeek).Info("peeking incoming stream")

go.mod

@@ -15,6 +15,7 @@ require (
 	github.com/golang/protobuf v1.4.3
 	github.com/google/uuid v1.1.2
 	github.com/jinzhu/copier v0.0.0-20170922082739-db4671f3a9b8
+	github.com/juju/ratelimit v1.0.1
 	github.com/kisielk/gotool v1.0.0 // indirect
 	github.com/kr/pretty v0.1.0
 	github.com/leodido/go-urn v1.2.1 // indirect

go.sum

@@ -164,6 +164,8 @@ github.com/jinzhu/copier v0.0.0-20170922082739-db4671f3a9b8 h1:+dKzeuiDYbD/Cfi/s
 github.com/jinzhu/copier v0.0.0-20170922082739-db4671f3a9b8/go.mod h1:yL958EeXv8Ylng6IfnvG4oflryUi3vgA3xPs9hmII1s=
 github.com/json-iterator/go v1.1.6/go.mod h1:+SdeFBvtyEkXs7REEP0seUULqWtbJapLOCVDaaPEHmU=
 github.com/json-iterator/go v1.1.7/go.mod h1:KdQUCv79m/52Kvf8AW2vK1V8akMuk1QjK/uOdHXbAo4=
+github.com/juju/ratelimit v1.0.1 h1:+7AIFJVQ0EQgq/K9+0Krm7m530Du7tIz0METWzN0RgY=
+github.com/juju/ratelimit v1.0.1/go.mod h1:qapgC/Gy+xNh9UxzV13HGGl/6UXNN+ct+vwSgWNm/qk=
 github.com/julienschmidt/httprouter v1.2.0/go.mod h1:SYymIcj16QtmaHHD7aYtjjsJG7VTCxuUUipMqKk8s4w=
 github.com/jung-kurt/gofpdf v1.0.3-0.20190309125859-24315acbbda5/go.mod h1:7Id9E/uU8ce6rXgefFLlgrJj/GYY22cpxn+r32jIOes=
 github.com/k0kubun/colorstring v0.0.0-20150214042306-9440f1994b88 h1:uC1QfSlInpQF+M0ao65imhwqKnz3Q2z/d8PWZRMQvDM=

packaging/deb/Dockerfile

@@ -1,9 +1,13 @@
 FROM debian:latest
 
+# binutils are for cross-compilation to work in bullseye
 RUN apt-get update && apt-get install -y \
 	build-essential \
     devscripts \
-    dh-exec
+    dh-exec \
+	binutils-aarch64-linux-gnu \
+	binutils-arm-linux-gnueabihf  \
+	binutils-i686-linux-gnu
 
 RUN mkdir -p /build/src && chmod -R 0777 /build
 

platformtest/tests/recvRollback.go

@@ -40,6 +40,7 @@ func ReceiveForceRollbackWorksUnencrypted(ctx *platformtest.Context) {
 
 	sendStream, err := zfs.ZFSSend(ctx, sendArgs)
 	require.NoError(ctx, err)
+	defer sendStream.Close()
 
 	recvOpts := zfs.RecvOptions{
 		RollbackAndForceRecv: true,

platformtest/tests/replication.go

@@ -7,6 +7,7 @@ import (
 	"os"
 	"path"
 	"sort"
+	"strings"
 	"time"
 
 	"github.com/kr/pretty"
@@ -20,6 +21,7 @@ import (
 	"github.com/zrepl/zrepl/replication/logic"
 	"github.com/zrepl/zrepl/replication/logic/pdu"
 	"github.com/zrepl/zrepl/replication/report"
+	"github.com/zrepl/zrepl/util/bandwidthlimit"
 	"github.com/zrepl/zrepl/util/limitio"
 	"github.com/zrepl/zrepl/util/nodefault"
 	"github.com/zrepl/zrepl/zfs"
@@ -63,9 +65,10 @@ func (i replicationInvocation) Do(ctx *platformtest.Context) *report.Report {
 	}
 
 	senderConfig := endpoint.SenderConfig{
-		FSF:     i.sfilter.AsFilter(),
-		Encrypt: &nodefault.Bool{B: false},
-		JobID:   i.sjid,
+		FSF:            i.sfilter.AsFilter(),
+		Encrypt:        &nodefault.Bool{B: false},
+		JobID:          i.sjid,
+		BandwidthLimit: bandwidthlimit.NoLimitConfig(),
 	}
 	if i.senderConfigHook != nil {
 		i.senderConfigHook(&senderConfig)
@@ -75,6 +78,7 @@ func (i replicationInvocation) Do(ctx *platformtest.Context) *report.Report {
 		JobID:                      i.rjid,
 		AppendClientIdentity:       false,
 		RootWithoutClientComponent: mustDatasetPath(i.rfsRoot),
+		BandwidthLimit:             bandwidthlimit.NoLimitConfig(),
 	}
 	if i.receiverConfigHook != nil {
 		i.receiverConfigHook(&receiverConfig)
@@ -90,6 +94,7 @@ func (i replicationInvocation) Do(ctx *platformtest.Context) *report.Report {
 		ReplicationConfig: &pdu.ReplicationConfig{
 			Protection: i.guarantee,
 		},
+		SizeEstimationConcurrency: 1,
 	}
 
 	report, wait := replication.Do(
@@ -1054,6 +1059,9 @@ func ReplicationPropertyReplicationWorks(ctx *platformtest.Context) {
 		rep := fsByName[fs]
 		require.Len(ctx, rep.Steps, 1)
 		require.Nil(ctx, rep.PlanError)
+		if rep.StepError != nil && strings.Contains(rep.StepError.Error(), "invalid option 'x'") {
+			ctx.SkipNow() // XXX feature detection
+		}
 		require.Nil(ctx, rep.StepError)
 		require.Len(ctx, rep.Steps, 1)
 		require.Equal(ctx, 1, rep.CurrentStep)

rpc/dataconn/dataconn_server.go

@@ -241,7 +241,7 @@ func (s *Server) serveConnRequest(ctx context.Context, endpoint string, c *strea
 		err := c.SendStream(ctx, sendStream, ZFSStream)
 		closeErr := sendStream.Close()
 		if closeErr != nil {
-			s.log.WithError(err).Error("cannot close send stream")
+			s.log.WithError(closeErr).Error("cannot close send stream")
 		}
 		if err != nil {
 			s.log.WithError(err).Error("cannot write send stream")

rpc/transportmux/transportmux.go

@@ -7,6 +7,7 @@ package transportmux
 
 import (
 	"context"
+	"os"
 	"sync/atomic"
 	"syscall"
 
@@ -15,6 +16,7 @@ import (
 	"net"
 	"time"
 
+	"github.com/kr/pretty"
 	"github.com/zrepl/zrepl/daemon/logging"
 	"github.com/zrepl/zrepl/logger"
 	"github.com/zrepl/zrepl/transport"
@@ -151,6 +153,7 @@ func Demux(ctx context.Context, rawListener transport.AuthenticatedListener, lab
 
 			rawConn, err := rawListener.Accept(ctx)
 			if err != nil {
+				fmt.Fprintf(os.Stderr, "transportmux.Demux: rawListener.Accept() returned error: %T %s\n%s\n", err, err, pretty.Sprint(err))
 				if ctx.Err() != nil {
 					return
 				}

rpc/versionhandshake/versionhandshake.go

@@ -10,9 +10,12 @@ import (
 	"fmt"
 	"io"
 	"net"
+	"os"
 	"strings"
 	"time"
 	"unicode/utf8"
+
+	"github.com/kr/pretty"
 )
 
 type HandshakeMessage struct {
@@ -93,6 +96,7 @@ func (m *HandshakeMessage) Encode() ([]byte, error) {
 func (m *HandshakeMessage) DecodeReader(r io.Reader, maxLen int) error {
 	var lenAndSpace [11]byte
 	if _, err := io.ReadFull(r, lenAndSpace[:]); err != nil {
+		fmt.Fprintf(os.Stderr, "HandshakeMessage.DecodeReader error: %T\n%s", err, pretty.Sprint(err))
 		return hsIOErr(err, "error reading protocol banner length: %s", err)
 	}
 	if !utf8.Valid(lenAndSpace[:]) {

rpc/versionhandshake/versionhandshake_transport_wrappers.go

@@ -2,7 +2,9 @@ package versionhandshake
 
 import (
 	"context"
+	"fmt"
 	"net"
+	"os"
 	"time"
 
 	"github.com/zrepl/zrepl/transport"
@@ -23,6 +25,7 @@ func (c HandshakeConnecter) Connect(ctx context.Context) (transport.Wire, error)
 		dl = time.Now().Add(c.timeout)
 	}
 	if err := DoHandshakeCurrentVersion(conn, dl); err != nil {
+		fmt.Fprintf(os.Stderr, "HandshakeConnecter error: %T\n\t%s\n\t%v\n\t%#v\n\n", err, err, err, err)
 		conn.Close()
 		return nil, err
 	}

transport/tcp/serve_tcp_ipmap_test.go

@@ -91,15 +91,12 @@ func TestIPMap(t *testing.T) {
 				"fde4:8dba:82e1::/64": "sub64-*",
 			},
 			expect: map[string]testCaseExpect{
-				"10.1.2.3":        {expectNoMapping: true},
-				"192.168.23.1":    {expectIdent: "db-192.168.23.1"},
-				"192.168.23.23":   {expectIdent: "db-twentythree"},
-				"192.168.023.001": {expectIdent: "db-192.168.23.1"},
-				"10.1.4.5":        {expectIdent: "my-10.1.4.5-server"},
+				"10.1.2.3":      {expectNoMapping: true},
+				"192.168.23.1":  {expectIdent: "db-192.168.23.1"},
+				"192.168.42.1":  {expectIdent: "web-192.168.42.1"},
+				"192.168.23.23": {expectIdent: "db-twentythree"},
+				"10.1.4.5":      {expectIdent: "my-10.1.4.5-server"},
 
-				// normalization
-				"192.168.42.1":    {expectIdent: "web-192.168.42.1"},
-				"192.168.042.001": {expectIdent: "web-192.168.42.1"},
 				// v6 matching
 				"fe80::23:42%eth1": {expectIdent: "san-fe80::23:42-eth1"},
 				"fe80::23:42%eth2": {expectNoMapping: true},
@@ -179,7 +176,8 @@ func TestIPMap(t *testing.T) {
 			for input, expect := range c.expect {
 				// reuse newIPMapEntry to parse test case input
 				// "test" is not used during testing but must not be empty.
-				ipMapEntry, _ := newIPMapEntry(input, "test")
+				ipMapEntry, err := newIPMapEntry(input, "test")
+				require.NoError(t, err)
 				ones, bits := ipMapEntry.subnet.Mask.Size()
 				require.Equal(t, bits, net.IPv6len*8, "and we know ipMapEntry always expands its IPs to 16bytes")
 				require.Equal(t, ones, net.IPv6len*8, "test case addresses must be fully specified")

transport/tls/connect_tls.go

@@ -3,7 +3,9 @@ package tls
 import (
 	"context"
 	"crypto/tls"
+	"fmt"
 	"net"
+	"os"
 
 	"github.com/pkg/errors"
 
@@ -48,6 +50,7 @@ func TLSConnecterFromConfig(in *config.TLSConnect) (*TLSConnecter, error) {
 func (c *TLSConnecter) Connect(dialCtx context.Context) (transport.Wire, error) {
 	conn, err := c.dialer.DialContext(dialCtx, "tcp", c.Address)
 	if err != nil {
+		fmt.Fprintf(os.Stderr, "tls connecter error %T\n\t%s\n\t%v\n\t%#v\n\n", err, err, err, err)
 		return nil, err
 	}
 	tcpConn := conn.(*net.TCPConn)

transport/tls/issue-517/main.go

@@ -0,0 +1,140 @@
+package main
+
+import (
+	"context"
+	"flag"
+	"io"
+	"log"
+	"os"
+	"strings"
+	"time"
+
+	"github.com/zrepl/zrepl/config"
+	"github.com/zrepl/zrepl/transport/tls"
+)
+
+var servConf = config.TLSServe{
+	ServeCommon: config.ServeCommon{
+		Type: "tls",
+	},
+	HandshakeTimeout: 10 * time.Second,
+}
+
+var clientConf = config.TLSConnect{
+	ConnectCommon: config.ConnectCommon{
+		Type: "",
+	},
+	Address:     "",
+	Ca:          "",
+	Cert:        "",
+	Key:         "",
+	ServerCN:    "",
+	DialTimeout: 10 * time.Second,
+}
+
+var ca string
+var mode string
+
+func main() {
+
+	flag.StringVar(&mode, "mode", "", "server|client")
+
+	flag.StringVar(&ca, "ca", "", "path")
+
+	flag.StringVar(&servConf.Listen, "serve.listen", "", "")
+	flag.StringVar(&servConf.Cert, "serve.cert", "", "path")
+	flag.StringVar(&servConf.Key, "serve.key", "", "path")
+	var clientCN string
+	flag.StringVar(&clientCN, "serve.client_cn", "", "")
+
+	flag.StringVar(&clientConf.Address, "client.address", "", "")
+	flag.StringVar(&clientConf.Cert, "client.cert", "", "path")
+	flag.StringVar(&clientConf.Key, "client.key", "", "path")
+	flag.StringVar(&clientConf.ServerCN, "client.server_cn", "", "")
+
+	flag.Parse()
+
+	servConf.ClientCNs = append(servConf.ClientCNs, clientCN)
+
+	servConf.Ca = ca
+	clientConf.Ca = ca
+
+	switch mode {
+	case "server":
+		server()
+	case "client":
+		client()
+	default:
+		panic(mode)
+	}
+
+}
+
+func server() {
+
+	servFactory, err := tls.TLSListenerFactoryFromConfig(nil, &servConf)
+	if err != nil {
+		panic(err)
+	}
+
+	listener, err := servFactory()
+	if err != nil {
+		panic(err)
+	}
+
+	ctx := context.Background()
+
+	for {
+		conn, err := listener.Accept(ctx)
+		if err != nil {
+			log.Printf("accept error: %s", err)
+			continue
+		}
+		go func() {
+			defer conn.Close()
+
+			log.Printf("handling connection %s", conn.RemoteAddr())
+			_, err = io.Copy(conn, strings.NewReader("here is the server\n"))
+			if err != nil {
+				log.Printf("%s: respond to client error: %s", conn.RemoteAddr(), err)
+				return
+			}
+
+			err = conn.CloseWrite()
+			if err != nil {
+				log.Printf("%s: failed to close write connection: err", conn.RemoteAddr(), err)
+			}
+
+			log.Printf("%s: waiting for client to close connection", conn.RemoteAddr())
+
+			_, err = io.Copy(io.Discard, conn)
+			if err != nil {
+				log.Printf("%s: error draining client connection: %s", conn.RemoteAddr(), err)
+				return
+			}
+
+			log.Printf("%s: done", conn.RemoteAddr())
+			return
+		}()
+	}
+
+}
+
+func client() {
+	connecter, err := tls.TLSConnecterFromConfig(&clientConf)
+	if err != nil {
+		panic(err)
+	}
+
+	ctx := context.Background()
+	conn, err := connecter.Connect(ctx)
+	if err != nil {
+		panic(err)
+	}
+	defer conn.Close()
+
+	_, err = io.Copy(os.Stdout, conn)
+	if err != nil {
+		panic(err)
+	}
+}

transport/tls/serve_tls.go

@@ -4,8 +4,10 @@ import (
 	"context"
 	"crypto/tls"
 	"fmt"
+	"os"
 	"time"
 
+	"github.com/kr/pretty"
 	"github.com/pkg/errors"
 
 	"github.com/zrepl/zrepl/config"
@@ -68,6 +70,7 @@ type tlsAuthListener struct {
 func (l tlsAuthListener) Accept(ctx context.Context) (*transport.AuthConn, error) {
 	tcpConn, tlsConn, cn, err := l.ClientAuthListener.Accept()
 	if err != nil {
+		fmt.Fprintf(os.Stderr, "tlsAuthListener.Accept: l.ClientAuthListener.Accept returned error %T %s\n%s\n", err, err, pretty.Sprint(err))
 		return nil, err
 	}
 	if _, ok := l.clientCNs[cn]; !ok {

util/bandwidthlimit/bandwidthlimit.go

@@ -0,0 +1,79 @@
+package bandwidthlimit
+
+import (
+	"errors"
+	"io"
+
+	"github.com/juju/ratelimit"
+)
+
+type Wrapper interface {
+	WrapReadCloser(io.ReadCloser) io.ReadCloser
+}
+
+type Config struct {
+	// Units in this struct are in _bytes_.
+
+	Max            int64 // < 0 means no limit, BucketCapacity is irrelevant then
+	BucketCapacity int64
+}
+
+func NoLimitConfig() Config {
+	return Config{
+		Max:            -1,
+		BucketCapacity: -1,
+	}
+}
+
+func ValidateConfig(conf Config) error {
+	if conf.BucketCapacity == 0 {
+		return errors.New("BucketCapacity must not be zero")
+	}
+	return nil
+}
+
+func WrapperFromConfig(conf Config) Wrapper {
+	if err := ValidateConfig(conf); err != nil {
+		panic(err)
+	}
+
+	if conf.Max < 0 {
+		return noLimit{}
+	}
+
+	return &withLimit{
+		bucket: ratelimit.NewBucketWithRate(float64(conf.Max), conf.BucketCapacity),
+	}
+}
+
+type noLimit struct{}
+
+func (_ noLimit) WrapReadCloser(rc io.ReadCloser) io.ReadCloser { return rc }
+
+type withLimit struct {
+	bucket *ratelimit.Bucket
+}
+
+func (l *withLimit) WrapReadCloser(rc io.ReadCloser) io.ReadCloser {
+	return WrapReadCloser(rc, l.bucket)
+}
+
+type withLimitReadCloser struct {
+	orig    io.Closer
+	limited io.Reader
+}
+
+func (r *withLimitReadCloser) Read(buf []byte) (int, error) {
+	return r.limited.Read(buf)
+}
+
+func (r *withLimitReadCloser) Close() error {
+	return r.orig.Close()
+}
+
+func WrapReadCloser(rc io.ReadCloser, bucket *ratelimit.Bucket) io.ReadCloser {
+	return &withLimitReadCloser{
+		limited: ratelimit.Reader(rc, bucket),
+		orig:    rc,
+	}
+}

util/bandwidthlimit/bandwidthlimit_test.go

@@ -0,0 +1,19 @@
+package bandwidthlimit
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestNoLimitConfig(t *testing.T) {
+
+	conf := NoLimitConfig()
+
+	err := ValidateConfig(conf)
+	require.NoError(t, err)
+
+	require.NotPanics(t, func() {
+		_ = WrapperFromConfig(conf)
+	})
+}

util/datasizeunit/datasizeunit.go

@@ -0,0 +1,86 @@
+package datasizeunit
+
+import (
+	"errors"
+	"fmt"
+	"math"
+	"regexp"
+	"strconv"
+	"strings"
+)
+
+type Bits struct {
+	bits float64
+}
+
+func (b Bits) ToBits() float64    { return b.bits }
+func (b Bits) ToBytes() float64   { return b.bits / 8 }
+func FromBytesInt64(i int64) Bits { return Bits{float64(i) * 8} }
+
+var datarateRegex = regexp.MustCompile(`^([-0-9\.]*)\s*(bit|(|K|Ki|M|Mi|G|Gi|T|Ti)([bB]))$`)
+
+func (r *Bits) UnmarshalYAML(u func(interface{}, bool) error) (_ error) {
+
+	var s string
+	err := u(&s, false)
+	if err != nil {
+		return err
+	}
+
+	genericErr := func(err error) error {
+		var buf strings.Builder
+		fmt.Fprintf(&buf, "cannot parse %q using regex %s", s, datarateRegex)
+		if err != nil {
+			fmt.Fprintf(&buf, ": %s", err)
+		}
+		return errors.New(buf.String())
+	}
+
+	match := datarateRegex.FindStringSubmatch(s)
+	if match == nil {
+		return genericErr(nil)
+	}
+
+	bps, err := strconv.ParseFloat(match[1], 64)
+	if err != nil {
+		return genericErr(err)
+	}
+
+	if match[2] == "bit" {
+		if math.Round(bps) != bps {
+			return genericErr(fmt.Errorf("unit bit must be an integer value"))
+		}
+		r.bits = bps
+		return nil
+	}
+
+	factorMap := map[string]uint64{
+		"": 1,
+
+		"K": 1e3,
+		"M": 1e6,
+		"G": 1e9,
+		"T": 1e12,
+
+		"Ki": 1 << 10,
+		"Mi": 1 << 20,
+		"Gi": 1 << 30,
+		"Ti": 1 << 40,
+	}
+	factor, ok := factorMap[match[3]]
+	if !ok {
+		panic(match)
+	}
+
+	baseUnitFactorMap := map[string]uint64{
+		"b": 1,
+		"B": 8,
+	}
+	baseUnitFactor, ok := baseUnitFactorMap[match[4]]
+	if !ok {
+		panic(match)
+	}
+
+	r.bits = bps * float64(factor) * float64(baseUnitFactor)
+	return nil
+}

util/datasizeunit/datasizeunit_test.go

@@ -0,0 +1,57 @@
+package datasizeunit
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"github.com/zrepl/yaml-config"
+)
+
+func TestBits(t *testing.T) {
+
+	tcs := []struct {
+		input      string
+		expectRate float64
+		expectErr  string
+	}{
+		{`23 bit`, 23, ""}, // bit special case works
+		{`23bit`, 23, ""},  // also without space
+
+		{`10MiB`, 10 * (1 << 20) * 8, ""},  // integer unit without space
+		{`10 MiB`, 8 * 10 * (1 << 20), ""}, // integer unit with space
+
+		{`10.5 Kib`, 10.5 * (1 << 10), ""}, // floating point with bit unit works with space
+		{`10.5Kib`, 10.5 * (1 << 10), ""},  // floating point with bit unit works without space
+
+		// unit checks
+		{`1 bit`, 1, ""},
+		{`1 B`, 1 * 8, ""},
+		{`1 Kb`, 1e3, ""},
+		{`1 Kib`, 1 << 10, ""},
+		{`1 Mb`, 1e6, ""},
+		{`1 Mib`, 1 << 20, ""},
+		{`1 Gb`, 1e9, ""},
+		{`1 Gib`, 1 << 30, ""},
+		{`1 Tb`, 1e12, ""},
+		{`1 Tib`, 1 << 40, ""},
+	}
+
+	for _, tc := range tcs {
+		t.Run(tc.input, func(t *testing.T) {
+
+			var bits Bits
+			err := yaml.Unmarshal([]byte(tc.input), &bits)
+			if tc.expectErr != "" {
+				assert.Error(t, err)
+				assert.Regexp(t, tc.expectErr, err.Error())
+				assert.Zero(t, bits.bits)
+			} else {
+				require.NoError(t, err)
+				assert.Equal(t, tc.expectRate, bits.bits)
+			}
+		})
+
+	}
+
+}