transport/ssh: bump go-netssh version to fix ssh client process leaks

fixes #322

Makefile

@@ -23,7 +23,8 @@ GOHOSTARCH ?= $(shell bash -c 'source <($(GO) env) && echo "$$GOHOSTARCH"')
 GO_ENV_VARS := GO111MODULE=on
 GO_LDFLAGS := "-X github.com/zrepl/zrepl/version.zreplVersion=$(_ZREPL_VERSION)"
 GO_MOD_READONLY := -mod=readonly
-GO_BUILDFLAGS := $(GO_MOD_READONLY)
+GO_EXTRA_BUILDFLAGS :=
+GO_BUILDFLAGS := $(GO_MOD_READONLY) $(GO_EXTRA_BUILDFLAGS)
 GO_BUILD := $(GO_ENV_VARS) $(GO) build $(GO_BUILDFLAGS) -ldflags $(GO_LDFLAGS)
 GOLANGCI_LINT := golangci-lint
 ifneq ($(GOARM),)

client/migrate.go

@@ -211,7 +211,7 @@ func doMigrateReplicationCursorFS(ctx context.Context, v1CursorJobs []job.Job, f
 	}
 	fmt.Printf("identified owning job %q\n", owningJob.Name())
 
-	bookmarks, err := zfs.ZFSListFilesystemVersions(fs, zfs.ListFilesystemVersionsOptions{
+	bookmarks, err := zfs.ZFSListFilesystemVersions(ctx, fs, zfs.ListFilesystemVersionsOptions{
 		Types: zfs.Bookmarks,
 	})
 	if err != nil {

config/samples/source.yml

@@ -5,7 +5,11 @@ jobs:
     type: tcp
     listen: "0.0.0.0:8888"
     clients: {
-      "192.168.122.123" : "client1"
+      "192.168.122.123" :               "mysql01",
+      "192.168.122.42" :                "mx01",
+      "2001:0db8:85a3::8a2e:0370:7334": "gateway",
+      "10.23.42.0/24":       "cluster-*",
+      "fde4:8dba:82e1::/64": "san-*",
     }
   filesystems: {
     "<": true,

daemon/snapper/snapper.go

@@ -485,7 +485,7 @@ var findSyncPointFSNoFilesystemVersionsErr = fmt.Errorf("no filesystem versions"
 
 func findSyncPointFSNextOptimalSnapshotTime(l Logger, now time.Time, interval time.Duration, prefix string, d *zfs.DatasetPath) (time.Time, error) {
 
-	fsvs, err := zfs.ZFSListFilesystemVersions(d, zfs.ListFilesystemVersionsOptions{
+	fsvs, err := zfs.ZFSListFilesystemVersions(context.TODO(), d, zfs.ListFilesystemVersionsOptions{
 		Types:           zfs.Snapshots,
 		ShortnamePrefix: prefix,
 	})

docs/configuration/transports.rst

@@ -52,8 +52,15 @@ Serve
         listen: ":8888"
         listen_freebind: true # optional, default false
         clients: {
-          "192.168.122.123" : "mysql01"
-          "192.168.122.123" : "mx01"
+          "192.168.122.123" :               "mysql01",
+          "192.168.122.42" :                "mx01",
+          "2001:0db8:85a3::8a2e:0370:7334": "gateway",
+
+          # CIDR masks require a '*' in the client identity string
+          # that is expanded to the client's IP address
+
+          "10.23.42.0/24":       "cluster-*"
+          "fde4:8dba:82e1::/64": "san-*"
         }
       ...
 

docs/installation.rst

@@ -1,138 +1,19 @@
-.. _binary releases: https://github.com/zrepl/zrepl/releases
 
-.. _installation:
+.. _installation_toc:
 
+************
 Installation
-============
+************
 
 .. TIP::
 
     Note: check out the :ref:`tutorial` if you want a first impression of zrepl.
 
-User Privileges
----------------
+.. toctree::
 
-It is possible to run zrepl as an unprivileged user in combination with
-`ZFS delegation <https://www.freebsd.org/doc/handbook/zfs-zfs-allow.html>`_.
-Also, there is the possibility to run it in a jail on FreeBSD by delegating a dataset to the jail.
-However, until we get around documenting those setups, you will have to run zrepl as root or experiment yourself :)
-
-Packages
---------
-
-zrepl source releases are signed & tagged by the author in the git repository.
-Your OS vendor may provide binary packages of zrepl through the package manager.
-Additionally, `binary releases`_ are provided on GitHub.
-The following list may be incomplete, feel free to submit a PR with an update:
-
-.. list-table::
-    :header-rows: 1
-
-    * - OS / Distro
-      - Install Command
-      - Link
-    * - FreeBSD
-      - ``pkg install zrepl``
-      - `<https://www.freshports.org/sysutils/zrepl/>`_
-    * - MacOS
-      - ``brew install zrepl``
-      - Available on `homebrew <https://brew.sh>`_
-    * - Arch Linux
-      - ``yay install zrepl``
-      - Available on `AUR <https://aur.archlinux.org/packages/zrepl>`_
-    * - Fedora
-      - ``dnf install zrepl``
-      - Available on `COPR <https://copr.fedorainfracloud.org/coprs/poettlerric/zrepl/>`_
-    * - CentOS/RHEL
-      - ``yum install zrepl``
-      - Available on `COPR <https://copr.fedorainfracloud.org/coprs/poettlerric/zrepl/>`_
-    * - Debian + Ubuntu
-      - ``apt install zrepl``
-      - APT repository config :ref:`see below <installation-apt-repos>`
-    * - OmniOS
-      - ``pkg install zrepl``
-      - Available since `r151030 <https://pkg.omniosce.org/r151030/extra/en/search.shtml?token=zrepl&action=Search>`_
-    * - Void Linux
-      - ``xbps-install zrepl``
-      - Available since `a88a2a4 <https://github.com/void-linux/void-packages/commit/a88a2a4d7bf56072dadf61ab56b8424e39155890>`_
-    * - Others
-      -
-      - Use `binary releases`_ or build from source.
-
-.. _installation-apt-repos:
-
-Debian / Ubuntu APT repositories
-~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
-
-We maintain APT repositories for Debian, Ubuntu and derivatives.
-The fingerprint of the signing key is ``E101 418F D3D6 FBCB 9D65  A62D 7086 99FC 5F2E BF16``.
-It is available at `<https://zrepl.cschwarz.com/apt/apt-key.asc>`_ .
-Please open an issue `in the packaging repository <https://github.com/zrepl/debian-binary-packaging>`_ if you encounter any issues with the repository.
-
-The following snippet configure the repository for your Debian or Ubuntu release:
-
-::
-
-    apt update && apt install curl gnupg lsb-release; \
-    ARCH="$(dpkg --print-architecture)"; \
-    CODENAME="$(lsb_release -i -s | tr '[:upper:]' '[:lower:]') $(lsb_release -c -s | tr '[:upper:]' '[:lower:]')"; \
-    echo "Using Distro and Codename: $CODENAME"; \
-    (curl https://zrepl.cschwarz.com/apt/apt-key.asc | apt-key add -) && \
-    (echo "deb [arch=$ARCH] https://zrepl.cschwarz.com/apt/$CODENAME main" > /etc/apt/sources.list.d/zrepl.list) && \
-    apt update
-
-
-.. NOTE::
-
-   Until zrepl reaches 1.0, all APT repositories will be updated to the latest zrepl release immediately.
-   This includes breaking changes between zrepl versions.
-   Use ``apt-mark hold zrepl`` to prevent upgrades of zrepl.
-
-Compile From Source
-~~~~~~~~~~~~~~~~~~~
-
-Producing a release requires **Go 1.11** or newer and **Python 3** + **pip3** + ``docs/requirements.txt`` for the Sphinx documentation.
-A tutorial to install Go is available over at `golang.org <https://golang.org/doc/install>`_.
-Python and pip3 should probably be installed via your distro's package manager.
-
-Alternatively, you can use the Docker build process:
-it is used to produce the official zrepl `binary releases`_
-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 . && \
-    sudo docker run -it --rm \
-        -v "${PWD}:/src" \
-        --user "$(id -u):$(id -g)" \
-        zrepl_build make release
-
-Alternatively, you can install build dependencies on your local system and then build in your ``$GOPATH``:
-
-::
-
-    mkdir -p "${GOPATH}/src/github.com/zrepl/zrepl"
-    git clone https://github.com/zrepl/zrepl.git "${GOPATH}/src/github.com/zrepl/zrepl"
-    cd "${GOPATH}/src/github.com/zrepl/zrepl"
-    python3 -m venv3
-    source venv3/bin/activate
-    ./lazy.sh devsetup
-    make release
-
-The Python venv is used for the documentation build dependencies.
-If you just want to build the zrepl binary, leave it out and use `./lazy.sh godep` instead.
-Either way, all build results are located in the ``artifacts/`` directory.
-
-.. NOTE::
-
-    It is your job to install the appropriate binary in the zrepl users's ``$PATH``, e.g. ``/usr/local/bin/zrepl``.
-    Otherwise, the examples in the :ref:`tutorial` may need to be adjusted.
-
-What next?
-----------
-
-Read the :ref:`configuration chapter<configuration_toc>` and then continue with the :ref:`usage chapter<usage>`.
-
-**Reminder**: If you want a quick introduction, please read the :ref:`tutorial`.
+    installation/user-privileges
+    installation/packages
+    installation/apt-repos
+    installation/compile-from-source
+    installation/freebsd-jail-with-iocage
+    installation/what-next

docs/installation/apt-repos.rst

@@ -0,0 +1,29 @@
+
+.. _installation-apt-repos:
+
+Debian / Ubuntu APT repositories
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+We maintain APT repositories for Debian, Ubuntu and derivatives.
+The fingerprint of the signing key is ``E101 418F D3D6 FBCB 9D65  A62D 7086 99FC 5F2E BF16``.
+It is available at `<https://zrepl.cschwarz.com/apt/apt-key.asc>`_ .
+Please open an issue `in the packaging repository <https://github.com/zrepl/debian-binary-packaging>`_ if you encounter any issues with the repository.
+
+The following snippet configure the repository for your Debian or Ubuntu release:
+
+::
+
+    apt update && apt install curl gnupg lsb-release; \
+    ARCH="$(dpkg --print-architecture)"; \
+    CODENAME="$(lsb_release -i -s | tr '[:upper:]' '[:lower:]') $(lsb_release -c -s | tr '[:upper:]' '[:lower:]')"; \
+    echo "Using Distro and Codename: $CODENAME"; \
+    (curl https://zrepl.cschwarz.com/apt/apt-key.asc | apt-key add -) && \
+    (echo "deb [arch=$ARCH] https://zrepl.cschwarz.com/apt/$CODENAME main" > /etc/apt/sources.list.d/zrepl.list) && \
+    apt update
+
+
+.. NOTE::
+
+   Until zrepl reaches 1.0, all APT repositories will be updated to the latest zrepl release immediately.
+   This includes breaking changes between zrepl versions.
+   Use ``apt-mark hold zrepl`` to prevent upgrades of zrepl.

docs/installation/compile-from-source.rst

@@ -0,0 +1,45 @@
+.. _binary releases: https://github.com/zrepl/zrepl/releases
+
+.. _installation-compile-from-source:
+
+Compile From Source
+~~~~~~~~~~~~~~~~~~~
+
+Producing a release requires **Go 1.11** or newer and **Python 3** + **pip3** + ``docs/requirements.txt`` for the Sphinx documentation.
+A tutorial to install Go is available over at `golang.org <https://golang.org/doc/install>`_.
+Python and pip3 should probably be installed via your distro's package manager.
+
+Alternatively, you can use the Docker build process:
+it is used to produce the official zrepl `binary releases`_
+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 . && \
+    sudo docker run -it --rm \
+        -v "${PWD}:/src" \
+        --user "$(id -u):$(id -g)" \
+        zrepl_build make release
+
+Alternatively, you can install build dependencies on your local system and then build in your ``$GOPATH``:
+
+::
+
+    mkdir -p "${GOPATH}/src/github.com/zrepl/zrepl"
+    git clone https://github.com/zrepl/zrepl.git "${GOPATH}/src/github.com/zrepl/zrepl"
+    cd "${GOPATH}/src/github.com/zrepl/zrepl"
+    python3 -m venv3
+    source venv3/bin/activate
+    ./lazy.sh devsetup
+    make release
+
+The Python venv is used for the documentation build dependencies.
+If you just want to build the zrepl binary, leave it out and use `./lazy.sh godep` instead.
+Either way, all build results are located in the ``artifacts/`` directory.
+
+.. NOTE::
+
+    It is your job to install the appropriate binary in the zrepl users's ``$PATH``, e.g. ``/usr/local/bin/zrepl``.
+    Otherwise, the examples in the :ref:`tutorial` may need to be adjusted.

docs/installation/freebsd-jail-with-iocage.rst

@@ -0,0 +1,140 @@
+.. include:: ../global.rst.inc
+
+.. _installation-freebsd-jail-with-iocage:
+
+FreeBSD Jail With iocage
+========================
+
+
+This tutorial shows how zrepl can be installed on FreeBSD, or FreeNAS in a jail using iocage.
+While this tutorial focuses on using iocage, much of the setup would be similar
+using a different jail manager.
+
+.. NOTE::
+
+   From a security perspective, just keep in mind that ``zfs send``/``recv`` was never designed with
+   jails in mind, an attacker could probably crash the receive-side kernel or worse induce stateful
+   damage to the receive-side pool if they were able to get access to the jail.
+
+   The jail doesn't provide security benefits, but only management ones.
+
+Requirements
+------------
+
+A dataset that will be delegated to the jail needs to be created if one does not already exist.
+For the tutorial ``tank/zrepl`` will be used.
+
+.. code-block:: bash
+
+   zfs create -o mountpoint=none tank/zrepl
+
+The only software requirements on the host system are ``iocage``, which can be installed
+from ports or packages.
+
+.. code-block:: bash
+
+   pkg install py37-iocage
+
+.. NOTE::
+
+   By default ``iocage`` will "activate" on first use which will set up some defaults such as
+   which pool will be used. To activate ``iocage`` manually the ``iocage activate`` command can be used.
+
+Jail Creation
+-------------
+
+There are two options for jail creation using FreeBSD.
+
+1. Manually set up the jail from scratch
+2. Create the jail using the ``zrepl`` plugin. On FreeNAS this is possible from the user interface using the community index.
+
+Manual Jail
+###########
+
+Create a jail, using the same release as the host, called ``zrepl`` that will be automatically started at boot.
+The jail will have ``tank/zrepl`` delegated into it.
+
+.. code-block:: bash
+
+   iocage create --release "$(freebsd-version -k | cut -d '-' -f '1,2')" --name zrepl \
+          boot=on nat=1 \
+          jail_zfs=on \
+          jail_zfs_dataset=zrepl \
+          jail_zfs_mountpoint='none'
+
+Enter the jail:
+
+.. code-block:: bash
+
+   iocage console zrepl
+
+Install ``zrepl``
+
+.. code-block:: bash
+
+   pkg update && pkg upgrade
+   pkg install zrepl
+
+Create the log file ``/var/log/zrepl.log``
+
+.. code-block:: bash
+
+   touch /var/log/zrepl.log && service newsyslog restart
+
+Tell syslogd to redirect facility local0 to the ``zrepl.log`` file:
+
+.. code-block:: bash
+
+   service syslogd reload
+
+Enable the zrepl daemon to start automatically at boot:
+
+.. code-block:: bash
+
+   sysrc zrepl_enable="YES"
+
+
+Plugin
+######
+
+When using the plugin, ``zrepl`` will be installed for you in a jail using the following ``iocage`` properties.
+
+* ``nat=1``
+* ``jail_zfs=on``
+* ``jail_zfs_mountpoint=none``
+
+Additionally the delegated dataset should be specified upon creation, and optionally start on boot can be set.
+This can also be done from the FreeNAS webui.
+
+.. code-block:: bash
+
+   fetch https://raw.githubusercontent.com/ix-plugin-hub/iocage-plugin-index/master/zrepl.json -o /tmp/zrepl.json
+   iocage fetch -P /tmp/zrepl.json --name zrepl jail_zfs_dataset=zrepl boot=on
+
+Configuration
+-------------
+
+Now ``zrepl`` can be configured.
+
+Enter the jail.
+
+.. code-block:: bash
+
+   iocage console zrepl
+
+Modify the ``/usr/local/etc/zrepl/zrepl.yml`` configuration file.
+
+.. TIP::
+
+    Note: check out the :ref:`tutorial` for examples of a ``sink`` job.
+
+Now ``zrepl`` can be started.
+
+.. code-block:: bash
+
+   service zrepl start
+
+Summary
+-------
+
+Congratulations, you have a working jail!

docs/installation/packages.rst

@@ -0,0 +1,50 @@
+.. _installation-packages:
+
+.. _binary releases: https://github.com/zrepl/zrepl/releases
+
+Packages
+--------
+
+zrepl source releases are signed & tagged by the author in the git repository.
+Your OS vendor may provide binary packages of zrepl through the package manager.
+Additionally, `binary releases`_ are provided on GitHub.
+The following list may be incomplete, feel free to submit a PR with an update:
+
+.. list-table::
+    :header-rows: 1
+
+    * - OS / Distro
+      - Install Command
+      - Link
+    * - FreeBSD
+      - ``pkg install zrepl``
+      - `<https://www.freshports.org/sysutils/zrepl/>`_
+      
+        :ref:`installation-freebsd-jail-with-iocage`
+    * - FreeNAS
+      -
+      - :ref:`installation-freebsd-jail-with-iocage`
+    * - MacOS
+      - ``brew install zrepl``
+      - Available on `homebrew <https://brew.sh>`_
+    * - Arch Linux
+      - ``yay install zrepl``
+      - Available on `AUR <https://aur.archlinux.org/packages/zrepl>`_
+    * - Fedora
+      - ``dnf install zrepl``
+      - Available on `COPR <https://copr.fedorainfracloud.org/coprs/poettlerric/zrepl/>`_
+    * - CentOS/RHEL
+      - ``yum install zrepl``
+      - Available on `COPR <https://copr.fedorainfracloud.org/coprs/poettlerric/zrepl/>`_
+    * - Debian + Ubuntu
+      - ``apt install zrepl``
+      - APT repository config :ref:`see below <installation-apt-repos>`
+    * - OmniOS
+      - ``pkg install zrepl``
+      - Available since `r151030 <https://pkg.omniosce.org/r151030/extra/en/search.shtml?token=zrepl&action=Search>`_
+    * - Void Linux
+      - ``xbps-install zrepl``
+      - Available since `a88a2a4 <https://github.com/void-linux/void-packages/commit/a88a2a4d7bf56072dadf61ab56b8424e39155890>`_
+    * - Others
+      -
+      - Use `binary releases`_ or build from source.

docs/installation/user-privileges.rst

@@ -0,0 +1,12 @@
+.. _installation-user-privileges:
+
+User Privileges
+---------------
+
+It is possible to run zrepl as an unprivileged user in combination with
+`ZFS delegation <https://www.freebsd.org/doc/handbook/zfs-zfs-allow.html>`_.
+Also, there is the possibility to run it in a jail on FreeBSD by delegating a dataset to the jail.
+
+.. TIP::
+
+    Note: check out the :ref:`installation-freebsd-jail-with-iocage` for FreeBSD jail setup instructions.

docs/installation/what-next.rst

@@ -0,0 +1,8 @@
+.. _installation-what-next:
+
+What next?
+----------
+
+Read the :ref:`configuration chapter<configuration_toc>` and then continue with the :ref:`usage chapter<usage>`.
+
+**Reminder**: If you want a quick introduction, please read the :ref:`tutorial`.

docs/tutorial.rst

@@ -47,7 +47,7 @@ We can model this situation as two jobs:
 Install zrepl
 -------------
 
-Follow the :ref:`OS-specific installation instructions <installation>` and come back here.
+Follow the :ref:`OS-specific installation instructions <installation_toc>` and come back here.
 
 Generate TLS Certificates
 -------------------------

endpoint/endpoint.go

@@ -96,7 +96,7 @@ func (s *Sender) ListFilesystemVersions(ctx context.Context, r *pdu.ListFilesyst
 	if err != nil {
 		return nil, err
 	}
-	fsvs, err := zfs.ZFSListFilesystemVersions(lp, zfs.ListFilesystemVersionsOptions{})
+	fsvs, err := zfs.ZFSListFilesystemVersions(ctx, lp, zfs.ListFilesystemVersionsOptions{})
 	if err != nil {
 		return nil, err
 	}
@@ -648,7 +648,7 @@ func (s *Receiver) ListFilesystemVersions(ctx context.Context, req *pdu.ListFile
 	}
 	// TODO share following code with sender
 
-	fsvs, err := zfs.ZFSListFilesystemVersions(lp, zfs.ListFilesystemVersionsOptions{})
+	fsvs, err := zfs.ZFSListFilesystemVersions(ctx, lp, zfs.ListFilesystemVersionsOptions{})
 	if err != nil {
 		return nil, err
 	}
@@ -806,7 +806,7 @@ func (s *Receiver) Receive(ctx context.Context, req *pdu.ReceiveReq, receive zfs
 	if err := zfs.ZFSRecv(ctx, lp.ToString(), to, receive, recvOpts); err != nil {
 		getLogger(ctx).
 			WithError(err).
-			WithField("opts", recvOpts).
+			WithField("opts", fmt.Sprintf("%#v", recvOpts)).
 			Error("zfs receive failed")
 		return nil, err
 	}

endpoint/endpoint_zfs_abstraction.go

@@ -467,7 +467,7 @@ func (e ListAbstractionsErrors) Error() string {
 	}
 	msgs := make([]string, len(e))
 	for i := range e {
-		msgs[i] = e.Error()
+		msgs[i] = e[i].Error()
 	}
 	return fmt.Sprintf("list endpoint abstractions: multiple errors:\n%s", strings.Join(msgs, "\n"))
 }
@@ -573,7 +573,7 @@ func listAbstractionsImplFS(ctx context.Context, fs string, query *ListZFSHoldsA
 			whatTypes[zfs.Snapshot] = true
 		}
 	}
-	fsvs, err := zfs.ZFSListFilesystemVersions(fsp, zfs.ListFilesystemVersionsOptions{
+	fsvs, err := zfs.ZFSListFilesystemVersions(ctx, fsp, zfs.ListFilesystemVersionsOptions{
 		Types: whatTypes,
 	})
 	if err != nil {
@@ -731,6 +731,9 @@ func listStaleFiltering(abs []Abstraction, sinceBound *CreateTXGRangeBound) *Sta
 	}
 	stepFirstNotStaleCandidates := make(map[fsAndJobId]stepFirstNotStaleCandidate) // empty map => will always return nil
 	for _, a := range abs {
+		if a.GetJobID() == nil {
+			continue // already put those into always-live list noJobId in above loop
+		}
 		key := fsAndJobId{a.GetFS(), *a.GetJobID()}
 		c := stepFirstNotStaleCandidates[key]
 

go.mod

@@ -21,7 +21,7 @@ require (
 	github.com/onsi/gomega v1.7.0 // indirect
 	github.com/pkg/errors v0.8.1
 	github.com/pkg/profile v1.2.1
-	github.com/problame/go-netssh v0.0.0-20191209123953-18d8aa6923c7
+	github.com/problame/go-netssh v0.0.0-20200601114649-26439f9f0dc5
 	github.com/prometheus/client_golang v1.2.1
 	github.com/sergi/go-diff v1.0.1-0.20180205163309-da645544ed44 // go1.12 thinks it needs this
 	github.com/spf13/cobra v0.0.2

go.sum

@@ -208,6 +208,8 @@ github.com/problame/go-netssh v0.0.0-20191026123024-f34099f4f6b1 h1:HH8yzlaZq/A8
 github.com/problame/go-netssh v0.0.0-20191026123024-f34099f4f6b1/go.mod h1:JRs3nyBjvOv/9YHC+Vlw9z1Jfzl5s5t0BNnTzmclj1c=
 github.com/problame/go-netssh v0.0.0-20191209123953-18d8aa6923c7 h1:Oik8tLrLhoMq4fduDRuNNOAQ+q0M0dXkjAYLUf4+mAc=
 github.com/problame/go-netssh v0.0.0-20191209123953-18d8aa6923c7/go.mod h1:Ba6RaFpK1+IHWs1wLZ6sCBuXkt4iAVLeOG5GinXHusM=
+github.com/problame/go-netssh v0.0.0-20200601114649-26439f9f0dc5 h1:1eSrO2WAeSXjMol8WRKW9LekL1252VIMaKQwWkmEdvs=
+github.com/problame/go-netssh v0.0.0-20200601114649-26439f9f0dc5/go.mod h1:Ba6RaFpK1+IHWs1wLZ6sCBuXkt4iAVLeOG5GinXHusM=
 github.com/prometheus/client_golang v0.0.0-20180410130117-e11c6ff8170b/go.mod h1:7SWBe2y4D6OKWSNQJUaRYU/AaXPKyh/dDVn+NZz0KFw=
 github.com/prometheus/client_golang v0.9.1/go.mod h1:7SWBe2y4D6OKWSNQJUaRYU/AaXPKyh/dDVn+NZz0KFw=
 github.com/prometheus/client_golang v1.0.0/go.mod h1:db9x61etRT2tGnBNRi70OPL5FsnadC4Ky3P0J6CfImo=
@@ -263,6 +265,7 @@ github.com/spf13/viper v1.0.2/go.mod h1:A8kyI5cUJhb8N+3pkfONlcEcZbueH6nhAm0Fq7Sr
 github.com/spf13/viper v1.3.2/go.mod h1:ZiWeW+zYFKm7srdB9IoDzzZXaJaI5eL9QjNiN/DMA2s=
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
 github.com/stretchr/objx v0.1.1/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
+github.com/stretchr/testify v1.2.1/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs=
 github.com/stretchr/testify v1.2.2 h1:bSDNvY7ZPG5RlJ8otE/7V6gMiyenm9RtJ7IUVIAoJ1w=
 github.com/stretchr/testify v1.2.2/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs=
 github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=

platformtest/platformtest.go

@@ -27,6 +27,7 @@ func (c *Context) Logf(format string, args ...interface{}) {
 
 func (c *Context) Errorf(format string, args ...interface{}) {
 	GetLog(c).Printf(format, args...)
+	c.FailNow()
 }
 
 func (c *Context) FailNow() {

platformtest/tests/listFilesystemVersions.go

@@ -43,7 +43,7 @@ func ListFilesystemVersionsTypeFilteringAndPrefix(t *platformtest.Context) {
 	fs := fmt.Sprintf("%s/foo bar", t.RootDataset)
 
 	// no options := all types
-	vs, err := zfs.ZFSListFilesystemVersions(mustDatasetPath(fs), zfs.ListFilesystemVersionsOptions{})
+	vs, err := zfs.ZFSListFilesystemVersions(t, mustDatasetPath(fs), zfs.ListFilesystemVersionsOptions{})
 	require.NoError(t, err)
 	require.Equal(t, []string{
 		"#blup 1", "#bookfoo 1", "#bookfoo 2", "#foo 1", "#foo 2",
@@ -51,21 +51,21 @@ func ListFilesystemVersionsTypeFilteringAndPrefix(t *platformtest.Context) {
 	}, versionRelnamesSorted(vs))
 
 	// just snapshots
-	vs, err = zfs.ZFSListFilesystemVersions(mustDatasetPath(fs), zfs.ListFilesystemVersionsOptions{
+	vs, err = zfs.ZFSListFilesystemVersions(t, mustDatasetPath(fs), zfs.ListFilesystemVersionsOptions{
 		Types: zfs.Snapshots,
 	})
 	require.NoError(t, err)
 	require.Equal(t, []string{"@ foo with leading whitespace", "@blup 1", "@foo 1", "@foo 2"}, versionRelnamesSorted(vs))
 
 	// just bookmarks
-	vs, err = zfs.ZFSListFilesystemVersions(mustDatasetPath(fs), zfs.ListFilesystemVersionsOptions{
+	vs, err = zfs.ZFSListFilesystemVersions(t, mustDatasetPath(fs), zfs.ListFilesystemVersionsOptions{
 		Types: zfs.Bookmarks,
 	})
 	require.NoError(t, err)
 	require.Equal(t, []string{"#blup 1", "#bookfoo 1", "#bookfoo 2", "#foo 1", "#foo 2"}, versionRelnamesSorted(vs))
 
 	// just with prefix foo
-	vs, err = zfs.ZFSListFilesystemVersions(mustDatasetPath(fs), zfs.ListFilesystemVersionsOptions{
+	vs, err = zfs.ZFSListFilesystemVersions(t, mustDatasetPath(fs), zfs.ListFilesystemVersionsOptions{
 		ShortnamePrefix: "foo",
 	})
 	require.NoError(t, err)
@@ -82,7 +82,7 @@ func ListFilesystemVersionsZeroExistIsNotAnError(t *platformtest.Context) {
 
 	fs := fmt.Sprintf("%s/foo bar", t.RootDataset)
 
-	vs, err := zfs.ZFSListFilesystemVersions(mustDatasetPath(fs), zfs.ListFilesystemVersionsOptions{})
+	vs, err := zfs.ZFSListFilesystemVersions(t, mustDatasetPath(fs), zfs.ListFilesystemVersionsOptions{})
 	require.Empty(t, vs)
 	require.NoError(t, err)
 	dsne, ok := err.(*zfs.DatasetDoesNotExist)
@@ -98,7 +98,7 @@ func ListFilesystemVersionsFilesystemNotExist(t *platformtest.Context) {
 
 	nonexistentFS := fmt.Sprintf("%s/not existent", t.RootDataset)
 
-	vs, err := zfs.ZFSListFilesystemVersions(mustDatasetPath(nonexistentFS), zfs.ListFilesystemVersionsOptions{})
+	vs, err := zfs.ZFSListFilesystemVersions(t, mustDatasetPath(nonexistentFS), zfs.ListFilesystemVersionsOptions{})
 	require.Empty(t, vs)
 	require.Error(t, err)
 	t.Logf("err = %T\n%s", err, err)
@@ -141,7 +141,7 @@ func ListFilesystemVersionsUserrefs(t *platformtest.Context) {
 
 	fs := fmt.Sprintf("%s/foo bar", t.RootDataset)
 
-	vs, err := zfs.ZFSListFilesystemVersions(mustDatasetPath(fs), zfs.ListFilesystemVersionsOptions{})
+	vs, err := zfs.ZFSListFilesystemVersions(t, mustDatasetPath(fs), zfs.ListFilesystemVersionsOptions{})
 	require.NoError(t, err)
 
 	type expectation struct {

replication/driver/replication_driver.go

@@ -355,7 +355,7 @@ func (a *attempt) do(ctx context.Context, prev *attempt) {
 	// invariant: prevs contains an entry for each unambiguous correspondence
 
 	stepQueue := newStepQueue()
-	defer stepQueue.Start(1)() // TODO parallel replication
+	defer stepQueue.Start(envconst.Int("ZREPL_REPLICATION_EXPERIMENTAL_REPLICATION_CONCURRENCY", 1))() // TODO parallel replication
 	var fssesDone sync.WaitGroup
 	for _, f := range a.fss {
 		fssesDone.Add(1)
@@ -370,54 +370,57 @@ func (a *attempt) do(ctx context.Context, prev *attempt) {
 	a.finishedAt = time.Now()
 }
 
-func (fs *fs) do(ctx context.Context, pq *stepQueue, prev *fs) {
+func (f *fs) debug(format string, args ...interface{}) {
+	debugPrefix("fs=%s", f.fs.ReportInfo().Name)(format, args...)
+}
 
-	defer fs.l.Lock().Unlock()
+func (f *fs) do(ctx context.Context, pq *stepQueue, prev *fs) {
+
+	defer f.l.Lock().Unlock()
 
 	// get planned steps from replication logic
 	var psteps []Step
 	var errTime time.Time
 	var err error
-	fs.l.DropWhile(func() {
+	f.l.DropWhile(func() {
 		// TODO hacky
 		// choose target time that is earlier than any snapshot, so fs planning is always prioritized
 		targetDate := time.Unix(0, 0)
-		defer pq.WaitReady(fs, targetDate)()
-		psteps, err = fs.fs.PlanFS(ctx) // no shadow
-		errTime = time.Now()            // no shadow
+		defer pq.WaitReady(f, targetDate)()
+		psteps, err = f.fs.PlanFS(ctx) // no shadow
+		errTime = time.Now()           // no shadow
 	})
-	debug := debugPrefix("fs=%s", fs.fs.ReportInfo().Name)
-	fs.planning.done = true
+	f.planning.done = true
 	if err != nil {
-		fs.planning.err = newTimedError(err, errTime)
+		f.planning.err = newTimedError(err, errTime)
 		return
 	}
 	for _, pstep := range psteps {
 		step := &step{
-			l:    fs.l,
+			l:    f.l,
 			step: pstep,
 		}
-		fs.planned.steps = append(fs.planned.steps, step)
+		f.planned.steps = append(f.planned.steps, step)
 	}
-	debug("initial len(fs.planned.steps) = %d", len(fs.planned.steps))
+	f.debug("initial len(fs.planned.steps) = %d", len(f.planned.steps))
 
 	// for not-first attempts, only allow fs.planned.steps
 	// up to including the originally planned target snapshot
 	if prev != nil && prev.planning.done && prev.planning.err == nil {
 		prevUncompleted := prev.planned.steps[prev.planned.step:]
 		if len(prevUncompleted) == 0 {
-			debug("prevUncompleted is empty")
+			f.debug("prevUncompleted is empty")
 			return
 		}
-		if len(fs.planned.steps) == 0 {
-			debug("fs.planned.steps is empty")
+		if len(f.planned.steps) == 0 {
+			f.debug("fs.planned.steps is empty")
 			return
 		}
 		prevFailed := prevUncompleted[0]
-		curFirst := fs.planned.steps[0]
+		curFirst := f.planned.steps[0]
 		// we assume that PlanFS retries prevFailed (using curFirst)
 		if !prevFailed.step.TargetEquals(curFirst.step) {
-			debug("Targets don't match")
+			f.debug("Targets don't match")
 			// Two options:
 			// A: planning algorithm is broken
 			// B: manual user intervention inbetween
@@ -433,43 +436,43 @@ func (fs *fs) do(ctx context.Context, pq *stepQueue, prev *fs) {
 			}
 			msg := fmt.Sprintf("last attempt's uncompleted step %s does not correspond to this attempt's first planned step %s",
 				stepFmt(prevFailed), stepFmt(curFirst))
-			fs.planned.stepErr = newTimedError(errors.New(msg), time.Now())
+			f.planned.stepErr = newTimedError(errors.New(msg), time.Now())
 			return
 		}
 		// only allow until step targets diverge
 		min := len(prevUncompleted)
-		if min > len(fs.planned.steps) {
-			min = len(fs.planned.steps)
+		if min > len(f.planned.steps) {
+			min = len(f.planned.steps)
 		}
 		diverge := 0
 		for ; diverge < min; diverge++ {
-			debug("diverge compare iteration %d", diverge)
-			if !fs.planned.steps[diverge].step.TargetEquals(prevUncompleted[diverge].step) {
+			f.debug("diverge compare iteration %d", diverge)
+			if !f.planned.steps[diverge].step.TargetEquals(prevUncompleted[diverge].step) {
 				break
 			}
 		}
-		debug("diverge is %d", diverge)
-		fs.planned.steps = fs.planned.steps[0:diverge]
+		f.debug("diverge is %d", diverge)
+		f.planned.steps = f.planned.steps[0:diverge]
 	}
-	debug("post-prev-merge len(fs.planned.steps) = %d", len(fs.planned.steps))
+	f.debug("post-prev-merge len(fs.planned.steps) = %d", len(f.planned.steps))
 
-	for i, s := range fs.planned.steps {
+	for i, s := range f.planned.steps {
 		var (
 			err     error
 			errTime time.Time
 		)
 		// lock must not be held while executing step in order for reporting to work
-		fs.l.DropWhile(func() {
+		f.l.DropWhile(func() {
 			targetDate := s.step.TargetDate()
-			defer pq.WaitReady(fs, targetDate)()
+			defer pq.WaitReady(f, targetDate)()
 			err = s.step.Step(ctx) // no shadow
 			errTime = time.Now()   // no shadow
 		})
 		if err != nil {
-			fs.planned.stepErr = newTimedError(err, errTime)
+			f.planned.stepErr = newTimedError(err, errTime)
 			break
 		}
-		fs.planned.step = i + 1 // fs.planned.step must be == len(fs.planned.steps) if all went OK
+		f.planned.step = i + 1 // fs.planned.step must be == len(fs.planned.steps) if all went OK
 	}
 }
 

replication/logic/replication_logic.go

@@ -602,7 +602,7 @@ func (s *Step) doReplication(ctx context.Context) error {
 
 	fs := s.parent.Path
 
-	log := getLogger(ctx)
+	log := getLogger(ctx).WithField("filesystem", fs)
 	sr := s.buildSendRequest(false)
 
 	log.Debug("initiate send request")

transport/tcp/serve_tcp.go

@@ -11,39 +11,6 @@ import (
 	"github.com/zrepl/zrepl/util/tcpsock"
 )
 
-type ipMapEntry struct {
-	ip    net.IP
-	ident string
-}
-
-type ipMap struct {
-	entries []ipMapEntry
-}
-
-func ipMapFromConfig(clients map[string]string) (*ipMap, error) {
-	entries := make([]ipMapEntry, 0, len(clients))
-	for clientIPString, clientIdent := range clients {
-		clientIP := net.ParseIP(clientIPString)
-		if clientIP == nil {
-			return nil, errors.Errorf("cannot parse client IP %q", clientIPString)
-		}
-		if err := transport.ValidateClientIdentity(clientIdent); err != nil {
-			return nil, errors.Wrapf(err, "invalid client identity for IP %q", clientIPString)
-		}
-		entries = append(entries, ipMapEntry{clientIP, clientIdent})
-	}
-	return &ipMap{entries: entries}, nil
-}
-
-func (m *ipMap) Get(ip net.IP) (string, error) {
-	for _, e := range m.entries {
-		if e.ip.Equal(ip) {
-			return e.ident, nil
-		}
-	}
-	return "", errors.Errorf("no identity mapping for client IP %s", ip)
-}
-
 func TCPListenerFactoryFromConfig(c *config.Global, in *config.TCPServe) (transport.AuthenticatedListenerFactory, error) {
 	clientMap, err := ipMapFromConfig(in.Clients)
 	if err != nil {
@@ -69,10 +36,13 @@ func (f *TCPAuthListener) Accept(ctx context.Context) (*transport.AuthConn, erro
 	if err != nil {
 		return nil, err
 	}
-	clientIP := nc.RemoteAddr().(*net.TCPAddr).IP
-	clientIdent, err := f.clientMap.Get(clientIP)
+	clientAddr := &net.IPAddr{
+		IP:   nc.RemoteAddr().(*net.TCPAddr).IP,
+		Zone: nc.RemoteAddr().(*net.TCPAddr).Zone,
+	}
+	clientIdent, err := f.clientMap.Get(clientAddr)
 	if err != nil {
-		transport.GetLogger(ctx).WithField("ip", clientIP).Error("client IP not in client map")
+		transport.GetLogger(ctx).WithField("ipaddr", clientAddr).Error("client IP not in client map")
 		nc.Close()
 		return nil, err
 	}

transport/tcp/serve_tcp_ipmap.go

@@ -0,0 +1,166 @@
+package tcp
+
+import (
+	"bytes"
+	"fmt"
+	"net"
+	"sort"
+	"strings"
+
+	"github.com/pkg/errors"
+	"github.com/zrepl/zrepl/transport"
+	"golang.org/x/sys/unix"
+)
+
+type ipMapEntry struct {
+	subnet *net.IPNet
+	// ident is always not empty
+	ident string
+	// zone may be empty (e.g. for IPv4)
+	zone string
+}
+
+type ipMap struct {
+	entries []*ipMapEntry
+}
+
+func ipMapFromConfig(clients map[string]string) (*ipMap, error) {
+
+	entries := make([]*ipMapEntry, 0, len(clients))
+	for clientInput, clientIdent := range clients {
+		userIPMapEntry, err := newIPMapEntry(clientInput, clientIdent)
+		if err != nil {
+			return nil, errors.Wrapf(err, "cannot not parse %q:%q", clientInput, clientIdent)
+		}
+
+		entries = append(entries, userIPMapEntry)
+	}
+
+	sort.Sort(byPrefixlen(entries))
+
+	return &ipMap{entries: entries}, nil
+}
+
+func (m *ipMap) Get(ipAddr *net.IPAddr) (string, error) {
+	for _, e := range m.entries {
+		if e.zone != ipAddr.Zone {
+			continue
+		}
+		if e.subnet.Contains(ipAddr.IP) {
+			return zfsDatasetPathComponentCompatibleRepresentation(e.ident, ipAddr), nil
+		}
+	}
+	return "", errors.Errorf("no identity mapping for client IP: %s%%%s", ipAddr.IP, ipAddr.Zone)
+}
+
+var ipv6FullySpecifiedMask = bytes.Repeat([]byte{0xff}, net.IPv6len)
+
+func newIPMapEntry(input string, ident string) (*ipMapEntry, error) {
+	ip := input
+	var zone string
+
+	ipZoneSplit := strings.SplitN(input, "%", 2)
+	if len(ipZoneSplit) > 1 {
+		ip = ipZoneSplit[0]
+		zone = ipZoneSplit[1]
+	}
+
+	_, subnet, err := net.ParseCIDR(ip)
+	if err != nil {
+		// expect full IP, no '*' placeholder expansion
+
+		if strings.Count(ident, "*") != 0 {
+			return nil, fmt.Errorf("non-CIDR matches must not contain '*' placeholder")
+		}
+
+		if err := transport.ValidateClientIdentity(ident); err != nil {
+			return nil, errors.Wrapf(err, "invalid client identity %q for IP %q", ident, ip)
+		}
+
+		parsedIP := net.ParseIP(ip)
+		if parsedIP == nil {
+			return nil, errors.Wrapf(err, "invalid client address %q", ip)
+		}
+		parsedIP = parsedIP.To16()
+
+		return &ipMapEntry{
+			subnet: &net.IPNet{
+				IP:   parsedIP,
+				Mask: ipv6FullySpecifiedMask,
+			},
+			zone:  zone,
+			ident: ident,
+		}, nil
+	}
+
+	// expect CIDR and '*' placeholder expansion
+
+	if strings.Count(ident, "*") != 1 {
+		err = fmt.Errorf("CIDRs require 1 IP placeholder")
+		return nil, errors.Wrapf(err, "invalid client identity %q for IP %q", ident, ip)
+	}
+
+	longestIPAddr := net.IPAddr{
+		IP:   net.IP(bytes.Repeat([]byte{0xff}, net.IPv6len)),
+		Zone: strings.Repeat("i", unix.IFNAMSIZ),
+	}
+	longestIdent := zfsDatasetPathComponentCompatibleRepresentation(ident, &longestIPAddr)
+
+	if err := transport.ValidateClientIdentity(longestIdent); err != nil {
+		return nil, errors.Wrapf(err, "invalid client identity for IP %q", ip)
+	}
+
+	ones, _ := subnet.Mask.Size()
+	preExpansionAddrlen := len(subnet.IP) * 8
+	expanded := subnet.IP.To16()
+	postExpansionAddrlen := len(expanded) * 8
+	return &ipMapEntry{
+		subnet: &net.IPNet{
+			IP:   expanded,
+			Mask: net.CIDRMask(postExpansionAddrlen-preExpansionAddrlen+ones, postExpansionAddrlen),
+		},
+		zone:  zone,
+		ident: ident,
+	}, nil
+}
+
+func zfsDatasetPathComponentCompatibleRepresentation(identityWithWildcard string, addr *net.IPAddr) string {
+	// If a Zone exists we append it after the IP using a "-" because "%"
+	// is a zfs dataset forbidden char.
+	if addr.Zone != "" {
+		return strings.Replace(identityWithWildcard, "*", addr.IP.String()+"-"+addr.Zone, 1)
+	}
+	// newIPMapEntry validates that the line contains exactly one "*"
+	return strings.Replace(identityWithWildcard, "*", addr.IP.String(), 1)
+}
+
+func (e *ipMapEntry) String() string {
+	return fmt.Sprintf("&ipMapEntry{subnet=%q, ident=%q, zone=%q}", e.subnet.String(), e.ident, e.zone)
+}
+
+func (e *ipMapEntry) PrefixLen() int {
+	ones, bits := e.subnet.Mask.Size()
+	if bits != net.IPv6len*8 {
+		panic(fmt.Sprintf("impl error: we represent all addresses as 16byte internally ones=%v bits=%v: %s", ones, bits, e))
+	}
+	return ones
+}
+
+// newtype to support sorting by prefixlength
+type byPrefixlen []*ipMapEntry
+
+func (m byPrefixlen) Len() int { return len(m) }
+func (m byPrefixlen) Less(i, j int) bool {
+
+	if m[i].PrefixLen() != m[j].PrefixLen() {
+		return m[i].PrefixLen() > m[j].PrefixLen()
+	}
+
+	addrCmp := bytes.Compare(m[i].subnet.IP.To16(), m[j].subnet.IP.To16())
+	if addrCmp != 0 {
+		return addrCmp < 0
+	}
+
+	return strings.Compare(m[i].zone, m[j].zone) < 0
+}
+func (m byPrefixlen) Swap(i, j int) { m[i], m[j] = m[j], m[i] }

transport/tcp/serve_tcp_ipmap_test.go

@@ -0,0 +1,204 @@
+package tcp
+
+import (
+	"net"
+	"os"
+	"testing"
+
+	"github.com/kr/pretty"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestIPMap(t *testing.T) {
+	type testCaseExpect struct {
+		expectNoMapping bool
+		expectIdent     string
+	}
+	type testCase struct {
+		name          string
+		config        map[string]string
+		expectInitErr bool
+		expect        map[string]testCaseExpect
+	}
+
+	cases := []testCase{
+		{
+			name:          "regular ips",
+			expectInitErr: false,
+			config: map[string]string{
+				"192.168.123.234": "ident1",
+				"192.168.20.23":   "ident2",
+			},
+			expect: map[string]testCaseExpect{
+				"192.168.123.234": {expectIdent: "ident1"},
+				"192.168.20.23":   {expectIdent: "ident2"},
+				"192.168.20.10":   {expectNoMapping: true},
+			},
+		},
+		{
+			name:          "full wildcard",
+			expectInitErr: false,
+			config: map[string]string{
+				"0.0.0.0/0": "*",
+				"0::/0":     "*",
+			},
+			expect: map[string]testCaseExpect{
+				"10.123.234.24": {expectIdent: "10.123.234.24"},
+				// '[' and ']' are forbiddenin dataset names
+				"fe80::23:42": {expectIdent: "fe80::23:42"},
+			},
+		},
+		{
+			name:          "longest prefix matching",
+			expectInitErr: false,
+			config: map[string]string{
+				"10.1.2.3":              "specific-host",
+				"10.1.2.0/24":           "subnet-one-two-*",
+				"10.1.1.0/24":           "subnet-one-one-*",
+				"10.1.0.0/24":           "subnet-one-zero-*",
+				"10.1.0.0/16":           "subnet-one-*",
+				"fde4:8dba:82e1:1::1":   "v6-48-1-specialhost",
+				"fde4:8dba:82e1::/48":   "v6-48-*",
+				"fde4:8dba:82e1:1::/64": "v6-64-1-*",
+				"fde4:8dba:82e1:2::/64": "v6-64-2-*",
+			},
+			expect: map[string]testCaseExpect{
+				"10.1.2.3":             {expectIdent: "specific-host"},
+				"10.1.2.1":             {expectIdent: "subnet-one-two-10.1.2.1"},
+				"10.1.1.1":             {expectIdent: "subnet-one-one-10.1.1.1"},
+				"10.1.0.23":            {expectIdent: "subnet-one-zero-10.1.0.23"},
+				"10.1.3.1":             {expectIdent: "subnet-one-10.1.3.1"},
+				"10.2.1.1":             {expectNoMapping: true},
+				"fde4:8dba:82e1:1::1":  {expectIdent: "v6-48-1-specialhost"},
+				"fde4:8dba:82e1:23::1": {expectIdent: "v6-48-fde4:8dba:82e1:23::1"},
+				"fde4:8dba:82e1:1::2":  {expectIdent: "v6-64-1-fde4:8dba:82e1:1::2"},
+				"fde4:8dba:82e1:2::1":  {expectIdent: "v6-64-2-fde4:8dba:82e1:2::1"},
+				"fde4:8dba:82e2::1":    {expectNoMapping: true},
+			},
+		},
+		{
+			name:          "different prefixes, mixed ipv4 ipv6, with interface ids",
+			expectInitErr: false,
+			config: map[string]string{
+				"192.168.23.0/24": "db-*",
+				"192.168.23.23":   "db-twentythree",
+				"192.168.42.0/24": "web-*",
+				"10.1.4.0/24":     "my-*-server",
+				"2001:0db8:85a3:0000:0000:8a2e:0370:7334":      "aspecifichost",
+				"2001:0db8:85a3:0000:0000:8a2e:0370:7334%eth1": "aspecifichost",
+				"fe80::/16%eth1":      "san-*",
+				"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"},
+
+				// 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},
+
+				// v6 subnet matching
+				"fde4:8dba:82e1::1": {expectIdent: "sub64-fde4:8dba:82e1::1"},
+				// v6 subnet matching with suffix that matches another allowed IPv4
+				"fde4:8dba:82e1::c0a8:1717": {expectIdent: "sub64-fde4:8dba:82e1::c0a8:1717"},
+
+				"2001:0db8:85a3:0000:0000:8a2e:0370:7334":      {expectIdent: "aspecifichost"},
+				"2001:0db8:85a3:0000:0000:8a2e:0370:7334%eth1": {expectIdent: "aspecifichost"},
+				"2001:0db8:85a3:0000:0000:8a2e:0370:7334%eth2": {expectNoMapping: true},
+			},
+		},
+		{
+			name:          "invalid user input: non ip or cidr",
+			expectInitErr: true,
+			config: map[string]string{
+				"jimmy": "db",
+			},
+		},
+		{
+			name:          "invalid user input: v4 ip with an identity containing *",
+			expectInitErr: true,
+			config: map[string]string{
+				"192.168.1.2": "db-*",
+			},
+		},
+		{
+			name:          "invalid user input: v4 subnet without an identity containing *",
+			expectInitErr: true,
+			config: map[string]string{
+				"192.168.1.0/24": "db-",
+			},
+		},
+		{
+			name:          "invalid user input: v6 ip with an identity containing *",
+			expectInitErr: true,
+			config: map[string]string{
+				"2001:0db8:85a3:0000:0000:8a2e:0370:7334": "aspecifichost*",
+			},
+		},
+		{
+			name:          "invalid user input: v6 subnet without identity containing *",
+			expectInitErr: true,
+			config: map[string]string{
+				"fe80::/16%eth1": "db-",
+			},
+		},
+		{
+			name:          "invalid user input with subnet match: client identity with forbidden zfs dataset name char @",
+			expectInitErr: true,
+			config: map[string]string{
+				"fe80::/16": "db@-*",
+			},
+		},
+		{
+			name:          "invalid user input with IP match: client identity with forbidden zfs dataset name char @",
+			expectInitErr: true,
+			config: map[string]string{
+				"fe80::1": "db@foo",
+			},
+		},
+	}
+
+	for i := range cases {
+		c := cases[i]
+		t.Run(c.name, func(t *testing.T) {
+			pretty.Fprintf(os.Stderr, "running %#v\n", c)
+			m, err := ipMapFromConfig(c.config)
+			if c.expectInitErr {
+				require.Error(t, err)
+			} else {
+				require.NoError(t, err)
+				require.NotNil(t, m)
+			}
+			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")
+				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")
+				require.NotNil(t, ipMapEntry)
+				ident, err := m.Get(&net.IPAddr{
+					IP:   ipMapEntry.subnet.IP,
+					Zone: ipMapEntry.zone,
+				})
+				if expect.expectNoMapping {
+					assert.Empty(t, ident)
+				} else {
+					assert.NoError(t, err)
+					assert.Equal(t, expect.expectIdent, ident)
+				}
+			}
+		})
+	}
+}
+
+func TestPackageNetAssumptions(t *testing.T) {
+
+}

transport/transport.go

@@ -4,10 +4,10 @@ package transport
 
 import (
 	"context"
-	"errors"
 	"net"
 	"syscall"
 
+	"github.com/pkg/errors"
 	"github.com/zrepl/zrepl/logger"
 	"github.com/zrepl/zrepl/rpc/dataconn/timeoutconn"
 	"github.com/zrepl/zrepl/zfs"
@@ -55,13 +55,10 @@ type Connecter interface {
 }
 
 // A client identity must be a single component in a ZFS filesystem path
-func ValidateClientIdentity(in string) (err error) {
-	path, err := zfs.NewDatasetPath(in)
+func ValidateClientIdentity(in string) error {
+	err := zfs.ComponentNamecheck(in)
 	if err != nil {
-		return err
-	}
-	if path.Length() != 1 {
-		return errors.New("client identity must be a single path component (not empty, no '/')")
+		return errors.Wrap(err, "client identity must be usable as a single dataset path component")
 	}
 	return nil
 }

zfs/versions.go

@@ -206,13 +206,13 @@ func (o *ListFilesystemVersionsOptions) matches(v FilesystemVersion) bool {
 }
 
 // returned versions are sorted by createtxg FIXME drop sort by createtxg requirement
-func ZFSListFilesystemVersions(fs *DatasetPath, options ListFilesystemVersionsOptions) (res []FilesystemVersion, err error) {
+func ZFSListFilesystemVersions(ctx context.Context, fs *DatasetPath, options ListFilesystemVersionsOptions) (res []FilesystemVersion, err error) {
 	listResults := make(chan ZFSListResult)
 
 	promTimer := prometheus.NewTimer(prom.ZFSListFilesystemVersionDuration.WithLabelValues(fs.ToString()))
 	defer promTimer.ObserveDuration()
 
-	ctx, cancel := context.WithCancel(context.Background())
+	ctx, cancel := context.WithCancel(ctx)
 	defer cancel()
 	go ZFSListChan(ctx, listResults,
 		[]string{"name", "guid", "createtxg", "creation", "userrefs"},

zfs/zfs.go

@@ -1084,7 +1084,7 @@ func ZFSRecv(ctx context.Context, fs string, v *ZFSSendArgVersion, streamCopier
 	if opts.RollbackAndForceRecv {
 		// destroy all snapshots before `recv -F` because `recv -F`
 		// does not perform a rollback unless `send -R` was used (which we assume hasn't been the case)
-		snaps, err := ZFSListFilesystemVersions(fsdp, ListFilesystemVersionsOptions{
+		snaps, err := ZFSListFilesystemVersions(ctx, fsdp, ListFilesystemVersionsOptions{
 			Types: Snapshots,
 		})
 		if err != nil {

zfs/zfscmd/zfscmd.go

@@ -18,10 +18,10 @@ import (
 )
 
 type Cmd struct {
-	cmd                       *exec.Cmd
-	ctx                       context.Context
-	mtx                       sync.RWMutex
-	startedAt, waitReturnedAt time.Time
+	cmd                                      *exec.Cmd
+	ctx                                      context.Context
+	mtx                                      sync.RWMutex
+	startedAt, waitStartedAt, waitReturnedAt time.Time
 }
 
 func CommandContext(ctx context.Context, name string, arg ...string) *Cmd {
@@ -119,19 +119,65 @@ func (c *Cmd) startPost(err error) {
 }
 
 func (c *Cmd) waitPre() {
-	waitPreLogging(c, time.Now())
+	now := time.Now()
+
+	// ignore duplicate waits
+	c.mtx.Lock()
+	// ignore duplicate waits
+	if !c.waitStartedAt.IsZero() {
+		c.mtx.Unlock()
+		return
+	}
+	c.waitStartedAt = now
+	c.mtx.Unlock()
+
+	waitPreLogging(c, now)
+}
+
+type usage struct {
+	total_secs, system_secs, user_secs float64
 }
 
 func (c *Cmd) waitPost(err error) {
 	now := time.Now()
 
 	c.mtx.Lock()
+	// ignore duplicate waits
+	if !c.waitReturnedAt.IsZero() {
+		c.mtx.Unlock()
+		return
+	}
 	c.waitReturnedAt = now
 	c.mtx.Unlock()
 
-	waitPostReport(c, now)
-	waitPostLogging(c, err, now)
-	waitPostPrometheus(c, err, now)
+	// build usage
+	var u usage
+	{
+		var s *os.ProcessState
+		if err == nil {
+			s = c.cmd.ProcessState
+		} else if ee, ok := err.(*exec.ExitError); ok {
+			s = ee.ProcessState
+		}
+
+		if s == nil {
+			u = usage{
+				total_secs:  c.Runtime().Seconds(),
+				system_secs: -1,
+				user_secs:   -1,
+			}
+		} else {
+			u = usage{
+				total_secs:  c.Runtime().Seconds(),
+				system_secs: s.SystemTime().Seconds(),
+				user_secs:   s.UserTime().Seconds(),
+			}
+		}
+	}
+
+	waitPostReport(c, u, now)
+	waitPostLogging(c, u, err, now)
+	waitPostPrometheus(c, u, err, now)
 }
 
 // returns 0 if the command did not yet finish

zfs/zfscmd/zfscmd_logging.go

@@ -1,7 +1,6 @@
 package zfscmd
 
 import (
-	"os/exec"
 	"time"
 )
 
@@ -28,23 +27,12 @@ func waitPreLogging(c *Cmd, now time.Time) {
 	c.log().Debug("start waiting")
 }
 
-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
-	}
+func waitPostLogging(c *Cmd, u usage, err error, now time.Time) {
 
 	log := c.log().
-		WithField("total_time_s", total).
-		WithField("systemtime_s", system).
-		WithField("usertime_s", user)
+		WithField("total_time_s", u.total_secs).
+		WithField("systemtime_s", u.system_secs).
+		WithField("usertime_s", u.user_secs)
 
 	if err == nil {
 		log.Info("command exited without error")

zfs/zfscmd/zfscmd_prometheus.go

@@ -46,7 +46,7 @@ func RegisterMetrics(r prometheus.Registerer) {
 	r.MustRegister(metrics.usertime)
 }
 
-func waitPostPrometheus(c *Cmd, err error, now time.Time) {
+func waitPostPrometheus(c *Cmd, u usage, err error, now time.Time) {
 
 	if len(c.cmd.Args) < 2 {
 		getLogger(c.ctx).WithField("args", c.cmd.Args).
@@ -64,10 +64,10 @@ func waitPostPrometheus(c *Cmd, err error, now time.Time) {
 
 	metrics.totaltime.
 		WithLabelValues(labelValues...).
-		Observe(c.Runtime().Seconds())
+		Observe(u.total_secs)
 	metrics.systemtime.WithLabelValues(labelValues...).
-		Observe(c.cmd.ProcessState.SystemTime().Seconds())
+		Observe(u.system_secs)
 	metrics.usertime.WithLabelValues(labelValues...).
-		Observe(c.cmd.ProcessState.UserTime().Seconds())
+		Observe(u.user_secs)
 
 }

zfs/zfscmd/zfscmd_report.go

@@ -57,7 +57,7 @@ func startPostReport(c *Cmd, err error, now time.Time) {
 	active.mtx.Unlock()
 }
 
-func waitPostReport(c *Cmd, now time.Time) {
+func waitPostReport(c *Cmd, _ usage, now time.Time) {
 	active.mtx.Lock()
 	defer active.mtx.Unlock()
 	prev := active.cmds[c]