endpoint: Receiver.Receive: save a peek of recv stream to a temporary dataset if placeholder + encryption recv -F error is detected

fixes #277

.circleci/config.yml

@@ -60,10 +60,10 @@ jobs:
 
       - run: make zrepl-bin
       - run: make vet
+      - run: make test
       - run: make lint
       - run: make release
-      - run: make test-go
-      # cannot run test-platform because circle-ci runs in linux containers
+
 
       - store_artifacts:
           path: ./artifacts/release
@@ -92,21 +92,11 @@ jobs:
             REPO="zrepl/zrepl"
             COMMIT="${CIRCLE_SHA1}"
             JOB_NAME="${CIRCLE_JOB}"
-            curl "https://api.github.com/repos/$REPO/statuses/$COMMIT" \
+            curl "https://api.github.com/repos/$REPO/statuses/$COMMIT?access_token=$GITHUB_COMMIT_STATUS_TOKEN" \
               -H "Content-Type: application/json" \
-              -H "Authorization: token $GITHUB_COMMIT_STATUS_TOKEN" \
               -X POST \
               -d '{"context":"zrepl/publish-ci-artifacts", "state": "success", "description":"CI Build Artifacts for '"$JOB_NAME"'", "target_url":"https://minio.cschwarz.com/minio/zrepl-ci-artifacts/'"$COMMIT"'/"}'
 
-      - run:
-          shell: /bin/bash -euo pipefail
-          command: |
-            # Trigger Debian Package Build
-            curl -v -X POST https://api.github.com/repos/zrepl/debian-binary-packaging/dispatches \
-              -H 'Accept: application/vnd.github.v3+json' \
-              -H "Authorization: token $ZREPL_DEBIAN_BINARYPACKAGIN_TRIGGER_BUILD_GITHUB_TOKEN" \
-              --data '{"event_type": "push", "client_payload": { "zrepl_main_repo_commit": "'"$CIRCLE_SHA1"'", "go_version": "'"${CIRCLE_JOB##build-}"'" }}'
-
   build-1.11:
     <<: *build-latest
     docker:

Makefile

@@ -27,8 +27,6 @@ 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
-GOCOVMERGE := gocovmerge
-
 ifneq ($(GOARM),)
 	ZREPL_TARGET_TUPLE := $(GOOS)-$(GOARCH)v$(GOARM)
 else
@@ -47,7 +45,7 @@ printvars:
 
 release: clean
 	# no cross-platform support for target test
-	$(MAKE) test-go
+	$(MAKE) test
 	$(MAKE) bins-all
 	$(MAKE) noarch
 	$(MAKE) wrapup-and-checksum
@@ -101,9 +99,9 @@ clean: docs-clean
 	rm -rf "$(ARTIFACTDIR)"
 
 ##################### BINARIES #####################
-.PHONY: bins-all lint test-go test-platform cover-merge cover-html vet zrepl-bin test-platform-bin generate-platform-test-list
+.PHONY: bins-all lint test vet zrepl-bin platformtest-bin
 
-BINS_ALL_TARGETS := zrepl-bin test-platform-bin vet lint
+BINS_ALL_TARGETS := zrepl-bin platformtest-bin vet lint
 GO_SUPPORTS_ILLUMOS := $(shell $(GO) version | gawk -F '.' '/^go version /{split($$0, comps, " "); split(comps[3], v, "."); if (v[1] == "go1" && v[2] >= 13) { print "illumos"; } else { print "noillumos"; }}')
 bins-all:
 	$(MAKE) $(BINS_ALL_TARGETS) GOOS=freebsd   GOARCH=amd64
@@ -124,34 +122,27 @@ endif
 
 lint:
 	$(GO_ENV_VARS) $(GOLANGCI_LINT) run ./...
-
+test:
+	$(GO_ENV_VARS) $(GO) test $(GO_BUILDFLAGS) ./...
 vet:
 	$(GO_ENV_VARS) $(GO) vet $(GO_BUILDFLAGS) ./...
 
-test-go: $(ARTIFACTDIR)
-	rm -f "$(ARTIFACTDIR)/gotest.cover"
-ifeq ($(COVER),1)
-	$(GO_ENV_VARS) $(GO) test $(GO_BUILDFLAGS) \
-		-coverpkg github.com/zrepl/zrepl/... \
-		-covermode atomic \
-		-coverprofile "$(ARTIFACTDIR)/gotest.cover" \
-		./...
-else
-	$(GO_ENV_VARS) $(GO) test $(GO_BUILDFLAGS) \
-		./...
-endif
-
 zrepl-bin:
 	$(GO_BUILD) -o "$(ARTIFACTDIR)/zrepl-$(ZREPL_TARGET_TUPLE)"
 
-generate-platform-test-list:
-	$(GO_BUILD) -o $(ARTIFACTDIR)/generate-platform-test-list ./platformtest/tests/gen
+platformtest-bin:
+	$(GO_BUILD) -o "$(ARTIFACTDIR)/platformtest-$(ZREPL_TARGET_TUPLE)" ./platformtest/harness
 
-test-platform-bin:
-	$(GO_ENV_VARS) $(GO) test $(GO_BUILDFLAGS) \
-		-c -o "$(ARTIFACTDIR)/platformtest-$(ZREPL_TARGET_TUPLE)" \
-		-covermode=atomic -cover -coverpkg github.com/zrepl/zrepl/... \
-		./platformtest/harness
+##################### DEV TARGETS #####################
+# not part of the build, must do that manually
+.PHONY: generate format platformtest
+
+generate:
+	protoc -I=replication/logic/pdu --go_out=plugins=grpc:replication/logic/pdu replication/logic/pdu/pdu.proto
+	$(GO_ENV_VARS) $(GO) generate $(GO_BUILDFLAGS) -x ./...
+
+format:
+	goimports -srcdir . -local 'github.com/zrepl/zrepl' -w $(shell find . -type f -name '*.go' -not -path "./vendor/*" -not -name '*.pb.go' -not -name '*_enumer.go')
 
 ZREPL_PLATFORMTEST_POOLNAME := zreplplatformtest
 ZREPL_PLATFORMTEST_IMAGEPATH := /tmp/zreplplatformtest.pool.img
@@ -159,42 +150,16 @@ ZREPL_PLATFORMTEST_MOUNTPOINT := /tmp/zreplplatformtest.pool
 ZREPL_PLATFORMTEST_ZFS_LOG := /tmp/zreplplatformtest.zfs.log
 # ZREPL_PLATFORMTEST_STOP_AND_KEEP := -failure.stop-and-keep-pool
 ZREPL_PLATFORMTEST_ARGS := 
-test-platform: $(ARTIFACTDIR) # do not track dependency on test-platform-bin to allow build of platformtest outside of test VM
+platformtest: # do not track dependency on platformtest-bin to allow build of platformtest outside of test VM
 	rm -f "$(ZREPL_PLATFORMTEST_ZFS_LOG)"
-	rm -f "$(ARTIFACTDIR)/platformtest.cover"
 	platformtest/logmockzfs/logzfsenv "$(ZREPL_PLATFORMTEST_ZFS_LOG)" `which zfs` \
 	"$(ARTIFACTDIR)/platformtest-$(ZREPL_TARGET_TUPLE)" \
-		-test.coverprofile "$(ARTIFACTDIR)/platformtest.cover" \
-		-test.v \
-		__DEVEL--i-heard-you-like-tests \
 		-poolname "$(ZREPL_PLATFORMTEST_POOLNAME)" \
 		-imagepath "$(ZREPL_PLATFORMTEST_IMAGEPATH)" \
 		-mountpoint "$(ZREPL_PLATFORMTEST_MOUNTPOINT)" \
 		$(ZREPL_PLATFORMTEST_STOP_AND_KEEP) \
 		$(ZREPL_PLATFORMTEST_ARGS)
 
-cover-merge: $(ARTIFACTDIR)
-	$(GOCOVMERGE) $(ARTIFACTDIR)/platformtest.cover $(ARTIFACTDIR)/gotest.cover > $(ARTIFACTDIR)/merged.cover
-cover-html: cover-merge
-	$(GO) tool cover -html "$(ARTIFACTDIR)/merged.cover" -o "$(ARTIFACTDIR)/merged.cover.html"
-
-test-full:
-	test "$$(id -u)" = "0" || echo "MUST RUN AS ROOT" 1>&2
-	$(MAKE) test-go COVER=1
-	$(MAKE) test-platform
-	$(MAKE) cover-html
-
-##################### DEV TARGETS #####################
-# not part of the build, must do that manually
-.PHONY: generate format
-
-generate: generate-platform-test-list
-	protoc -I=replication/logic/pdu --go_out=plugins=grpc:replication/logic/pdu replication/logic/pdu/pdu.proto
-	$(GO_ENV_VARS) $(GO) generate $(GO_BUILDFLAGS) -x ./...
-
-format:
-	goimports -srcdir . -local 'github.com/zrepl/zrepl' -w $(shell find . -type f -name '*.go' -not -path "./vendor/*" -not -name '*.pb.go' -not -name '*_enumer.go')
-
 ##################### NOARCH #####################
 .PHONY: noarch $(ARTIFACTDIR)/bash_completion $(ARTIFACTDIR)/_zrepl.zsh_completion $(ARTIFACTDIR)/go_env.txt docs docs-clean
 

README.md

@@ -39,7 +39,7 @@ Check out the *Coding Workflow* section below for details.
 * Ship other material provided in `./dist`, e.g. in `/usr/share/zrepl/`.
 * Use `make release ZREPL_VERSION='mydistro-1.2.3_1'`
     * Your distro's name and any versioning supplemental to zrepl's (e.g. package revision) should be in this string
-* Use `sudo make test-platform` **on a test system** to validate that zrepl's abstractions on top of ZFS work with the system ZFS.
+* Use `make platformtest` **on a test system** to validate that zrepl's abstractions on top of ZFS work with the system ZFS.
 * Make sure you are informed about new zrepl versions, e.g. by subscribing to GitHub's release RSS feed.
 
 ## Developer Documentation
@@ -47,18 +47,10 @@ Check out the *Coding Workflow* section below for details.
 zrepl is written in [Go](https://golang.org) and uses [Go modules](https://github.com/golang/go/wiki/Modules) to manage dependencies.
 The documentation is written in [ReStructured Text](http://docutils.sourceforge.net/rst.html) using the [Sphinx](https://www.sphinx-doc.org) framework.
 
-Install **build dependencies** using  `./lazy.sh devsetup`.
+To get started, run `./lazy.sh devsetup` to easily install build dependencies and read `docs/installation.rst -> Compiling from Source`.
 `lazy.sh` uses `python3-pip` to fetch the build dependencies for the docs - you might want to use a [venv](https://docs.python.org/3/library/venv.html).
 If you just want to install the Go dependencies, run `./lazy.sh godep`.
 
-The **test suite** is split into pure **Go tests** (`make test-go`) and **platform tests** that interact with ZFS and thus generally **require root privileges** (`sudo make test-platform`).
-Platform tests run on their own pool with the name `zreplplatformtest`, which is created using the file vdev in `/tmp`.
-
-For a full **code coverage** profile, run `make test-go COVER=1 && sudo make test-platform && make cover-merge`.
-An HTML report can be generated using `make cover-html`.
-
-**Code generation** is triggered by `make generate`. Generated code is committed to the source tree.
-
 ### Project Structure
 
 ```

build/go.mod

@@ -28,7 +28,6 @@ require (
 	github.com/spf13/jwalterweatherman v1.1.0 // indirect
 	github.com/spf13/viper v1.3.2 // indirect
 	github.com/stretchr/testify v1.4.0 // indirect
-	github.com/wadey/gocovmerge v0.0.0-20160331181800-b5bfa59ec0ad // indirect
 	golang.org/x/net v0.0.0-20190613194153-d28f0bde5980 // indirect
 	golang.org/x/sys v0.0.0-20191010194322-b09406accb47 // indirect
 	golang.org/x/tools v0.0.0-20190524210228-3d17549cdc6b

build/go.sum

@@ -214,8 +214,6 @@ github.com/valyala/bytebufferpool v1.0.0/go.mod h1:6bBcMArwyJ5K/AmCkWv1jt77kVWyC
 github.com/valyala/fasthttp v1.2.0/go.mod h1:4vX61m6KN+xDduDNwXrhIAVZaZaZiQ1luJk8LWSxF3s=
 github.com/valyala/quicktemplate v1.1.1/go.mod h1:EH+4AkTd43SvgIbQHYu59/cJyxDoOVRUAfrukLPuGJ4=
 github.com/valyala/tcplisten v0.0.0-20161114210144-ceec8f93295a/go.mod h1:v3UYOV9WzVtRmSR+PDvWpU/qWl4Wa5LApYYX4ZtKbio=
-github.com/wadey/gocovmerge v0.0.0-20160331181800-b5bfa59ec0ad h1:W0LEBv82YCGEtcmPA3uNZBI33/qF//HAAs3MawDjRa0=
-github.com/wadey/gocovmerge v0.0.0-20160331181800-b5bfa59ec0ad/go.mod h1:Hy8o65+MXnS6EwGElrSRjUzQDLXreJlzYLlWiHtt8hM=
 github.com/xordataexchange/crypt v0.0.3-0.20170626215501-b2862e3d0a77/go.mod h1:aYKd//L2LvnjZzWKhF00oedf4jCCReLcmhLdhm1A27Q=
 golang.org/x/crypto v0.0.0-20181203042331-505ab145d0a9/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4=
 golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=

build/tools.go

@@ -6,7 +6,6 @@ import (
 	_ "github.com/alvaroloes/enumer"
 	_ "github.com/golang/protobuf/protoc-gen-go"
 	_ "github.com/golangci/golangci-lint/cmd/golangci-lint"
-	_ "github.com/wadey/gocovmerge"
 	_ "golang.org/x/tools/cmd/goimports"
 	_ "golang.org/x/tools/cmd/stringer"
 )

cli/cli.go

@@ -1,13 +1,11 @@
 package cli
 
 import (
-	"context"
 	"fmt"
 	"os"
 
 	"github.com/spf13/cobra"
 	"github.com/spf13/pflag"
-	"github.com/zrepl/zrepl/daemon/logging/trace"
 
 	"github.com/zrepl/zrepl/config"
 )
@@ -77,7 +75,7 @@ type Subcommand struct {
 	Short            string
 	Example          string
 	NoRequireConfig  bool
-	Run              func(ctx context.Context, subcommand *Subcommand, args []string) error
+	Run              func(subcommand *Subcommand, args []string) error
 	SetupFlags       func(f *pflag.FlagSet)
 	SetupSubcommands func() []*Subcommand
 
@@ -98,11 +96,7 @@ func (s *Subcommand) Config() *config.Config {
 
 func (s *Subcommand) run(cmd *cobra.Command, args []string) {
 	s.tryParseConfig()
-	ctx := context.Background()
-	endTask := trace.WithTaskFromStackUpdateCtx(&ctx)
-	defer endTask()
-	err := s.Run(ctx, s, args)
-	endTask()
+	err := s.Run(s, args)
 	if err != nil {
 		fmt.Fprintf(os.Stderr, "%s\n", err)
 		os.Exit(1)

client/configcheck.go

@@ -1,7 +1,6 @@
 package client
 
 import (
-	"context"
 	"encoding/json"
 	"fmt"
 	"os"
@@ -30,7 +29,7 @@ var ConfigcheckCmd = &cli.Subcommand{
 		f.StringVar(&configcheckArgs.format, "format", "", "dump parsed config object [pretty|yaml|json]")
 		f.StringVar(&configcheckArgs.what, "what", "all", "what to print [all|config|jobs|logging]")
 	},
-	Run: func(ctx context.Context, subcommand *cli.Subcommand, args []string) error {
+	Run: func(subcommand *cli.Subcommand, args []string) error {
 		formatMap := map[string]func(interface{}){
 			"": func(i interface{}) {},
 			"pretty": func(i interface{}) {

client/migrate.go

@@ -48,13 +48,14 @@ var migratePlaceholder0_1Args struct {
 	dryRun bool
 }
 
-func doMigratePlaceholder0_1(ctx context.Context, sc *cli.Subcommand, args []string) error {
+func doMigratePlaceholder0_1(sc *cli.Subcommand, args []string) error {
 	if len(args) != 0 {
 		return fmt.Errorf("migration does not take arguments, got %v", args)
 	}
 
 	cfg := sc.Config()
 
+	ctx := context.Background()
 	allFSS, err := zfs.ZFSListMapping(ctx, zfs.NoFilter())
 	if err != nil {
 		return errors.Wrap(err, "cannot list filesystems")
@@ -123,7 +124,7 @@ var fail = color.New(color.FgRed)
 
 var migrateReplicationCursorSkipSentinel = fmt.Errorf("skipping this filesystem")
 
-func doMigrateReplicationCursor(ctx context.Context, sc *cli.Subcommand, args []string) error {
+func doMigrateReplicationCursor(sc *cli.Subcommand, args []string) error {
 	if len(args) != 0 {
 		return fmt.Errorf("migration does not take arguments, got %v", args)
 	}
@@ -136,6 +137,8 @@ func doMigrateReplicationCursor(ctx context.Context, sc *cli.Subcommand, args []
 		return fmt.Errorf("exiting migration after error")
 	}
 
+	ctx := context.Background()
+
 	v1cursorJobs := make([]job.Job, 0, len(cfg.Jobs))
 	for i, j := range cfg.Jobs {
 		if jobs[i].Name() != j.Name() {

client/pprof.go

@@ -1,10 +1,67 @@
 package client
 
-import "github.com/zrepl/zrepl/cli"
+import (
+	"errors"
+	"log"
+	"os"
+
+	"github.com/zrepl/zrepl/cli"
+	"github.com/zrepl/zrepl/config"
+	"github.com/zrepl/zrepl/daemon"
+)
+
+var pprofArgs struct {
+	daemon.PprofServerControlMsg
+}
 
 var PprofCmd = &cli.Subcommand{
-	Use: "pprof",
-	SetupSubcommands: func() []*cli.Subcommand {
-		return []*cli.Subcommand{PprofListenCmd, pprofActivityTraceCmd}
+	Use:   "pprof off | [on TCP_LISTEN_ADDRESS]",
+	Short: "start a http server exposing go-tool-compatible profiling endpoints at TCP_LISTEN_ADDRESS",
+	Run: func(subcommand *cli.Subcommand, args []string) error {
+		if len(args) < 1 {
+			goto enargs
+		}
+		switch args[0] {
+		case "on":
+			pprofArgs.Run = true
+			if len(args) != 2 {
+				return errors.New("must specify TCP_LISTEN_ADDRESS as second positional argument")
+			}
+			pprofArgs.HttpListenAddress = args[1]
+		case "off":
+			if len(args) != 1 {
+				goto enargs
+			}
+			pprofArgs.Run = false
+		}
+
+		RunPProf(subcommand.Config())
+		return nil
+	enargs:
+		return errors.New("invalid number of positional arguments")
+
 	},
 }
+
+func RunPProf(conf *config.Config) {
+	log := log.New(os.Stderr, "", 0)
+
+	die := func() {
+		log.Printf("exiting after error")
+		os.Exit(1)
+	}
+
+	log.Printf("connecting to zrepl daemon")
+
+	httpc, err := controlHttpClient(conf.Global.Control.SockPath)
+	if err != nil {
+		log.Printf("error creating http client: %s", err)
+		die()
+	}
+	err = jsonRequestResponse(httpc, daemon.ControlJobEndpointPProf, pprofArgs.PprofServerControlMsg, struct{}{})
+	if err != nil {
+		log.Printf("error sending control message: %s", err)
+		die()
+	}
+	log.Printf("finished")
+}

client/pprof_activity_trace.go

@@ -1,44 +0,0 @@
-package client
-
-import (
-	"context"
-	"io"
-	"log"
-	"os"
-
-	"golang.org/x/net/websocket"
-
-	"github.com/zrepl/zrepl/cli"
-)
-
-var pprofActivityTraceCmd = &cli.Subcommand{
-	Use:   "activity-trace ZREPL_PPROF_HOST:ZREPL_PPROF_PORT",
-	Short: "attach to zrepl daemon with activated pprof listener and dump an activity-trace to stdout",
-	Run:   runPProfActivityTrace,
-}
-
-func runPProfActivityTrace(ctx context.Context, subcommand *cli.Subcommand, args []string) error {
-	log := log.New(os.Stderr, "", 0)
-
-	die := func() {
-		log.Printf("exiting after error")
-		os.Exit(1)
-	}
-
-	if len(args) != 1 {
-		log.Printf("exactly one positional argument is required")
-		die()
-	}
-
-	url := "ws://" + args[0] + "/debug/zrepl/activity-trace" // FIXME dont' repeat that
-
-	log.Printf("attaching to activity trace stream %s", url)
-	ws, err := websocket.Dial(url, "", url)
-	if err != nil {
-		log.Printf("error: %s", err)
-		die()
-	}
-
-	_, err = io.Copy(os.Stdout, ws)
-	return err
-}

client/pprof_listen.go

@@ -1,68 +0,0 @@
-package client
-
-import (
-	"context"
-	"errors"
-	"log"
-	"os"
-
-	"github.com/zrepl/zrepl/cli"
-	"github.com/zrepl/zrepl/config"
-	"github.com/zrepl/zrepl/daemon"
-)
-
-var pprofListenCmd struct {
-	daemon.PprofServerControlMsg
-}
-
-var PprofListenCmd = &cli.Subcommand{
-	Use:   "listen off | [on TCP_LISTEN_ADDRESS]",
-	Short: "start a http server exposing go-tool-compatible profiling endpoints at TCP_LISTEN_ADDRESS",
-	Run: func(ctx context.Context, subcommand *cli.Subcommand, args []string) error {
-		if len(args) < 1 {
-			goto enargs
-		}
-		switch args[0] {
-		case "on":
-			pprofListenCmd.Run = true
-			if len(args) != 2 {
-				return errors.New("must specify TCP_LISTEN_ADDRESS as second positional argument")
-			}
-			pprofListenCmd.HttpListenAddress = args[1]
-		case "off":
-			if len(args) != 1 {
-				goto enargs
-			}
-			pprofListenCmd.Run = false
-		}
-
-		RunPProf(subcommand.Config())
-		return nil
-	enargs:
-		return errors.New("invalid number of positional arguments")
-
-	},
-}
-
-func RunPProf(conf *config.Config) {
-	log := log.New(os.Stderr, "", 0)
-
-	die := func() {
-		log.Printf("exiting after error")
-		os.Exit(1)
-	}
-
-	log.Printf("connecting to zrepl daemon")
-
-	httpc, err := controlHttpClient(conf.Global.Control.SockPath)
-	if err != nil {
-		log.Printf("error creating http client: %s", err)
-		die()
-	}
-	err = jsonRequestResponse(httpc, daemon.ControlJobEndpointPProf, pprofListenCmd.PprofServerControlMsg, struct{}{})
-	if err != nil {
-		log.Printf("error sending control message: %s", err)
-		die()
-	}
-	log.Printf("finished")
-}

client/signal.go

@@ -1,8 +1,6 @@
 package client
 
 import (
-	"context"
-
 	"github.com/pkg/errors"
 
 	"github.com/zrepl/zrepl/cli"
@@ -13,7 +11,7 @@ import (
 var SignalCmd = &cli.Subcommand{
 	Use:   "signal [wakeup|reset] JOB",
 	Short: "wake up a job from wait state or abort its current invocation",
-	Run: func(ctx context.Context, subcommand *cli.Subcommand, args []string) error {
+	Run: func(subcommand *cli.Subcommand, args []string) error {
 		return runSignalCmd(subcommand.Config(), args)
 	},
 }

client/status.go

@@ -1,7 +1,6 @@
 package client
 
 import (
-	"context"
 	"fmt"
 	"io"
 	"math"
@@ -181,7 +180,7 @@ var StatusCmd = &cli.Subcommand{
 	Run: runStatus,
 }
 
-func runStatus(ctx context.Context, s *cli.Subcommand, args []string) error {
+func runStatus(s *cli.Subcommand, args []string) error {
 	httpc, err := controlHttpClient(s.Config().Global.Control.SockPath)
 	if err != nil {
 		return err

client/stdinserver.go

@@ -17,7 +17,7 @@ import (
 var StdinserverCmd = &cli.Subcommand{
 	Use:   "stdinserver CLIENT_IDENTITY",
 	Short: "stdinserver transport mode (started from authorized_keys file as forced command)",
-	Run: func(ctx context.Context, subcommand *cli.Subcommand, args []string) error {
+	Run: func(subcommand *cli.Subcommand, args []string) error {
 		return runStdinserver(subcommand.Config(), args)
 	},
 }

client/testcmd.go

@@ -39,7 +39,7 @@ var testFilter = &cli.Subcommand{
 	Run: runTestFilterCmd,
 }
 
-func runTestFilterCmd(ctx context.Context, subcommand *cli.Subcommand, args []string) error {
+func runTestFilterCmd(subcommand *cli.Subcommand, args []string) error {
 
 	if testFilterArgs.job == "" {
 		return fmt.Errorf("must specify --job flag")
@@ -49,6 +49,7 @@ func runTestFilterCmd(ctx context.Context, subcommand *cli.Subcommand, args []st
 	}
 
 	conf := subcommand.Config()
+	ctx := context.Background()
 
 	var confFilter config.FilesystemsFilter
 	job, err := conf.Job(testFilterArgs.job)
@@ -136,9 +137,10 @@ var testPlaceholder = &cli.Subcommand{
 	Run: runTestPlaceholder,
 }
 
-func runTestPlaceholder(ctx context.Context, subcommand *cli.Subcommand, args []string) error {
+func runTestPlaceholder(subcommand *cli.Subcommand, args []string) error {
 
 	var checkDPs []*zfs.DatasetPath
+	ctx := context.Background()
 
 	// all actions first
 	if testPlaceholderArgs.all {
@@ -195,11 +197,11 @@ var testDecodeResumeToken = &cli.Subcommand{
 	Run: runTestDecodeResumeTokenCmd,
 }
 
-func runTestDecodeResumeTokenCmd(ctx context.Context, subcommand *cli.Subcommand, args []string) error {
+func runTestDecodeResumeTokenCmd(subcommand *cli.Subcommand, args []string) error {
 	if testDecodeResumeTokenArgs.token == "" {
 		return fmt.Errorf("token argument must be specified")
 	}
-	token, err := zfs.ParseResumeToken(ctx, testDecodeResumeTokenArgs.token)
+	token, err := zfs.ParseResumeToken(context.Background(), testDecodeResumeTokenArgs.token)
 	if err != nil {
 		return err
 	}

client/version.go

@@ -1,7 +1,6 @@
 package client
 
 import (
-	"context"
 	"fmt"
 	"os"
 
@@ -26,7 +25,7 @@ var VersionCmd = &cli.Subcommand{
 	SetupFlags: func(f *pflag.FlagSet) {
 		f.StringVar(&versionArgs.Show, "show", "", "version info to show (client|daemon)")
 	},
-	Run: func(ctx context.Context, subcommand *cli.Subcommand, args []string) error {
+	Run: func(subcommand *cli.Subcommand, args []string) error {
 		versionArgs.Config = subcommand.Config()
 		versionArgs.ConfigErr = subcommand.ConfigParsingError()
 		return runVersionCmd()

client/zfsabstractions_create_step_hold.go

@@ -28,7 +28,7 @@ var zabsCmdCreateStepHold = &cli.Subcommand{
 	},
 }
 
-func doZabsCreateStep(ctx context.Context, sc *cli.Subcommand, args []string) error {
+func doZabsCreateStep(sc *cli.Subcommand, args []string) error {
 	if len(args) > 0 {
 		return errors.New("subcommand takes no arguments")
 	}
@@ -44,6 +44,8 @@ func doZabsCreateStep(ctx context.Context, sc *cli.Subcommand, args []string) er
 		return errors.Errorf("jobid must be set")
 	}
 
+	ctx := context.Background()
+
 	v, err := zfs.ZFSGetFilesystemVersion(ctx, f.target)
 	if err != nil {
 		return errors.Wrapf(err, "get info about target %q", f.target)

client/zfsabstractions_list.go

@@ -32,8 +32,9 @@ var zabsCmdList = &cli.Subcommand{
 	},
 }
 
-func doZabsList(ctx context.Context, sc *cli.Subcommand, args []string) error {
+func doZabsList(sc *cli.Subcommand, args []string) error {
 	var err error
+	ctx := context.Background()
 
 	if len(args) > 0 {
 		return errors.New("this subcommand takes no positional arguments")

client/zfsabstractions_release.go

@@ -43,8 +43,9 @@ var zabsCmdReleaseStale = &cli.Subcommand{
 	SetupFlags:      registerZabsReleaseFlags,
 }
 
-func doZabsReleaseAll(ctx context.Context, sc *cli.Subcommand, args []string) error {
+func doZabsReleaseAll(sc *cli.Subcommand, args []string) error {
 	var err error
+	ctx := context.Background()
 
 	if len(args) > 0 {
 		return errors.New("this subcommand takes no positional arguments")
@@ -67,9 +68,10 @@ func doZabsReleaseAll(ctx context.Context, sc *cli.Subcommand, args []string) er
 	return doZabsRelease_Common(ctx, abstractions)
 }
 
-func doZabsReleaseStale(ctx context.Context, sc *cli.Subcommand, args []string) error {
+func doZabsReleaseStale(sc *cli.Subcommand, args []string) error {
 
 	var err error
+	ctx := context.Background()
 
 	if len(args) > 0 {
 		return errors.New("this subcommand takes no positional arguments")

config/config.go

@@ -309,11 +309,6 @@ type PruneKeepNotReplicated struct {
 	KeepSnapshotAtCursor bool   `yaml:"keep_snapshot_at_cursor,optional,default=true"`
 }
 
-type PruneKeepStepHolds struct {
-	Type             string   `yaml:"type"`
-	AdditionalJobIds []string `yaml:"additional_job_ids,optional"`
-}
-
 type PruneKeepLastN struct {
 	Type  string `yaml:"type"`
 	Count int    `yaml:"count"`
@@ -485,7 +480,6 @@ func (t *ServeEnum) UnmarshalYAML(u func(interface{}, bool) error) (err error) {
 
 func (t *PruningEnum) UnmarshalYAML(u func(interface{}, bool) error) (err error) {
 	t.Ret, err = enumUnmarshal(u, map[string]interface{}{
-		"step_holds":     &PruneKeepStepHolds{},
 		"not_replicated": &PruneKeepNotReplicated{},
 		"last_n":         &PruneKeepLastN{},
 		"grid":           &PruneGrid{},

config/samples/source.yml

@@ -5,11 +5,7 @@ jobs:
     type: tcp
     listen: "0.0.0.0:8888"
     clients: {
-      "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-*",
+      "192.168.122.123" : "client1"
     }
   filesystems: {
     "<": true,

daemon/daemon.go

@@ -12,8 +12,6 @@ import (
 
 	"github.com/pkg/errors"
 	"github.com/prometheus/client_golang/prometheus"
-	"github.com/zrepl/zrepl/daemon/logging/trace"
-	"github.com/zrepl/zrepl/endpoint"
 
 	"github.com/zrepl/zrepl/config"
 	"github.com/zrepl/zrepl/daemon/job"
@@ -25,8 +23,9 @@ import (
 	"github.com/zrepl/zrepl/zfs/zfscmd"
 )
 
-func Run(ctx context.Context, conf *config.Config) error {
-	ctx, cancel := context.WithCancel(ctx)
+func Run(conf *config.Config) error {
+
+	ctx, cancel := context.WithCancel(context.Background())
 
 	defer cancel()
 	sigChan := make(chan os.Signal, 1)
@@ -40,7 +39,6 @@ func Run(ctx context.Context, conf *config.Config) error {
 	if err != nil {
 		return errors.Wrap(err, "cannot build logging from config")
 	}
-	outlets.Add(newPrometheusLogOutlet(), logger.Debug)
 
 	confJobs, err := job.JobsFromConfig(conf)
 	if err != nil {
@@ -50,23 +48,14 @@ func Run(ctx context.Context, conf *config.Config) error {
 	log := logger.NewLogger(outlets, 1*time.Second)
 	log.Info(version.NewZreplVersionInformation().String())
 
-	ctx = logging.WithLoggers(ctx, logging.SubsystemLoggersWithUniversalLogger(log))
-	trace.RegisterCallback(trace.Callback{
-		OnBegin: func(ctx context.Context) { logging.GetLogger(ctx, logging.SubsysTraceData).Debug("begin span") },
-		OnEnd: func(ctx context.Context, spanInfo trace.SpanInfo) {
-			logging.
-				GetLogger(ctx, logging.SubsysTraceData).
-				WithField("duration_s", spanInfo.EndedAt().Sub(spanInfo.StartedAt()).Seconds()).
-				Debug("finished span " + spanInfo.TaskAndSpanStack(trace.SpanStackKindAnnotation))
-		},
-	})
-
 	for _, job := range confJobs {
 		if IsInternalJobName(job.Name()) {
 			panic(fmt.Sprintf("internal job name used for config job '%s'", job.Name())) //FIXME
 		}
 	}
 
+	ctx = job.WithLogger(ctx, log)
+
 	jobs := newJobs()
 
 	// start control socket
@@ -95,8 +84,6 @@ func Run(ctx context.Context, conf *config.Config) error {
 
 	// register global (=non job-local) metrics
 	zfscmd.RegisterMetrics(prometheus.DefaultRegisterer)
-	trace.RegisterMetrics(prometheus.DefaultRegisterer)
-	endpoint.RegisterMetrics(prometheus.DefaultRegisterer)
 
 	log.Info("starting daemon")
 
@@ -111,8 +98,6 @@ func Run(ctx context.Context, conf *config.Config) error {
 	case <-ctx.Done():
 		log.WithError(ctx.Err()).Info("context finished")
 	}
-	log.Info("waiting for jobs to finish")
-	<-jobs.wait()
 	log.Info("daemon exiting")
 	return nil
 }
@@ -135,11 +120,14 @@ func newJobs() *jobs {
 	}
 }
 
+const (
+	logJobField string = "job"
+)
+
 func (s *jobs) wait() <-chan struct{} {
 	ch := make(chan struct{})
 	go func() {
 		s.wg.Wait()
-		close(ch)
 	}()
 	return ch
 }
@@ -214,8 +202,9 @@ func (s *jobs) start(ctx context.Context, j job.Job, internal bool) {
 	s.m.Lock()
 	defer s.m.Unlock()
 
-	ctx = logging.WithInjectedField(ctx, logging.JobField, j.Name())
-
+	jobLog := job.GetLogger(ctx).
+		WithField(logJobField, j.Name()).
+		WithOutlet(newPrometheusLogOutlet(j.Name()), logger.Debug)
 	jobName := j.Name()
 	if !internal && IsInternalJobName(jobName) {
 		panic(fmt.Sprintf("internal job name used for non-internal job %s", jobName))
@@ -230,6 +219,7 @@ func (s *jobs) start(ctx context.Context, j job.Job, internal bool) {
 	j.RegisterMetrics(prometheus.DefaultRegisterer)
 
 	s.jobs[jobName] = j
+	ctx = job.WithLogger(ctx, jobLog)
 	ctx = zfscmd.WithJobID(ctx, j.Name())
 	ctx, wakeup := wakeup.Context(ctx)
 	ctx, resetFunc := reset.Context(ctx)
@@ -239,8 +229,8 @@ func (s *jobs) start(ctx context.Context, j job.Job, internal bool) {
 	s.wg.Add(1)
 	go func() {
 		defer s.wg.Done()
-		job.GetLogger(ctx).Info("starting job")
-		defer job.GetLogger(ctx).Info("job exited")
+		jobLog.Info("starting job")
+		defer jobLog.Info("job exited")
 		j.Run(ctx)
 	}()
 }

daemon/hooks/hook_logging.go

@@ -6,17 +6,29 @@ import (
 	"context"
 	"sync"
 
-	"github.com/zrepl/zrepl/daemon/logging"
 	"github.com/zrepl/zrepl/logger"
 	"github.com/zrepl/zrepl/util/envconst"
 )
 
+type contextKey int
+
+const (
+	contextKeyLog contextKey = 0
+)
+
 type Logger = logger.Logger
 
+func WithLogger(ctx context.Context, log Logger) context.Context {
+	return context.WithValue(ctx, contextKeyLog, log)
+}
+
 func GetLogger(ctx context.Context) Logger { return getLogger(ctx) }
 
 func getLogger(ctx context.Context) Logger {
-	return logging.GetLogger(ctx, logging.SubsysHooks)
+	if log, ok := ctx.Value(contextKeyLog).(Logger); ok {
+		return log
+	}
+	return logger.NewNullLogger()
 }
 
 const MAX_HOOK_LOG_SIZE_DEFAULT int = 1 << 20

daemon/hooks/hooks_test.go

@@ -10,11 +10,9 @@ import (
 	"text/template"
 
 	"github.com/stretchr/testify/require"
-	"github.com/zrepl/zrepl/daemon/logging/trace"
 
 	"github.com/zrepl/zrepl/config"
 	"github.com/zrepl/zrepl/daemon/hooks"
-	"github.com/zrepl/zrepl/daemon/logging"
 	"github.com/zrepl/zrepl/logger"
 	"github.com/zrepl/zrepl/zfs"
 )
@@ -71,9 +69,6 @@ func curry(f comparisonAssertionFunc, expected interface{}, right bool) (ret val
 }
 
 func TestHooks(t *testing.T) {
-	ctx, end := trace.WithTaskFromStack(context.Background())
-	defer end()
-
 	testFSName := "testpool/testdataset"
 	testSnapshotName := "testsnap"
 
@@ -423,8 +418,9 @@ jobs:
 
 			cbReached = false
 
+			ctx := context.Background()
 			if testing.Verbose() && !tt.SuppressOutput {
-				ctx = logging.WithLoggers(ctx, logging.SubsystemLoggersWithUniversalLogger(log))
+				ctx = hooks.WithLogger(ctx, log)
 			}
 			plan.Run(ctx, false)
 			report := plan.Report()

daemon/job/active.go

@@ -8,13 +8,12 @@ import (
 
 	"github.com/pkg/errors"
 	"github.com/prometheus/client_golang/prometheus"
-	"github.com/prometheus/common/log"
-	"github.com/zrepl/zrepl/daemon/logging/trace"
 
 	"github.com/zrepl/zrepl/config"
 	"github.com/zrepl/zrepl/daemon/filters"
 	"github.com/zrepl/zrepl/daemon/job/reset"
 	"github.com/zrepl/zrepl/daemon/job/wakeup"
+	"github.com/zrepl/zrepl/daemon/logging"
 	"github.com/zrepl/zrepl/daemon/pruner"
 	"github.com/zrepl/zrepl/daemon/snapper"
 	"github.com/zrepl/zrepl/endpoint"
@@ -80,7 +79,7 @@ func (a *ActiveSide) updateTasks(u func(*activeSideTasks)) activeSideTasks {
 }
 
 type activeMode interface {
-	ConnectEndpoints(ctx context.Context, connecter transport.Connecter)
+	ConnectEndpoints(rpcLoggers rpc.Loggers, connecter transport.Connecter)
 	DisconnectEndpoints()
 	SenderReceiver() (logic.Sender, logic.Receiver)
 	Type() Type
@@ -99,14 +98,14 @@ type modePush struct {
 	snapper       *snapper.PeriodicOrManual
 }
 
-func (m *modePush) ConnectEndpoints(ctx context.Context, connecter transport.Connecter) {
+func (m *modePush) ConnectEndpoints(loggers rpc.Loggers, connecter transport.Connecter) {
 	m.setupMtx.Lock()
 	defer m.setupMtx.Unlock()
 	if m.receiver != nil || m.sender != nil {
 		panic("inconsistent use of ConnectEndpoints and DisconnectEndpoints")
 	}
 	m.sender = endpoint.NewSender(*m.senderConfig)
-	m.receiver = rpc.NewClient(connecter, rpc.GetLoggersOrPanic(ctx))
+	m.receiver = rpc.NewClient(connecter, loggers)
 }
 
 func (m *modePush) DisconnectEndpoints() {
@@ -177,14 +176,14 @@ type modePull struct {
 	interval       config.PositiveDurationOrManual
 }
 
-func (m *modePull) ConnectEndpoints(ctx context.Context, connecter transport.Connecter) {
+func (m *modePull) ConnectEndpoints(loggers rpc.Loggers, connecter transport.Connecter) {
 	m.setupMtx.Lock()
 	defer m.setupMtx.Unlock()
 	if m.receiver != nil || m.sender != nil {
 		panic("inconsistent use of ConnectEndpoints and DisconnectEndpoints")
 	}
 	m.receiver = endpoint.NewReceiver(m.receiverConfig)
-	m.sender = rpc.NewClient(connecter, rpc.GetLoggersOrPanic(ctx))
+	m.sender = rpc.NewClient(connecter, loggers)
 }
 
 func (m *modePull) DisconnectEndpoints() {
@@ -376,31 +375,16 @@ func (j *ActiveSide) SenderConfig() *endpoint.SenderConfig {
 	return push.senderConfig
 }
 
-// The active side of a replication uses one end (sender or receiver)
-// directly by method invocation, without going through a transport that
-// provides a client identity.
-// However, in order to avoid the need to distinguish between direct-method-invocating
-// clients and RPC client, we use an invalid client identity as a sentinel value.
-func FakeActiveSideDirectMethodInvocationClientIdentity(jobId endpoint.JobID) string {
-	return fmt.Sprintf("<local><active><job><client><identity><job=%q>", jobId.String())
-}
-
 func (j *ActiveSide) Run(ctx context.Context) {
-	ctx, endTask := trace.WithTaskAndSpan(ctx, "active-side-job", j.Name())
-	defer endTask()
-
-	ctx = context.WithValue(ctx, endpoint.ClientIdentityKey, FakeActiveSideDirectMethodInvocationClientIdentity(j.name))
-
 	log := GetLogger(ctx)
+	ctx = logging.WithSubsystemLoggers(ctx, log)
 
 	defer log.Info("job exiting")
 
 	periodicDone := make(chan struct{})
 	ctx, cancel := context.WithCancel(ctx)
 	defer cancel()
-	periodicCtx, endTask := trace.WithTask(ctx, "periodic")
-	defer endTask()
-	go j.mode.RunPeriodic(periodicCtx, periodicDone)
+	go j.mode.RunPeriodic(ctx, periodicDone)
 
 	invocationCount := 0
 outer:
@@ -416,15 +400,17 @@ outer:
 		case <-periodicDone:
 		}
 		invocationCount++
-		invocationCtx, endSpan := trace.WithSpan(ctx, fmt.Sprintf("invocation-%d", invocationCount))
-		j.do(invocationCtx)
-		endSpan()
+		invLog := log.WithField("invocation", invocationCount)
+		j.do(WithLogger(ctx, invLog))
 	}
 }
 
 func (j *ActiveSide) do(ctx context.Context) {
 
-	j.mode.ConnectEndpoints(ctx, j.connecter)
+	log := GetLogger(ctx)
+	ctx = logging.WithSubsystemLoggers(ctx, log)
+	loggers := rpc.GetLoggersOrPanic(ctx) // filled by WithSubsystemLoggers
+	j.mode.ConnectEndpoints(loggers, j.connecter)
 	defer j.mode.DisconnectEndpoints()
 
 	// allow cancellation of an invocation (this function)
@@ -447,22 +433,20 @@ func (j *ActiveSide) do(ctx context.Context) {
 			return
 		default:
 		}
-		ctx, endSpan := trace.WithSpan(ctx, "replication")
 		ctx, repCancel := context.WithCancel(ctx)
 		var repWait driver.WaitFunc
 		j.updateTasks(func(tasks *activeSideTasks) {
 			// reset it
 			*tasks = activeSideTasks{}
-			tasks.replicationCancel = func() { repCancel(); endSpan() }
+			tasks.replicationCancel = repCancel
 			tasks.replicationReport, repWait = replication.Do(
 				ctx, logic.NewPlanner(j.promRepStateSecs, j.promBytesReplicated, sender, receiver, j.mode.PlannerPolicy()),
 			)
 			tasks.state = ActiveSideReplicating
 		})
-		GetLogger(ctx).Info("start replication")
+		log.Info("start replication")
 		repWait(true) // wait blocking
 		repCancel()   // always cancel to free up context resources
-		endSpan()
 	}
 
 	{
@@ -471,18 +455,16 @@ func (j *ActiveSide) do(ctx context.Context) {
 			return
 		default:
 		}
-		ctx, endSpan := trace.WithSpan(ctx, "prune_sender")
 		ctx, senderCancel := context.WithCancel(ctx)
 		tasks := j.updateTasks(func(tasks *activeSideTasks) {
 			tasks.prunerSender = j.prunerFactory.BuildSenderPruner(ctx, sender, sender)
-			tasks.prunerSenderCancel = func() { senderCancel(); endSpan() }
+			tasks.prunerSenderCancel = senderCancel
 			tasks.state = ActiveSidePruneSender
 		})
-		GetLogger(ctx).Info("start pruning sender")
+		log.Info("start pruning sender")
 		tasks.prunerSender.Prune()
-		GetLogger(ctx).Info("finished pruning sender")
+		log.Info("finished pruning sender")
 		senderCancel()
-		endSpan()
 	}
 	{
 		select {
@@ -490,18 +472,16 @@ func (j *ActiveSide) do(ctx context.Context) {
 			return
 		default:
 		}
-		ctx, endSpan := trace.WithSpan(ctx, "prune_recever")
 		ctx, receiverCancel := context.WithCancel(ctx)
 		tasks := j.updateTasks(func(tasks *activeSideTasks) {
 			tasks.prunerReceiver = j.prunerFactory.BuildReceiverPruner(ctx, receiver, sender)
-			tasks.prunerReceiverCancel = func() { receiverCancel(); endSpan() }
+			tasks.prunerReceiverCancel = receiverCancel
 			tasks.state = ActiveSidePruneReceiver
 		})
-		GetLogger(ctx).Info("start pruning receiver")
+		log.Info("start pruning receiver")
 		tasks.prunerReceiver.Prune()
-		GetLogger(ctx).Info("finished pruning receiver")
+		log.Info("finished pruning receiver")
 		receiverCancel()
-		endSpan()
 	}
 
 	j.updateTasks(func(tasks *activeSideTasks) {

daemon/job/active_test.go

@@ -1,19 +0,0 @@
-package job
-
-import (
-	"testing"
-
-	"github.com/stretchr/testify/assert"
-	"github.com/stretchr/testify/require"
-	"github.com/zrepl/zrepl/endpoint"
-	"github.com/zrepl/zrepl/transport"
-)
-
-func TestFakeActiveSideDirectMethodInvocationClientIdentityDoesNotPassValidityTest(t *testing.T) {
-	jobid, err := endpoint.MakeJobID("validjobname")
-	require.NoError(t, err)
-	clientIdentity := FakeActiveSideDirectMethodInvocationClientIdentity(jobid)
-	t.Logf("%v", clientIdentity)
-	err = transport.ValidateClientIdentity(clientIdentity)
-	assert.Error(t, err)
-}

daemon/job/job.go

@@ -7,7 +7,6 @@ import (
 
 	"github.com/prometheus/client_golang/prometheus"
 
-	"github.com/zrepl/zrepl/daemon/logging"
 	"github.com/zrepl/zrepl/endpoint"
 	"github.com/zrepl/zrepl/logger"
 	"github.com/zrepl/zrepl/zfs"
@@ -15,8 +14,21 @@ import (
 
 type Logger = logger.Logger
 
+type contextKey int
+
+const (
+	contextKeyLog contextKey = iota
+)
+
 func GetLogger(ctx context.Context) Logger {
-	return logging.GetLogger(ctx, logging.SubsysJob)
+	if l, ok := ctx.Value(contextKeyLog).(Logger); ok {
+		return l
+	}
+	return logger.NewNullLogger()
+}
+
+func WithLogger(ctx context.Context, l Logger) context.Context {
+	return context.WithValue(ctx, contextKeyLog, l)
 }
 
 type Job interface {

daemon/job/passive.go

@@ -6,7 +6,6 @@ import (
 
 	"github.com/pkg/errors"
 	"github.com/prometheus/client_golang/prometheus"
-	"github.com/zrepl/zrepl/daemon/logging/trace"
 
 	"github.com/zrepl/zrepl/config"
 	"github.com/zrepl/zrepl/daemon/filters"
@@ -165,14 +164,12 @@ func (j *PassiveSide) SenderConfig() *endpoint.SenderConfig {
 func (*PassiveSide) RegisterMetrics(registerer prometheus.Registerer) {}
 
 func (j *PassiveSide) Run(ctx context.Context) {
-	ctx, endTask := trace.WithTaskAndSpan(ctx, "passive-side-job", j.Name())
-	defer endTask()
+
 	log := GetLogger(ctx)
 	defer log.Info("job exiting")
+	ctx = logging.WithSubsystemLoggers(ctx, log)
 	{
-		ctx, endTask := trace.WithTask(ctx, "periodic") // shadowing
-		defer endTask()
-		ctx, cancel := context.WithCancel(ctx)
+		ctx, cancel := context.WithCancel(ctx) // shadowing
 		defer cancel()
 		go j.mode.RunPeriodic(ctx)
 	}
@@ -182,14 +179,8 @@ func (j *PassiveSide) Run(ctx context.Context) {
 		panic(fmt.Sprintf("implementation error: j.mode.Handler() returned nil: %#v", j))
 	}
 
-	ctxInterceptor := func(handlerCtx context.Context, info rpc.HandlerContextInterceptorData, handler func(ctx context.Context)) {
-		// the handlerCtx is clean => need to inherit logging and tracing config from job context
-		handlerCtx = logging.WithInherit(handlerCtx, ctx)
-		handlerCtx = trace.WithInherit(handlerCtx, ctx)
-
-		handlerCtx, endTask := trace.WithTaskAndSpan(handlerCtx, "handler", fmt.Sprintf("job=%q client=%q method=%q", j.Name(), info.ClientIdentity(), info.FullMethod()))
-		defer endTask()
-		handler(handlerCtx)
+	ctxInterceptor := func(handlerCtx context.Context) context.Context {
+		return logging.WithSubsystemLoggers(handlerCtx, log)
 	}
 
 	rpcLoggers := rpc.GetLoggersOrPanic(ctx) // WithSubsystemLoggers above

daemon/job/snapjob.go

@@ -2,16 +2,15 @@ package job
 
 import (
 	"context"
-	"fmt"
 	"sort"
 
 	"github.com/pkg/errors"
 	"github.com/prometheus/client_golang/prometheus"
-	"github.com/zrepl/zrepl/daemon/logging/trace"
 
 	"github.com/zrepl/zrepl/config"
 	"github.com/zrepl/zrepl/daemon/filters"
 	"github.com/zrepl/zrepl/daemon/job/wakeup"
+	"github.com/zrepl/zrepl/daemon/logging"
 	"github.com/zrepl/zrepl/daemon/pruner"
 	"github.com/zrepl/zrepl/daemon/snapper"
 	"github.com/zrepl/zrepl/endpoint"
@@ -90,18 +89,15 @@ func (j *SnapJob) OwnedDatasetSubtreeRoot() (rfs *zfs.DatasetPath, ok bool) {
 func (j *SnapJob) SenderConfig() *endpoint.SenderConfig { return nil }
 
 func (j *SnapJob) Run(ctx context.Context) {
-	ctx, endTask := trace.WithTaskAndSpan(ctx, "snap-job", j.Name())
-	defer endTask()
 	log := GetLogger(ctx)
+	ctx = logging.WithSubsystemLoggers(ctx, log)
 
 	defer log.Info("job exiting")
 
 	periodicDone := make(chan struct{})
 	ctx, cancel := context.WithCancel(ctx)
 	defer cancel()
-	periodicCtx, endTask := trace.WithTask(ctx, "snapshotting")
-	defer endTask()
-	go j.snapper.Run(periodicCtx, periodicDone)
+	go j.snapper.Run(ctx, periodicDone)
 
 	invocationCount := 0
 outer:
@@ -116,10 +112,8 @@ outer:
 		case <-periodicDone:
 		}
 		invocationCount++
-
-		invocationCtx, endSpan := trace.WithSpan(ctx, fmt.Sprintf("invocation-%d", invocationCount))
-		j.doPrune(invocationCtx)
-		endSpan()
+		invLog := log.WithField("invocation", invocationCount)
+		j.doPrune(WithLogger(ctx, invLog))
 	}
 }
 
@@ -167,9 +161,8 @@ func (h alwaysUpToDateReplicationCursorHistory) ListFilesystems(ctx context.Cont
 }
 
 func (j *SnapJob) doPrune(ctx context.Context) {
-	ctx, endSpan := trace.WithSpan(ctx, "snap-job-do-prune")
-	defer endSpan()
 	log := GetLogger(ctx)
+	ctx = logging.WithSubsystemLoggers(ctx, log)
 	sender := endpoint.NewSender(endpoint.SenderConfig{
 		JobID: j.name,
 		FSF:   j.fsfilter,

daemon/logging/build_logging.go

@@ -9,10 +9,20 @@ import (
 
 	"github.com/mattn/go-isatty"
 	"github.com/pkg/errors"
+
 	"github.com/zrepl/zrepl/config"
-	"github.com/zrepl/zrepl/daemon/logging/trace"
+	"github.com/zrepl/zrepl/daemon/hooks"
+	"github.com/zrepl/zrepl/daemon/pruner"
+	"github.com/zrepl/zrepl/daemon/snapper"
+	"github.com/zrepl/zrepl/endpoint"
 	"github.com/zrepl/zrepl/logger"
+	"github.com/zrepl/zrepl/replication/driver"
+	"github.com/zrepl/zrepl/replication/logic"
+	"github.com/zrepl/zrepl/rpc"
+	"github.com/zrepl/zrepl/rpc/transportmux"
 	"github.com/zrepl/zrepl/tlsconf"
+	"github.com/zrepl/zrepl/transport"
+	"github.com/zrepl/zrepl/zfs/zfscmd"
 )
 
 func OutletsFromConfig(in config.LoggingOutletEnumList) (*logger.Outlets, error) {
@@ -60,8 +70,6 @@ func OutletsFromConfig(in config.LoggingOutletEnumList) (*logger.Outlets, error)
 type Subsystem string
 
 const (
-	SubsysMeta         Subsystem = "meta"
-	SubsysJob          Subsystem = "job"
 	SubsysReplication  Subsystem = "repl"
 	SubsysEndpoint     Subsystem = "endpoint"
 	SubsysPruning      Subsystem = "pruning"
@@ -73,103 +81,30 @@ const (
 	SubsysRPCControl   Subsystem = "rpc.ctrl"
 	SubsysRPCData      Subsystem = "rpc.data"
 	SubsysZFSCmd       Subsystem = "zfs.cmd"
-	SubsysTraceData    Subsystem = "trace.data"
-	SubsysPlatformtest Subsystem = "platformtest"
 )
 
-var AllSubsystems = []Subsystem{
-	SubsysMeta,
-	SubsysJob,
-	SubsysReplication,
-	SubsysEndpoint,
-	SubsysPruning,
-	SubsysSnapshot,
-	SubsysHooks,
-	SubsysTransport,
-	SubsysTransportMux,
-	SubsysRPC,
-	SubsysRPCControl,
-	SubsysRPCData,
-	SubsysZFSCmd,
-	SubsysTraceData,
-	SubsysPlatformtest,
+func WithSubsystemLoggers(ctx context.Context, log logger.Logger) context.Context {
+	ctx = logic.WithLogger(ctx, log.WithField(SubsysField, SubsysReplication))
+	ctx = driver.WithLogger(ctx, log.WithField(SubsysField, SubsysReplication))
+	ctx = endpoint.WithLogger(ctx, log.WithField(SubsysField, SubsysEndpoint))
+	ctx = pruner.WithLogger(ctx, log.WithField(SubsysField, SubsysPruning))
+	ctx = snapper.WithLogger(ctx, log.WithField(SubsysField, SubsysSnapshot))
+	ctx = hooks.WithLogger(ctx, log.WithField(SubsysField, SubsysHooks))
+	ctx = transport.WithLogger(ctx, log.WithField(SubsysField, SubsysTransport))
+	ctx = transportmux.WithLogger(ctx, log.WithField(SubsysField, SubsysTransportMux))
+	ctx = zfscmd.WithLogger(ctx, log.WithField(SubsysField, SubsysZFSCmd))
+	ctx = rpc.WithLoggers(ctx,
+		rpc.Loggers{
+			General: log.WithField(SubsysField, SubsysRPC),
+			Control: log.WithField(SubsysField, SubsysRPCControl),
+			Data:    log.WithField(SubsysField, SubsysRPCData),
+		},
+	)
+	return ctx
 }
 
-type injectedField struct {
-	field  string
-	value  interface{}
-	parent *injectedField
-}
-
-func WithInjectedField(ctx context.Context, field string, value interface{}) context.Context {
-	var parent *injectedField
-	parentI := ctx.Value(contextKeyInjectedField)
-	if parentI != nil {
-		parent = parentI.(*injectedField)
-	}
-	// TODO sanity-check `field` now
-	this := &injectedField{field, value, parent}
-	return context.WithValue(ctx, contextKeyInjectedField, this)
-}
-
-func iterInjectedFields(ctx context.Context, cb func(field string, value interface{})) {
-	injI := ctx.Value(contextKeyInjectedField)
-	if injI == nil {
-		return
-	}
-	inj := injI.(*injectedField)
-	for ; inj != nil; inj = inj.parent {
-		cb(inj.field, inj.value)
-	}
-}
-
-type SubsystemLoggers map[Subsystem]logger.Logger
-
-func SubsystemLoggersWithUniversalLogger(l logger.Logger) SubsystemLoggers {
-	loggers := make(SubsystemLoggers)
-	for _, s := range AllSubsystems {
-		loggers[s] = l
-	}
-	return loggers
-}
-
-func WithLoggers(ctx context.Context, loggers SubsystemLoggers) context.Context {
-	return context.WithValue(ctx, contextKeyLoggers, loggers)
-}
-
-func GetLoggers(ctx context.Context) SubsystemLoggers {
-	loggers, ok := ctx.Value(contextKeyLoggers).(SubsystemLoggers)
-	if !ok {
-		return nil
-	}
-	return loggers
-}
-
-func GetLogger(ctx context.Context, subsys Subsystem) logger.Logger {
-	return getLoggerImpl(ctx, subsys, true)
-}
-
-func getLoggerImpl(ctx context.Context, subsys Subsystem, panicIfEnded bool) logger.Logger {
-	loggers, ok := ctx.Value(contextKeyLoggers).(SubsystemLoggers)
-	if !ok || loggers == nil {
-		return logger.NewNullLogger()
-	}
-	l, ok := loggers[subsys]
-	if !ok {
-		return logger.NewNullLogger()
-	}
-
-	l = l.WithField(SubsysField, subsys)
-
-	l = l.WithField(SpanField, trace.GetSpanStackOrDefault(ctx, *trace.StackKindId, "NOSPAN"))
-
-	fields := make(logger.Fields)
-	iterInjectedFields(ctx, func(field string, value interface{}) {
-		fields[field] = value
-	})
-	l = l.WithFields(fields)
-
-	return l
+func LogSubsystem(log logger.Logger, subsys Subsystem) logger.Logger {
+	return log.ReplaceField(SubsysField, subsys)
 }
 
 func parseLogFormat(i interface{}) (f EntryFormatter, err error) {

daemon/logging/logging_context.go

@@ -1,24 +0,0 @@
-package logging
-
-import "context"
-
-type contextKey int
-
-const (
-	contextKeyLoggers contextKey = 1 + iota
-	contextKeyInjectedField
-)
-
-var contextKeys = []contextKey{
-	contextKeyLoggers,
-	contextKeyInjectedField,
-}
-
-func WithInherit(ctx, inheritFrom context.Context) context.Context {
-	for _, k := range contextKeys {
-		if v := inheritFrom.Value(k); v != nil {
-			ctx = context.WithValue(ctx, k, v) // no shadow
-		}
-	}
-	return ctx
-}

daemon/logging/logging_formatters.go

@@ -22,7 +22,6 @@ const (
 const (
 	JobField    string = "job"
 	SubsysField string = "subsystem"
-	SpanField   string = "span"
 )
 
 type MetadataFlags int64
@@ -86,7 +85,7 @@ func (f *HumanFormatter) Format(e *logger.Entry) (out []byte, err error) {
 		fmt.Fprintf(&line, "[%s]", col.Sprint(e.Level.Short()))
 	}
 
-	prefixFields := []string{JobField, SubsysField, SpanField}
+	prefixFields := []string{JobField, SubsysField}
 	prefixed := make(map[string]bool, len(prefixFields)+2)
 	for _, field := range prefixFields {
 		val, ok := e.Fields[field]
@@ -175,8 +174,8 @@ func (f *LogfmtFormatter) Format(e *logger.Entry) ([]byte, error) {
 	}
 
 	// at least try and put job and task in front
-	prefixed := make(map[string]bool, 3)
-	prefix := []string{JobField, SubsysField, SpanField}
+	prefixed := make(map[string]bool, 2)
+	prefix := []string{JobField, SubsysField}
 	for _, pf := range prefix {
 		v, ok := e.Fields[pf]
 		if !ok {

daemon/logging/trace/trace.go

@@ -1,386 +0,0 @@
-// package trace provides activity tracing via ctx through Tasks and Spans
-//
-// Basic Concepts
-//
-// Tracing can be used to identify where a piece of code spends its time.
-//
-// The Go standard library provides package runtime/trace which is useful to identify CPU bottlenecks or
-// to understand what happens inside the Go runtime.
-// However, it is not ideal for application level tracing, in particular if those traces should be understandable
-// to tech-savvy users (albeit not developers).
-//
-// This package provides the concept of Tasks and Spans to express what activity is happening within an application:
-//
-//  - Neither task nor span is really tangible but instead contained within the context.Context tree
-//  - Tasks represent concurrent activity (i.e. goroutines).
-//  - Spans represent a semantic stack trace within a task.
-//
-// As a consequence, whenever a context is propagated across goroutine boundary, you need to create a child task:
-//
-//   go func(ctx context.Context) {
-//     ctx, endTask = WithTask(ctx, "what-happens-inside-the-child-task")
-//     defer endTask()
-//     // ...
-//   }(ctx)
-//
-// Within the task, you can open up a hierarchy of spans.
-// In contrast to tasks, which have can multiple concurrently running child tasks,
-// spans must nest and not cross the goroutine boundary.
-//
-//  ctx, endSpan = WithSpan(ctx, "copy-dir")
-//  defer endSpan()
-//  for _, f := range dir.Files() {
-//    func() {
-//      ctx, endSpan := WithSpan(ctx, fmt.Sprintf("copy-file %q", f))
-//      defer endspan()
-//      b, _ := ioutil.ReadFile(f)
-//      _ = ioutil.WriteFile(f + ".copy", b, 0600)
-//    }()
-//  }
-//
-// In combination:
-//  ctx, endTask = WithTask(ctx, "copy-dirs")
-//  defer endTask()
-//  for i := range dirs {
-//    go func(dir string) {
-//      ctx, endTask := WithTask(ctx, "copy-dir")
-//      defer endTask()
-//      for _, f := range filesIn(dir) {
-//        func() {
-//          ctx, endSpan := WithSpan(ctx, fmt.Sprintf("copy-file %q", f))
-//          defer endspan()
-//          b, _ := ioutil.ReadFile(f)
-//          _ = ioutil.WriteFile(f + ".copy", b, 0600)
-//        }()
-//      }
-//    }()
-//  }
-//
-// Note that a span ends at the time you call endSpan - not before and not after that.
-// If you violate the stack-like nesting of spans by forgetting an endSpan() invocation,
-// the out-of-order endSpan() will panic.
-//
-// A similar rule applies to the endTask closure returned by WithTask:
-// If a task has live child tasks at the time you call endTask(), the call will panic.
-//
-// Recovering from endSpan() or endTask() panics will corrupt the trace stack and lead to corrupt tracefile output.
-//
-//
-// Best Practices For Naming Tasks And Spans
-//
-// Tasks should always have string constants as names, and must not contain the `#` character. WHy?
-// First, the visualization by chrome://tracing draws a horizontal bar for each task in the trace.
-// Also, the package appends `#NUM` for each concurrently running instance of a task name.
-// Note that the `#NUM` suffix will be reused if a task has ended, in order to avoid an
-// infinite number of horizontal bars in the visualization.
-//
-//
-// Chrome-compatible Tracefile Support
-//
-// The activity trace generated by usage of WithTask and WithSpan can be rendered to a JSON output file
-// that can be loaded into chrome://tracing .
-// Apart from function GetSpanStackOrDefault, this is the main benefit of this package.
-//
-// First, there is a convenience environment variable 'ZREPL_ACTIVITY_TRACE' that can be set to an output path.
-// From process start onward, a trace is written to that path.
-//
-// More consumers can attach to the activity trace through the ChrometraceClientWebsocketHandler websocket handler.
-//
-// If a write error is encountered with any consumer (including the env-var based one), the consumer is closed and
-// will not receive further trace output.
-package trace
-
-import (
-	"context"
-	"fmt"
-	"strings"
-	"time"
-
-	"github.com/pkg/errors"
-	"github.com/prometheus/client_golang/prometheus"
-	"github.com/zrepl/zrepl/util/chainlock"
-)
-
-var metrics struct {
-	activeTasks prometheus.Gauge
-}
-var taskNamer *uniqueConcurrentTaskNamer = newUniqueTaskNamer()
-
-func init() {
-	metrics.activeTasks = prometheus.NewGauge(prometheus.GaugeOpts{
-		Namespace: "zrepl",
-		Subsystem: "trace",
-		Name:      "active_tasks",
-		Help:      "number of active (tracing-level) tasks in the daemon",
-	})
-}
-
-func RegisterMetrics(r prometheus.Registerer) {
-	r.MustRegister(metrics.activeTasks)
-}
-
-type traceNode struct {
-	id         string
-	annotation string
-	parentTask *traceNode
-
-	mtx chainlock.L
-
-	activeChildTasks int32 // only for task nodes, insignificant for span nodes
-	parentSpan       *traceNode
-	activeChildSpan  *traceNode // nil if task or span doesn't have an active child span
-
-	startedAt time.Time
-	endedAt   time.Time
-}
-
-func (s *traceNode) StartedAt() time.Time { return s.startedAt }
-func (s *traceNode) EndedAt() time.Time   { return s.endedAt }
-
-// Returned from WithTask or WithSpan.
-// Must be called once the task or span ends.
-// See package-level docs for nesting rules.
-// Wrong call order / forgetting to call it will result in panics.
-type DoneFunc func()
-
-var ErrTaskStillHasActiveChildTasks = fmt.Errorf("end task: task still has active child tasks")
-
-// Start a new root task or create a child task of an existing task.
-//
-// This is required when starting a new goroutine and
-// passing an existing task context to it.
-//
-// taskName should be a constantand must not contain '#'
-//
-// The implementation ensures that,
-// if multiple tasks with the same name exist simultaneously,
-// a unique suffix is appended to uniquely identify the task opened with this function.
-func WithTask(ctx context.Context, taskName string) (context.Context, DoneFunc) {
-
-	var parentTask *traceNode
-	nodeI := ctx.Value(contextKeyTraceNode)
-	if nodeI != nil {
-		node := nodeI.(*traceNode)
-		if node.parentSpan != nil {
-			parentTask = node.parentTask
-		} else {
-			parentTask = node
-		}
-	}
-	// find the first ancestor that hasn't ended yet (nil if need be)
-	if parentTask != nil {
-		parentTask.mtx.Lock()
-	}
-	for parentTask != nil && !parentTask.endedAt.IsZero() {
-		thisParent := parentTask
-		parentTask = parentTask.parentTask
-		// lock parent first such that it isn't modified by other callers to thisParent
-		if parentTask != nil {
-			parentTask.mtx.Lock()
-		}
-		thisParent.mtx.Unlock()
-	}
-	// invariant: either parentTask != nil and we hold the lock on parentTask, or parentTask is nil
-
-	taskName, taskNameDone := taskNamer.UniqueConcurrentTaskName(taskName)
-
-	this := &traceNode{
-		id:               genID(),
-		annotation:       taskName,
-		parentTask:       parentTask,
-		activeChildTasks: 0,
-		parentSpan:       nil,
-		activeChildSpan:  nil,
-
-		startedAt: time.Now(),
-		endedAt:   time.Time{},
-	}
-
-	if parentTask != nil {
-		this.parentTask.activeChildTasks++
-		parentTask.mtx.Unlock()
-	}
-
-	ctx = context.WithValue(ctx, contextKeyTraceNode, this)
-
-	chrometraceBeginTask(this)
-
-	metrics.activeTasks.Inc()
-
-	endTaskFunc := func() {
-
-		// only hold locks while manipulating the tree
-		// (trace writer might block too long and unlike spans, tasks are updated concurrently)
-		alreadyEnded := func() (alreadyEnded bool) {
-			if this.parentTask != nil {
-				defer this.parentTask.mtx.Lock().Unlock()
-			}
-			defer this.mtx.Lock().Unlock()
-
-			if this.activeChildTasks != 0 {
-				panic(errors.Wrapf(ErrTaskStillHasActiveChildTasks, "end task: %v active child tasks", this.activeChildTasks))
-			}
-
-			// support idempotent task ends
-			if !this.endedAt.IsZero() {
-				return true
-			}
-			this.endedAt = time.Now()
-
-			if this.parentTask != nil {
-				this.parentTask.activeChildTasks--
-				if this.parentTask.activeChildTasks < 0 {
-					panic("impl error: parent task with negative activeChildTasks count")
-				}
-			}
-			return false
-		}()
-		if alreadyEnded {
-			return
-		}
-
-		chrometraceEndTask(this)
-
-		metrics.activeTasks.Dec()
-
-		taskNameDone()
-	}
-
-	return ctx, endTaskFunc
-}
-
-var ErrAlreadyActiveChildSpan = fmt.Errorf("create child span: span already has an active child span")
-var ErrSpanStillHasActiveChildSpan = fmt.Errorf("end span: span still has active child spans")
-
-// Start a new span.
-// Important: ctx must have an active task (see WithTask)
-func WithSpan(ctx context.Context, annotation string) (context.Context, DoneFunc) {
-	var parentSpan, parentTask *traceNode
-	nodeI := ctx.Value(contextKeyTraceNode)
-	if nodeI != nil {
-		parentSpan = nodeI.(*traceNode)
-		if parentSpan.parentSpan == nil {
-			parentTask = parentSpan
-		} else {
-			parentTask = parentSpan.parentTask
-		}
-	} else {
-		panic("must be called from within a task")
-	}
-
-	this := &traceNode{
-		id:              genID(),
-		annotation:      annotation,
-		parentTask:      parentTask,
-		parentSpan:      parentSpan,
-		activeChildSpan: nil,
-
-		startedAt: time.Now(),
-		endedAt:   time.Time{},
-	}
-
-	parentSpan.mtx.HoldWhile(func() {
-		if parentSpan.activeChildSpan != nil {
-			panic(ErrAlreadyActiveChildSpan)
-		}
-		parentSpan.activeChildSpan = this
-	})
-
-	ctx = context.WithValue(ctx, contextKeyTraceNode, this)
-	chrometraceBeginSpan(this)
-	callbackEndSpan := callbackBeginSpan(ctx)
-
-	endTaskFunc := func() {
-
-		defer parentSpan.mtx.Lock().Unlock()
-		if parentSpan.activeChildSpan != this && this.endedAt.IsZero() {
-			panic("impl error: activeChildSpan should not change while != nil because there can only be one")
-		}
-
-		defer this.mtx.Lock().Unlock()
-		if this.activeChildSpan != nil {
-			panic(ErrSpanStillHasActiveChildSpan)
-		}
-
-		if !this.endedAt.IsZero() {
-			return // support idempotent span ends
-		}
-
-		parentSpan.activeChildSpan = nil
-		this.endedAt = time.Now()
-
-		chrometraceEndSpan(this)
-		callbackEndSpan(this)
-	}
-
-	return ctx, endTaskFunc
-}
-
-type StackKind struct {
-	symbolizeTask func(t *traceNode) string
-	symbolizeSpan func(s *traceNode) string
-}
-
-var (
-	StackKindId = &StackKind{
-		symbolizeTask: func(t *traceNode) string { return t.id },
-		symbolizeSpan: func(s *traceNode) string { return s.id },
-	}
-	SpanStackKindCombined = &StackKind{
-		symbolizeTask: func(t *traceNode) string { return fmt.Sprintf("(%s %q)", t.id, t.annotation) },
-		symbolizeSpan: func(s *traceNode) string { return fmt.Sprintf("(%s %q)", s.id, s.annotation) },
-	}
-	SpanStackKindAnnotation = &StackKind{
-		symbolizeTask: func(t *traceNode) string { return t.annotation },
-		symbolizeSpan: func(s *traceNode) string { return s.annotation },
-	}
-)
-
-func (n *traceNode) task() *traceNode {
-	task := n.parentTask
-	if n.parentSpan == nil {
-		task = n
-	}
-	return task
-}
-
-func (n *traceNode) TaskName() string {
-	task := n.task()
-	return task.annotation
-}
-
-func (this *traceNode) TaskAndSpanStack(kind *StackKind) (spanIdStack string) {
-	task := this.task()
-
-	var spansInTask []*traceNode
-	for s := this; s != nil; s = s.parentSpan {
-		spansInTask = append(spansInTask, s)
-	}
-
-	var tasks []*traceNode
-	for t := task; t != nil; t = t.parentTask {
-		tasks = append(tasks, t)
-	}
-
-	var taskIdsRev []string
-	for i := len(tasks) - 1; i >= 0; i-- {
-		taskIdsRev = append(taskIdsRev, kind.symbolizeTask(tasks[i]))
-	}
-
-	var spanIdsRev []string
-	for i := len(spansInTask) - 1; i >= 0; i-- {
-		spanIdsRev = append(spanIdsRev, kind.symbolizeSpan(spansInTask[i]))
-	}
-
-	taskStack := strings.Join(taskIdsRev, "$")
-	return fmt.Sprintf("%s$%s", taskStack, strings.Join(spanIdsRev, "."))
-}
-
-func GetSpanStackOrDefault(ctx context.Context, kind StackKind, def string) string {
-	if nI := ctx.Value(contextKeyTraceNode); nI != nil {
-		n := nI.(*traceNode)
-		return n.TaskAndSpanStack(StackKindId)
-	} else {
-		return def
-	}
-}

daemon/logging/trace/trace_callback.go

@@ -1,54 +0,0 @@
-package trace
-
-import (
-	"context"
-	"time"
-
-	"github.com/zrepl/zrepl/util/chainlock"
-)
-
-type SpanInfo interface {
-	StartedAt() time.Time
-	EndedAt() time.Time
-	TaskAndSpanStack(kind *StackKind) string
-}
-
-type Callback struct {
-	OnBegin func(ctx context.Context)
-	OnEnd   func(ctx context.Context, spanInfo SpanInfo)
-}
-
-var callbacks struct {
-	mtx chainlock.L
-	cs  []Callback
-}
-
-func RegisterCallback(c Callback) {
-	callbacks.mtx.HoldWhile(func() {
-		callbacks.cs = append(callbacks.cs, c)
-	})
-}
-
-func callbackBeginSpan(ctx context.Context) func(SpanInfo) {
-	// capture the current state of callbacks into a local variable
-	// this is safe because the slice is append-only and immutable
-
-	// (it is important that a callback registered _after_ callbackBeginSpin is called does not get called on OnEnd)
-
-	var cbs []Callback
-	callbacks.mtx.HoldWhile(func() {
-		cbs = callbacks.cs
-	})
-	for _, cb := range cbs {
-		if cb.OnBegin != nil {
-			cb.OnBegin(ctx)
-		}
-	}
-	return func(spanInfo SpanInfo) {
-		for _, cb := range cbs {
-			if cb.OnEnd != nil {
-				cb.OnEnd(ctx, spanInfo)
-			}
-		}
-	}
-}

daemon/logging/trace/trace_chrometrace.go

@@ -1,230 +0,0 @@
-package trace
-
-// The functions in this file are concerned with the generation
-// of trace files based on the information from WithTask and WithSpan.
-//
-// The emitted trace files are open-ended array of JSON objects
-// that follow the Chrome trace file format:
-//   https://docs.google.com/document/d/1CvAClvFfyA5R-PhYUmn5OOQtYMH4h6I0nSsKchNAySU/preview
-//
-// The emitted JSON can be loaded into Chrome's chrome://tracing view.
-//
-// The trace file can be written to a file whose path is specified in an env file,
-// and be written to web sockets established on ChrometraceHttpHandler
-
-import (
-	"bufio"
-	"bytes"
-	"encoding/json"
-	"fmt"
-	"io"
-	"os"
-	"sync"
-	"sync/atomic"
-
-	"golang.org/x/net/websocket"
-
-	"github.com/zrepl/zrepl/util/envconst"
-)
-
-var chrometracePID string
-
-func init() {
-	var err error
-	chrometracePID, err = os.Hostname()
-	if err != nil {
-		panic(err)
-	}
-	chrometracePID = fmt.Sprintf("%q", chrometracePID)
-}
-
-type chrometraceEvent struct {
-	Cat                       string   `json:"cat,omitempty"`
-	Name                      string   `json:"name"`
-	Stack                     []string `json:"stack,omitempty"`
-	Phase                     string   `json:"ph"`
-	TimestampUnixMicroseconds int64    `json:"ts"`
-	DurationMicroseconds      int64    `json:"dur,omitempty"`
-	Pid                       string   `json:"pid"`
-	Tid                       string   `json:"tid"`
-	Id                        string   `json:"id,omitempty"`
-}
-
-func chrometraceBeginSpan(s *traceNode) {
-	taskName := s.TaskName()
-	chrometraceWrite(chrometraceEvent{
-		Name:                      s.annotation,
-		Phase:                     "B",
-		TimestampUnixMicroseconds: s.startedAt.UnixNano() / 1000,
-		Pid:                       chrometracePID,
-		Tid:                       taskName,
-	})
-}
-
-func chrometraceEndSpan(s *traceNode) {
-	taskName := s.TaskName()
-	chrometraceWrite(chrometraceEvent{
-		Name:                      s.annotation,
-		Phase:                     "E",
-		TimestampUnixMicroseconds: s.endedAt.UnixNano() / 1000,
-		Pid:                       chrometracePID,
-		Tid:                       taskName,
-	})
-}
-
-var chrometraceFlowId uint64
-
-func chrometraceBeginTask(s *traceNode) {
-	chrometraceBeginSpan(s)
-
-	if s.parentTask == nil {
-		return
-	}
-	// beginning of a task that has a parent
-	// => use flow events to link parent and child
-
-	flowId := atomic.AddUint64(&chrometraceFlowId, 1)
-	flowIdStr := fmt.Sprintf("%x", flowId)
-
-	parentTask := s.parentTask.TaskName()
-	chrometraceWrite(chrometraceEvent{
-		Cat:                       "task", // seems to be necessary, otherwise the GUI shows some `indexOf` JS error
-		Name:                      "child-task",
-		Phase:                     "s",
-		TimestampUnixMicroseconds: s.startedAt.UnixNano() / 1000, // yes, the child's timestamp (=> from-point of the flow line is at right x-position of parent's bar)
-		Pid:                       chrometracePID,
-		Tid:                       parentTask,
-		Id:                        flowIdStr,
-	})
-
-	childTask := s.TaskName()
-	if parentTask == childTask {
-		panic(parentTask)
-	}
-	chrometraceWrite(chrometraceEvent{
-		Cat:                       "task", // seems to be necessary, otherwise the GUI shows some `indexOf` JS error
-		Name:                      "child-task",
-		Phase:                     "f",
-		TimestampUnixMicroseconds: s.startedAt.UnixNano() / 1000,
-		Pid:                       chrometracePID,
-		Tid:                       childTask,
-		Id:                        flowIdStr,
-	})
-}
-
-func chrometraceEndTask(s *traceNode) {
-	chrometraceEndSpan(s)
-}
-
-type chrometraceConsumerRegistration struct {
-	w io.Writer
-	// errored must have capacity 1, the writer thread will send to it non-blocking, then close it
-	errored chan error
-}
-
-var chrometraceConsumers struct {
-	register  chan chrometraceConsumerRegistration
-	consumers map[chrometraceConsumerRegistration]bool
-	write     chan []byte
-}
-
-func init() {
-	chrometraceConsumers.register = make(chan chrometraceConsumerRegistration)
-	chrometraceConsumers.consumers = make(map[chrometraceConsumerRegistration]bool)
-	chrometraceConsumers.write = make(chan []byte)
-	go func() {
-		kickConsumer := func(c chrometraceConsumerRegistration, err error) {
-			debug("chrometrace kicking consumer %#v after error %v", c, err)
-			select {
-			case c.errored <- err:
-			default:
-			}
-			close(c.errored)
-			delete(chrometraceConsumers.consumers, c)
-		}
-		for {
-			select {
-			case reg := <-chrometraceConsumers.register:
-				debug("registered chrometrace consumer %#v", reg)
-				chrometraceConsumers.consumers[reg] = true
-				n, err := reg.w.Write([]byte("[\n"))
-				if err != nil {
-					kickConsumer(reg, err)
-				} else if n != 2 {
-					kickConsumer(reg, fmt.Errorf("short write: %v", n))
-				}
-				// successfully registered
-
-			case buf := <-chrometraceConsumers.write:
-				debug("chrometrace write request: %s", string(buf))
-				var r bytes.Reader
-				for c := range chrometraceConsumers.consumers {
-					r.Reset(buf)
-					n, err := io.Copy(c.w, &r)
-					debug("chrometrace wrote n=%v bytes to consumer %#v", n, c)
-					if err != nil {
-						kickConsumer(c, err)
-					}
-				}
-			}
-		}
-	}()
-}
-
-func chrometraceWrite(i interface{}) {
-	var buf bytes.Buffer
-	err := json.NewEncoder(&buf).Encode(i)
-	if err != nil {
-		panic(err)
-	}
-	buf.WriteString(",")
-	chrometraceConsumers.write <- buf.Bytes()
-}
-
-func ChrometraceClientWebsocketHandler(conn *websocket.Conn) {
-	defer conn.Close()
-
-	var wg sync.WaitGroup
-	defer wg.Wait()
-	wg.Add(1)
-	go func() {
-		defer wg.Done()
-		r := bufio.NewReader(conn)
-		_, _, _ = r.ReadLine() // ignore errors
-		conn.Close()
-	}()
-
-	errored := make(chan error, 1)
-	chrometraceConsumers.register <- chrometraceConsumerRegistration{
-		w:       conn,
-		errored: errored,
-	}
-
-	wg.Add(1)
-	go func() {
-		defer wg.Done()
-		<-errored
-		conn.Close()
-	}()
-}
-
-var chrometraceFileConsumerPath = envconst.String("ZREPL_ACTIVITY_TRACE", "")
-
-func init() {
-	if chrometraceFileConsumerPath != "" {
-		var err error
-		f, err := os.Create(chrometraceFileConsumerPath)
-		if err != nil {
-			panic(err)
-		}
-		errored := make(chan error, 1)
-		chrometraceConsumers.register <- chrometraceConsumerRegistration{
-			w:       f,
-			errored: errored,
-		}
-		go func() {
-			<-errored
-			f.Close()
-		}()
-	}
-}

daemon/logging/trace/trace_context.go

@@ -1,27 +0,0 @@
-package trace
-
-import "context"
-
-type contextKey int
-
-const (
-	contextKeyTraceNode contextKey = 1 + iota
-)
-
-var contextKeys = []contextKey{
-	contextKeyTraceNode,
-}
-
-// WithInherit inherits the task hierarchy from inheritFrom into ctx.
-// The returned context is a child of ctx, but its task and span are those of inheritFrom.
-//
-// Note that in most use cases, callers most likely want to call WithTask since it will most likely
-// be in some sort of connection handler context.
-func WithInherit(ctx, inheritFrom context.Context) context.Context {
-	for _, k := range contextKeys {
-		if v := inheritFrom.Value(k); v != nil {
-			ctx = context.WithValue(ctx, k, v) // no shadow
-		}
-	}
-	return ctx
-}

daemon/logging/trace/trace_convenience.go

@@ -1,74 +0,0 @@
-package trace
-
-import (
-	"context"
-	"fmt"
-	"runtime"
-	"strings"
-	"sync"
-)
-
-// use like this:
-//
-//	defer WithSpanFromStackUpdateCtx(&existingCtx)()
-//
-//
-func WithSpanFromStackUpdateCtx(ctx *context.Context) DoneFunc {
-	childSpanCtx, end := WithSpan(*ctx, getMyCallerOrPanic())
-	*ctx = childSpanCtx
-	return end
-}
-
-// derive task name from call stack (caller's name)
-func WithTaskFromStack(ctx context.Context) (context.Context, DoneFunc) {
-	return WithTask(ctx, getMyCallerOrPanic())
-}
-
-// derive task name from call stack (caller's name) and update *ctx
-// to point to be the child task ctx
-func WithTaskFromStackUpdateCtx(ctx *context.Context) DoneFunc {
-	child, end := WithTask(*ctx, getMyCallerOrPanic())
-	*ctx = child
-	return end
-}
-
-// create a task and a span within it in one call
-func WithTaskAndSpan(ctx context.Context, task string, span string) (context.Context, DoneFunc) {
-	ctx, endTask := WithTask(ctx, task)
-	ctx, endSpan := WithSpan(ctx, fmt.Sprintf("%s %s", task, span))
-	return ctx, func() {
-		endSpan()
-		endTask()
-	}
-}
-
-// create a span during which several child tasks are spawned using the `add` function
-func WithTaskGroup(ctx context.Context, taskGroup string) (_ context.Context, add func(f func(context.Context)), waitEnd DoneFunc) {
-	var wg sync.WaitGroup
-	ctx, endSpan := WithSpan(ctx, taskGroup)
-	add = func(f func(context.Context)) {
-		wg.Add(1)
-		defer wg.Done()
-		ctx, endTask := WithTask(ctx, taskGroup)
-		defer endTask()
-		f(ctx)
-	}
-	waitEnd = func() {
-		wg.Wait()
-		endSpan()
-	}
-	return ctx, add, waitEnd
-}
-
-func getMyCallerOrPanic() string {
-	pc, _, _, ok := runtime.Caller(2)
-	if !ok {
-		panic("cannot get caller")
-	}
-	details := runtime.FuncForPC(pc)
-	if ok && details != nil {
-		const prefix = "github.com/zrepl/zrepl"
-		return strings.TrimPrefix(strings.TrimPrefix(details.Name(), prefix), "/")
-	}
-	return ""
-}

daemon/logging/trace/trace_convenience_test.go

@@ -1,16 +0,0 @@
-package trace
-
-import (
-	"testing"
-
-	"github.com/stretchr/testify/assert"
-)
-
-func TestGetCallerOrPanic(t *testing.T) {
-	withStackFromCtxMock := func() string {
-		return getMyCallerOrPanic()
-	}
-	ret := withStackFromCtxMock()
-	// zrepl prefix is stripped
-	assert.Equal(t, "daemon/logging/trace.TestGetCallerOrPanic", ret)
-}

daemon/logging/trace/trace_debug.go

@@ -1,15 +0,0 @@
-package trace
-
-import (
-	"fmt"
-	"os"
-)
-
-const debugEnabled = false
-
-func debug(format string, args ...interface{}) {
-	if !debugEnabled {
-		return
-	}
-	fmt.Fprintf(os.Stderr, format+"\n", args...)
-}

daemon/logging/trace/trace_genID.go

@@ -1,47 +0,0 @@
-package trace
-
-import (
-	"encoding/base64"
-	"math/rand"
-	"os"
-	"strings"
-	"time"
-
-	"github.com/zrepl/zrepl/util/envconst"
-)
-
-var genIdPRNG = rand.New(rand.NewSource(1))
-
-func init() {
-	genIdPRNG.Seed(time.Now().UnixNano())
-	genIdPRNG.Seed(int64(os.Getpid()))
-}
-
-var genIdNumBytes = envconst.Int("ZREPL_TRACE_ID_NUM_BYTES", 3)
-
-func init() {
-	if genIdNumBytes < 1 {
-		panic("trace node id byte length must be at least 1")
-	}
-}
-
-func genID() string {
-	var out strings.Builder
-	enc := base64.NewEncoder(base64.RawStdEncoding, &out)
-	buf := make([]byte, genIdNumBytes)
-	for i := 0; i < len(buf); {
-		n, err := genIdPRNG.Read(buf[i:])
-		if err != nil {
-			panic(err)
-		}
-		i += n
-	}
-	n, err := enc.Write(buf[:])
-	if err != nil || n != len(buf) {
-		panic(err)
-	}
-	if err := enc.Close(); err != nil {
-		panic(err)
-	}
-	return out.String()
-}

daemon/logging/trace/trace_test.go

@@ -1,205 +0,0 @@
-package trace
-
-import (
-	"context"
-	"fmt"
-	"testing"
-
-	"github.com/gitchander/permutation"
-	"github.com/pkg/errors"
-	"github.com/stretchr/testify/assert"
-	"github.com/stretchr/testify/require"
-)
-
-func TestRegularSpanUsage(t *testing.T) {
-	root, endRoot := WithTask(context.Background(), "root")
-	defer endRoot()
-
-	s1, endS1 := WithSpan(root, "parent")
-	s2, endS2 := WithSpan(s1, "child")
-	_, endS3 := WithSpan(s2, "grand-child")
-	require.NotPanics(t, func() { endS3() })
-	require.NotPanics(t, func() { endS2() })
-
-	// reuse
-	_, endS4 := WithSpan(s1, "child-2")
-	require.NotPanics(t, func() { endS4() })
-
-	// close parent
-	require.NotPanics(t, func() { endS1() })
-}
-
-func TestMultipleActiveChildSpansNotAllowed(t *testing.T) {
-	root, endRoot := WithTask(context.Background(), "root")
-	defer endRoot()
-
-	s1, _ := WithSpan(root, "s1")
-	_, endS2 := WithSpan(s1, "s1-child1")
-
-	require.PanicsWithValue(t, ErrAlreadyActiveChildSpan, func() {
-		_, _ = WithSpan(s1, "s1-child2")
-	})
-
-	endS2()
-
-	require.NotPanics(t, func() {
-		_, _ = WithSpan(s1, "s1-child2")
-	})
-}
-
-func TestForkingChildSpansNotAllowed(t *testing.T) {
-	root, endRoot := WithTask(context.Background(), "root")
-	defer endRoot()
-
-	s1, _ := WithSpan(root, "s1")
-	sc, endSC := WithSpan(s1, "s1-child")
-	_, _ = WithSpan(sc, "s1-child-child")
-
-	require.PanicsWithValue(t, ErrSpanStillHasActiveChildSpan, func() {
-		endSC()
-	})
-}
-
-func TestRegularTaskUsage(t *testing.T) {
-	// assert concurrent activities on different tasks can end in any order
-	closeOrder := []int{0, 1, 2}
-	closeOrders := permutation.New(permutation.IntSlice(closeOrder))
-	for closeOrders.Next() {
-		t.Run(fmt.Sprintf("%v", closeOrder), func(t *testing.T) {
-			root, endRoot := WithTask(context.Background(), "root")
-			defer endRoot()
-
-			c1, endC1 := WithTask(root, "c1")
-			defer endC1()
-			c2, endC2 := WithTask(root, "c2")
-			defer endC2()
-
-			// begin 3 concurrent activities
-			_, endAR := WithSpan(root, "aR")
-			_, endAC1 := WithSpan(c1, "aC1")
-			_, endAC2 := WithSpan(c2, "aC2")
-
-			endFuncs := []DoneFunc{endAR, endAC1, endAC2}
-			for _, i := range closeOrder {
-				require.NotPanics(t, func() {
-					endFuncs[i]()
-				}, "%v", i)
-			}
-		})
-	}
-}
-
-func TestTaskEndWithActiveChildTaskNotAllowed(t *testing.T) {
-	root, _ := WithTask(context.Background(), "root")
-	c, endC := WithTask(root, "child")
-	_, _ = WithTask(c, "grand-child")
-	func() {
-		defer func() {
-			r := recover()
-			require.NotNil(t, r)
-			err, ok := r.(error)
-			require.True(t, ok)
-			require.Equal(t, ErrTaskStillHasActiveChildTasks, errors.Cause(err))
-		}()
-		endC()
-	}()
-
-}
-
-func TestIdempotentEndTask(t *testing.T) {
-	_, end := WithTask(context.Background(), "root")
-	end()
-	require.NotPanics(t, func() { end() })
-}
-
-func TestSpansPanicIfNoParentTask(t *testing.T) {
-	require.Panics(t, func() { WithSpan(context.Background(), "taskless-span") })
-}
-
-func TestIdempotentEndSpan(t *testing.T) {
-	root, _ := WithTask(context.Background(), "root")
-	_, end := WithSpan(root, "span")
-	end()
-	require.NotPanics(t, func() { end() })
-}
-
-func logAndGetTraceNode(t *testing.T, descr string, ctx context.Context) *traceNode {
-	n, ok := ctx.Value(contextKeyTraceNode).(*traceNode)
-	require.True(t, ok)
-	t.Logf("% 20s %p %#v", descr, n, n)
-	return n
-}
-
-func TestWhiteboxHierachy(t *testing.T) {
-	root, e1 := WithTask(context.Background(), "root")
-	rootN := logAndGetTraceNode(t, "root", root)
-	assert.Nil(t, rootN.parentTask)
-	assert.Nil(t, rootN.parentSpan)
-
-	child, e2 := WithSpan(root, "child")
-	childN := logAndGetTraceNode(t, "child", child)
-	assert.Equal(t, rootN, childN.parentTask)
-	assert.Equal(t, rootN, childN.parentSpan)
-
-	grandchild, e3 := WithSpan(child, "grandchild")
-	grandchildN := logAndGetTraceNode(t, "grandchild", grandchild)
-	assert.Equal(t, rootN, grandchildN.parentTask)
-	assert.Equal(t, childN, grandchildN.parentSpan)
-
-	gcTask, e4 := WithTask(grandchild, "grandchild-task")
-	gcTaskN := logAndGetTraceNode(t, "grandchild-task", gcTask)
-	assert.Equal(t, rootN, gcTaskN.parentTask)
-	assert.Nil(t, gcTaskN.parentSpan)
-
-	// it is allowed that a child task outlives the _span_ in which it was created
-	// (albeit not its parent task)
-	e3()
-	e2()
-	gcTaskSpan, e5 := WithSpan(gcTask, "granschild-task-span")
-	gcTaskSpanN := logAndGetTraceNode(t, "granschild-task-span", gcTaskSpan)
-	assert.Equal(t, gcTaskN, gcTaskSpanN.parentTask)
-	assert.Equal(t, gcTaskN, gcTaskSpanN.parentSpan)
-	e5()
-
-	e4()
-	e1()
-}
-
-func TestOrphanTasksBecomeNewRootWhenAllAncestorsAreDead(t *testing.T) {
-	parent, e1 := WithTask(context.Background(), "parent")
-	_ = logAndGetTraceNode(t, "parent-task", parent)
-	e1()
-
-	var child context.Context
-	var e2 DoneFunc
-	require.NotPanics(t, func() {
-		child, e2 = WithTask(parent, "child")
-	})
-	childN := logAndGetTraceNode(t, "child-task", child)
-	assert.Nil(t, childN.parentTask)
-
-	grandchild, _ := WithTask(child, "grandchild")
-	grandchildN := logAndGetTraceNode(t, "grandchild-task", grandchild)
-	assert.Equal(t, childN, grandchildN.parentTask)
-
-	require.Panics(t, func() { e2() }, "if the parent was alive at creation of child task, wrong termination order remains a panicable offense though")
-}
-
-func TestOrphanTaskBecomesChildToNearestLiveAncestor(t *testing.T) {
-	parent, e1 := WithTask(context.Background(), "parent")
-	parentN := logAndGetTraceNode(t, "parent-task", parent)
-
-	child, e2 := WithTask(parent, "child")
-	childN := logAndGetTraceNode(t, "child-task", child)
-	assert.Equal(t, parentN, childN.parentTask)
-	e2()
-
-	grandchild, e3 := WithTask(child, "grandchild")
-	grandchildN := logAndGetTraceNode(t, "grandchild-task", grandchild)
-	assert.Equal(t, parentN, grandchildN.parentTask) // child is already dead
-
-	require.NotPanics(t, func() {
-		e3()
-		e1()
-	})
-}

daemon/logging/trace/trace_unique_concurrent_task_namer.go

@@ -1,53 +0,0 @@
-package trace
-
-import (
-	"fmt"
-	"strings"
-	"sync"
-
-	"github.com/willf/bitset"
-)
-
-type uniqueConcurrentTaskNamer struct {
-	mtx    sync.Mutex
-	active map[string]*bitset.BitSet
-}
-
-// bitvecLengthGauge may be nil
-func newUniqueTaskNamer() *uniqueConcurrentTaskNamer {
-	return &uniqueConcurrentTaskNamer{
-		active: make(map[string]*bitset.BitSet),
-	}
-}
-
-// appends `#%d` to `name` such that until `done` is called,
-// it is guaranteed that `#%d` is not returned a second time for the same `name`
-func (namer *uniqueConcurrentTaskNamer) UniqueConcurrentTaskName(name string) (uniqueName string, done func()) {
-	if strings.Contains(name, "#") {
-		panic(name)
-	}
-	namer.mtx.Lock()
-	act, ok := namer.active[name]
-	if !ok {
-		act = bitset.New(64) // FIXME magic const
-		namer.active[name] = act
-	}
-	id, ok := act.NextClear(0)
-	if !ok {
-		// if !ok, all bits are 1 and act.Len() returns the next bit
-		id = act.Len()
-		// FIXME unbounded growth without reclamation
-	}
-	act.Set(id)
-	namer.mtx.Unlock()
-
-	return fmt.Sprintf("%s#%d", name, id), func() {
-		namer.mtx.Lock()
-		defer namer.mtx.Unlock()
-		act, ok := namer.active[name]
-		if !ok {
-			panic("must be initialized upon entry")
-		}
-		act.Clear(id)
-	}
-}

daemon/logging/trace/trace_unique_concurrent_task_namer_test.go

@@ -1,58 +0,0 @@
-package trace
-
-import (
-	"fmt"
-	"sync"
-	"sync/atomic"
-	"testing"
-
-	"github.com/stretchr/testify/require"
-	"github.com/willf/bitset"
-)
-
-func TestBitsetFeaturesForUniqueConcurrentTaskNamer(t *testing.T) {
-	var b bitset.BitSet
-	require.Equal(t, uint(0), b.Len())
-	require.Equal(t, uint(0), b.Count())
-
-	b.Set(0)
-	require.Equal(t, uint(1), b.Len())
-	require.Equal(t, uint(1), b.Count())
-
-	b.Set(8)
-	require.Equal(t, uint(9), b.Len())
-	require.Equal(t, uint(2), b.Count())
-
-	b.Set(1)
-	require.Equal(t, uint(9), b.Len())
-	require.Equal(t, uint(3), b.Count())
-}
-
-func TestUniqueConcurrentTaskNamer(t *testing.T) {
-	namer := newUniqueTaskNamer()
-
-	var wg sync.WaitGroup
-
-	const N = 8128
-	const Q = 23
-
-	var fails uint32
-	var m sync.Map
-	wg.Add(N)
-	for i := 0; i < N; i++ {
-		go func(i int) {
-			defer wg.Done()
-			name := fmt.Sprintf("%d", i/Q)
-			uniqueName, done := namer.UniqueConcurrentTaskName(name)
-			act, _ := m.LoadOrStore(uniqueName, i)
-			if act.(int) != i {
-				atomic.AddUint32(&fails, 1)
-			}
-			m.Delete(uniqueName)
-			done()
-		}(i)
-	}
-	wg.Wait()
-
-	require.Equal(t, uint32(0), fails)
-}

daemon/main.go

@@ -1,8 +1,6 @@
 package daemon
 
 import (
-	"context"
-
 	"github.com/zrepl/zrepl/cli"
 	"github.com/zrepl/zrepl/logger"
 )
@@ -12,7 +10,7 @@ type Logger = logger.Logger
 var DaemonCmd = &cli.Subcommand{
 	Use:   "daemon",
 	Short: "run the zrepl daemon",
-	Run: func(ctx context.Context, subcommand *cli.Subcommand, args []string) error {
-		return Run(ctx, subcommand.Config())
+	Run: func(subcommand *cli.Subcommand, args []string) error {
+		return Run(subcommand.Config())
 	},
 }

daemon/pprof.go

@@ -8,11 +8,7 @@ import (
 	"net"
 	"net/http/pprof"
 
-	"github.com/prometheus/client_golang/prometheus/promhttp"
-	"golang.org/x/net/websocket"
-
 	"github.com/zrepl/zrepl/daemon/job"
-	"github.com/zrepl/zrepl/daemon/logging/trace"
 )
 
 type pprofServer struct {
@@ -68,8 +64,6 @@ outer:
 			mux.Handle("/debug/pprof/profile", http.HandlerFunc(pprof.Profile))
 			mux.Handle("/debug/pprof/symbol", http.HandlerFunc(pprof.Symbol))
 			mux.Handle("/debug/pprof/trace", http.HandlerFunc(pprof.Trace))
-			mux.Handle("/metrics", promhttp.Handler())
-			mux.Handle("/debug/zrepl/activity-trace", websocket.Handler(trace.ChrometraceClientWebsocketHandler))
 			go func() {
 				err := http.Serve(s.listener, mux)
 				if ctx.Err() != nil {

daemon/prometheus.go

@@ -10,7 +10,6 @@ import (
 
 	"github.com/zrepl/zrepl/config"
 	"github.com/zrepl/zrepl/daemon/job"
-	"github.com/zrepl/zrepl/daemon/logging"
 	"github.com/zrepl/zrepl/endpoint"
 	"github.com/zrepl/zrepl/logger"
 	"github.com/zrepl/zrepl/rpc/dataconn/frameconn"
@@ -87,19 +86,16 @@ func (j *prometheusJob) Run(ctx context.Context) {
 }
 
 type prometheusJobOutlet struct {
+	jobName string
 }
 
 var _ logger.Outlet = prometheusJobOutlet{}
 
-func newPrometheusLogOutlet() prometheusJobOutlet {
-	return prometheusJobOutlet{}
+func newPrometheusLogOutlet(jobName string) prometheusJobOutlet {
+	return prometheusJobOutlet{jobName}
 }
 
 func (o prometheusJobOutlet) WriteEntry(entry logger.Entry) error {
-	jobFieldVal, ok := entry.Fields[logging.JobField].(string)
-	if !ok {
-		jobFieldVal = "_nojobid"
-	}
-	prom.taskLogEntries.WithLabelValues(jobFieldVal, entry.Level.String()).Inc()
+	prom.taskLogEntries.WithLabelValues(o.jobName, entry.Level.String()).Inc()
 	return nil
 }

daemon/pruner.v2/fsstate_enumer.go

@@ -1,72 +0,0 @@
-// Code generated by "enumer -type=FSState -json"; DO NOT EDIT.
-
-//
-package pruner
-
-import (
-	"encoding/json"
-	"fmt"
-)
-
-const _FSStateName = "FSStateInitializedFSStatePlanningFSStatePlanErrFSStateExecutingFSStateExecuteErrFSStateExecuteSuccess"
-
-var _FSStateIndex = [...]uint8{0, 18, 33, 47, 63, 80, 101}
-
-func (i FSState) String() string {
-	if i < 0 || i >= FSState(len(_FSStateIndex)-1) {
-		return fmt.Sprintf("FSState(%d)", i)
-	}
-	return _FSStateName[_FSStateIndex[i]:_FSStateIndex[i+1]]
-}
-
-var _FSStateValues = []FSState{0, 1, 2, 3, 4, 5}
-
-var _FSStateNameToValueMap = map[string]FSState{
-	_FSStateName[0:18]:   0,
-	_FSStateName[18:33]:  1,
-	_FSStateName[33:47]:  2,
-	_FSStateName[47:63]:  3,
-	_FSStateName[63:80]:  4,
-	_FSStateName[80:101]: 5,
-}
-
-// FSStateString retrieves an enum value from the enum constants string name.
-// Throws an error if the param is not part of the enum.
-func FSStateString(s string) (FSState, error) {
-	if val, ok := _FSStateNameToValueMap[s]; ok {
-		return val, nil
-	}
-	return 0, fmt.Errorf("%s does not belong to FSState values", s)
-}
-
-// FSStateValues returns all values of the enum
-func FSStateValues() []FSState {
-	return _FSStateValues
-}
-
-// IsAFSState returns "true" if the value is listed in the enum definition. "false" otherwise
-func (i FSState) IsAFSState() bool {
-	for _, v := range _FSStateValues {
-		if i == v {
-			return true
-		}
-	}
-	return false
-}
-
-// MarshalJSON implements the json.Marshaler interface for FSState
-func (i FSState) MarshalJSON() ([]byte, error) {
-	return json.Marshal(i.String())
-}
-
-// UnmarshalJSON implements the json.Unmarshaler interface for FSState
-func (i *FSState) UnmarshalJSON(data []byte) error {
-	var s string
-	if err := json.Unmarshal(data, &s); err != nil {
-		return fmt.Errorf("FSState should be a string, got %s", data)
-	}
-
-	var err error
-	*i, err = FSStateString(s)
-	return err
-}

daemon/pruner.v2/pruner.go

@@ -1,454 +0,0 @@
-package pruner
-
-import (
-	"context"
-	"encoding/json"
-	"fmt"
-	"os"
-
-	"github.com/pkg/errors"
-	"github.com/zrepl/zrepl/daemon/logging"
-	"github.com/zrepl/zrepl/daemon/logging/trace"
-	"github.com/zrepl/zrepl/endpoint"
-	"github.com/zrepl/zrepl/pruning"
-	"github.com/zrepl/zrepl/zfs"
-)
-
-type Pruner struct {
-	fsfilter  endpoint.FSFilter
-	jid       endpoint.JobID
-	side      Side
-	keepRules []pruning.KeepRule
-
-	// all channels consumed by the run loop
-	reportReqs chan reportRequest
-	stopReqs   chan stopRequest
-	done       chan struct{}
-	fsListRes  chan fsListRes
-
-	state State
-
-	listFilesystemsError error       // only in state StateListFilesystemsError
-	fsPruners            []*FSPruner // only in state StateFanOutFilesystems
-}
-
-//go:generate enumer -type=State -json
-type State int
-
-const (
-	StateInitialized State = iota
-	StateListFilesystems
-	StateListFilesystemsError
-	StateFanOutFilesystems
-	StateDone
-)
-
-type Report struct {
-	State                State
-	ListFilesystemsError error       // only valid in StateListFilesystemsError
-	Filesystems          []*FSReport // valid from StateFanOutFilesystems
-}
-
-type reportRequest struct {
-	ctx   context.Context
-	reply chan *Report
-}
-
-type runRequest struct {
-	complete chan struct{}
-}
-
-type stopRequest struct {
-	complete chan struct{}
-}
-
-type fsListRes struct {
-	filesystems []*zfs.DatasetPath
-	err         error
-}
-
-type Side interface {
-	// may return both nil, indicating there is no replication position
-	GetReplicationPosition(ctx context.Context, fs string) (*zfs.FilesystemVersion, error)
-	isSide() Side
-}
-
-func NewPruner(fsfilter endpoint.FSFilter, jid endpoint.JobID, side Side, keepRules []pruning.KeepRule) *Pruner {
-	return &Pruner{
-		fsfilter,
-		jid,
-		side,
-		keepRules,
-		make(chan reportRequest),
-		make(chan stopRequest),
-		make(chan struct{}),
-		make(chan fsListRes),
-		StateInitialized,
-		nil,
-		nil,
-	}
-}
-
-func (p *Pruner) Run(ctx context.Context) *Report {
-
-	ctx, cancel := context.WithCancel(ctx)
-	defer cancel()
-
-	if p.state != StateInitialized {
-		panic("Run can onl[y be called once")
-	}
-
-	go func() {
-		fss, err := zfs.ZFSListMapping(ctx, p.fsfilter)
-		p.fsListRes <- fsListRes{fss, err}
-	}()
-
-	for {
-		select {
-		case res := <-p.fsListRes:
-			if res.err != nil {
-				p.state = StateListFilesystemsError
-				p.listFilesystemsError = res.err
-				close(p.done)
-				continue
-			}
-
-			p.state = StateFanOutFilesystems
-
-			p.fsPruners = make([]*FSPruner, len(res.filesystems))
-			_, add, end := trace.WithTaskGroup(ctx, "pruner-fan-out-fs")
-			for i, fs := range res.filesystems {
-				p.fsPruners[i] = NewFSPruner(p.jid, p.side, p.keepRules, fs)
-				add(func(ctx context.Context) {
-					p.fsPruners[i].Run(ctx)
-				})
-			}
-			go func() {
-				end()
-				close(p.done)
-			}()
-
-		case req := <-p.stopReqs:
-			cancel()
-			go func() {
-				<-p.done
-				close(req.complete)
-			}()
-		case req := <-p.reportReqs:
-			req.reply <- p.report(req.ctx)
-		case <-p.done:
-			p.state = StateDone
-			return p.report(ctx)
-		}
-	}
-}
-
-func (p *Pruner) Report(ctx context.Context) *Report {
-	req := reportRequest{
-		ctx:   ctx,
-		reply: make(chan *Report, 1),
-	}
-	select {
-	case p.reportReqs <- req:
-		return <-req.reply
-	case <-ctx.Done():
-		return nil
-	case <-p.done:
-		return nil
-	}
-}
-
-func (p *Pruner) report(ctx context.Context) *Report {
-	fsreports := make([]*FSReport, len(p.fsPruners))
-	for i := range fsreports {
-		fsreports[i] = p.fsPruners[i].report()
-	}
-	return &Report{
-		State:                p.state,
-		ListFilesystemsError: p.listFilesystemsError,
-		Filesystems:          fsreports,
-	}
-}
-
-// implements pruning.Snapshot
-type snapshot struct {
-	replicated bool
-	stepHolds  []pruning.StepHold
-	zfs.FilesystemVersion
-
-	state     SnapState
-	destroyOp *zfs.DestroySnapOp
-}
-
-//go:generate enumer -type=SnapState -json
-type SnapState int
-
-const (
-	SnapStateInitialized SnapState = iota
-	SnapStateKeeping
-	SnapStateDeletePending
-	SnapStateDeleteAttempted
-)
-
-// implements pruning.StepHold
-type stepHold struct {
-	endpoint.Abstraction
-}
-
-func (s snapshot) Replicated() bool              { return s.replicated }
-func (s snapshot) StepHolds() []pruning.StepHold { return s.stepHolds }
-
-func (s stepHold) GetJobID() endpoint.JobID { return *s.Abstraction.GetJobID() }
-
-type FSPruner struct {
-	jid       endpoint.JobID
-	side      Side
-	keepRules []pruning.KeepRule
-	fsp       *zfs.DatasetPath
-
-	state FSState
-
-	// all channels consumed by the run loop
-	planned    chan fsPlanRes
-	executed   chan fsExecuteRes
-	done       chan struct{}
-	reportReqs chan fsReportReq
-
-	keepList    []*snapshot // valid in FSStateExecuting and forward
-	destroyList []*snapshot // valid in FSStateExecuting and forward, field .destroyOp is invalid until FSStateExecuting is left
-
-}
-
-type fsPlanRes struct {
-	keepList    []*snapshot
-	destroyList []*snapshot
-	err         error
-}
-
-type fsExecuteRes struct {
-	completedDestroyOps []*zfs.DestroySnapOp // same len() as FSPruner.destroyList
-}
-
-type fsReportReq struct {
-	res chan *FSReport
-}
-
-type FSReport struct {
-	State    FSState
-	KeepList []*SnapReport
-	Destroy  []*SnapReport
-}
-
-type SnapReport struct {
-	State         SnapState
-	Name          string
-	Replicated    bool
-	StepHoldCount int
-	DestroyError  error
-}
-
-//go:generate enumer -type=FSState -json
-type FSState int
-
-const (
-	FSStateInitialized FSState = iota
-	FSStatePlanning
-	FSStatePlanErr
-	FSStateExecuting
-	FSStateExecuteErr
-	FSStateExecuteSuccess
-)
-
-func (s FSState) IsTerminal() bool {
-	return s == FSStatePlanErr || s == FSStateExecuteErr || s == FSStateExecuteSuccess
-}
-
-func NewFSPruner(jid endpoint.JobID, side Side, keepRules []pruning.KeepRule, fsp *zfs.DatasetPath) *FSPruner {
-	return &FSPruner{
-		jid, side, keepRules, fsp,
-		FSStateInitialized,
-		make(chan fsPlanRes),
-		make(chan fsExecuteRes),
-		make(chan struct{}),
-		make(chan fsReportReq),
-		nil, nil,
-	}
-}
-
-func (p *FSPruner) Run(ctx context.Context) *FSReport {
-
-	defer func() {
-	}()
-
-	p.state = FSStatePlanning
-
-	go func() { p.planned <- p.plan(ctx) }()
-
-out:
-	for !p.state.IsTerminal() {
-		select {
-		case res := <-p.planned:
-
-			if res.err != nil {
-				p.state = FSStatePlanErr
-				continue
-			}
-			p.state = FSStateExecuting
-			p.keepList = res.keepList
-			p.destroyList = res.destroyList
-
-			go func() { p.executed <- p.execute(ctx, p.destroyList) }()
-
-		case res := <-p.executed:
-
-			if len(res.completedDestroyOps) != len(p.destroyList) {
-				panic("impl error: completedDestroyOps is a vector corresponding to entries in p.destroyList")
-			}
-
-			var erronous []*zfs.DestroySnapOp
-			for i, op := range res.completedDestroyOps {
-				if *op.ErrOut != nil {
-					erronous = append(erronous, op)
-				}
-				p.destroyList[i].destroyOp = op
-				p.destroyList[i].state = SnapStateDeleteAttempted
-			}
-			if len(erronous) > 0 {
-				p.state = FSStateExecuteErr
-			} else {
-				p.state = FSStateExecuteSuccess
-			}
-
-			close(p.done)
-
-		case <-p.reportReqs:
-			panic("unimp")
-		case <-p.done:
-			break out
-		}
-	}
-
-	// TODO render last FS report
-	return nil
-}
-
-func (p *FSPruner) plan(ctx context.Context) fsPlanRes {
-	fs := p.fsp.ToString()
-	vs, err := zfs.ZFSListFilesystemVersions(ctx, p.fsp, zfs.ListFilesystemVersionsOptions{})
-	if err != nil {
-		return fsPlanRes{err: errors.Wrap(err, "list filesystem versions")}
-	}
-
-	allJobsStepHolds, absErrs, err := endpoint.ListAbstractions(ctx, endpoint.ListZFSHoldsAndBookmarksQuery{
-		FS: endpoint.ListZFSHoldsAndBookmarksQueryFilesystemFilter{
-			FS: &fs,
-		},
-		What: endpoint.AbstractionTypeSet{
-			endpoint.AbstractionStepHold: true,
-		},
-		Concurrency: 1,
-	})
-	if err != nil {
-		return fsPlanRes{err: errors.Wrap(err, "list abstractions")}
-	}
-	if len(absErrs) > 0 {
-		logging.GetLogger(ctx, logging.SubsysPruning).WithError(endpoint.ListAbstractionsErrors(absErrs)).
-			Error("error listing some step holds, prune attempt might fail with 'dataset is busy' errors")
-	}
-
-	repPos, err := p.side.GetReplicationPosition(ctx, p.fsp.ToString())
-	if err != nil {
-		return fsPlanRes{err: errors.Wrap(err, "get replication position")}
-	}
-
-	vsAsSnaps := make([]pruning.Snapshot, len(vs))
-	for i := range vs {
-		var repPosCreateTxgOrZero uint64
-		if repPos != nil {
-			repPosCreateTxgOrZero = repPos.GetCreateTXG()
-		}
-		s := &snapshot{
-			state:             SnapStateInitialized,
-			FilesystemVersion: vs[i],
-			replicated:        vs[i].GetCreateTXG() <= repPosCreateTxgOrZero,
-		}
-		for _, h := range allJobsStepHolds {
-			if zfs.FilesystemVersionEqualIdentity(vs[i], h.GetFilesystemVersion()) {
-				s.stepHolds = append(s.stepHolds, stepHold{h})
-			}
-		}
-		vsAsSnaps[i] = s
-	}
-
-	downcastToSnapshots := func(l []pruning.Snapshot) (r []*snapshot) {
-		r = make([]*snapshot, len(l))
-		for i, e := range l {
-			r[i] = e.(*snapshot)
-		}
-		return r
-	}
-	pruningResult := pruning.PruneSnapshots(vsAsSnaps, p.keepRules)
-	remove, keep := downcastToSnapshots(pruningResult.Remove), downcastToSnapshots(pruningResult.Keep)
-	if len(remove)+len(keep) != len(vsAsSnaps) {
-		for _, s := range vsAsSnaps {
-			r, _ := json.MarshalIndent(s.(*snapshot).report(), "", "  ")
-			fmt.Fprintf(os.Stderr, "%s\n", string(r))
-		}
-		panic("indecisive")
-	}
-
-	for _, s := range remove {
-		s.state = SnapStateDeletePending
-	}
-	for _, s := range keep {
-		s.state = SnapStateKeeping
-	}
-
-	return fsPlanRes{keepList: keep, destroyList: remove, err: nil}
-}
-
-func (p *FSPruner) execute(ctx context.Context, destroyList []*snapshot) fsExecuteRes {
-	ops := make([]*zfs.DestroySnapOp, len(destroyList))
-	for i, fsv := range p.destroyList {
-		ops[i] = &zfs.DestroySnapOp{
-			Filesystem: p.fsp.ToString(),
-			Name:       fsv.GetName(),
-			ErrOut:     new(error),
-		}
-	}
-	zfs.ZFSDestroyFilesystemVersions(ctx, ops)
-
-	return fsExecuteRes{completedDestroyOps: ops}
-}
-
-func (p *FSPruner) report() *FSReport {
-	return &FSReport{
-		State:    p.state,
-		KeepList: p.reportRenderSnapReports(p.keepList),
-		Destroy:  p.reportRenderSnapReports(p.destroyList),
-	}
-}
-
-func (p *FSPruner) reportRenderSnapReports(l []*snapshot) (r []*SnapReport) {
-	r = make([]*SnapReport, len(l))
-	for i := range l {
-		r[i] = l[i].report()
-	}
-	return r
-}
-
-func (s *snapshot) report() *SnapReport {
-	var snapErr error
-	if s.state == SnapStateDeleteAttempted {
-		if *s.destroyOp.ErrOut != nil {
-			snapErr = (*s.destroyOp.ErrOut)
-		}
-	}
-	return &SnapReport{
-		State:         s.state,
-		Name:          s.Name,
-		Replicated:    s.Replicated(),
-		StepHoldCount: len(s.stepHolds),
-		DestroyError:  snapErr,
-	}
-}

daemon/pruner.v2/pruner_side.go

@@ -1,27 +0,0 @@
-package pruner
-
-import (
-	"context"
-
-	"github.com/zrepl/zrepl/endpoint"
-	"github.com/zrepl/zrepl/zfs"
-)
-
-type SideSender struct {
-	jobID endpoint.JobID
-}
-
-func NewSideSender(jid endpoint.JobID) *SideSender {
-	return &SideSender{jid}
-}
-
-func (s *SideSender) isSide() Side { return nil }
-
-var _ Side = (*SideSender)(nil)
-
-func (s *SideSender) GetReplicationPosition(ctx context.Context, fs string) (*zfs.FilesystemVersion, error) {
-	if fs == "" {
-		panic("must not pass zero value for fs")
-	}
-	return endpoint.GetMostRecentReplicationCursorOfJob(ctx, fs, s.jobID)
-}

daemon/pruner.v2/snapstate_enumer.go

@@ -1,70 +0,0 @@
-// Code generated by "enumer -type=SnapState -json"; DO NOT EDIT.
-
-//
-package pruner
-
-import (
-	"encoding/json"
-	"fmt"
-)
-
-const _SnapStateName = "SnapStateInitializedSnapStateKeepingSnapStateDeletePendingSnapStateDeleteAttempted"
-
-var _SnapStateIndex = [...]uint8{0, 20, 36, 58, 82}
-
-func (i SnapState) String() string {
-	if i < 0 || i >= SnapState(len(_SnapStateIndex)-1) {
-		return fmt.Sprintf("SnapState(%d)", i)
-	}
-	return _SnapStateName[_SnapStateIndex[i]:_SnapStateIndex[i+1]]
-}
-
-var _SnapStateValues = []SnapState{0, 1, 2, 3}
-
-var _SnapStateNameToValueMap = map[string]SnapState{
-	_SnapStateName[0:20]:  0,
-	_SnapStateName[20:36]: 1,
-	_SnapStateName[36:58]: 2,
-	_SnapStateName[58:82]: 3,
-}
-
-// SnapStateString retrieves an enum value from the enum constants string name.
-// Throws an error if the param is not part of the enum.
-func SnapStateString(s string) (SnapState, error) {
-	if val, ok := _SnapStateNameToValueMap[s]; ok {
-		return val, nil
-	}
-	return 0, fmt.Errorf("%s does not belong to SnapState values", s)
-}
-
-// SnapStateValues returns all values of the enum
-func SnapStateValues() []SnapState {
-	return _SnapStateValues
-}
-
-// IsASnapState returns "true" if the value is listed in the enum definition. "false" otherwise
-func (i SnapState) IsASnapState() bool {
-	for _, v := range _SnapStateValues {
-		if i == v {
-			return true
-		}
-	}
-	return false
-}
-
-// MarshalJSON implements the json.Marshaler interface for SnapState
-func (i SnapState) MarshalJSON() ([]byte, error) {
-	return json.Marshal(i.String())
-}
-
-// UnmarshalJSON implements the json.Unmarshaler interface for SnapState
-func (i *SnapState) UnmarshalJSON(data []byte) error {
-	var s string
-	if err := json.Unmarshal(data, &s); err != nil {
-		return fmt.Errorf("SnapState should be a string, got %s", data)
-	}
-
-	var err error
-	*i, err = SnapStateString(s)
-	return err
-}

daemon/pruner.v2/state_enumer.go

@@ -1,71 +0,0 @@
-// Code generated by "enumer -type=State -json"; DO NOT EDIT.
-
-//
-package pruner
-
-import (
-	"encoding/json"
-	"fmt"
-)
-
-const _StateName = "StateInitializedStateListFilesystemsStateListFilesystemsErrorStateFanOutFilesystemsStateDone"
-
-var _StateIndex = [...]uint8{0, 16, 36, 61, 83, 92}
-
-func (i State) String() string {
-	if i < 0 || i >= State(len(_StateIndex)-1) {
-		return fmt.Sprintf("State(%d)", i)
-	}
-	return _StateName[_StateIndex[i]:_StateIndex[i+1]]
-}
-
-var _StateValues = []State{0, 1, 2, 3, 4}
-
-var _StateNameToValueMap = map[string]State{
-	_StateName[0:16]:  0,
-	_StateName[16:36]: 1,
-	_StateName[36:61]: 2,
-	_StateName[61:83]: 3,
-	_StateName[83:92]: 4,
-}
-
-// StateString retrieves an enum value from the enum constants string name.
-// Throws an error if the param is not part of the enum.
-func StateString(s string) (State, error) {
-	if val, ok := _StateNameToValueMap[s]; ok {
-		return val, nil
-	}
-	return 0, fmt.Errorf("%s does not belong to State values", s)
-}
-
-// StateValues returns all values of the enum
-func StateValues() []State {
-	return _StateValues
-}
-
-// IsAState returns "true" if the value is listed in the enum definition. "false" otherwise
-func (i State) IsAState() bool {
-	for _, v := range _StateValues {
-		if i == v {
-			return true
-		}
-	}
-	return false
-}
-
-// MarshalJSON implements the json.Marshaler interface for State
-func (i State) MarshalJSON() ([]byte, error) {
-	return json.Marshal(i.String())
-}
-
-// UnmarshalJSON implements the json.Unmarshaler interface for State
-func (i *State) UnmarshalJSON(data []byte) error {
-	var s string
-	if err := json.Unmarshal(data, &s); err != nil {
-		return fmt.Errorf("State should be a string, got %s", data)
-	}
-
-	var err error
-	*i, err = StateString(s)
-	return err
-}

daemon/pruner/pruner.go

@@ -12,7 +12,6 @@ import (
 	"github.com/prometheus/client_golang/prometheus"
 
 	"github.com/zrepl/zrepl/config"
-	"github.com/zrepl/zrepl/daemon/logging"
 	"github.com/zrepl/zrepl/logger"
 	"github.com/zrepl/zrepl/pruning"
 	"github.com/zrepl/zrepl/replication/logic/pdu"
@@ -20,14 +19,15 @@ import (
 )
 
 // Try to keep it compatible with github.com/zrepl/zrepl/endpoint.Endpoint
-type Endpoint interface {
+type History interface {
+	ReplicationCursor(ctx context.Context, req *pdu.ReplicationCursorReq) (*pdu.ReplicationCursorRes, error)
 	ListFilesystems(ctx context.Context, req *pdu.ListFilesystemReq) (*pdu.ListFilesystemRes, error)
-	ListFilesystemVersions(ctx context.Context, req *pdu.ListFilesystemVersionsReq) (*pdu.ListFilesystemVersionsRes, error)
 }
 
 // Try to keep it compatible with github.com/zrepl/zrepl/endpoint.Endpoint
 type Target interface {
-	Endpoint
+	ListFilesystems(ctx context.Context, req *pdu.ListFilesystemReq) (*pdu.ListFilesystemRes, error)
+	ListFilesystemVersions(ctx context.Context, req *pdu.ListFilesystemVersionsReq) (*pdu.ListFilesystemVersionsRes, error)
 	DestroySnapshots(ctx context.Context, req *pdu.DestroySnapshotsReq) (*pdu.DestroySnapshotsRes, error)
 }
 
@@ -35,24 +35,27 @@ type Logger = logger.Logger
 
 type contextKey int
 
-const (
-	contextKeyPruneSide contextKey = 1 + iota
-)
+const contextKeyLogger contextKey = 0
+
+func WithLogger(ctx context.Context, log Logger) context.Context {
+	return context.WithValue(ctx, contextKeyLogger, log)
+}
 
 func GetLogger(ctx context.Context) Logger {
-	pruneSide := ctx.Value(contextKeyPruneSide).(string)
-	return logging.GetLogger(ctx, logging.SubsysPruning).WithField("prune_side", pruneSide)
+	if l, ok := ctx.Value(contextKeyLogger).(Logger); ok {
+		return l
+	}
+	return logger.NewNullLogger()
 }
 
 type args struct {
-	ctx                              context.Context
-	target                           Target
-	sender, receiver                 Endpoint
-	rules                            []pruning.KeepRule
-	retryWait                        time.Duration
-	considerSnapAtCursorReplicated   bool
-	convertAnyStepHoldToStepBookmark bool
-	promPruneSecs                    prometheus.Observer
+	ctx                            context.Context
+	target                         Target
+	receiver                       History
+	rules                          []pruning.KeepRule
+	retryWait                      time.Duration
+	considerSnapAtCursorReplicated bool
+	promPruneSecs                  prometheus.Observer
 }
 
 type Pruner struct {
@@ -70,12 +73,11 @@ type Pruner struct {
 }
 
 type PrunerFactory struct {
-	senderRules                      []pruning.KeepRule
-	receiverRules                    []pruning.KeepRule
-	retryWait                        time.Duration
-	considerSnapAtCursorReplicated   bool
-	convertAnyStepHoldToStepBookmark bool
-	promPruneSecs                    *prometheus.HistogramVec
+	senderRules                    []pruning.KeepRule
+	receiverRules                  []pruning.KeepRule
+	retryWait                      time.Duration
+	considerSnapAtCursorReplicated bool
+	promPruneSecs                  *prometheus.HistogramVec
 }
 
 type LocalPrunerFactory struct {
@@ -123,35 +125,25 @@ func NewPrunerFactory(in config.PruningSenderReceiver, promPruneSecs *prometheus
 		}
 		considerSnapAtCursorReplicated = considerSnapAtCursorReplicated || !knr.KeepSnapshotAtCursor
 	}
-
-	convertAnyStepHoldToStepBookmark := false
-	for _, r := range in.KeepSender {
-		_, ok := r.Ret.(*config.PruneKeepStepHolds)
-		convertAnyStepHoldToStepBookmark = convertAnyStepHoldToStepBookmark || ok
-	}
-
 	f := &PrunerFactory{
-		senderRules:                      keepRulesSender,
-		receiverRules:                    keepRulesReceiver,
-		retryWait:                        envconst.Duration("ZREPL_PRUNER_RETRY_INTERVAL", 10*time.Second),
-		considerSnapAtCursorReplicated:   considerSnapAtCursorReplicated,
-		convertAnyStepHoldToStepBookmark: convertAnyStepHoldToStepBookmark,
-		promPruneSecs:                    promPruneSecs,
+		senderRules:                    keepRulesSender,
+		receiverRules:                  keepRulesReceiver,
+		retryWait:                      envconst.Duration("ZREPL_PRUNER_RETRY_INTERVAL", 10*time.Second),
+		considerSnapAtCursorReplicated: considerSnapAtCursorReplicated,
+		promPruneSecs:                  promPruneSecs,
 	}
 	return f, nil
 }
 
-func (f *PrunerFactory) BuildSenderPruner(ctx context.Context, sender Target, receiver Endpoint) *Pruner {
+func (f *PrunerFactory) BuildSenderPruner(ctx context.Context, target Target, receiver History) *Pruner {
 	p := &Pruner{
 		args: args{
-			context.WithValue(ctx, contextKeyPruneSide, "sender"),
-			sender,
-			sender,
+			WithLogger(ctx, GetLogger(ctx).WithField("prune_side", "sender")),
+			target,
 			receiver,
 			f.senderRules,
 			f.retryWait,
 			f.considerSnapAtCursorReplicated,
-			f.convertAnyStepHoldToStepBookmark,
 			f.promPruneSecs.WithLabelValues("sender"),
 		},
 		state: Plan,
@@ -159,17 +151,15 @@ func (f *PrunerFactory) BuildSenderPruner(ctx context.Context, sender Target, re
 	return p
 }
 
-func (f *PrunerFactory) BuildReceiverPruner(ctx context.Context, receiver Target, sender Endpoint) *Pruner {
+func (f *PrunerFactory) BuildReceiverPruner(ctx context.Context, target Target, receiver History) *Pruner {
 	p := &Pruner{
 		args: args{
-			context.WithValue(ctx, contextKeyPruneSide, "receiver"),
-			receiver,
-			sender,
+			WithLogger(ctx, GetLogger(ctx).WithField("prune_side", "receiver")),
+			target,
 			receiver,
 			f.receiverRules,
 			f.retryWait,
 			false, // senseless here anyways
-			false, // senseless here anyways
 			f.promPruneSecs.WithLabelValues("receiver"),
 		},
 		state: Plan,
@@ -177,17 +167,15 @@ func (f *PrunerFactory) BuildReceiverPruner(ctx context.Context, receiver Target
 	return p
 }
 
-func (f *LocalPrunerFactory) BuildLocalPruner(ctx context.Context, target Target) *Pruner {
+func (f *LocalPrunerFactory) BuildLocalPruner(ctx context.Context, target Target, receiver History) *Pruner {
 	p := &Pruner{
 		args: args{
-			context.WithValue(ctx, contextKeyPruneSide, "local"),
-			target,
-			target,
+			ctx,
 			target,
+			receiver,
 			f.keepRules,
 			f.retryWait,
 			false, // considerSnapAtCursorReplicated is not relevant for local pruning
-			false, // convertAnyStepHoldToStepBookmark is not relevant for local pruning
 			f.promPruneSecs.WithLabelValues("local"),
 		},
 		state: Plan,
@@ -356,13 +344,11 @@ func (s snapshot) Replicated() bool { return s.replicated }
 
 func (s snapshot) Date() time.Time { return s.date }
 
-func (s snapshot) CreateTXG() uint64 { return s.fsv.GetCreateTXG() }
-
 func doOneAttempt(a *args, u updater) {
 
-	ctx, sender, receiver, target := a.ctx, a.sender, a.receiver, a.target
+	ctx, target, receiver := a.ctx, a.target, a.receiver
 
-	sfssres, err := sender.ListFilesystems(ctx, &pdu.ListFilesystemReq{})
+	sfssres, err := receiver.ListFilesystems(ctx, &pdu.ListFilesystemReq{})
 	if err != nil {
 		u(func(p *Pruner) {
 			p.state = PlanErr
@@ -424,10 +410,6 @@ tfss_loop:
 
 		pfs.snaps = make([]pruning.Snapshot, 0, len(tfsvs))
 
-		receiver.ListFilesystemVersions(ctx, &pdu.ListFilesystemVersionsReq{
-			Filesystem: tfs.Path,
-		})
-
 		rcReq := &pdu.ReplicationCursorReq{
 			Filesystem: tfs.Path,
 		}
@@ -436,7 +418,6 @@ tfss_loop:
 			pfsPlanErrAndLog(err, "cannot get replication cursor bookmark")
 			continue tfss_loop
 		}
-
 		if rc.GetNotexist() {
 			err := errors.New("replication cursor bookmark does not exist (one successful replication is required before pruning works)")
 			pfsPlanErrAndLog(err, "")

daemon/snapper/snapper.go

@@ -8,12 +8,10 @@ import (
 	"time"
 
 	"github.com/pkg/errors"
-	"github.com/zrepl/zrepl/daemon/logging/trace"
 
 	"github.com/zrepl/zrepl/config"
 	"github.com/zrepl/zrepl/daemon/filters"
 	"github.com/zrepl/zrepl/daemon/hooks"
-	"github.com/zrepl/zrepl/daemon/logging"
 	"github.com/zrepl/zrepl/logger"
 	"github.com/zrepl/zrepl/util/envconst"
 	"github.com/zrepl/zrepl/zfs"
@@ -47,6 +45,7 @@ type snapProgress struct {
 
 type args struct {
 	ctx            context.Context
+	log            Logger
 	prefix         string
 	interval       time.Duration
 	fsf            *filters.DatasetMapFilter
@@ -103,10 +102,23 @@ func (s State) sf() state {
 type updater func(u func(*Snapper)) State
 type state func(a args, u updater) state
 
+type contextKey int
+
+const (
+	contextKeyLog contextKey = 0
+)
+
 type Logger = logger.Logger
 
+func WithLogger(ctx context.Context, log Logger) context.Context {
+	return context.WithValue(ctx, contextKeyLog, log)
+}
+
 func getLogger(ctx context.Context) Logger {
-	return logging.GetLogger(ctx, logging.SubsysSnapshot)
+	if log, ok := ctx.Value(contextKeyLog).(Logger); ok {
+		return log
+	}
+	return logger.NewNullLogger()
 }
 
 func PeriodicFromConfig(g *config.Global, fsf *filters.DatasetMapFilter, in *config.SnapshottingPeriodic) (*Snapper, error) {
@@ -134,12 +146,13 @@ func PeriodicFromConfig(g *config.Global, fsf *filters.DatasetMapFilter, in *con
 }
 
 func (s *Snapper) Run(ctx context.Context, snapshotsTaken chan<- struct{}) {
-	defer trace.WithSpanFromStackUpdateCtx(&ctx)()
+
 	getLogger(ctx).Debug("start")
 	defer getLogger(ctx).Debug("stop")
 
 	s.args.snapshotsTaken = snapshotsTaken
 	s.args.ctx = ctx
+	s.args.log = getLogger(ctx)
 	s.args.dryRun = false // for future expansion
 
 	u := func(u func(*Snapper)) State {
@@ -177,7 +190,7 @@ func onErr(err error, u updater) state {
 		case Snapshotting:
 			s.state = ErrorWait
 		}
-		getLogger(s.args.ctx).WithError(err).WithField("pre_state", preState).WithField("post_state", s.state).Error("snapshotting error")
+		s.args.log.WithError(err).WithField("pre_state", preState).WithField("post_state", s.state).Error("snapshotting error")
 	}).sf()
 }
 
@@ -196,7 +209,7 @@ func syncUp(a args, u updater) state {
 	if err != nil {
 		return onErr(err, u)
 	}
-	syncPoint, err := findSyncPoint(a.ctx, fss, a.prefix, a.interval)
+	syncPoint, err := findSyncPoint(a.log, fss, a.prefix, a.interval)
 	if err != nil {
 		return onErr(err, u)
 	}
@@ -253,18 +266,18 @@ func snapshot(a args, u updater) state {
 		suffix := time.Now().In(time.UTC).Format("20060102_150405_000")
 		snapname := fmt.Sprintf("%s%s", a.prefix, suffix)
 
-		ctx := logging.WithInjectedField(a.ctx, "fs", fs.ToString())
-		ctx = logging.WithInjectedField(ctx, "snap", snapname)
+		l := a.log.
+			WithField("fs", fs.ToString()).
+			WithField("snap", snapname)
 
 		hookEnvExtra := hooks.Env{
 			hooks.EnvFS:       fs.ToString(),
 			hooks.EnvSnapshot: snapname,
 		}
 
-		jobCallback := hooks.NewCallbackHookForFilesystem("snapshot", fs, func(ctx context.Context) (err error) {
-			l := getLogger(ctx)
+		jobCallback := hooks.NewCallbackHookForFilesystem("snapshot", fs, func(_ context.Context) (err error) {
 			l.Debug("create snapshot")
-			err = zfs.ZFSSnapshot(ctx, fs, snapname, false) // TODO propagate context to ZFSSnapshot
+			err = zfs.ZFSSnapshot(a.ctx, fs, snapname, false) // TODO propagate context to ZFSSnapshot
 			if err != nil {
 				l.WithError(err).Error("cannot create snapshot")
 			}
@@ -277,7 +290,7 @@ func snapshot(a args, u updater) state {
 		{
 			filteredHooks, err := a.hooks.CopyFilteredForFilesystem(fs)
 			if err != nil {
-				getLogger(ctx).WithError(err).Error("unexpected filter error")
+				l.WithError(err).Error("unexpected filter error")
 				fsHadErr = true
 				goto updateFSState
 			}
@@ -290,7 +303,7 @@ func snapshot(a args, u updater) state {
 			plan, planErr = hooks.NewPlan(&filteredHooks, hooks.PhaseSnapshot, jobCallback, hookEnvExtra)
 			if planErr != nil {
 				fsHadErr = true
-				getLogger(ctx).WithError(planErr).Error("cannot create job hook plan")
+				l.WithError(planErr).Error("cannot create job hook plan")
 				goto updateFSState
 			}
 		}
@@ -301,14 +314,15 @@ func snapshot(a args, u updater) state {
 			progress.state = SnapStarted
 		})
 		{
-			getLogger(ctx).WithField("report", plan.Report().String()).Debug("begin run job plan")
-			plan.Run(ctx, a.dryRun)
+			l := hooks.GetLogger(a.ctx).WithField("fs", fs.ToString()).WithField("snap", snapname)
+			l.WithField("report", plan.Report().String()).Debug("begin run job plan")
+			plan.Run(hooks.WithLogger(a.ctx, l), a.dryRun)
 			planReport = plan.Report()
 			fsHadErr = planReport.HadError() // not just fatal errors
 			if fsHadErr {
-				getLogger(ctx).WithField("report", planReport.String()).Error("end run job plan with error")
+				l.WithField("report", planReport.String()).Error("end run job plan with error")
 			} else {
-				getLogger(ctx).WithField("report", planReport.String()).Info("end run job plan successful")
+				l.WithField("report", planReport.String()).Info("end run job plan successful")
 			}
 		}
 
@@ -328,7 +342,7 @@ func snapshot(a args, u updater) state {
 	case a.snapshotsTaken <- struct{}{}:
 	default:
 		if a.snapshotsTaken != nil {
-			getLogger(a.ctx).Warn("callback channel is full, discarding snapshot update event")
+			a.log.Warn("callback channel is full, discarding snapshot update event")
 		}
 	}
 
@@ -341,7 +355,7 @@ func snapshot(a args, u updater) state {
 					break
 				}
 			}
-			getLogger(a.ctx).WithField("hook", h.String()).WithField("hook_number", hookIdx+1).Warn("hook did not match any snapshotted filesystems")
+			a.log.WithField("hook", h.String()).WithField("hook_number", hookIdx+1).Warn("hook did not match any snapshotted filesystems")
 		}
 	}
 
@@ -362,7 +376,7 @@ func wait(a args, u updater) state {
 		lastTick := snapper.lastInvocation
 		snapper.sleepUntil = lastTick.Add(a.interval)
 		sleepUntil = snapper.sleepUntil
-		log := getLogger(a.ctx).WithField("sleep_until", sleepUntil).WithField("duration", a.interval)
+		log := a.log.WithField("sleep_until", sleepUntil).WithField("duration", a.interval)
 		logFunc := log.Debug
 		if snapper.state == ErrorWait || snapper.state == SyncUpErrWait {
 			logFunc = log.Error
@@ -390,7 +404,7 @@ func listFSes(ctx context.Context, mf *filters.DatasetMapFilter) (fss []*zfs.Dat
 var syncUpWarnNoSnapshotUntilSyncupMinDuration = envconst.Duration("ZREPL_SNAPPER_SYNCUP_WARN_MIN_DURATION", 1*time.Second)
 
 // see docs/snapshotting.rst
-func findSyncPoint(ctx context.Context, fss []*zfs.DatasetPath, prefix string, interval time.Duration) (syncPoint time.Time, err error) {
+func findSyncPoint(log Logger, fss []*zfs.DatasetPath, prefix string, interval time.Duration) (syncPoint time.Time, err error) {
 
 	const (
 		prioHasVersions int = iota
@@ -412,10 +426,10 @@ func findSyncPoint(ctx context.Context, fss []*zfs.DatasetPath, prefix string, i
 
 	now := time.Now()
 
-	getLogger(ctx).Debug("examine filesystem state to find sync point")
+	log.Debug("examine filesystem state to find sync point")
 	for _, d := range fss {
-		ctx := logging.WithInjectedField(ctx, "fs", d.ToString())
-		syncPoint, err := findSyncPointFSNextOptimalSnapshotTime(ctx, now, interval, prefix, d)
+		l := log.WithField("fs", d.ToString())
+		syncPoint, err := findSyncPointFSNextOptimalSnapshotTime(l, now, interval, prefix, d)
 		if err == findSyncPointFSNoFilesystemVersionsErr {
 			snaptimes = append(snaptimes, snapTime{
 				ds:   d,
@@ -424,9 +438,9 @@ func findSyncPoint(ctx context.Context, fss []*zfs.DatasetPath, prefix string, i
 			})
 		} else if err != nil {
 			hardErrs++
-			getLogger(ctx).WithError(err).Error("cannot determine optimal sync point for this filesystem")
+			l.WithError(err).Error("cannot determine optimal sync point for this filesystem")
 		} else {
-			getLogger(ctx).WithField("syncPoint", syncPoint).Debug("found optimal sync point for this filesystem")
+			l.WithField("syncPoint", syncPoint).Debug("found optimal sync point for this filesystem")
 			snaptimes = append(snaptimes, snapTime{
 				ds:   d,
 				prio: prioHasVersions,
@@ -453,7 +467,7 @@ func findSyncPoint(ctx context.Context, fss []*zfs.DatasetPath, prefix string, i
 	})
 
 	winnerSyncPoint := snaptimes[0].time
-	l := getLogger(ctx).WithField("syncPoint", winnerSyncPoint.String())
+	l := log.WithField("syncPoint", winnerSyncPoint.String())
 	l.Info("determined sync point")
 	if winnerSyncPoint.Sub(now) > syncUpWarnNoSnapshotUntilSyncupMinDuration {
 		for _, st := range snaptimes {
@@ -469,9 +483,9 @@ func findSyncPoint(ctx context.Context, fss []*zfs.DatasetPath, prefix string, i
 
 var findSyncPointFSNoFilesystemVersionsErr = fmt.Errorf("no filesystem versions")
 
-func findSyncPointFSNextOptimalSnapshotTime(ctx context.Context, now time.Time, interval time.Duration, prefix string, d *zfs.DatasetPath) (time.Time, error) {
+func findSyncPointFSNextOptimalSnapshotTime(l Logger, now time.Time, interval time.Duration, prefix string, d *zfs.DatasetPath) (time.Time, error) {
 
-	fsvs, err := zfs.ZFSListFilesystemVersions(ctx, d, zfs.ListFilesystemVersionsOptions{
+	fsvs, err := zfs.ZFSListFilesystemVersions(context.TODO(), d, zfs.ListFilesystemVersionsOptions{
 		Types:           zfs.Snapshots,
 		ShortnamePrefix: prefix,
 	})
@@ -488,7 +502,7 @@ func findSyncPointFSNextOptimalSnapshotTime(ctx context.Context, now time.Time,
 	})
 
 	latest := fsvs[len(fsvs)-1]
-	getLogger(ctx).WithField("creation", latest.Creation).Debug("found latest snapshot")
+	l.WithField("creation", latest.Creation).Debug("found latest snapshot")
 
 	since := now.Sub(latest.Creation)
 	if since < 0 {

docs/configuration/transports.rst

@@ -52,15 +52,8 @@ Serve
         listen: ":8888"
         listen_freebind: true # optional, default false
         clients: {
-          "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-*"
+          "192.168.122.123" : "mysql01"
+          "192.168.122.123" : "mx01"
         }
       ...
 

docs/installation.rst

@@ -1,19 +1,138 @@
+.. _binary releases: https://github.com/zrepl/zrepl/releases
 
-.. _installation_toc:
+.. _installation:
 
-************
 Installation
-************
+============
 
 .. TIP::
 
     Note: check out the :ref:`tutorial` if you want a first impression of zrepl.
 
-.. toctree::
+User Privileges
+---------------
 
-    installation/user-privileges
-    installation/packages
-    installation/apt-repos
-    installation/compile-from-source
-    installation/freebsd-jail-with-iocage
-    installation/what-next
+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`.

docs/installation/apt-repos.rst

@@ -1,29 +0,0 @@
-
-.. _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

@@ -1,45 +0,0 @@
-.. _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

@@ -1,140 +0,0 @@
-.. 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

@@ -1,50 +0,0 @@
-.. _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

@@ -1,12 +0,0 @@
-.. _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

@@ -1,8 +0,0 @@
-.. _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/supporters.rst

@@ -32,9 +32,6 @@ We would like to thank the following people and organizations for supporting zre
 
    <div class="fa fa-code" style="width: 1em;"></div>
 
-* |supporter-std| Henning Kessler
-* |supporter-std| `John Ramsden <https://github.com/johnramsden>`_
-* |supporter-std| `DrLuke <https://github.com/drluke>`_
 * |supporter-gold| `Mateusz Kwiatkowski (runhyve.app) <https://runhyve.app>`_
 * |supporter-std| `Gaelan D'costa <https://www.robot-disco.net>`_
 * |supporter-std| `Tenzin Lhakhang <https://confluence.tenzin.io>`_

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_toc>` and come back here.
+Follow the :ref:`OS-specific installation instructions <installation>` and come back here.
 
 Generate TLS Certificates
 -------------------------

endpoint/context.go

@@ -3,18 +3,25 @@ package endpoint
 import (
 	"context"
 
-	"github.com/zrepl/zrepl/daemon/logging"
 	"github.com/zrepl/zrepl/logger"
 )
 
 type contextKey int
 
 const (
-	ClientIdentityKey contextKey = iota
+	contextKeyLogger contextKey = iota
+	ClientIdentityKey
 )
 
 type Logger = logger.Logger
 
-func getLogger(ctx context.Context) Logger {
-	return logging.GetLogger(ctx, logging.SubsysEndpoint)
+func WithLogger(ctx context.Context, log Logger) context.Context {
+	return context.WithValue(ctx, contextKeyLogger, log)
+}
+
+func getLogger(ctx context.Context) Logger {
+	if l, ok := ctx.Value(contextKeyLogger).(Logger); ok {
+		return l
+	}
+	return logger.NewNullLogger()
 }

endpoint/endpoint.go

@@ -7,10 +7,9 @@ import (
 	"fmt"
 	"io"
 	"path"
+	"sync"
 
-	"github.com/kr/pretty"
 	"github.com/pkg/errors"
-	"github.com/zrepl/zrepl/daemon/logging/trace"
 
 	"github.com/zrepl/zrepl/replication/logic/pdu"
 	"github.com/zrepl/zrepl/util/chainedio"
@@ -74,8 +73,6 @@ func (s *Sender) filterCheckFS(fs string) (*zfs.DatasetPath, error) {
 }
 
 func (s *Sender) ListFilesystems(ctx context.Context, r *pdu.ListFilesystemReq) (*pdu.ListFilesystemRes, error) {
-	defer trace.WithSpanFromStackUpdateCtx(&ctx)()
-
 	fss, err := zfs.ZFSListMapping(ctx, s.FSFilter)
 	if err != nil {
 		return nil, err
@@ -98,8 +95,6 @@ func (s *Sender) ListFilesystems(ctx context.Context, r *pdu.ListFilesystemReq)
 }
 
 func (s *Sender) ListFilesystemVersions(ctx context.Context, r *pdu.ListFilesystemVersionsReq) (*pdu.ListFilesystemVersionsRes, error) {
-	defer trace.WithSpanFromStackUpdateCtx(&ctx)()
-
 	lp, err := s.filterCheckFS(r.GetFilesystem())
 	if err != nil {
 		return nil, err
@@ -112,20 +107,123 @@ func (s *Sender) ListFilesystemVersions(ctx context.Context, r *pdu.ListFilesyst
 	for i := range fsvs {
 		rfsvs[i] = pdu.FilesystemVersionFromZFS(&fsvs[i])
 	}
-
-	sendAbstractions := sendAbstractionsCacheSingleton.GetByFS(ctx, lp.ToString())
-	rSabsInfo := make([]*pdu.SendAbstraction, len(sendAbstractions))
-	for i := range rSabsInfo {
-		rSabsInfo[i] = SendAbstractionToPDU(sendAbstractions[i])
-	}
-
 	res := &pdu.ListFilesystemVersionsRes{Versions: rfsvs}
 	return res, nil
 
 }
 
-var maxConcurrentZFSSend = envconst.Int64("ZREPL_ENDPOINT_MAX_CONCURRENT_SEND", 10)
-var maxConcurrentZFSSendSemaphore = semaphore.New(maxConcurrentZFSSend)
+func (p *Sender) HintMostRecentCommonAncestor(ctx context.Context, r *pdu.HintMostRecentCommonAncestorReq) (*pdu.HintMostRecentCommonAncestorRes, error) {
+
+	fsp, err := p.filterCheckFS(r.GetFilesystem())
+	if err != nil {
+		return nil, err
+	}
+	fs := fsp.ToString()
+
+	log := getLogger(ctx).WithField("fs", fs).WithField("hinted_most_recent", fmt.Sprintf("%#v", r.GetSenderVersion()))
+
+	log.WithField("full_hint", r).Debug("full hint")
+
+	if r.GetSenderVersion() == nil {
+		// no common ancestor found, likely due to failed prior replication attempt
+		// => release stale step holds to prevent them from accumulating
+		//    (they can accumulate on initial replication because each inital replication step might hold a different `to`)
+		// => replication cursors cannot accumulate because we always _move_ the replication cursor
+		log.Debug("releasing all step holds on the filesystem")
+		TryReleaseStepStaleFS(ctx, fs, p.jobId)
+		return &pdu.HintMostRecentCommonAncestorRes{}, nil
+	}
+
+	// we were hinted a specific common ancestor
+
+	mostRecentVersion, err := sendArgsFromPDUAndValidateExistsAndGetVersion(ctx, fs, r.GetSenderVersion())
+	if err != nil {
+		msg := "HintMostRecentCommonAncestor rpc with nonexistent most recent version"
+		log.Warn(msg)
+		return nil, errors.Wrap(err, msg)
+	}
+
+	// move replication cursor to this position
+	destroyedCursors, err := MoveReplicationCursor(ctx, fs, mostRecentVersion, p.jobId)
+	if err == zfs.ErrBookmarkCloningNotSupported {
+		log.Debug("not creating replication cursor from bookmark because ZFS does not support it")
+		// fallthrough
+	} else if err != nil {
+		return nil, errors.Wrap(err, "cannot set replication cursor to hinted version")
+	}
+
+	// take care of stale step holds
+	log.WithField("step-holds-cleanup-mode", senderHintMostRecentCommonAncestorStepCleanupMode).
+		Debug("taking care of possibly stale step holds")
+	doStepCleanup := false
+	var stepCleanupSince *CreateTXGRangeBound
+	switch senderHintMostRecentCommonAncestorStepCleanupMode {
+	case StepCleanupNoCleanup:
+		doStepCleanup = false
+	case StepCleanupRangeSinceUnbounded:
+		doStepCleanup = true
+		stepCleanupSince = nil
+	case StepCleanupRangeSinceReplicationCursor:
+		doStepCleanup = true
+		// Use the destroyed replication cursors as indicator how far the previous replication got.
+		// To be precise: We limit the amount of visisted snapshots to exactly those snapshots
+		// created since the last successful replication cursor movement (i.e. last successful replication step)
+		//
+		// If we crash now, we'll leak the step we are about to release, but the performance gain
+		// of limiting the amount of snapshots we visit makes up for that.
+		// Users have the `zrepl holds release-stale` command to cleanup leaked step holds.
+		for _, destroyed := range destroyedCursors {
+			if stepCleanupSince == nil {
+				stepCleanupSince = &CreateTXGRangeBound{
+					CreateTXG: destroyed.GetCreateTXG(),
+					Inclusive: &zfs.NilBool{B: true},
+				}
+			} else if destroyed.GetCreateTXG() < stepCleanupSince.CreateTXG {
+				stepCleanupSince.CreateTXG = destroyed.GetCreateTXG()
+			}
+		}
+	default:
+		panic(senderHintMostRecentCommonAncestorStepCleanupMode)
+	}
+	if !doStepCleanup {
+		log.Info("skipping cleanup of prior invocations' step holds due to environment variable setting")
+	} else {
+		if err := ReleaseStepCummulativeInclusive(ctx, fs, stepCleanupSince, mostRecentVersion, p.jobId); err != nil {
+			return nil, errors.Wrap(err, "cannot cleanup prior invocation's step holds and bookmarks")
+		} else {
+			log.Info("step hold cleanup done")
+		}
+	}
+
+	return &pdu.HintMostRecentCommonAncestorRes{}, nil
+}
+
+type HintMostRecentCommonAncestorStepCleanupMode struct{ string }
+
+var (
+	StepCleanupRangeSinceReplicationCursor = HintMostRecentCommonAncestorStepCleanupMode{"range-since-replication-cursor"}
+	StepCleanupRangeSinceUnbounded         = HintMostRecentCommonAncestorStepCleanupMode{"range-since-unbounded"}
+	StepCleanupNoCleanup                   = HintMostRecentCommonAncestorStepCleanupMode{"no-cleanup"}
+)
+
+func (m HintMostRecentCommonAncestorStepCleanupMode) String() string { return string(m.string) }
+func (m *HintMostRecentCommonAncestorStepCleanupMode) Set(s string) error {
+	switch s {
+	case StepCleanupRangeSinceReplicationCursor.String():
+		*m = StepCleanupRangeSinceReplicationCursor
+	case StepCleanupRangeSinceUnbounded.String():
+		*m = StepCleanupRangeSinceUnbounded
+	case StepCleanupNoCleanup.String():
+		*m = StepCleanupNoCleanup
+	default:
+		return fmt.Errorf("unknown step cleanup mode %q", s)
+	}
+	return nil
+}
+
+var senderHintMostRecentCommonAncestorStepCleanupMode = *envconst.Var("ZREPL_ENDPOINT_SENDER_HINT_MOST_RECENT_STEP_HOLD_CLEANUP_MODE", &StepCleanupRangeSinceReplicationCursor).(*HintMostRecentCommonAncestorStepCleanupMode)
+
+var maxConcurrentZFSSendSemaphore = semaphore.New(envconst.Int64("ZREPL_ENDPOINT_MAX_CONCURRENT_SEND", 10))
 
 func uncheckedSendArgsFromPDU(fsv *pdu.FilesystemVersion) *zfs.ZFSSendArgVersion {
 	if fsv == nil {
@@ -147,7 +245,6 @@ func sendArgsFromPDUAndValidateExistsAndGetVersion(ctx context.Context, fs strin
 }
 
 func (s *Sender) Send(ctx context.Context, r *pdu.SendReq) (*pdu.SendRes, io.ReadCloser, error) {
-	defer trace.WithSpanFromStackUpdateCtx(&ctx)()
 
 	_, err := s.filterCheckFS(r.Filesystem)
 	if err != nil {
@@ -215,11 +312,10 @@ func (s *Sender) Send(ctx context.Context, r *pdu.SendReq) (*pdu.SendRes, io.Rea
 		return res, nil, nil
 	}
 
-	// create a replication cursor for `From` (usually an idempotent no-op because SendCompleted already created it before)
-	var fromReplicationCursor Abstraction
+	// update replication cursor
 	if sendArgs.From != nil {
 		// For all but the first replication, this should always be a no-op because SendCompleted already moved the cursor
-		fromReplicationCursor, err = CreateReplicationCursor(ctx, sendArgs.FS, *sendArgs.FromVersion, s.jobId) // no shadow
+		_, err = MoveReplicationCursor(ctx, sendArgs.FS, sendArgs.FromVersion, s.jobId)
 		if err == zfs.ErrBookmarkCloningNotSupported {
 			getLogger(ctx).Debug("not creating replication cursor from bookmark because ZFS does not support it")
 			// fallthrough
@@ -228,10 +324,9 @@ func (s *Sender) Send(ctx context.Context, r *pdu.SendReq) (*pdu.SendRes, io.Rea
 		}
 	}
 
-	var fromHold, toHold Abstraction
 	// make sure `From` doesn't go away in order to make this step resumable
 	if sendArgs.From != nil {
-		fromHold, err = HoldStep(ctx, sendArgs.FS, *sendArgs.FromVersion, s.jobId) // no shadow
+		_, err := HoldStep(ctx, sendArgs.FS, *sendArgs.FromVersion, s.jobId)
 		if err == zfs.ErrBookmarkCloningNotSupported {
 			getLogger(ctx).Debug("not creating step bookmark because ZFS does not support it")
 			// fallthrough
@@ -240,76 +335,21 @@ func (s *Sender) Send(ctx context.Context, r *pdu.SendReq) (*pdu.SendRes, io.Rea
 		}
 	}
 	// make sure `To` doesn't go away in order to make this step resumable
-	toHold, err = HoldStep(ctx, sendArgs.FS, sendArgs.ToVersion, s.jobId)
+	_, err = HoldStep(ctx, sendArgs.FS, sendArgs.ToVersion, s.jobId)
 	if err != nil {
 		return nil, nil, errors.Wrapf(err, "cannot hold `to` version %q before starting send", sendArgs.ToVersion)
 	}
 
-	// cleanup the mess that _this function_ might have created in prior failed attempts:
-	//
-	// In summary, we delete every endpoint ZFS abstraction created on this filesystem for this job id,
-	// except for the ones we just created above.
-	//
-	// This is the most robust approach to avoid leaking (= forgetting to clean up) endpoint ZFS abstractions,
-	// all under the assumption that there will only ever be one send for a (jobId,fs) combination at any given time.
-	//
-	// Note that the SendCompleted rpc can't be relied upon for this purpose:
-	// - it might be lost due to network errors,
-	// - or never be sent by a potentially malicious or buggy client,
-	// - or never be send because the replication step failed at some point
-	//   (potentially leaving a resumable state on the receiver, which is the case where we really do not want to blow away the step holds too soon.)
-	//
-	// Note further that a resuming send, due to the idempotent nature of func CreateReplicationCursor and HoldStep,
-	// will never lose its step holds because we just (idempotently re-)created them above, before attempting the cleanup.
-	func() {
-		ctx, endSpan := trace.WithSpan(ctx, "cleanup-stale-abstractions")
-		defer endSpan()
-
-		liveAbs := []Abstraction{fromHold, toHold, fromReplicationCursor}
-		keep := func(a Abstraction) (keep bool) {
-			keep = false
-			for _, k := range liveAbs {
-				keep = keep || AbstractionEquals(a, k)
-			}
-			return keep
-		}
-		check := func(obsoleteAbs []Abstraction) {
-			// last line of defense: check that we don't destroy the incremental `from` and `to`
-			// if we did that, we might be about to blow away the last common filesystem version between sender and receiver
-			mustLiveVersions := []zfs.FilesystemVersion{sendArgs.ToVersion}
-			if sendArgs.FromVersion != nil {
-				mustLiveVersions = append(mustLiveVersions, *sendArgs.FromVersion)
-			}
-			for _, staleVersion := range obsoleteAbs {
-				for _, mustLiveVersion := range mustLiveVersions {
-					if zfs.FilesystemVersionEqualIdentity(mustLiveVersion, staleVersion.GetFilesystemVersion()) {
-						panic(fmt.Sprintf("impl error: %q would be destroyed because it is considered stale but it is part of of sendArgs=%s", mustLiveVersion.String(), pretty.Sprint(sendArgs)))
-					}
-				}
-			}
-		}
-		sendAbstractionsCacheSingleton.TryBatchDestroy(ctx, s.jobId, sendArgs.FS, keep, check)
-	}()
-
-	if fromHold != nil {
-		sendAbstractionsCacheSingleton.Put(fromHold)
-	}
-	sendAbstractionsCacheSingleton.Put(toHold)
-	if fromReplicationCursor != nil {
-		sendAbstractionsCacheSingleton.Put(fromReplicationCursor)
-	}
+	// step holds & replication cursor released / moved forward in s.SendCompleted => s.moveCursorAndReleaseSendHolds
 
 	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")
 	}
-
 	return res, sendStream, nil
 }
 
 func (p *Sender) SendCompleted(ctx context.Context, r *pdu.SendCompletedReq) (*pdu.SendCompletedRes, error) {
-	defer trace.WithSpanFromStackUpdateCtx(&ctx)()
 
 	orig := r.GetOriginalReq() // may be nil, always use proto getters
 	fsp, err := p.filterCheckFS(orig.GetFilesystem())
@@ -331,43 +371,76 @@ func (p *Sender) SendCompleted(ctx context.Context, r *pdu.SendCompletedReq) (*p
 		return nil, errors.Wrap(err, "validate `to` exists")
 	}
 
-	log := func(ctx context.Context) Logger {
-		log := getLogger(ctx).WithField("to_guid", to.Guid).
-			WithField("fs", fs).
-			WithField("to", to.RelName)
-		if from != nil {
-			log = log.WithField("from", from.RelName).WithField("from_guid", from.Guid)
-		}
-		return log
+	log := getLogger(ctx).WithField("to_guid", to.Guid).
+		WithField("fs", fs).
+		WithField("to", to.RelName)
+	if from != nil {
+		log = log.WithField("from", from.RelName).WithField("from_guid", from.Guid)
 	}
 
-	toReplicationCursor, err := CreateReplicationCursor(ctx, fs, to, p.jobId)
+	log.Debug("move replication cursor to most recent common version")
+	destroyedCursors, err := MoveReplicationCursor(ctx, fs, to, p.jobId)
 	if err != nil {
 		if err == zfs.ErrBookmarkCloningNotSupported {
-			log(ctx).Debug("not setting replication cursor, bookmark cloning not supported")
+			log.Debug("not setting replication cursor, bookmark cloning not supported")
 		} else {
 			msg := "cannot move replication cursor, keeping hold on `to` until successful"
-			log(ctx).WithError(err).Error(msg)
+			log.WithError(err).Error(msg)
 			err = errors.Wrap(err, msg)
-			// it is correct to not destroy from and to step holds if we can't move the cursor!
+			// it is correct to not release the hold if we can't move the cursor!
 			return &pdu.SendCompletedRes{}, err
 		}
 	} else {
-		log(ctx).WithField("to_cursor", toReplicationCursor.String()).Info("successfully created `to` replication cursor")
+		log.Info("successfully moved replication cursor")
 	}
 
-	keep := func(a Abstraction) bool {
-		return AbstractionEquals(a, toReplicationCursor)
-	}
-	sendAbstractionsCacheSingleton.TryBatchDestroy(ctx, p.jobId, fs, keep, nil)
+	// kick off releasing of step holds / bookmarks
+	// if we fail to release them, don't bother the caller:
+	// they are merely an implementation detail on the sender for better resumability
+	var wg sync.WaitGroup
+	wg.Add(2)
+	go func() {
+		defer wg.Done()
+		log.Debug("release step-hold of or step-bookmark on `to`")
+		err = ReleaseStep(ctx, fs, to, p.jobId)
+		if err != nil {
+			log.WithError(err).Error("cannot release step-holds on or destroy step-bookmark of `to`")
+		} else {
+			log.Info("successfully released step-holds on or destroyed step-bookmark of `to`")
+		}
+
+	}()
+	go func() {
+		defer wg.Done()
+		if from == nil {
+			return
+		}
+		log.Debug("release step-hold of or step-bookmark on `from`")
+		err := ReleaseStep(ctx, fs, *from, p.jobId)
+		if err != nil {
+			if dne, ok := err.(*zfs.DatasetDoesNotExist); ok {
+				// If bookmark cloning is not supported, `from` might be the old replication cursor
+				// and thus have already been destroyed by MoveReplicationCursor above
+				// In that case, nonexistence of `from` is not an error, otherwise it is.
+				for _, c := range destroyedCursors {
+					if c.GetFullPath() == dne.Path {
+						log.Info("`from` was a replication cursor and has already been destroyed")
+						return
+					}
+				}
+				// fallthrough
+			}
+			log.WithError(err).Error("cannot release step-holds on or destroy step-bookmark of `from`")
+		} else {
+			log.Info("successfully released step-holds on or destroyed step-bookmark of `from`")
+		}
+	}()
+	wg.Wait()
 
 	return &pdu.SendCompletedRes{}, nil
-
 }
 
 func (p *Sender) DestroySnapshots(ctx context.Context, req *pdu.DestroySnapshotsReq) (*pdu.DestroySnapshotsRes, error) {
-	defer trace.WithSpanFromStackUpdateCtx(&ctx)()
-
 	dp, err := p.filterCheckFS(req.Filesystem)
 	if err != nil {
 		return nil, err
@@ -376,8 +449,6 @@ func (p *Sender) DestroySnapshots(ctx context.Context, req *pdu.DestroySnapshots
 }
 
 func (p *Sender) Ping(ctx context.Context, req *pdu.PingReq) (*pdu.PingRes, error) {
-	defer trace.WithSpanFromStackUpdateCtx(&ctx)()
-
 	res := pdu.PingRes{
 		Echo: req.GetMessage(),
 	}
@@ -385,20 +456,14 @@ func (p *Sender) Ping(ctx context.Context, req *pdu.PingReq) (*pdu.PingRes, erro
 }
 
 func (p *Sender) PingDataconn(ctx context.Context, req *pdu.PingReq) (*pdu.PingRes, error) {
-	defer trace.WithSpanFromStackUpdateCtx(&ctx)()
-
 	return p.Ping(ctx, req)
 }
 
 func (p *Sender) WaitForConnectivity(ctx context.Context) error {
-	defer trace.WithSpanFromStackUpdateCtx(&ctx)()
-
 	return nil
 }
 
 func (p *Sender) ReplicationCursor(ctx context.Context, req *pdu.ReplicationCursorReq) (*pdu.ReplicationCursorRes, error) {
-	defer trace.WithSpanFromStackUpdateCtx(&ctx)()
-
 	dp, err := p.filterCheckFS(req.Filesystem)
 	if err != nil {
 		return nil, err
@@ -529,15 +594,6 @@ func (f subroot) MapToLocal(fs string) (*zfs.DatasetPath, error) {
 }
 
 func (s *Receiver) ListFilesystems(ctx context.Context, req *pdu.ListFilesystemReq) (*pdu.ListFilesystemRes, error) {
-	defer trace.WithSpanFromStackUpdateCtx(&ctx)()
-
-	// first make sure that root_fs is imported
-	if rphs, err := zfs.ZFSGetFilesystemPlaceholderState(ctx, s.conf.RootWithoutClientComponent); err != nil {
-		return nil, errors.Wrap(err, "cannot determine whether root_fs exists")
-	} else if !rphs.FSExists {
-		return nil, errors.New("root_fs does not exist")
-	}
-
 	root := s.clientRootFromCtx(ctx)
 	filtered, err := zfs.ZFSListMapping(ctx, subroot{root})
 	if err != nil {
@@ -588,8 +644,6 @@ func (s *Receiver) ListFilesystems(ctx context.Context, req *pdu.ListFilesystemR
 }
 
 func (s *Receiver) ListFilesystemVersions(ctx context.Context, req *pdu.ListFilesystemVersionsReq) (*pdu.ListFilesystemVersionsRes, error) {
-	defer trace.WithSpanFromStackUpdateCtx(&ctx)()
-
 	root := s.clientRootFromCtx(ctx)
 	lp, err := subroot{root}.MapToLocal(req.GetFilesystem())
 	if err != nil {
@@ -611,8 +665,6 @@ func (s *Receiver) ListFilesystemVersions(ctx context.Context, req *pdu.ListFile
 }
 
 func (s *Receiver) Ping(ctx context.Context, req *pdu.PingReq) (*pdu.PingRes, error) {
-	defer trace.WithSpanFromStackUpdateCtx(&ctx)()
-
 	res := pdu.PingRes{
 		Echo: req.GetMessage(),
 	}
@@ -620,30 +672,24 @@ func (s *Receiver) Ping(ctx context.Context, req *pdu.PingReq) (*pdu.PingRes, er
 }
 
 func (s *Receiver) PingDataconn(ctx context.Context, req *pdu.PingReq) (*pdu.PingRes, error) {
-	defer trace.WithSpanFromStackUpdateCtx(&ctx)()
 	return s.Ping(ctx, req)
 }
 
 func (s *Receiver) WaitForConnectivity(ctx context.Context) error {
-	defer trace.WithSpanFromStackUpdateCtx(&ctx)()
 	return nil
 }
 
-func (s *Receiver) ReplicationCursor(ctx context.Context, _ *pdu.ReplicationCursorReq) (*pdu.ReplicationCursorRes, error) {
-	defer trace.WithSpanFromStackUpdateCtx(&ctx)()
+func (s *Receiver) ReplicationCursor(context.Context, *pdu.ReplicationCursorReq) (*pdu.ReplicationCursorRes, error) {
 	return nil, fmt.Errorf("ReplicationCursor not implemented for Receiver")
 }
 
 func (s *Receiver) Send(ctx context.Context, req *pdu.SendReq) (*pdu.SendRes, io.ReadCloser, error) {
-	defer trace.WithSpanFromStackUpdateCtx(&ctx)()
 	return nil, nil, fmt.Errorf("receiver does not implement Send()")
 }
 
 var maxConcurrentZFSRecvSemaphore = semaphore.New(envconst.Int64("ZREPL_ENDPOINT_MAX_CONCURRENT_RECV", 10))
 
 func (s *Receiver) Receive(ctx context.Context, req *pdu.ReceiveReq, receive io.ReadCloser) (*pdu.ReceiveRes, error) {
-	defer trace.WithSpanFromStackUpdateCtx(&ctx)()
-
 	getLogger(ctx).Debug("incoming Receive")
 	defer receive.Close()
 
@@ -766,9 +812,9 @@ func (s *Receiver) Receive(ctx context.Context, req *pdu.ReceiveReq, receive io.
 	}
 	defer guard.Release()
 
+	log.Info("peeking 1M ahead")
 	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")
 	if _, err := io.Copy(&peek, io.LimitReader(receive, MaxPeek)); err != nil {
 		log.WithError(err).Error("cannot read peek-buffer from send stream")
 	}
@@ -872,8 +918,6 @@ func (s *Receiver) Receive(ctx context.Context, req *pdu.ReceiveReq, receive io.
 }
 
 func (s *Receiver) DestroySnapshots(ctx context.Context, req *pdu.DestroySnapshotsReq) (*pdu.DestroySnapshotsRes, error) {
-	defer trace.WithSpanFromStackUpdateCtx(&ctx)()
-
 	root := s.clientRootFromCtx(ctx)
 	lp, err := subroot{root}.MapToLocal(req.Filesystem)
 	if err != nil {
@@ -882,9 +926,16 @@ func (s *Receiver) DestroySnapshots(ctx context.Context, req *pdu.DestroySnapsho
 	return doDestroySnapshots(ctx, lp, req.Snapshots)
 }
 
-func (p *Receiver) SendCompleted(ctx context.Context, _ *pdu.SendCompletedReq) (*pdu.SendCompletedRes, error) {
-	defer trace.WithSpanFromStackUpdateCtx(&ctx)()
+func (p *Receiver) HintMostRecentCommonAncestor(ctx context.Context, r *pdu.HintMostRecentCommonAncestorReq) (*pdu.HintMostRecentCommonAncestorRes, error) {
+	// we don't move last-received-hold as part of this hint
+	// because that wouldn't give us any benefit wrt resumability.
+	//
+	// Other reason: the replication logic that issues this RPC would require refactoring
+	// to include the receiver's FilesystemVersion in the request)
+	return &pdu.HintMostRecentCommonAncestorRes{}, nil
+}
 
+func (p *Receiver) SendCompleted(context.Context, *pdu.SendCompletedReq) (*pdu.SendCompletedRes, error) {
 	return &pdu.SendCompletedRes{}, nil
 }
 
@@ -906,7 +957,7 @@ func doDestroySnapshots(ctx context.Context, lp *zfs.DatasetPath, snaps []*pdu.F
 			ErrOut:     &errs[i],
 		}
 	}
-	zfs.ZFSDestroyFilesystemVersions(ctx, reqs)
+	zfs.ZFSDestroyFilesystemVersions(reqs)
 	for i := range reqs {
 		if errs[i] != nil {
 			if de, ok := errs[i].(*zfs.DestroySnapshotsError); ok && len(de.Reason) == 1 {

endpoint/endpoint_metrics.go

@@ -1,7 +0,0 @@
-package endpoint
-
-import "github.com/prometheus/client_golang/prometheus"
-
-func RegisterMetrics(r prometheus.Registerer) {
-	r.MustRegister(sendAbstractionsCacheMetrics.count)
-}

endpoint/endpoint_send_abstractions_cache.go

@@ -1,232 +0,0 @@
-package endpoint
-
-import (
-	"context"
-	"fmt"
-	"sync"
-
-	"github.com/prometheus/client_golang/prometheus"
-	"github.com/zrepl/zrepl/daemon/logging/trace"
-	"github.com/zrepl/zrepl/util/chainlock"
-)
-
-var sendAbstractionsCacheMetrics struct {
-	count prometheus.Gauge
-}
-
-func init() {
-	sendAbstractionsCacheMetrics.count = prometheus.NewGauge(prometheus.GaugeOpts{
-		Namespace: "zrepl",
-		Subsystem: "endpoint",
-		Name:      "send_abstractions_cache_entry_count",
-		Help:      "number of send abstractions tracked in the sendAbstractionsCache data structure",
-	})
-}
-
-var sendAbstractionsCacheSingleton = newSendAbstractionsCache()
-
-func SendAbstractionsCacheInvalidate(fs string) {
-	sendAbstractionsCacheSingleton.InvalidateFSCache(fs)
-}
-
-type sendAbstractionsCacheDidLoadFSState int
-
-const (
-	sendAbstractionsCacheDidLoadFSStateNo sendAbstractionsCacheDidLoadFSState = iota // 0-value has meaning
-	sendAbstractionsCacheDidLoadFSStateInProgress
-	sendAbstractionsCacheDidLoadFSStateDone
-)
-
-type sendAbstractionsCache struct {
-	mtx              chainlock.L
-	abstractions     []Abstraction
-	didLoadFS        map[string]sendAbstractionsCacheDidLoadFSState
-	didLoadFSChanged *sync.Cond
-}
-
-func newSendAbstractionsCache() *sendAbstractionsCache {
-	c := &sendAbstractionsCache{
-		didLoadFS: make(map[string]sendAbstractionsCacheDidLoadFSState),
-	}
-	c.didLoadFSChanged = c.mtx.NewCond()
-	return c
-}
-
-func (s *sendAbstractionsCache) Put(a Abstraction) {
-	defer s.mtx.Lock().Unlock()
-
-	var zeroJobId JobID
-	if a.GetJobID() == nil {
-		panic("abstraction must not have nil job id")
-	} else if *a.GetJobID() == zeroJobId {
-		panic(fmt.Sprintf("abstraction must not have zero-value job id: %s", a))
-	}
-
-	s.abstractions = append(s.abstractions, a)
-	sendAbstractionsCacheMetrics.count.Set(float64(len(s.abstractions)))
-}
-
-func (s *sendAbstractionsCache) InvalidateFSCache(fs string) {
-	// FIXME: O(n)
-	newAbs := make([]Abstraction, 0, len(s.abstractions))
-	for _, a := range s.abstractions {
-		if a.GetFS() != fs {
-			newAbs = append(newAbs, a)
-		}
-	}
-	s.abstractions = newAbs
-	sendAbstractionsCacheMetrics.count.Set(float64(len(s.abstractions)))
-
-	s.didLoadFS[fs] = sendAbstractionsCacheDidLoadFSStateNo
-	s.didLoadFSChanged.Broadcast()
-
-}
-
-func (s *sendAbstractionsCache) GetByFS(ctx context.Context, fs string) (ret []Abstraction) {
-	defer s.mtx.Lock().Unlock()
-	defer trace.WithSpanFromStackUpdateCtx(&ctx)()
-	if fs == "" {
-		panic("must not pass zero-value fs")
-	}
-
-	s.tryLoadOnDiskSendAbstractions(ctx, fs)
-
-	for _, a := range s.abstractions {
-		if a.GetFS() == fs {
-			ret = append(ret, a)
-		}
-	}
-
-	return ret
-}
-
-// - logs errors in getting on-disk abstractions
-// - only fetches on-disk abstractions once, but every time from the in-memory store
-//
-// That means that for precise results, all abstractions created by the endpoint must be .Put into this cache.
-func (s *sendAbstractionsCache) GetAndDeleteByJobIDAndFS(ctx context.Context, jobID JobID, fs string) (ret []Abstraction) {
-	defer s.mtx.Lock().Unlock()
-	defer trace.WithSpanFromStackUpdateCtx(&ctx)()
-	var zeroJobId JobID
-	if jobID == zeroJobId {
-		panic("must not pass zero-value job id")
-	}
-	if fs == "" {
-		panic("must not pass zero-value fs")
-	}
-
-	s.tryLoadOnDiskSendAbstractions(ctx, fs)
-
-	// FIXME O(n)
-	var remaining []Abstraction
-	for _, a := range s.abstractions {
-		aJobId := *a.GetJobID()
-		aFS := a.GetFS()
-		if aJobId == jobID && aFS == fs {
-			ret = append(ret, a)
-		} else {
-			remaining = append(remaining, a)
-		}
-	}
-	s.abstractions = remaining
-	sendAbstractionsCacheMetrics.count.Set(float64(len(s.abstractions)))
-
-	return ret
-}
-
-// caller must hold s.mtx
-func (s *sendAbstractionsCache) tryLoadOnDiskSendAbstractions(ctx context.Context, fs string) {
-	for s.didLoadFS[fs] != sendAbstractionsCacheDidLoadFSStateDone {
-		if s.didLoadFS[fs] == sendAbstractionsCacheDidLoadFSStateInProgress {
-			s.didLoadFSChanged.Wait()
-			continue
-		}
-		if s.didLoadFS[fs] != sendAbstractionsCacheDidLoadFSStateNo {
-			panic(fmt.Sprintf("unreachable: %v", s.didLoadFS[fs]))
-		}
-
-		s.didLoadFS[fs] = sendAbstractionsCacheDidLoadFSStateInProgress
-		defer s.didLoadFSChanged.Broadcast()
-
-		var onDiskAbs []Abstraction
-		var err error
-		s.mtx.DropWhile(func() {
-			onDiskAbs, err = s.tryLoadOnDiskSendAbstractionsImpl(ctx, fs) // no shadow
-		})
-
-		if err != nil {
-			s.didLoadFS[fs] = sendAbstractionsCacheDidLoadFSStateNo
-			getLogger(ctx).WithField("fs", fs).WithError(err).Error("cannot list send step abstractions for filesystem")
-		} else {
-			s.didLoadFS[fs] = sendAbstractionsCacheDidLoadFSStateDone
-			s.abstractions = append(s.abstractions, onDiskAbs...)
-			getLogger(ctx).WithField("fs", fs).WithField("abstractions", onDiskAbs).Debug("loaded step abstractions for filesystem")
-		}
-		return
-	}
-}
-
-// caller should _not hold s.mtx
-func (s *sendAbstractionsCache) tryLoadOnDiskSendAbstractionsImpl(ctx context.Context, fs string) ([]Abstraction, error) {
-	defer trace.WithSpanFromStackUpdateCtx(&ctx)()
-
-	q := ListZFSHoldsAndBookmarksQuery{
-		FS: ListZFSHoldsAndBookmarksQueryFilesystemFilter{
-			FS: &fs,
-		},
-		JobID: nil,
-		What: AbstractionTypeSet{
-			AbstractionStepHold:                    true,
-			AbstractionStepBookmark:                true,
-			AbstractionReplicationCursorBookmarkV2: true,
-		},
-		Concurrency: 1,
-	}
-	abs, absErrs, err := ListAbstractions(ctx, q)
-	if err != nil {
-		return nil, err
-	}
-	// safe to ignore absErrs here, this is best-effort cleanup
-	if len(absErrs) > 0 {
-		return nil, ListAbstractionsErrors(absErrs)
-	}
-	return abs, nil
-}
-
-func (s *sendAbstractionsCache) TryBatchDestroy(ctx context.Context, jobId JobID, fs string, keep func(a Abstraction) bool, check func(willDestroy []Abstraction)) {
-	// no s.mtx, we only use the public interface in this function
-
-	defer trace.WithSpanFromStackUpdateCtx(&ctx)()
-
-	allSendStepAbstractions := s.GetAndDeleteByJobIDAndFS(ctx, jobId, fs)
-
-	var obsoleteAbs []Abstraction
-	for _, a := range allSendStepAbstractions {
-		if !keep(a) {
-			obsoleteAbs = append(obsoleteAbs, a)
-		}
-	}
-
-	if check != nil {
-		check(obsoleteAbs)
-	}
-
-	hadErr := false
-	for res := range BatchDestroy(ctx, obsoleteAbs) {
-		if res.DestroyErr != nil {
-			hadErr = true
-			getLogger(ctx).
-				WithField("abstraction", res.Abstraction).
-				WithError(res.DestroyErr).
-				Error("cannot destroy stale send step abstraction")
-		} else {
-			getLogger(ctx).
-				WithField("abstraction", res.Abstraction).
-				Info("destroyed stale send step abstraction")
-		}
-	}
-	if hadErr {
-		s.InvalidateFSCache(fs)
-	}
-
-}

endpoint/endpoint_zfs_abstraction.go

@@ -11,7 +11,6 @@ import (
 
 	"github.com/pkg/errors"
 
-	"github.com/zrepl/zrepl/daemon/logging/trace"
 	"github.com/zrepl/zrepl/util/envconst"
 	"github.com/zrepl/zrepl/util/semaphore"
 	"github.com/zrepl/zrepl/zfs"
@@ -54,30 +53,6 @@ type Abstraction interface {
 	json.Marshaler
 }
 
-func AbstractionEquals(a, b Abstraction) bool {
-	if (a != nil) != (b != nil) {
-		return false
-	}
-	if a == nil && b == nil {
-		return true
-	}
-	var aJobId, bJobId JobID
-	if aJid := a.GetJobID(); aJid != nil {
-		aJobId = *aJid
-	}
-	if bJid := b.GetJobID(); bJid != nil {
-		bJobId = *bJid
-	}
-	return a.GetType() == b.GetType() &&
-		a.GetFS() == b.GetFS() &&
-		a.GetName() == b.GetName() &&
-		a.GetFullPath() == b.GetFullPath() &&
-		aJobId == bJobId &&
-		a.GetCreateTXG() == b.GetCreateTXG() &&
-		zfs.FilesystemVersionEqualIdentity(a.GetFilesystemVersion(), b.GetFilesystemVersion()) &&
-		a.String() == b.String()
-}
-
 func (t AbstractionType) Validate() error {
 	switch t {
 	case AbstractionStepBookmark:
@@ -492,7 +467,7 @@ func (e ListAbstractionsErrors) Error() string {
 	}
 	msgs := make([]string, len(e))
 	for i := range e {
-		msgs[i] = e[i].Error()
+		msgs[i] = e.Error()
 	}
 	return fmt.Sprintf("list endpoint abstractions: multiple errors:\n%s", strings.Join(msgs, "\n"))
 }
@@ -554,16 +529,15 @@ func ListAbstractionsStreamed(ctx context.Context, query ListZFSHoldsAndBookmark
 	}
 
 	sem := semaphore.New(int64(query.Concurrency))
-	ctx, endTask := trace.WithTask(ctx, "list-abstractions-streamed-producer")
 	go func() {
-		defer endTask()
 		defer close(out)
 		defer close(outErrs)
-
-		_, add, wait := trace.WithTaskGroup(ctx, "list-abstractions-impl-fs")
-		defer wait()
+		var wg sync.WaitGroup
+		defer wg.Wait()
 		for i := range fss {
-			add(func(ctx context.Context) {
+			wg.Add(1)
+			go func(i int) {
+				defer wg.Done()
 				g, err := sem.Acquire(ctx)
 				if err != nil {
 					errCb(err, fss[i], err.Error())
@@ -573,7 +547,7 @@ func ListAbstractionsStreamed(ctx context.Context, query ListZFSHoldsAndBookmark
 					defer g.Release()
 					listAbstractionsImplFS(ctx, fss[i], &query, emitAbstraction, errCb)
 				}()
-			})
+			}(i)
 		}
 	}()
 
@@ -614,23 +588,23 @@ func listAbstractionsImplFS(ctx context.Context, fs string, query *ListZFSHoldsA
 			panic("implementation error: extractors misconfigured for " + at)
 		}
 		for _, v := range fsvs {
+			var a Abstraction
 			if v.Type == zfs.Bookmark && bmE != nil {
-				if a := bmE(fsp, v); a != nil {
-					emitCandidate(a)
-				}
+				a = bmE(fsp, v)
 			}
-			if v.Type == zfs.Snapshot && holdE != nil && query.CreateTXG.Contains(v.GetCreateTXG()) && (!v.UserRefs.Valid || v.UserRefs.Value > 0) { // FIXME review v.UserRefsValid
+			if v.Type == zfs.Snapshot && holdE != nil && query.CreateTXG.Contains(v.GetCreateTXG()) && (!v.UserRefs.Valid || v.UserRefs.Value > 0) {
 				holds, err := zfs.ZFSHolds(ctx, fsp.ToString(), v.Name)
 				if err != nil {
 					errCb(err, v.ToAbsPath(fsp), "get hold on snap")
 					continue
 				}
 				for _, tag := range holds {
-					if a := holdE(fsp, v, tag); a != nil {
-						emitCandidate(a)
-					}
+					a = holdE(fsp, v, tag)
 				}
 			}
+			if a != nil {
+				emitCandidate(a)
+			}
 		}
 	}
 }
@@ -757,9 +731,6 @@ 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]
 

endpoint/endpoint_zfs_abstraction_cursor_and_last_received_hold.go

@@ -118,32 +118,31 @@ func GetReplicationCursors(ctx context.Context, dp *zfs.DatasetPath, jobID JobID
 	return candidates, nil
 }
 
-// idempotently create a replication cursor targeting `target`
+type ReplicationCursorTarget interface {
+	IsSnapshot() bool
+	GetGuid() uint64
+	GetCreateTXG() uint64
+	ToSendArgVersion() zfs.ZFSSendArgVersion
+}
+
+// `target` is validated before replication cursor is set. if validation fails, the cursor is not moved.
 //
 // returns ErrBookmarkCloningNotSupported if version is a bookmark and bookmarking bookmarks is not supported by ZFS
-func CreateReplicationCursor(ctx context.Context, fs string, target zfs.FilesystemVersion, jobID JobID) (a Abstraction, err error) {
+func MoveReplicationCursor(ctx context.Context, fs string, target ReplicationCursorTarget, jobID JobID) (destroyedCursors []Abstraction, err error) {
+
+	if !target.IsSnapshot() {
+		return nil, zfs.ErrBookmarkCloningNotSupported
+	}
 
 	bookmarkname, err := ReplicationCursorBookmarkName(fs, target.GetGuid(), jobID)
 	if err != nil {
 		return nil, errors.Wrap(err, "determine replication cursor name")
 	}
 
-	if target.IsBookmark() && target.GetName() == bookmarkname {
-		return &bookmarkBasedAbstraction{
-			Type:              AbstractionReplicationCursorBookmarkV2,
-			FS:                fs,
-			FilesystemVersion: target,
-			JobID:             jobID,
-		}, nil
-	}
+	// idempotently create bookmark (guid is encoded in it, hence we'll most likely add a new one
+	// cleanup the old one afterwards
 
-	if !target.IsSnapshot() {
-		return nil, zfs.ErrBookmarkCloningNotSupported
-	}
-
-	// idempotently create bookmark (guid is encoded in it)
-
-	cursorBookmark, err := zfs.ZFSBookmark(ctx, fs, target, bookmarkname)
+	err = zfs.ZFSBookmark(ctx, fs, target.ToSendArgVersion(), bookmarkname)
 	if err != nil {
 		if err == zfs.ErrBookmarkCloningNotSupported {
 			return nil, err // TODO go1.13 use wrapping
@@ -151,17 +150,65 @@ func CreateReplicationCursor(ctx context.Context, fs string, target zfs.Filesyst
 		return nil, errors.Wrapf(err, "cannot create bookmark")
 	}
 
-	return &bookmarkBasedAbstraction{
-		Type:              AbstractionReplicationCursorBookmarkV2,
-		FS:                fs,
-		FilesystemVersion: cursorBookmark,
-		JobID:             jobID,
-	}, nil
+	destroyedCursors, err = DestroyObsoleteReplicationCursors(ctx, fs, target, jobID)
+	if err != nil {
+		return nil, errors.Wrap(err, "destroy obsolete replication cursors")
+	}
+
+	return destroyedCursors, nil
 }
 
-const (
-	ReplicationCursorBookmarkNamePrefix = "zrepl_last_received_J_"
-)
+type ReplicationCursor interface {
+	GetCreateTXG() uint64
+}
+
+func DestroyObsoleteReplicationCursors(ctx context.Context, fs string, current ReplicationCursor, jobID JobID) (_ []Abstraction, err error) {
+
+	q := ListZFSHoldsAndBookmarksQuery{
+		FS: ListZFSHoldsAndBookmarksQueryFilesystemFilter{
+			FS: &fs,
+		},
+		What: AbstractionTypeSet{
+			AbstractionReplicationCursorBookmarkV2: true,
+		},
+		JobID: &jobID,
+		CreateTXG: CreateTXGRange{
+			Since: nil,
+			Until: &CreateTXGRangeBound{
+				CreateTXG: current.GetCreateTXG(),
+				Inclusive: &zfs.NilBool{B: false},
+			},
+		},
+		Concurrency: 1,
+	}
+	abs, absErr, err := ListAbstractions(ctx, q)
+	if err != nil {
+		return nil, errors.Wrap(err, "list abstractions")
+	}
+	if len(absErr) > 0 {
+		return nil, errors.Wrap(ListAbstractionsErrors(absErr), "list abstractions")
+	}
+
+	var destroyed []Abstraction
+	var errs []error
+	for res := range BatchDestroy(ctx, abs) {
+		log := getLogger(ctx).
+			WithField("replication_cursor_bookmark", res.Abstraction)
+		if res.DestroyErr != nil {
+			errs = append(errs, res.DestroyErr)
+			log.WithError(err).
+				Error("cannot destroy obsolete replication cursor bookmark")
+		} else {
+			destroyed = append(destroyed, res.Abstraction)
+			log.Info("destroyed obsolete replication cursor bookmark")
+		}
+	}
+	if len(errs) == 0 {
+		return destroyed, nil
+	} else {
+		return destroyed, errorarray.Wrap(errs, "destroy obsolete replication cursor")
+	}
+}
 
 var lastReceivedHoldTagRE = regexp.MustCompile("^zrepl_last_received_J_(.+)$")
 
@@ -183,22 +230,22 @@ func LastReceivedHoldTag(jobID JobID) (string, error) {
 }
 
 func lastReceivedHoldImpl(jobid string) (string, error) {
-	tag := fmt.Sprintf("%s%s", ReplicationCursorBookmarkNamePrefix, jobid)
+	tag := fmt.Sprintf("zrepl_last_received_J_%s", jobid)
 	if err := zfs.ValidHoldTag(tag); err != nil {
 		return "", err
 	}
 	return tag, nil
 }
 
-func CreateLastReceivedHold(ctx context.Context, fs string, to zfs.FilesystemVersion, jobID JobID) (Abstraction, error) {
+func MoveLastReceivedHold(ctx context.Context, fs string, to zfs.FilesystemVersion, jobID JobID) error {
 
 	if !to.IsSnapshot() {
-		return nil, errors.Errorf("last-received-hold: target must be a snapshot: %s", to.FullPath(fs))
+		return errors.Errorf("last-received-hold: target must be a snapshot: %s", to.FullPath(fs))
 	}
 
 	tag, err := LastReceivedHoldTag(jobID)
 	if err != nil {
-		return nil, errors.Wrap(err, "last-received-hold: hold tag")
+		return errors.Wrap(err, "last-received-hold: hold tag")
 	}
 
 	// we never want to be without a hold
@@ -206,23 +253,7 @@ func CreateLastReceivedHold(ctx context.Context, fs string, to zfs.FilesystemVer
 
 	err = zfs.ZFSHold(ctx, fs, to, tag)
 	if err != nil {
-		return nil, errors.Wrap(err, "last-received-hold: hold newly received")
-	}
-
-	return &holdBasedAbstraction{
-		Type:              AbstractionLastReceivedHold,
-		FS:                fs,
-		FilesystemVersion: to,
-		JobID:             jobID,
-		Tag:               tag,
-	}, nil
-}
-
-func MoveLastReceivedHold(ctx context.Context, fs string, to zfs.FilesystemVersion, jobID JobID) error {
-
-	_, err := CreateLastReceivedHold(ctx, fs, to, jobID)
-	if err != nil {
-		return err
+		return errors.Wrap(err, "last-received-hold: hold newly received")
 	}
 
 	q := ListZFSHoldsAndBookmarksQuery{

endpoint/endpoint_zfs_abstraction_pdu.go

@@ -1,28 +0,0 @@
-package endpoint
-
-import "github.com/zrepl/zrepl/replication/logic/pdu"
-
-func SendAbstractionToPDU(a Abstraction) *pdu.SendAbstraction {
-	var ty pdu.SendAbstraction_SendAbstractionType
-	switch a.GetType() {
-	case AbstractionLastReceivedHold:
-		panic(a)
-	case AbstractionReplicationCursorBookmarkV1:
-		panic(a)
-	case AbstractionReplicationCursorBookmarkV2:
-		ty = pdu.SendAbstraction_ReplicationCursorV2
-	case AbstractionStepHold:
-		ty = pdu.SendAbstraction_StepHold
-	case AbstractionStepBookmark:
-		ty = pdu.SendAbstraction_StepBookmark
-	default:
-		panic(a)
-	}
-
-	version := a.GetFilesystemVersion()
-	return &pdu.SendAbstraction{
-		Type:    ty,
-		JobID:   (*a.GetJobID()).String(),
-		Version: pdu.FilesystemVersionFromZFS(&version),
-	}
-}

endpoint/endpoint_zfs_abstraction_step.go

@@ -7,6 +7,7 @@ import (
 
 	"github.com/pkg/errors"
 
+	"github.com/zrepl/zrepl/util/errorarray"
 	"github.com/zrepl/zrepl/zfs"
 )
 
@@ -92,7 +93,7 @@ func HoldStep(ctx context.Context, fs string, v zfs.FilesystemVersion, jobID Job
 		return nil, errors.Wrap(err, "create step bookmark: determine bookmark name")
 	}
 	// idempotently create bookmark
-	stepBookmark, err := zfs.ZFSBookmark(ctx, fs, v, bmname)
+	err = zfs.ZFSBookmark(ctx, fs, v.ToSendArgVersion(), bmname)
 	if err != nil {
 		if err == zfs.ErrBookmarkCloningNotSupported {
 			// TODO we could actually try to find a local snapshot that has the requested GUID
@@ -107,11 +108,137 @@ func HoldStep(ctx context.Context, fs string, v zfs.FilesystemVersion, jobID Job
 	return &bookmarkBasedAbstraction{
 		Type:              AbstractionStepBookmark,
 		FS:                fs,
-		FilesystemVersion: stepBookmark,
+		FilesystemVersion: v,
 		JobID:             jobID,
 	}, nil
 }
 
+// idempotently release the step-hold on v if v is a snapshot
+// or idempotently destroy the step-bookmark  of v if v is a bookmark
+//
+// note that this operation leaves v itself untouched, unless v is the step-bookmark itself, in which case v is destroyed
+//
+// returns an instance of *zfs.DatasetDoesNotExist if `v` does not exist
+func ReleaseStep(ctx context.Context, fs string, v zfs.FilesystemVersion, jobID JobID) error {
+
+	if v.IsSnapshot() {
+		tag, err := StepHoldTag(jobID)
+		if err != nil {
+			return errors.Wrap(err, "step release tag")
+		}
+
+		if err := zfs.ZFSRelease(ctx, tag, v.FullPath(fs)); err != nil {
+			return errors.Wrap(err, "step release: zfs")
+		}
+
+		return nil
+	}
+	if !v.IsBookmark() {
+		panic(fmt.Sprintf("impl error: expecting version to be a bookmark, got %#v", v))
+	}
+
+	bmname, err := StepBookmarkName(fs, v.Guid, jobID)
+	if err != nil {
+		return errors.Wrap(err, "step release: determine bookmark name")
+	}
+	// idempotently destroy bookmark
+
+	if err := zfs.ZFSDestroyIdempotent(ctx, bmname); err != nil {
+		return errors.Wrap(err, "step release: bookmark destroy: zfs")
+	}
+
+	return nil
+}
+
+// release {step holds, step bookmarks} earlier and including `mostRecent`
+func ReleaseStepCummulativeInclusive(ctx context.Context, fs string, since *CreateTXGRangeBound, mostRecent zfs.FilesystemVersion, jobID JobID) error {
+	q := ListZFSHoldsAndBookmarksQuery{
+		What: AbstractionTypeSet{
+			AbstractionStepHold:     true,
+			AbstractionStepBookmark: true,
+		},
+		FS: ListZFSHoldsAndBookmarksQueryFilesystemFilter{
+			FS: &fs,
+		},
+		JobID: &jobID,
+		CreateTXG: CreateTXGRange{
+			Since: since,
+			Until: &CreateTXGRangeBound{
+				CreateTXG: mostRecent.CreateTXG,
+				Inclusive: &zfs.NilBool{B: true},
+			},
+		},
+		Concurrency: 1,
+	}
+	abs, absErrs, err := ListAbstractions(ctx, q)
+	if err != nil {
+		return errors.Wrap(err, "step release cummulative: list")
+	}
+	if len(absErrs) > 0 {
+		return errors.Wrap(ListAbstractionsErrors(absErrs), "step release cummulative: list")
+	}
+
+	getLogger(ctx).WithField("step_holds_and_bookmarks", fmt.Sprintf("%s", abs)).Debug("releasing step holds and bookmarks")
+
+	var errs []error
+	for res := range BatchDestroy(ctx, abs) {
+		log := getLogger(ctx).
+			WithField("step_hold_or_bookmark", res.Abstraction)
+		if res.DestroyErr != nil {
+			errs = append(errs, res.DestroyErr)
+			log.WithError(err).
+				Error("cannot release step hold or bookmark")
+		} else {
+			log.Info("released step hold or bookmark")
+		}
+	}
+	if len(errs) == 0 {
+		return nil
+	} else {
+		return errorarray.Wrap(errs, "step release cummulative: release")
+	}
+}
+
+func TryReleaseStepStaleFS(ctx context.Context, fs string, jobID JobID) {
+
+	q := ListZFSHoldsAndBookmarksQuery{
+		FS: ListZFSHoldsAndBookmarksQueryFilesystemFilter{
+			FS: &fs,
+		},
+		JobID: &jobID,
+		What: AbstractionTypeSet{
+			AbstractionStepHold:                    true,
+			AbstractionStepBookmark:                true,
+			AbstractionReplicationCursorBookmarkV2: true,
+		},
+		Concurrency: 1,
+	}
+	staleness, err := ListStale(ctx, q)
+	if _, ok := err.(*ListStaleQueryError); ok {
+		panic(err)
+	} else if err != nil {
+		getLogger(ctx).WithError(err).Error("cannot list stale step holds and bookmarks")
+		return
+	}
+	for _, s := range staleness.Stale {
+		getLogger(ctx).WithField("stale_step_hold_or_bookmark", s).Info("batch-destroying stale step hold or bookmark")
+	}
+	for res := range BatchDestroy(ctx, staleness.Stale) {
+		if res.DestroyErr != nil {
+			getLogger(ctx).
+				WithField("stale_step_hold_or_bookmark", res.Abstraction).
+				WithError(res.DestroyErr).
+				Error("cannot destroy stale step-hold or bookmark")
+		} else {
+			getLogger(ctx).
+				WithField("stale_step_hold_or_bookmark", res.Abstraction).
+				WithError(res.DestroyErr).
+				Info("destroyed stale step-hold or bookmark")
+		}
+	}
+
+}
+
 var _ BookmarkExtractor = StepBookmarkExtractor
 
 func StepBookmarkExtractor(fs *zfs.DatasetPath, v zfs.FilesystemVersion) (_ Abstraction) {

go.mod

@@ -5,7 +5,6 @@ go 1.12
 require (
 	github.com/fatih/color v1.7.0
 	github.com/gdamore/tcell v1.2.0
-	github.com/gitchander/permutation v0.0.0-20181107151852-9e56b92e9909
 	github.com/go-logfmt/logfmt v0.4.0
 	github.com/go-sql-driver/mysql v1.4.1-0.20190907122137-b2c03bcae3d4
 	github.com/golang/protobuf v1.3.2
@@ -24,19 +23,16 @@ require (
 	github.com/pkg/profile v1.2.1
 	github.com/problame/go-netssh v0.0.0-20191209123953-18d8aa6923c7
 	github.com/prometheus/client_golang v1.2.1
-	github.com/prometheus/common v0.7.0
 	github.com/sergi/go-diff v1.0.1-0.20180205163309-da645544ed44 // go1.12 thinks it needs this
 	github.com/spf13/cobra v0.0.2
 	github.com/spf13/pflag v1.0.5
 	github.com/stretchr/testify v1.4.0
-	github.com/willf/bitset v1.1.10
 	github.com/yudai/gojsondiff v0.0.0-20170107030110-7b1b7adf999d
 	github.com/yudai/golcs v0.0.0-20170316035057-ecda9a501e82 // go1.12 thinks it needs this
 	github.com/zrepl/yaml-config v0.0.0-20191220194647-cbb6b0cf4bdd
 	golang.org/x/net v0.0.0-20190613194153-d28f0bde5980
 	golang.org/x/sync v0.0.0-20190423024810-112230192c58
 	golang.org/x/sys v0.0.0-20191026070338-33540a1f6037
-	golang.org/x/tools v0.0.0-20190206041539-40960b6deb8e
 	gonum.org/v1/gonum v0.7.0 // indirect
 	google.golang.org/grpc v1.17.0
 )

go.sum

@@ -7,10 +7,8 @@ github.com/OpenPeeDeeP/depguard v0.0.0-20181229194401-1f388ab2d810/go.mod h1:7/4
 github.com/StackExchange/wmi v0.0.0-20180116203802-5d049714c4a6/go.mod h1:3eOhrUMpNV+6aFIbp5/iudMxNCF27Vw2OZgy4xEx0Fg=
 github.com/ajstarks/svgo v0.0.0-20180226025133-644b8db467af/go.mod h1:K08gAheRH3/J6wwsYMMT4xOr94bZjxIelGM0+d/wbFw=
 github.com/alecthomas/template v0.0.0-20160405071501-a0175ee3bccc/go.mod h1:LOuyumcjzFXgccqObfd/Ljyb9UuFJ6TxHnclSeseNhc=
-github.com/alecthomas/template v0.0.0-20190718012654-fb15b899a751 h1:JYp7IbQjafoB+tBA3gMyHYHrpOtNuDiK/uB5uXxq5wM=
 github.com/alecthomas/template v0.0.0-20190718012654-fb15b899a751/go.mod h1:LOuyumcjzFXgccqObfd/Ljyb9UuFJ6TxHnclSeseNhc=
 github.com/alecthomas/units v0.0.0-20151022065526-2efee857e7cf/go.mod h1:ybxpYRFXyAe+OPACYpWeL0wqObRcbAqCMya13uyzqw0=
-github.com/alecthomas/units v0.0.0-20190717042225-c3de453c63f4 h1:Hs82Z41s6SdL1CELW+XaDYmOH4hkBN4/N9og/AsOv7E=
 github.com/alecthomas/units v0.0.0-20190717042225-c3de453c63f4/go.mod h1:ybxpYRFXyAe+OPACYpWeL0wqObRcbAqCMya13uyzqw0=
 github.com/alvaroloes/enumer v1.1.1/go.mod h1:FxrjvuXoDAx9isTJrv4c+T410zFi0DtXIT0m65DJ+Wo=
 github.com/armon/consul-api v0.0.0-20180202201655-eb2c6b5be1b6/go.mod h1:grANhF5doyWs3UAsr3K4I6qtAmlQcZDesFNEHPZAzj8=
@@ -40,8 +38,6 @@ github.com/gdamore/encoding v1.0.0 h1:+7OoQ1Bc6eTm5niUzBa0Ctsh6JbMW6Ra+YNuAtDBdk
 github.com/gdamore/encoding v1.0.0/go.mod h1:alR0ol34c49FCSBLjhosxzcPHQbf2trDkoo5dl+VrEg=
 github.com/gdamore/tcell v1.2.0 h1:ikixzsxc8K8o3V2/CEmyoEW8mJZaNYQQ3NP3VIQdUe4=
 github.com/gdamore/tcell v1.2.0/go.mod h1:Hjvr+Ofd+gLglo7RYKxxnzCBmev3BzsS67MebKS4zMM=
-github.com/gitchander/permutation v0.0.0-20181107151852-9e56b92e9909 h1:9NC8seTx6/zRmMTAdsHj/uOMi0EGHGQtjyLafBjk77Q=
-github.com/gitchander/permutation v0.0.0-20181107151852-9e56b92e9909/go.mod h1:lP+DW8LR6Rw3ru9Vo2/y/3iiLaLWmofYql/va+7zJOk=
 github.com/go-critic/go-critic v0.3.4/go.mod h1:AHR42Lk/E/aOznsrYdMYeIQS5RH10HZHSqP+rD6AJrc=
 github.com/go-critic/go-critic v0.3.5-0.20190526074819-1df300866540/go.mod h1:+sE8vrLDS2M0pZkBk0wy6+nLdKexVDrl/jBqQOTDThA=
 github.com/go-kit/kit v0.8.0/go.mod h1:xBxKIO96dXMWWy0MnWVtmwkA9/13aqxPnvrjFYMA2as=
@@ -246,7 +242,6 @@ github.com/shurcooL/go-goon v0.0.0-20170922171312-37c2f522c041/go.mod h1:N5mDOms
 github.com/sirupsen/logrus v1.0.5/go.mod h1:pMByvHTf9Beacp5x1UXfOR9xyW/9antXMhjMPG0dEzc=
 github.com/sirupsen/logrus v1.2.0/go.mod h1:LxeOpSwHxABJmUn/MG1IvRgCAasNZTLOkJPxbbu5VWo=
 github.com/sirupsen/logrus v1.4.0/go.mod h1:LxeOpSwHxABJmUn/MG1IvRgCAasNZTLOkJPxbbu5VWo=
-github.com/sirupsen/logrus v1.4.2 h1:SPIRibHv4MatM3XXNO2BJeFLZwZ2LvZgfQ5+UNI2im4=
 github.com/sirupsen/logrus v1.4.2/go.mod h1:tLMulIdttU9McNUspp0xgXVQah82FyeX6MwdIuYE2rE=
 github.com/sourcegraph/go-diff v0.5.1/go.mod h1:j2dHj3m8aZgQO8lMTcTnBcXkRRRqi34cd2MNlA9u1mE=
 github.com/spf13/afero v1.1.0/go.mod h1:j4pytiNVoe2o6bmDsKpLACNPDBIoEAkihy7loJ1B0CQ=
@@ -268,7 +263,6 @@ 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=
@@ -282,8 +276,6 @@ github.com/valyala/bytebufferpool v1.0.0/go.mod h1:6bBcMArwyJ5K/AmCkWv1jt77kVWyC
 github.com/valyala/fasthttp v1.2.0/go.mod h1:4vX61m6KN+xDduDNwXrhIAVZaZaZiQ1luJk8LWSxF3s=
 github.com/valyala/quicktemplate v1.1.1/go.mod h1:EH+4AkTd43SvgIbQHYu59/cJyxDoOVRUAfrukLPuGJ4=
 github.com/valyala/tcplisten v0.0.0-20161114210144-ceec8f93295a/go.mod h1:v3UYOV9WzVtRmSR+PDvWpU/qWl4Wa5LApYYX4ZtKbio=
-github.com/willf/bitset v1.1.10 h1:NotGKqX0KwQ72NUzqrjZq5ipPNDQex9lo3WpaS8L2sc=
-github.com/willf/bitset v1.1.10/go.mod h1:RjeCKbqT1RxIR/KWY6phxZiaY1IyutSBfGjNPySAYV4=
 github.com/xordataexchange/crypt v0.0.3-0.20170626215501-b2862e3d0a77/go.mod h1:aYKd//L2LvnjZzWKhF00oedf4jCCReLcmhLdhm1A27Q=
 github.com/yudai/gojsondiff v0.0.0-20170107030110-7b1b7adf999d h1:yJIizrfO599ot2kQ6Af1enICnwBD3XoxgX3MrMwot2M=
 github.com/yudai/gojsondiff v0.0.0-20170107030110-7b1b7adf999d/go.mod h1:AY32+k2cwILAkW1fbgxQ5mUmMiZFgLIV+FBNExI05xg=
@@ -351,7 +343,6 @@ golang.org/x/tools v0.0.0-20181117154741-2ddaf7f79a09/go.mod h1:n7NCudcB/nEzxVGm
 golang.org/x/tools v0.0.0-20181205014116-22934f0fdb62/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
 golang.org/x/tools v0.0.0-20190110163146-51295c7ec13a/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
 golang.org/x/tools v0.0.0-20190121143147-24cd39ecf745/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
-golang.org/x/tools v0.0.0-20190206041539-40960b6deb8e h1:Io7mpb+aUAGF0MKxbyQ7HQl1VgB+cL6ZJZUFaFNqVV4=
 golang.org/x/tools v0.0.0-20190206041539-40960b6deb8e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
 golang.org/x/tools v0.0.0-20190213192042-740235f6c0d8/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
 golang.org/x/tools v0.0.0-20190311212946-11955173bddd/go.mod h1:LCzVGOaR6xXOjkQ3onu1FJEFr0SW1gC7cKk1uF8kGRs=
@@ -374,7 +365,6 @@ google.golang.org/grpc v1.16.0/go.mod h1:0JHn/cJsOMiMfNA9+DeHDlAU7KAAB5GDlYFpa9M
 google.golang.org/grpc v1.17.0 h1:TRJYBgMclJvGYn2rIMjj+h9KtMt5r1Ij7ODVRIZkwhk=
 google.golang.org/grpc v1.17.0/go.mod h1:6QZJwpn2B+Zp71q/5VxRsJ6NXXVCE5NRUHRo+f3cWCs=
 gopkg.in/airbrake/gobrake.v2 v2.0.9/go.mod h1:/h5ZAUhDkGaJfjzjKLSjv6zCL6O0LLBxU4K+aSYdM/U=
-gopkg.in/alecthomas/kingpin.v2 v2.2.6 h1:jMFz6MfLP0/4fUyZle81rXUoxOBFi19VUFKVDOQfozc=
 gopkg.in/alecthomas/kingpin.v2 v2.2.6/go.mod h1:FMv+mEhP44yOT+4EoQTLFTRgOQ1FBLkstjWtayDeSgw=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=

lazy.sh

@@ -36,10 +36,9 @@ godep() {
     go build -v -mod=readonly -o "$GOPATH/bin/enumer"        github.com/alvaroloes/enumer
     go build -v -mod=readonly -o "$GOPATH/bin/goimports"     golang.org/x/tools/cmd/goimports
     go build -v -mod=readonly -o "$GOPATH/bin/golangci-lint" github.com/golangci/golangci-lint/cmd/golangci-lint
-    go build -v -mod=readonly -o "$GOPATH/bin/gocovmerge"    github.com/wadey/gocovmerge
     set +x
     popd
-    if ! type stringer || ! type protoc-gen-go || ! type enumer || ! type goimports || ! type golangci-lint || ! type gocovmerge; then
+    if ! type stringer || ! type protoc-gen-go || ! type enumer || ! type goimports || ! type golangci-lint; then
         echo "Installed dependencies but can't find them in \$PATH, adjust it to contain \$GOPATH/bin" 1>&2
         exit 1
     fi

platformtest/harness/harness.go

@@ -11,7 +11,6 @@ import (
 
 	"github.com/fatih/color"
 	"github.com/pkg/errors"
-	"github.com/zrepl/zrepl/daemon/logging/trace"
 
 	"github.com/zrepl/zrepl/config"
 	"github.com/zrepl/zrepl/daemon/logging"
@@ -24,36 +23,33 @@ var bold = color.New(color.Bold)
 var boldRed = color.New(color.Bold, color.FgHiRed)
 var boldGreen = color.New(color.Bold, color.FgHiGreen)
 
-const DefaultPoolImageSize = 200 * (1 << 20)
+var args struct {
+	createArgs            platformtest.ZpoolCreateArgs
+	stopAndKeepPoolOnFail bool
+
+	run   string
+	runRE *regexp.Regexp
+}
 
 func main() {
-
-	var args HarnessArgs
-
-	flag.StringVar(&args.CreateArgs.PoolName, "poolname", "", "")
-	flag.StringVar(&args.CreateArgs.ImagePath, "imagepath", "", "")
-	flag.Int64Var(&args.CreateArgs.ImageSize, "imagesize", DefaultPoolImageSize, "")
-	flag.StringVar(&args.CreateArgs.Mountpoint, "mountpoint", "", "")
-	flag.BoolVar(&args.StopAndKeepPoolOnFail, "failure.stop-and-keep-pool", false, "if a test case fails, stop test execution and keep pool as it was when the test failed")
-	flag.StringVar(&args.Run, "run", "", "")
-	flag.Parse()
-
-	if err := HarnessRun(args); err != nil {
+	if err := doMain(); err != nil {
 		os.Exit(1)
 	}
 }
 
 var exitWithErr = fmt.Errorf("exit with error")
 
-type HarnessArgs struct {
-	CreateArgs            platformtest.ZpoolCreateArgs
-	StopAndKeepPoolOnFail bool
-	Run                   string
-}
+func doMain() error {
 
-func HarnessRun(args HarnessArgs) error {
+	flag.StringVar(&args.createArgs.PoolName, "poolname", "", "")
+	flag.StringVar(&args.createArgs.ImagePath, "imagepath", "", "")
+	flag.Int64Var(&args.createArgs.ImageSize, "imagesize", 200*(1<<20), "")
+	flag.StringVar(&args.createArgs.Mountpoint, "mountpoint", "", "")
+	flag.BoolVar(&args.stopAndKeepPoolOnFail, "failure.stop-and-keep-pool", false, "if a test case fails, stop test execution and keep pool as it was when the test failed")
+	flag.StringVar(&args.run, "run", "", "")
+	flag.Parse()
 
-	runRE := regexp.MustCompile(args.Run)
+	args.runRE = regexp.MustCompile(args.run)
 
 	outlets := logger.NewOutlets()
 	outlet, level, err := logging.ParseOutlet(config.LoggingOutletEnum{Ret: &config.StdoutLoggingOutlet{
@@ -68,13 +64,11 @@ func HarnessRun(args HarnessArgs) error {
 	outlets.Add(outlet, level)
 	logger := logger.NewLogger(outlets, 1*time.Second)
 
-	if err := args.CreateArgs.Validate(); err != nil {
+	if err := args.createArgs.Validate(); err != nil {
 		logger.Error(err.Error())
 		panic(err)
 	}
-	ctx := context.Background()
-	defer trace.WithTaskFromStackUpdateCtx(&ctx)()
-	ctx = logging.WithLoggers(ctx, logging.SubsystemLoggersWithUniversalLogger(logger))
+	ctx := platformtest.WithLogger(context.Background(), logger)
 	ex := platformtest.NewEx(logger)
 
 	type invocation struct {
@@ -84,7 +78,7 @@ func HarnessRun(args HarnessArgs) error {
 
 	invocations := make([]*invocation, 0, len(tests.Cases))
 	for _, c := range tests.Cases {
-		if runRE.MatchString(c.String()) {
+		if args.runRE.MatchString(c.String()) {
 			invocations = append(invocations, &invocation{runFunc: c})
 		}
 	}
@@ -93,7 +87,7 @@ func HarnessRun(args HarnessArgs) error {
 
 		bold.Printf("BEGIN TEST CASE %s\n", inv.runFunc.String())
 
-		pool, err := platformtest.CreateOrReplaceZpool(ctx, ex, args.CreateArgs)
+		pool, err := platformtest.CreateOrReplaceZpool(ctx, ex, args.createArgs)
 		if err != nil {
 			panic(errors.Wrap(err, "create test pool"))
 		}
@@ -109,7 +103,7 @@ func HarnessRun(args HarnessArgs) error {
 			fmt.Printf("%+v\n", res.failedStack) // print with stack trace
 		}
 
-		if res.failed && args.StopAndKeepPoolOnFail {
+		if res.failed && args.stopAndKeepPoolOnFail {
 			boldRed.Printf("STOPPING TEST RUN AT FAILING TEST PER USER REQUEST\n")
 			return exitWithErr
 		}

platformtest/harness/harness_test.go

@@ -1,7 +1,6 @@
 package main
 
 import (
-	"fmt"
 	"os"
 	"strings"
 	"testing"
@@ -22,29 +21,22 @@ import (
 //  https://github.com/wadey/gocovmerge
 
 func TestMain(t *testing.T) {
-	fmt.Println("incoming args: ", os.Args)
-
 	var (
-		args             []string
-		run              bool
-		startCaptureArgs bool
+		args []string
+		run  bool
 	)
 
-	for i, arg := range os.Args {
+	for _, arg := range os.Args {
 		switch {
 		case arg == "__DEVEL--i-heard-you-like-tests":
 			run = true
-			startCaptureArgs = true
 		case strings.HasPrefix(arg, "-test"):
 		case strings.HasPrefix(arg, "__DEVEL"):
-		case i == 0:
-			args = append(args, arg)
-		case startCaptureArgs:
+		default:
 			args = append(args, arg)
 		}
 	}
 	os.Args = args
-	fmt.Println("using args: ", os.Args)
 
 	if run {
 		main()

platformtest/platformtest.go

@@ -27,7 +27,6 @@ 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/platformtest_logging.go

@@ -3,12 +3,22 @@ package platformtest
 import (
 	"context"
 
-	"github.com/zrepl/zrepl/daemon/logging"
 	"github.com/zrepl/zrepl/logger"
 )
 
 type Logger = logger.Logger
 
-func GetLog(ctx context.Context) Logger {
-	return logging.GetLogger(ctx, logging.SubsysPlatformtest)
+type contextKey int
+
+const (
+	contextKeyLogger contextKey = iota
+)
+
+func WithLogger(ctx context.Context, logger Logger) context.Context {
+	ctx = context.WithValue(ctx, contextKeyLogger, logger)
+	return ctx
+}
+
+func GetLog(ctx context.Context) Logger {
+	return ctx.Value(contextKeyLogger).(Logger)
 }

platformtest/platformtest_zpool.go

@@ -58,14 +58,6 @@ func CreateOrReplaceZpool(ctx context.Context, e Execer, args ZpoolCreateArgs) (
 		}
 	}
 
-	// clear the mountpoint dir
-	if err := os.RemoveAll(args.Mountpoint); err != nil {
-		return nil, errors.Wrapf(err, "remove mountpoint dir %q", args.Mountpoint)
-	}
-	if err := os.Mkdir(args.Mountpoint, 0700); err != nil {
-		return nil, errors.Wrapf(err, "create mountpoint dir %q", args.Mountpoint)
-	}
-
 	// idempotently (re)create the pool image
 	image, err := os.OpenFile(args.ImagePath, os.O_CREATE|os.O_RDWR, 0600)
 	if err != nil {
@@ -101,9 +93,5 @@ func (p *Zpool) Destroy(ctx context.Context, e Execer) error {
 		return errors.Wrapf(err, "remove pool image")
 	}
 
-	if err := os.RemoveAll(p.args.Mountpoint); err != nil {
-		return errors.Wrapf(err, "remove mountpoint dir %q", p.args.Mountpoint)
-	}
-
 	return nil
 }

platformtest/tests/batchDestroy.go

@@ -2,9 +2,8 @@ package tests
 
 import (
 	"fmt"
+	"strings"
 
-	"github.com/kr/pretty"
-	"github.com/stretchr/testify/require"
 	"github.com/zrepl/zrepl/platformtest"
 	"github.com/zrepl/zrepl/zfs"
 )
@@ -18,9 +17,7 @@ func BatchDestroy(ctx *platformtest.Context) {
 		+  "foo bar@1"
 		+  "foo bar@2"
 		+  "foo bar@3"
-		+  "foo bar@4"
 		R  zfs hold zrepl_platformtest "${ROOTDS}/foo bar@2"
-		R  zfs hold zrepl_platformtest "${ROOTDS}/foo bar@4"
 	`)
 
 	reqs := []*zfs.DestroySnapOp{
@@ -34,40 +31,24 @@ func BatchDestroy(ctx *platformtest.Context) {
 			Filesystem: fmt.Sprintf("%s/foo bar", ctx.RootDataset),
 			Name:       "2",
 		},
-		&zfs.DestroySnapOp{
-			ErrOut:     new(error),
-			Filesystem: fmt.Sprintf("%s/foo bar", ctx.RootDataset),
-			Name:       "non existent",
-		},
-		&zfs.DestroySnapOp{
-			ErrOut:     new(error),
-			Filesystem: fmt.Sprintf("%s/foo bar", ctx.RootDataset),
-			Name:       "4",
-		},
 	}
-	zfs.ZFSDestroyFilesystemVersions(ctx, reqs)
-
-	pretty.Println(reqs)
-
+	zfs.ZFSDestroyFilesystemVersions(reqs)
 	if *reqs[0].ErrOut != nil {
 		panic("expecting no error")
 	}
-
-	eBusy, ok := (*reqs[1].ErrOut).(*zfs.ErrDestroySnapshotDatasetIsBusy)
-	require.True(ctx, ok)
-	require.Equal(ctx, reqs[1].Name, eBusy.Name)
-
-	require.Nil(ctx, *reqs[2].ErrOut, "destroying non-existent snap is not an error (idempotence)")
-
-	eBusy, ok = (*reqs[3].ErrOut).(*zfs.ErrDestroySnapshotDatasetIsBusy)
-	require.True(ctx, ok)
-	require.Equal(ctx, reqs[3].Name, eBusy.Name)
+	err := (*reqs[1].ErrOut).Error()
+	if !strings.Contains(err, fmt.Sprintf("%s/foo bar@2", ctx.RootDataset)) {
+		panic(fmt.Sprintf("expecting error about being unable to destroy @2: %T\n%s", err, err))
+	}
 
 	platformtest.Run(ctx, platformtest.PanicErr, ctx.RootDataset, `
 	!N "foo bar@3"
 	!E "foo bar@1"
 	!E "foo bar@2"
-	!E "foo bar@4"
+	R zfs release zrepl_platformtest "${ROOTDS}/foo bar@2"
+	-  "foo bar@2"
+	-  "foo bar@1"
+	-  "foo bar"
 	`)
 
 }

platformtest/tests/gen/gen.go

@@ -1,140 +0,0 @@
-package main
-
-import (
-	"bytes"
-	"go/ast"
-	"go/format"
-	"go/parser"
-	"go/token"
-	"io/ioutil"
-	"os"
-	"sort"
-	"strings"
-	"text/template"
-
-	"golang.org/x/tools/go/packages"
-)
-
-func check(err error) {
-	if err != nil {
-		panic(err)
-	}
-}
-
-type platformtestFuncDeclFinder struct {
-	pkg       *packages.Package
-	testFuncs []*ast.FuncDecl
-}
-
-func isPlatformtestFunc(n *ast.FuncDecl) bool {
-	if !n.Name.IsExported() {
-		return false
-	}
-	if n.Recv != nil {
-		return false
-	}
-	if n.Type.Results.NumFields() != 0 {
-		return false
-	}
-	if n.Type.Params.NumFields() != 1 {
-		return false
-	}
-	se, ok := n.Type.Params.List[0].Type.(*ast.StarExpr)
-	if !ok {
-		return false
-	}
-	sel, ok := se.X.(*ast.SelectorExpr)
-	if !ok {
-		return false
-	}
-	x, ok := sel.X.(*ast.Ident)
-	if !ok {
-		return false
-	}
-	if x.Name != "platformtest" || sel.Sel.Name != "Context" {
-		return false
-	}
-	return true
-}
-
-func (e *platformtestFuncDeclFinder) Visit(n2 ast.Node) ast.Visitor {
-	switch n := n2.(type) {
-	case *ast.File:
-		return e
-	case *ast.FuncDecl:
-		if isPlatformtestFunc(n) {
-			e.testFuncs = append(e.testFuncs, n)
-		}
-		return nil
-	default:
-		return nil
-	}
-}
-
-func main() {
-
-	// TODO safeguards that prevent us from deleting non-generated generated_cases.go
-	os.Remove("generated_cases.go")
-	// (no error handling to easily cover the case where the file doesn't exist)
-
-	pkgs, err := packages.Load(
-		&packages.Config{
-			Mode:  packages.LoadFiles,
-			Tests: false,
-		},
-		os.Args[1],
-	)
-	check(err)
-
-	if len(pkgs) != 1 {
-		panic(pkgs)
-	}
-
-	p := pkgs[0]
-
-	var tests []*ast.FuncDecl
-
-	for _, f := range p.GoFiles {
-		s := token.NewFileSet()
-		a, err := parser.ParseFile(s, f, nil, parser.AllErrors)
-		check(err)
-		finder := &platformtestFuncDeclFinder{
-			pkg: p,
-		}
-		ast.Walk(finder, a)
-		tests = append(tests, finder.testFuncs...)
-	}
-
-	sort.Slice(tests, func(i, j int) bool {
-		return strings.Compare(tests[i].Name.Name, tests[j].Name.Name) < 0
-	})
-
-	{
-		casesTemplate := `
-// Code generated by zrepl tooling; DO NOT EDIT.
-
-package tests
-
-var Cases = []Case {
-{{- range . -}}
-	{{ .Name }},
-{{ end -}}
-}
-
-	`
-		t, err := template.New("CaseFunc").Parse(casesTemplate)
-		check(err)
-
-		var buf bytes.Buffer
-		err = t.Execute(&buf, tests)
-		check(err)
-
-		formatted, err := format.Source(buf.Bytes())
-		check(err)
-
-		err = ioutil.WriteFile("generated_cases.go", formatted, 0664)
-		check(err)
-
-	}
-
-}

platformtest/tests/generated_cases.go

@@ -1,30 +0,0 @@
-// Code generated by zrepl tooling; DO NOT EDIT.
-
-package tests
-
-var Cases = []Case{BatchDestroy,
-	CreateReplicationCursor,
-	GetNonexistent,
-	HoldsWork,
-	IdempotentBookmark,
-	IdempotentDestroy,
-	IdempotentHold,
-	ListFilesystemVersionsFilesystemNotExist,
-	ListFilesystemVersionsTypeFilteringAndPrefix,
-	ListFilesystemVersionsUserrefs,
-	ListFilesystemVersionsZeroExistIsNotAnError,
-	ListFilesystemsNoFilter,
-	Pruner2NotReplicated,
-	ReceiveForceIntoEncryptedErr,
-	ReceiveForceRollbackWorksUnencrypted,
-	ReplicationIncrementalCleansUpStaleAbstractionsWithCacheOnSecondReplication,
-	ReplicationIncrementalCleansUpStaleAbstractionsWithoutCacheOnSecondReplication,
-	ReplicationIncrementalIsPossibleIfCommonSnapshotIsDestroyed,
-	ReplicationIsResumableFullSend,
-	ResumableRecvAndTokenHandling,
-	ResumeTokenParsing,
-	SendArgsValidationEncryptedSendOfUnencryptedDatasetForbidden,
-	SendArgsValidationResumeTokenDifferentFilesystemForbidden,
-	SendArgsValidationResumeTokenEncryptionMismatchForbidden,
-	UndestroyableSnapshotParsing,
-}

platformtest/tests/holds.go

@@ -1,32 +0,0 @@
-package tests
-
-import (
-	"path"
-
-	"github.com/stretchr/testify/require"
-	"github.com/zrepl/zrepl/platformtest"
-	"github.com/zrepl/zrepl/zfs"
-)
-
-func HoldsWork(ctx *platformtest.Context) {
-	platformtest.Run(ctx, platformtest.PanicErr, ctx.RootDataset, `
-		DESTROYROOT
-		CREATEROOT
-		+  "foo bar"
-		+  "foo bar@snap name"
-	`)
-
-	fs := path.Join(ctx.RootDataset, "foo bar")
-
-	err := zfs.ZFSHold(ctx, fs, fsversion(ctx, fs, "@snap name"), "tag 1")
-	require.NoError(ctx, err)
-
-	err = zfs.ZFSHold(ctx, fs, fsversion(ctx, fs, "@snap name"), "tag 2")
-	require.NoError(ctx, err)
-
-	holds, err := zfs.ZFSHolds(ctx, fs, "snap name")
-	require.NoError(ctx, err)
-	require.Len(ctx, holds, 2)
-	require.Contains(ctx, holds, "tag 1")
-	require.Contains(ctx, holds, "tag 2")
-}

platformtest/tests/idempotentBookmark.go

@@ -3,7 +3,6 @@ package tests
 import (
 	"fmt"
 
-	"github.com/stretchr/testify/assert"
 	"github.com/zrepl/zrepl/platformtest"
 	"github.com/zrepl/zrepl/zfs"
 )
@@ -20,23 +19,22 @@ func IdempotentBookmark(ctx *platformtest.Context) {
 
 	fs := fmt.Sprintf("%s/foo bar", ctx.RootDataset)
 
-	asnap := fsversion(ctx, fs, "@a snap")
-	anotherSnap := fsversion(ctx, fs, "@another snap")
+	asnap := sendArgVersion(ctx, fs, "@a snap")
+	anotherSnap := sendArgVersion(ctx, fs, "@another snap")
 
-	aBookmark, err := zfs.ZFSBookmark(ctx, fs, asnap, "a bookmark")
+	err := zfs.ZFSBookmark(ctx, fs, asnap, "a bookmark")
 	if err != nil {
 		panic(err)
 	}
 
 	// do it again, should be idempotent
-	aBookmarkIdemp, err := zfs.ZFSBookmark(ctx, fs, asnap, "a bookmark")
+	err = zfs.ZFSBookmark(ctx, fs, asnap, "a bookmark")
 	if err != nil {
 		panic(err)
 	}
-	assert.Equal(ctx, aBookmark, aBookmarkIdemp)
 
 	// should fail for another snapshot
-	_, err = zfs.ZFSBookmark(ctx, fs, anotherSnap, "a bookmark")
+	err = zfs.ZFSBookmark(ctx, fs, anotherSnap, "a bookmark")
 	if err == nil {
 		panic(err)
 	}
@@ -50,7 +48,7 @@ func IdempotentBookmark(ctx *platformtest.Context) {
 	}
 
 	// do it again, should fail with special error type
-	_, err = zfs.ZFSBookmark(ctx, fs, asnap, "a bookmark")
+	err = zfs.ZFSBookmark(ctx, fs, asnap, "a bookmark")
 	if err == nil {
 		panic(err)
 	}

platformtest/tests/idempotentDestroy.go

@@ -18,8 +18,8 @@ func IdempotentDestroy(ctx *platformtest.Context) {
 	`)
 
 	fs := fmt.Sprintf("%s/foo bar", ctx.RootDataset)
-	asnap := fsversion(ctx, fs, "@a snap")
-	_, err := zfs.ZFSBookmark(ctx, fs, asnap, "a bookmark")
+	asnap := sendArgVersion(ctx, fs, "@a snap")
+	err := zfs.ZFSBookmark(ctx, fs, asnap, "a bookmark")
 	if err != nil {
 		panic(err)
 	}

platformtest/tests/listFilesystemVersions.go

@@ -85,6 +85,9 @@ func ListFilesystemVersionsZeroExistIsNotAnError(t *platformtest.Context) {
 	vs, err := zfs.ZFSListFilesystemVersions(t, mustDatasetPath(fs), zfs.ListFilesystemVersionsOptions{})
 	require.Empty(t, vs)
 	require.NoError(t, err)
+	dsne, ok := err.(*zfs.DatasetDoesNotExist)
+	require.True(t, ok)
+	require.Equal(t, fs, dsne.Path)
 }
 
 func ListFilesystemVersionsFilesystemNotExist(t *platformtest.Context) {

platformtest/tests/pruner.go.deact

@@ -1,331 +0,0 @@
-package tests
-
-import (
-	"encoding/json"
-	"fmt"
-	"time"
-
-	"github.com/prometheus/client_golang/prometheus"
-	"github.com/stretchr/testify/require"
-	"github.com/zrepl/zrepl/config"
-	"github.com/zrepl/zrepl/daemon/filters"
-	"github.com/zrepl/zrepl/daemon/pruner"
-	"github.com/zrepl/zrepl/endpoint"
-	"github.com/zrepl/zrepl/platformtest"
-	"github.com/zrepl/zrepl/zfs"
-)
-
-func PrunerNotReplicated(ctx *platformtest.Context) {
-
-	platformtest.Run(ctx, platformtest.PanicErr, ctx.RootDataset, `
-        DESTROYROOT
-        CREATEROOT
-        +  "foo bar"
-        +  "foo bar@1"
-        +  "foo bar@2"
-        +  "foo bar@3"
-        +  "foo bar@4"
-        +  "foo bar@5"
-    `)
-
-	c, err := config.ParseConfigBytes([]byte(fmt.Sprintf(`
-jobs:
-- name: prunetest
-  type: push
-  filesystems: {
-      "%s/foo bar<": true
-  }
-  connect:
-    type: tcp
-    address: 255.255.255.255:255
-  snapshotting:
-    type: manual
-  pruning:
-    keep_sender:
-    - type: not_replicated
-    - type: last_n
-      count: 1
-    keep_receiver:
-    - type: last_n
-      count: 2
-    `, ctx.RootDataset)))
-	require.NoError(ctx, err)
-
-	pushJob := c.Jobs[0].Ret.(*config.PushJob)
-
-	dummyHistVec := prometheus.NewHistogramVec(prometheus.HistogramOpts{
-		Namespace: "foo",
-		Subsystem: "foo",
-		Name:      "foo",
-		Help:      "foo",
-	}, []string{"foo"})
-
-	prunerFactory, err := pruner.NewPrunerFactory(pushJob.Pruning, dummyHistVec)
-	require.NoError(ctx, err)
-
-	senderJid := endpoint.MustMakeJobID("sender-job")
-
-	fsfilter, err := filters.DatasetMapFilterFromConfig(pushJob.Filesystems)
-	require.NoError(ctx, err)
-
-	sender := endpoint.NewSender(endpoint.SenderConfig{
-		FSF: fsfilter,
-		Encrypt: &zfs.NilBool{
-			B: false,
-		},
-		JobID: senderJid,
-	})
-
-	fs := ctx.RootDataset + "/foo bar"
-
-	// create a replication cursor to make pruning work at all
-	_, err = endpoint.CreateReplicationCursor(ctx, fs, fsversion(ctx, fs, "@2"), senderJid)
-	require.NoError(ctx, err)
-
-	p := prunerFactory.BuildSenderPruner(ctx, sender, sender)
-
-	p.Prune()
-
-	report := p.Report()
-
-	reportJSON, err := json.MarshalIndent(report, "", "  ")
-	require.NoError(ctx, err)
-	ctx.Logf("%s\n", string(reportJSON))
-
-	require.Equal(ctx, pruner.Done.String(), report.State)
-	require.Len(ctx, report.Completed, 1)
-	fsReport := report.Completed[0]
-	require.Equal(ctx, fs, fsReport.Filesystem)
-	require.Empty(ctx, fsReport.SkipReason)
-	require.Empty(ctx, fsReport.LastError)
-	require.Len(ctx, fsReport.DestroyList, 1)
-	require.Equal(ctx, fsReport.DestroyList[0], pruner.SnapshotReport{
-		Name:       "1",
-		Replicated: true,
-		Date:       fsReport.DestroyList[0].Date,
-	})
-
-}
-
-func PrunerNoKeepNotReplicatedNoKeepStepHoldConvertsAnyStepHoldToBookmark(ctx *platformtest.Context) {
-
-	platformtest.Run(ctx, platformtest.PanicErr, ctx.RootDataset, `
-        DESTROYROOT
-        CREATEROOT
-        +  "foo bar"
-        +  "foo bar@1"
-        +  "foo bar@2"
-        +  "foo bar@3"
-        +  "foo bar@4"
-        +  "foo bar@5"
-    `)
-
-	c, err := config.ParseConfigBytes([]byte(fmt.Sprintf(`
-jobs:
-- name: prunetest
-  type: push
-  filesystems: {
-      "%s/foo bar<": true
-  }
-  connect:
-    type: tcp
-    address: 255.255.255.255:255
-  snapshotting:
-    type: manual
-  pruning:
-    keep_sender:
-    - type: last_n
-      count: 1
-    keep_receiver:
-    - type: last_n
-      count: 2
-    `, ctx.RootDataset)))
-	require.NoError(ctx, err)
-
-	pushJob := c.Jobs[0].Ret.(*config.PushJob)
-
-	dummyHistVec := prometheus.NewHistogramVec(prometheus.HistogramOpts{
-		Namespace: "foo",
-		Subsystem: "foo",
-		Name:      "foo",
-		Help:      "foo",
-	}, []string{"foo"})
-
-	prunerFactory, err := pruner.NewPrunerFactory(pushJob.Pruning, dummyHistVec)
-	require.NoError(ctx, err)
-
-	senderJid := endpoint.MustMakeJobID("sender-job")
-
-	fsfilter, err := filters.DatasetMapFilterFromConfig(pushJob.Filesystems)
-	require.NoError(ctx, err)
-
-	sender := endpoint.NewSender(endpoint.SenderConfig{
-		FSF: fsfilter,
-		Encrypt: &zfs.NilBool{
-			B: false,
-		},
-        JobID: senderJid,
-	})
-
-	fs := ctx.RootDataset + "/foo bar"
-
-	// create a replication cursor to make pruning work at all
-	_, err = endpoint.CreateReplicationCursor(ctx, fs, fsversion(ctx, fs, "@2"), senderJid)
-	require.NoError(ctx, err)
-
-	// create step holds for the incremental @2->@3
-	endpoint.HoldStep(ctx, fs, fsversion(ctx, fs, "@2"), senderJid)
-	endpoint.HoldStep(ctx, fs, fsversion(ctx, fs, "@3"), senderJid)
-	// create step holds for another job
-	otherJid := endpoint.MustMakeJobID("other-job")
-	endpoint.HoldStep(ctx, fs, fsversion(ctx, fs, "@2"), otherJid)
-	endpoint.HoldStep(ctx, fs, fsversion(ctx, fs, "@3"), otherJid)
-
-	p := prunerFactory.BuildSenderPruner(ctx, sender, sender)
-
-	p.Prune()
-
-	report := p.Report()
-
-	reportJSON, err := json.MarshalIndent(report, "", "  ")
-	require.NoError(ctx, err)
-	ctx.Logf("%s\n", string(reportJSON))
-
-	require.Equal(ctx, pruner.Done.String(), report.State)
-	require.Len(ctx, report.Completed, 1)
-	fsReport := report.Completed[0]
-	require.Equal(ctx, fs, fsReport.Filesystem)
-	require.Empty(ctx, fsReport.SkipReason)
-	require.Empty(ctx, fsReport.LastError)
-	expectDestroyList := []pruner.SnapshotReport{
-		{
-			Name:       "1",
-			Replicated: true,
-		},
-		{
-			Name:       "2",
-			Replicated: true,
-		},
-		{
-			Name:       "3",
-			Replicated: true,
-		},
-		{
-			Name:       "4",
-			Replicated: true,
-		},
-	}
-	for _, d := range fsReport.DestroyList {
-		d.Date = time.Time{}
-	}
-	require.Subset(ctx, fsReport.DestroyList, expectDestroyList)
-}
-
-
-
-func PrunerNoKeepNotReplicatedButKeepStepHold(ctx *platformtest.Context) {
-
-	platformtest.Run(ctx, platformtest.PanicErr, ctx.RootDataset, `
-        DESTROYROOT
-        CREATEROOT
-        +  "foo bar"
-        +  "foo bar@1"
-        +  "foo bar@2"
-        +  "foo bar@3"
-        +  "foo bar@4"
-        +  "foo bar@5"
-    `)
-
-	c, err := config.ParseConfigBytes([]byte(fmt.Sprintf(`
-jobs:
-- name: prunetest
-  type: push
-  filesystems: {
-      "%s/foo bar<": true
-  }
-  connect:
-    type: tcp
-    address: 255.255.255.255:255
-  snapshotting:
-    type: manual
-  pruning:
-    keep_sender:
-    - type: step_holds
-    - type: last_n
-      count: 1
-    keep_receiver:
-    - type: last_n
-      count: 2
-    `, ctx.RootDataset)))
-	require.NoError(ctx, err)
-
-	pushJob := c.Jobs[0].Ret.(*config.PushJob)
-
-	dummyHistVec := prometheus.NewHistogramVec(prometheus.HistogramOpts{
-		Namespace: "foo",
-		Subsystem: "foo",
-		Name:      "foo",
-		Help:      "foo",
-	}, []string{"foo"})
-
-	prunerFactory, err := pruner.NewPrunerFactory(pushJob.Pruning, dummyHistVec)
-	require.NoError(ctx, err)
-
-	senderJid := endpoint.MustMakeJobID("sender-job")
-
-	fsfilter, err := filters.DatasetMapFilterFromConfig(pushJob.Filesystems)
-	require.NoError(ctx, err)
-
-	sender := endpoint.NewSender(endpoint.SenderConfig{
-		FSF: fsfilter,
-		Encrypt: &zfs.NilBool{
-			B: false,
-		},
-        JobID: senderJid,
-	})
-
-	fs := ctx.RootDataset + "/foo bar"
-
-	// create a replication cursor to make pruning work at all
-	_, err = endpoint.CreateReplicationCursor(ctx, fs, fsversion(ctx, fs, "@2"), senderJid)
-	require.NoError(ctx, err)
-
-	// create step holds for the incremental @2->@3
-	endpoint.HoldStep(ctx, fs, fsversion(ctx, fs, "@2"), senderJid)
-	endpoint.HoldStep(ctx, fs, fsversion(ctx, fs, "@3"), senderJid)
-	// create step holds for another job
-	otherJid := endpoint.MustMakeJobID("other-job")
-	endpoint.HoldStep(ctx, fs, fsversion(ctx, fs, "@2"), otherJid)
-	endpoint.HoldStep(ctx, fs, fsversion(ctx, fs, "@3"), otherJid)
-
-	p := prunerFactory.BuildSenderPruner(ctx, sender, sender)
-
-	p.Prune()
-
-	report := p.Report()
-
-	reportJSON, err := json.MarshalIndent(report, "", "  ")
-	require.NoError(ctx, err)
-	ctx.Logf("%s\n", string(reportJSON))
-
-	require.Equal(ctx, pruner.Done.String(), report.State)
-	require.Len(ctx, report.Completed, 1)
-	fsReport := report.Completed[0]
-	require.Equal(ctx, fs, fsReport.Filesystem)
-	require.Empty(ctx, fsReport.SkipReason)
-	require.Empty(ctx, fsReport.LastError)
-	expectDestroyList := []pruner.SnapshotReport{
-		{
-			Name:       "1",
-			Replicated: true,
-		},
-		{
-			Name:       "4",
-			Replicated: true,
-		},
-	}
-	for _, d := range fsReport.DestroyList {
-		d.Date = time.Time{}
-	}
-	require.Subset(ctx, fsReport.DestroyList, expectDestroyList)
-}

platformtest/tests/pruner.v2.go

@@ -1,137 +0,0 @@
-package tests
-
-import (
-	"encoding/json"
-	"fmt"
-	"time"
-
-	"github.com/kr/pretty"
-	"github.com/stretchr/testify/require"
-	"github.com/zrepl/zrepl/config"
-	"github.com/zrepl/zrepl/daemon/filters"
-	"github.com/zrepl/zrepl/daemon/pruner.v2"
-	"github.com/zrepl/zrepl/endpoint"
-	"github.com/zrepl/zrepl/platformtest"
-	"github.com/zrepl/zrepl/pruning"
-	"github.com/zrepl/zrepl/zfs"
-)
-
-func Pruner2NotReplicated(ctx *platformtest.Context) {
-
-	platformtest.Run(ctx, platformtest.PanicErr, ctx.RootDataset, `
-        DESTROYROOT
-        CREATEROOT
-        +  "foo bar"
-        +  "foo bar@1"
-        +  "foo bar@2"
-        +  "foo bar@3"
-        +  "foo bar@4"
-        +  "foo bar@5"
-	`)
-
-	fs := ctx.RootDataset + "/foo bar"
-	senderJid := endpoint.MustMakeJobID("sender-job")
-	otherJid1 := endpoint.MustMakeJobID("other-job-1")
-	otherJid2 := endpoint.MustMakeJobID("other-job-2")
-
-
-	// create step holds for the incremental @2->@3
-	endpoint.HoldStep(ctx, fs, fsversion(ctx, fs, "@2"), senderJid)
-	endpoint.HoldStep(ctx, fs, fsversion(ctx, fs, "@3"), senderJid)
-	// create step holds for other-job-1 @2 -> @3
-	endpoint.HoldStep(ctx, fs, fsversion(ctx, fs, "@2"), otherJid1)
-	endpoint.HoldStep(ctx, fs, fsversion(ctx, fs, "@3"), otherJid1)
-	// create step hold for other-job-2 @1 (will be pruned)
-	endpoint.HoldStep(ctx, fs, fsversion(ctx, fs, "@1"), otherJid2)
-
-	c, err := config.ParseConfigBytes([]byte(fmt.Sprintf(`
-jobs:
-- name: prunetest
-  type: push
-  filesystems: {
-      "%s/foo bar": true
-  }
-  connect:
-    type: tcp
-    address: 255.255.255.255:255
-  snapshotting:
-    type: manual
-  pruning:
-    keep_sender:
-    #- type: not_replicated
-    - type: step_holds
-    - type: last_n
-      count: 1
-    keep_receiver:
-    - type: last_n
-      count: 2
-    `, ctx.RootDataset)))
-	require.NoError(ctx, err)
-
-	pushJob := c.Jobs[0].Ret.(*config.PushJob)
-
-	require.NoError(ctx, err)
-
-	fsfilter, err := filters.DatasetMapFilterFromConfig(pushJob.Filesystems)
-	require.NoError(ctx, err)
-
-	matchedFilesystems, err := zfs.ZFSListMapping(ctx, fsfilter)
-	ctx.Logf("%s", pretty.Sprint(matchedFilesystems))
-	require.NoError(ctx, err)
-	require.Len(ctx, matchedFilesystems, 1)
-
-	sideSender := pruner.NewSideSender(senderJid)
-
-	keepRules, err := pruning.RulesFromConfig(senderJid, pushJob.Pruning.KeepSender)
-	require.NoError(ctx, err)
-	p := pruner.NewPruner(fsfilter, senderJid, sideSender, keepRules)
-
-	runDone := make(chan *pruner.Report)
-	go func() {
-		runDone <- p.Run(ctx)
-	}()
-
-	var report *pruner.Report
-	// concurrency stress
-out:
-	for {
-		select {
-		case <-time.After(10 * time.Millisecond):
-			p.Report(ctx)
-		case report = <-runDone:
-			break out
-		}
-	}
-	ctx.Logf("%s\n", pretty.Sprint(report))
-
-	reportJSON, err := json.MarshalIndent(report, "", "  ")
-	require.NoError(ctx, err)
-	ctx.Logf("%s\n", string(reportJSON))
-
-	ctx.FailNow()
-	// fs := ctx.RootDataset + "/foo bar"
-
-	// // create a replication cursor to make pruning work at all
-	// _, err = endpoint.CreateReplicationCursor(ctx, fs, fsversion(ctx, fs, "@2"), senderJid)
-	// require.NoError(ctx, err)
-
-	// p := prunerFactory.BuildSenderPruner(ctx, sender, sender)
-
-	// p.Prune()
-
-	// report := p.Report()
-
-	// require.Equal(ctx, pruner.Done.String(), report.State)
-	// require.Len(ctx, report.Completed, 1)
-	// fsReport := report.Completed[0]
-	// require.Equal(ctx, fs, fsReport.Filesystem)
-	// require.Empty(ctx, fsReport.SkipReason)
-	// require.Empty(ctx, fsReport.LastError)
-	// require.Len(ctx, fsReport.DestroyList, 1)
-	// require.Equal(ctx, fsReport.DestroyList[0], pruner.SnapshotReport{
-	// 	Name:       "1",
-	// 	Replicated: true,
-	// 	Date:       fsReport.DestroyList[0].Date,
-	// })
-
-}

platformtest/tests/recvForceIntoEncryptedErr.go

@@ -1,45 +0,0 @@
-package tests
-
-import (
-	"fmt"
-
-	"github.com/stretchr/testify/require"
-	"github.com/zrepl/zrepl/platformtest"
-	"github.com/zrepl/zrepl/zfs"
-)
-
-func ReceiveForceIntoEncryptedErr(ctx *platformtest.Context) {
-	platformtest.Run(ctx, platformtest.PanicErr, ctx.RootDataset, `
-		DESTROYROOT
-		CREATEROOT
-		+  "foo bar" encrypted
-		+  "sender" encrypted
-		+  "sender@1"
-	`)
-
-	rfs := fmt.Sprintf("%s/foo bar", ctx.RootDataset)
-	sfs := fmt.Sprintf("%s/sender", ctx.RootDataset)
-	sfsSnap1 := sendArgVersion(ctx, sfs, "@1")
-
-	sendArgs, err := zfs.ZFSSendArgsUnvalidated{
-		FS:          sfs,
-		Encrypted:   &zfs.NilBool{B: false},
-		From:        nil,
-		To:          &sfsSnap1,
-		ResumeToken: "",
-	}.Validate(ctx)
-	require.NoError(ctx, err)
-
-	sendStream, err := zfs.ZFSSend(ctx, sendArgs)
-	require.NoError(ctx, err)
-
-	recvOpts := zfs.RecvOptions{
-		RollbackAndForceRecv: true,
-		SavePartialRecvState: false,
-	}
-	err = zfs.ZFSRecv(ctx, rfs, &zfs.ZFSSendArgVersion{RelName: "@1", GUID: sfsSnap1.GUID}, sendStream, recvOpts)
-	require.Error(ctx, err)
-	re, ok := err.(*zfs.RecvDestroyOrOverwriteEncryptedErr)
-	require.True(ctx, ok)
-	require.Contains(ctx, re.Error(), "zfs receive -F cannot be used to destroy an encrypted filesystem or overwrite an unencrypted one with an encrypted on")
-}

platformtest/tests/recvRollback.go

@@ -1,54 +0,0 @@
-package tests
-
-import (
-	"fmt"
-
-	"github.com/stretchr/testify/assert"
-	"github.com/stretchr/testify/require"
-	"github.com/zrepl/zrepl/platformtest"
-	"github.com/zrepl/zrepl/zfs"
-)
-
-func ReceiveForceRollbackWorksUnencrypted(ctx *platformtest.Context) {
-	platformtest.Run(ctx, platformtest.PanicErr, ctx.RootDataset, `
-		DESTROYROOT
-		CREATEROOT
-		+  "foo bar"
-		+  "foo bar@a snap"
-		+  "foo bar@another snap"
-		+  "foo bar@snap3"
-		+  "sender"
-		+  "sender@1"
-	`)
-
-	rfs := fmt.Sprintf("%s/foo bar", ctx.RootDataset)
-	sfs := fmt.Sprintf("%s/sender", ctx.RootDataset)
-	sfsSnap1 := sendArgVersion(ctx, sfs, "@1")
-
-	sendArgs, err := zfs.ZFSSendArgsUnvalidated{
-		FS:          sfs,
-		Encrypted:   &zfs.NilBool{B: false},
-		From:        nil,
-		To:          &sfsSnap1,
-		ResumeToken: "",
-	}.Validate(ctx)
-	require.NoError(ctx, err)
-
-	sendStream, err := zfs.ZFSSend(ctx, sendArgs)
-	require.NoError(ctx, err)
-
-	recvOpts := zfs.RecvOptions{
-		RollbackAndForceRecv: true,
-		SavePartialRecvState: false,
-	}
-	err = zfs.ZFSRecv(ctx, rfs, &zfs.ZFSSendArgVersion{RelName: "@1", GUID: sfsSnap1.GUID}, sendStream, recvOpts)
-	require.NoError(ctx, err)
-
-	// assert exists on receiver
-	rfsSnap1 := fsversion(ctx, rfs, "@1")
-	// assert it's the only one (rollback and force-recv should be blowing away the other filesystems)
-	rfsVersions, err := zfs.ZFSListFilesystemVersions(ctx, mustDatasetPath(rfs), zfs.ListFilesystemVersionsOptions{})
-	require.NoError(ctx, err)
-	assert.Len(ctx, rfsVersions, 1)
-	assert.Equal(ctx, rfsVersions[0], rfsSnap1)
-}

platformtest/tests/replication.go

@@ -1,396 +0,0 @@
-package tests
-
-import (
-	"context"
-	"fmt"
-	"io"
-	"path"
-
-	"github.com/kr/pretty"
-	"github.com/stretchr/testify/require"
-	"github.com/zrepl/zrepl/daemon/filters"
-	"github.com/zrepl/zrepl/endpoint"
-	"github.com/zrepl/zrepl/platformtest"
-	"github.com/zrepl/zrepl/replication"
-	"github.com/zrepl/zrepl/replication/logic"
-	"github.com/zrepl/zrepl/replication/logic/pdu"
-	"github.com/zrepl/zrepl/replication/report"
-	"github.com/zrepl/zrepl/util/limitio"
-	"github.com/zrepl/zrepl/zfs"
-)
-
-// mimics the replication invocations of an active-side job
-// for a single sender-receiver filesystem pair
-//
-// each invocation of method Do results in the construction
-// of a new sender and receiver instance and one blocking invocation
-// of the replication engine without encryption
-type replicationInvocation struct {
-	sjid, rjid      endpoint.JobID
-	sfs             string
-	rfsRoot         string
-	interceptSender func(e *endpoint.Sender) logic.Sender
-}
-
-func (i replicationInvocation) Do(ctx *platformtest.Context) *report.Report {
-
-	if i.interceptSender == nil {
-		i.interceptSender = func(e *endpoint.Sender) logic.Sender { return e }
-	}
-
-	sfilter := filters.NewDatasetMapFilter(1, true)
-	err := sfilter.Add(i.sfs, "ok")
-	require.NoError(ctx, err)
-	sender := i.interceptSender(endpoint.NewSender(endpoint.SenderConfig{
-		FSF:     sfilter.AsFilter(),
-		Encrypt: &zfs.NilBool{B: false},
-		JobID:   i.sjid,
-	}))
-	receiver := endpoint.NewReceiver(endpoint.ReceiverConfig{
-		JobID:                      i.rjid,
-		AppendClientIdentity:       false,
-		RootWithoutClientComponent: mustDatasetPath(i.rfsRoot),
-		UpdateLastReceivedHold:     true,
-	})
-	plannerPolicy := logic.PlannerPolicy{
-		EncryptedSend: logic.TriFromBool(false),
-	}
-
-	report, wait := replication.Do(
-		ctx,
-		logic.NewPlanner(nil, nil, sender, receiver, plannerPolicy),
-	)
-	wait(true)
-	return report()
-}
-
-func (i replicationInvocation) ReceiveSideFilesystem() string {
-	return path.Join(i.rfsRoot, i.sfs)
-}
-
-func ReplicationIncrementalIsPossibleIfCommonSnapshotIsDestroyed(ctx *platformtest.Context) {
-
-	platformtest.Run(ctx, platformtest.PanicErr, ctx.RootDataset, `
-		CREATEROOT
-		+  "sender"
-		+  "sender@1"
-		+  "receiver"
-		R  zfs create -p "${ROOTDS}/receiver/${ROOTDS}"
-	`)
-
-	sjid := endpoint.MustMakeJobID("sender-job")
-	rjid := endpoint.MustMakeJobID("receiver-job")
-
-	sfs := ctx.RootDataset + "/sender"
-	rfsRoot := ctx.RootDataset + "/receiver"
-	snap1 := fsversion(ctx, sfs, "@1")
-
-	rep := replicationInvocation{
-		sjid:    sjid,
-		rjid:    rjid,
-		sfs:     sfs,
-		rfsRoot: rfsRoot,
-	}
-	rfs := rep.ReceiveSideFilesystem()
-
-	// first replication
-	report := rep.Do(ctx)
-	ctx.Logf("\n%s", pretty.Sprint(report))
-
-	// assert @1 exists on receiver
-	_ = fsversion(ctx, rfs, "@1")
-
-	// cut off the common base between sender and receiver
-	// (replication engine guarantees resumability through bookmarks)
-	err := zfs.ZFSDestroy(ctx, snap1.FullPath(sfs))
-	require.NoError(ctx, err)
-
-	// assert that the replication cursor has been created
-	snap1CursorName, err := endpoint.ReplicationCursorBookmarkName(sfs, snap1.Guid, sjid)
-	require.NoError(ctx, err)
-	snap1CursorInfo, err := zfs.ZFSGetFilesystemVersion(ctx, sfs+"#"+snap1CursorName)
-	require.NoError(ctx, err)
-	require.True(ctx, snap1CursorInfo.IsBookmark())
-
-	// second replication of a new snapshot, should use the cursor
-	mustSnapshot(ctx, sfs+"@2")
-	report = rep.Do(ctx)
-	ctx.Logf("\n%s", pretty.Sprint(report))
-	_ = fsversion(ctx, rfs, "@2")
-
-}
-
-func ReplicationIncrementalCleansUpStaleAbstractionsWithCacheOnSecondReplication(ctx *platformtest.Context) {
-	implReplicationIncrementalCleansUpStaleAbstractions(ctx, true)
-}
-
-func ReplicationIncrementalCleansUpStaleAbstractionsWithoutCacheOnSecondReplication(ctx *platformtest.Context) {
-	implReplicationIncrementalCleansUpStaleAbstractions(ctx, false)
-}
-
-func implReplicationIncrementalCleansUpStaleAbstractions(ctx *platformtest.Context, invalidateCacheBeforeSecondReplication bool) {
-
-	platformtest.Run(ctx, platformtest.PanicErr, ctx.RootDataset, `
-		CREATEROOT
-		+  "sender"
-		+  "sender@1"
-		+  "sender@2"
-		+  "sender#2" "sender@2"
-		+  "sender@3"
-		+  "receiver"
-		R  zfs create -p "${ROOTDS}/receiver/${ROOTDS}"
-	`)
-
-	sjid := endpoint.MustMakeJobID("sender-job")
-	ojid := endpoint.MustMakeJobID("other-job")
-	rjid := endpoint.MustMakeJobID("receiver-job")
-
-	sfs := ctx.RootDataset + "/sender"
-	rfsRoot := ctx.RootDataset + "/receiver"
-
-	rep := replicationInvocation{
-		sjid:    sjid,
-		rjid:    rjid,
-		sfs:     sfs,
-		rfsRoot: rfsRoot,
-	}
-	rfs := rep.ReceiveSideFilesystem()
-
-	// first replication
-	report := rep.Do(ctx)
-	ctx.Logf("\n%s", pretty.Sprint(report))
-
-	// assert most recent send-side version @3 exists on receiver (=replication succeeded)
-	rSnap3 := fsversion(ctx, rfs, "@3")
-	// assert the source-side versions not managed by zrepl still exist
-	snap1 := fsversion(ctx, sfs, "@1")
-	snap2 := fsversion(ctx, sfs, "@2")
-	_ = fsversion(ctx, sfs, "#2") // non-replicationc-cursor bookmarks should not be affected
-	snap3 := fsversion(ctx, sfs, "@3")
-	// assert a replication cursor is in place
-	snap3CursorName, err := endpoint.ReplicationCursorBookmarkName(sfs, snap3.Guid, sjid)
-	require.NoError(ctx, err)
-	_ = fsversion(ctx, sfs, "#"+snap3CursorName)
-	// assert a last-received hold is in place
-	expectRjidHoldTag, err := endpoint.LastReceivedHoldTag(rjid)
-	require.NoError(ctx, err)
-	holds, err := zfs.ZFSHolds(ctx, rfs, rSnap3.Name)
-	require.NoError(ctx, err)
-	require.Contains(ctx, holds, expectRjidHoldTag)
-
-	// create artifical stale replication cursors
-	createArtificalStaleAbstractions := func(jobId endpoint.JobID) []endpoint.Abstraction {
-		snap2Cursor, err := endpoint.CreateReplicationCursor(ctx, sfs, snap2, jobId) // no shadow
-		require.NoError(ctx, err)
-		// create artifical stale step holds jobId
-		snap1Hold, err := endpoint.HoldStep(ctx, sfs, snap1, jobId) // no shadow
-		require.NoError(ctx, err)
-		snap2Hold, err := endpoint.HoldStep(ctx, sfs, snap2, jobId) // no shadow
-		require.NoError(ctx, err)
-		return []endpoint.Abstraction{snap2Cursor, snap1Hold, snap2Hold}
-	}
-	createArtificalStaleAbstractions(sjid)
-	ojidSendAbstractions := createArtificalStaleAbstractions(ojid)
-
-	snap3ojidLastReceivedHold, err := endpoint.CreateLastReceivedHold(ctx, rfs, fsversion(ctx, rfs, "@3"), ojid)
-	require.NoError(ctx, err)
-	require.True(ctx, zfs.FilesystemVersionEqualIdentity(fsversion(ctx, rfs, "@3"), snap3ojidLastReceivedHold.GetFilesystemVersion()))
-
-	// take another 2 snapshots
-	mustSnapshot(ctx, sfs+"@4")
-	mustSnapshot(ctx, sfs+"@5")
-	snap5 := fsversion(ctx, sfs, "@5")
-
-	if invalidateCacheBeforeSecondReplication {
-		endpoint.SendAbstractionsCacheInvalidate(sfs)
-	}
-
-	// do another replication
-	// - ojid's abstractions should not be affected on either side
-	// - stale abstractions of sjid and rjid should be cleaned up
-	// - 1 replication cursors and 1 last-received hold should be present
-
-	checkOjidAbstractionsExist := func() {
-		var expectedOjidAbstractions []endpoint.Abstraction
-		expectedOjidAbstractions = append(expectedOjidAbstractions, ojidSendAbstractions...)
-		expectedOjidAbstractions = append(expectedOjidAbstractions, snap3ojidLastReceivedHold)
-
-		sfsAndRfsFilter := filters.NewDatasetMapFilter(2, true)
-		require.NoError(ctx, sfsAndRfsFilter.Add(sfs, "ok"))
-		require.NoError(ctx, sfsAndRfsFilter.Add(rfs, "ok"))
-		rAbs, rAbsErrs, err := endpoint.ListAbstractions(ctx, endpoint.ListZFSHoldsAndBookmarksQuery{
-			FS:          endpoint.ListZFSHoldsAndBookmarksQueryFilesystemFilter{Filter: sfsAndRfsFilter},
-			JobID:       &ojid,
-			What:        endpoint.AbstractionTypesAll,
-			Concurrency: 1,
-		})
-		require.NoError(ctx, err)
-		require.Len(ctx, rAbsErrs, 0)
-		ctx.Logf("rAbs=%s", rAbs)
-		ctx.Logf("expectedOjidAbstractions=%s", expectedOjidAbstractions)
-		require.Equal(ctx, len(expectedOjidAbstractions), len(rAbs))
-		for _, ea := range expectedOjidAbstractions {
-			ctx.Logf("looking for %s %#v", ea, ea.GetFilesystemVersion())
-			found := false
-			for _, a := range rAbs {
-				eq := endpoint.AbstractionEquals(ea, a)
-				ctx.Logf("comp=%v for %s %#v", eq, a, a.GetFilesystemVersion())
-				found = found || eq
-			}
-			require.True(ctx, found, "%s", ea)
-		}
-	}
-	checkOjidAbstractionsExist()
-
-	report = rep.Do(ctx)
-	ctx.Logf("\n%s", pretty.Sprint(report))
-
-	checkOjidAbstractionsExist()
-
-	_ = fsversion(ctx, sfs, "@1")
-	_ = fsversion(ctx, sfs, "@2")
-	_ = fsversion(ctx, sfs, "#2")
-	_ = fsversion(ctx, sfs, "@3")
-	_ = fsversion(ctx, sfs, "@4")
-	_ = fsversion(ctx, sfs, "@5")
-
-	_ = fsversion(ctx, rfs, "@3")
-	_ = fsversion(ctx, rfs, "@4")
-	_ = fsversion(ctx, rfs, "@5")
-
-	// check bookmark situation
-	{
-		sBms, err := zfs.ZFSListFilesystemVersions(ctx, mustDatasetPath(sfs), zfs.ListFilesystemVersionsOptions{
-			Types: zfs.Bookmarks,
-		})
-		ctx.Logf("sbms=%s", sBms)
-		require.NoError(ctx, err)
-
-		snap5SjidCursorName, err := endpoint.ReplicationCursorBookmarkName(sfs, snap5.Guid, sjid)
-		require.NoError(ctx, err)
-		snap2SjidCursorName, err := endpoint.ReplicationCursorBookmarkName(sfs, snap2.Guid, sjid)
-		require.NoError(ctx, err)
-		snap2OjidCursorName, err := endpoint.ReplicationCursorBookmarkName(sfs, snap2.Guid, ojid)
-		require.NoError(ctx, err)
-		var bmNames []string
-		for _, bm := range sBms {
-			bmNames = append(bmNames, bm.Name)
-		}
-
-		if invalidateCacheBeforeSecondReplication {
-			require.Len(ctx, sBms, 3)
-			require.Contains(ctx, bmNames, snap5SjidCursorName)
-			require.Contains(ctx, bmNames, snap2OjidCursorName)
-			require.Contains(ctx, bmNames, "2")
-		} else {
-			require.Len(ctx, sBms, 4)
-			require.Contains(ctx, bmNames, snap5SjidCursorName)
-			require.Contains(ctx, bmNames, snap2SjidCursorName)
-			require.Contains(ctx, bmNames, snap2OjidCursorName)
-			require.Contains(ctx, bmNames, "2")
-		}
-	}
-
-	// check last-received hold moved
-	{
-		rAbs, rAbsErrs, err := endpoint.ListAbstractions(ctx, endpoint.ListZFSHoldsAndBookmarksQuery{
-			FS:          endpoint.ListZFSHoldsAndBookmarksQueryFilesystemFilter{FS: &rfs},
-			JobID:       &rjid,
-			What:        endpoint.AbstractionTypesAll,
-			Concurrency: 1,
-		})
-		require.NoError(ctx, err)
-		require.Len(ctx, rAbsErrs, 0)
-		require.Len(ctx, rAbs, 1)
-		require.Equal(ctx, rAbs[0].GetType(), endpoint.AbstractionLastReceivedHold)
-		require.Equal(ctx, *rAbs[0].GetJobID(), rjid)
-		require.Equal(ctx, rAbs[0].GetFilesystemVersion().GetGuid(), snap5.GetGuid())
-	}
-
-}
-
-type PartialSender struct {
-	*endpoint.Sender
-	failAfterByteCount int64
-}
-
-var _ logic.Sender = (*PartialSender)(nil)
-
-func (s *PartialSender) Send(ctx context.Context, r *pdu.SendReq) (r1 *pdu.SendRes, r2 io.ReadCloser, r3 error) {
-	r1, r2, r3 = s.Sender.Send(ctx, r)
-	r2 = limitio.ReadCloser(r2, s.failAfterByteCount)
-	return r1, r2, r3
-}
-
-func ReplicationIsResumableFullSend(ctx *platformtest.Context) {
-
-	platformtest.Run(ctx, platformtest.PanicErr, ctx.RootDataset, `
-		CREATEROOT
-		+  "sender"
-		+  "receiver"
-		R  zfs create -p "${ROOTDS}/receiver/${ROOTDS}"
-	`)
-
-	sjid := endpoint.MustMakeJobID("sender-job")
-	rjid := endpoint.MustMakeJobID("receiver-job")
-
-	sfs := ctx.RootDataset + "/sender"
-	rfsRoot := ctx.RootDataset + "/receiver"
-
-	sfsmp, err := zfs.ZFSGetMountpoint(ctx, sfs)
-	require.NoError(ctx, err)
-	require.True(ctx, sfsmp.Mounted)
-
-	writeDummyData(path.Join(sfsmp.Mountpoint, "dummy.data"), 1<<22)
-	mustSnapshot(ctx, sfs+"@1")
-	snap1 := fsversion(ctx, sfs, "@1")
-
-	rep := replicationInvocation{
-		sjid:    sjid,
-		rjid:    rjid,
-		sfs:     sfs,
-		rfsRoot: rfsRoot,
-		interceptSender: func(e *endpoint.Sender) logic.Sender {
-			return &PartialSender{Sender: e, failAfterByteCount: 1 << 20}
-		},
-	}
-	rfs := rep.ReceiveSideFilesystem()
-
-	for i := 2; i < 10; i++ {
-		report := rep.Do(ctx)
-		ctx.Logf("\n%s", pretty.Sprint(report))
-
-		// always attempt to destroy the incremental source
-		err := zfs.ZFSDestroy(ctx, snap1.FullPath(sfs))
-		if i < 4 {
-			// we configured the PartialSender to fail after 1<<20 bytes
-			// and we wrote dummy data 1<<22 bytes, thus at least
-			// for the first 4 times this should not be possible
-			// due to step holds
-			require.Error(ctx, err)
-			require.Contains(ctx, err.Error(), "dataset is busy")
-		}
-
-		// and create some additional snapshots that could
-		// confuse a naive implementation that doesn't take into
-		// account resume state when planning replication
-		if i == 2 || i == 3 {
-			// no significant size to avoid making this test run longer than necessary
-			mustSnapshot(ctx, fmt.Sprintf("%s@%d", sfs, i))
-		}
-
-		require.Len(ctx, report.Attempts, 1)
-		require.Nil(ctx, report.Attempts[0].PlanError)
-		require.Len(ctx, report.Attempts[0].Filesystems, 1)
-		if len(report.Attempts[0].Filesystems[0].Steps) == 0 {
-			break
-		}
-	}
-
-	// make sure all the filesystem versions we created
-	// were replicated by the replication loop
-	_ = fsversion(ctx, rfs, "@1")
-	_ = fsversion(ctx, rfs, "@2")
-	_ = fsversion(ctx, rfs, "@3")
-
-}

platformtest/tests/replicationCursor.go

@@ -11,7 +11,7 @@ import (
 	"github.com/zrepl/zrepl/zfs"
 )
 
-func CreateReplicationCursor(ctx *platformtest.Context) {
+func ReplicationCursor(ctx *platformtest.Context) {
 
 	platformtest.Run(ctx, platformtest.PanicErr, ctx.RootDataset, `
 		CREATEROOT
@@ -19,9 +19,9 @@ func CreateReplicationCursor(ctx *platformtest.Context) {
 		+  "foo bar@1 with space"
 		R  zfs bookmark "${ROOTDS}/foo bar@1 with space" "${ROOTDS}/foo bar#1 with space"
 		+  "foo bar@2 with space"
-		R  zfs bookmark "${ROOTDS}/foo bar@2 with space" "${ROOTDS}/foo bar#2 with space"
+		R  zfs bookmark "${ROOTDS}/foo bar@1 with space" "${ROOTDS}/foo bar#2 with space"
 		+  "foo bar@3 with space"
-		R  zfs bookmark "${ROOTDS}/foo bar@3 with space" "${ROOTDS}/foo bar#3 with space"
+		R  zfs bookmark "${ROOTDS}/foo bar@1 with space" "${ROOTDS}/foo bar#3 with space"
 	`)
 
 	jobid := endpoint.MustMakeJobID("zreplplatformtest")
@@ -32,29 +32,13 @@ func CreateReplicationCursor(ctx *platformtest.Context) {
 	}
 
 	fs := ds.ToString()
-
-	checkCreateCursor := func(createErr error, c endpoint.Abstraction, references zfs.FilesystemVersion) {
-		assert.NoError(ctx, createErr)
-		expectName, err := endpoint.ReplicationCursorBookmarkName(fs, references.Guid, jobid)
-		assert.NoError(ctx, err)
-		require.Equal(ctx, expectName, c.GetFilesystemVersion().Name)
-	}
-
 	snap := fsversion(ctx, fs, "@1 with space")
-	book := fsversion(ctx, fs, "#1 with space")
 
-	// create first cursor
-	cursorOfSnap, err := endpoint.CreateReplicationCursor(ctx, fs, snap, jobid)
-	checkCreateCursor(err, cursorOfSnap, snap)
-	// check CreateReplicationCursor is idempotent (for snapshot target)
-	cursorOfSnapIdemp, err := endpoint.CreateReplicationCursor(ctx, fs, snap, jobid)
-	checkCreateCursor(err, cursorOfSnap, snap)
-	// ... for target = non-cursor bookmark
-	_, err = endpoint.CreateReplicationCursor(ctx, fs, book, jobid)
-	assert.Equal(ctx, zfs.ErrBookmarkCloningNotSupported, err)
-	// ... for target = replication cursor bookmark to be created
-	cursorOfCursor, err := endpoint.CreateReplicationCursor(ctx, fs, cursorOfSnapIdemp.GetFilesystemVersion(), jobid)
-	checkCreateCursor(err, cursorOfCursor, cursorOfSnap.GetFilesystemVersion())
+	destroyed, err := endpoint.MoveReplicationCursor(ctx, fs, &snap, jobid)
+	if err != nil {
+		panic(err)
+	}
+	assert.Empty(ctx, destroyed)
 
 	snapProps, err := zfs.ZFSGetFilesystemVersion(ctx, snap.FullPath(fs))
 	if err != nil {
@@ -72,4 +56,14 @@ func CreateReplicationCursor(ctx *platformtest.Context) {
 		panic(fmt.Sprintf("guids do not match: %v != %v", bm.Guid, snapProps.Guid))
 	}
 
+	// try moving
+	cursor1BookmarkName, err := endpoint.ReplicationCursorBookmarkName(fs, snap.Guid, jobid)
+	require.NoError(ctx, err)
+
+	snap2 := fsversion(ctx, fs, "@2 with space")
+	destroyed, err = endpoint.MoveReplicationCursor(ctx, fs, &snap2, jobid)
+	require.NoError(ctx, err)
+	require.Equal(ctx, 1, len(destroyed))
+	require.Equal(ctx, endpoint.AbstractionReplicationCursorBookmarkV2, destroyed[0].GetType())
+	require.Equal(ctx, cursor1BookmarkName, destroyed[0].GetName())
 }

platformtest/tests/tests.go

@@ -13,4 +13,22 @@ func (c Case) String() string {
 	return runtime.FuncForPC(reflect.ValueOf(c).Pointer()).Name()
 }
 
-//go:generate ../../artifacts/generate-platform-test-list github.com/zrepl/zrepl/platformtest/tests
+var Cases = []Case{
+	BatchDestroy,
+	UndestroyableSnapshotParsing,
+	GetNonexistent,
+	ReplicationCursor,
+	IdempotentHold,
+	IdempotentBookmark,
+	IdempotentDestroy,
+	ResumeTokenParsing,
+	ResumableRecvAndTokenHandling,
+	SendArgsValidationEncryptedSendOfUnencryptedDatasetForbidden,
+	SendArgsValidationResumeTokenEncryptionMismatchForbidden,
+	SendArgsValidationResumeTokenDifferentFilesystemForbidden,
+	ListFilesystemVersionsTypeFilteringAndPrefix,
+	ListFilesystemVersionsFilesystemNotExist,
+	ListFilesystemVersionsFilesystemNotExist,
+	ListFilesystemVersionsUserrefs,
+	ListFilesystemsNoFilter,
+}

platformtest/tests/undestroyableSnapshotParsing.go

@@ -17,12 +17,10 @@ func UndestroyableSnapshotParsing(t *platformtest.Context) {
 		+  "foo bar@1 2 3"
 		+  "foo bar@4 5 6"
 		+  "foo bar@7 8 9"
-		+  "foo bar@10 11 12"
 		R  zfs hold zrepl_platformtest "${ROOTDS}/foo bar@4 5 6"
-		R  zfs hold zrepl_platformtest "${ROOTDS}/foo bar@7 8 9"
 	`)
 
-	err := zfs.ZFSDestroy(t, fmt.Sprintf("%s/foo bar@1 2 3,4 5 6,7 8 9,10 11 12", t.RootDataset))
+	err := zfs.ZFSDestroy(t, fmt.Sprintf("%s/foo bar@1 2 3,4 5 6,7 8 9", t.RootDataset))
 	if err == nil {
 		panic("expecting destroy error due to hold")
 	}
@@ -32,9 +30,8 @@ func UndestroyableSnapshotParsing(t *platformtest.Context) {
 		if dse.Filesystem != fmt.Sprintf("%s/foo bar", t.RootDataset) {
 			panic(dse.Filesystem)
 		}
-		expectUndestroyable := []string{"4 5 6", "7 8 9"}
-		require.Len(t, dse.Undestroyable, len(expectUndestroyable))
-		require.Subset(t, dse.Undestroyable, expectUndestroyable)
+		require.Equal(t, []string{"4 5 6"}, dse.Undestroyable)
+		require.Equal(t, []string{"dataset is busy"}, dse.Reason)
 	}
 
 }

pruning/keep_grid.go

@@ -66,26 +66,24 @@ type retentionGridAdaptor struct {
 	Snapshot
 }
 
-func (a retentionGridAdaptor) Date() time.Time { return a.Snapshot.GetCreation() }
-
 func (a retentionGridAdaptor) LessThan(b retentiongrid.Entry) bool {
 	return a.Date().Before(b.Date())
 }
 
 // Prune filters snapshots with the retention grid.
-func (p *KeepGrid) KeepRule(snaps []Snapshot) PruneSnapshotsResult {
+func (p *KeepGrid) KeepRule(snaps []Snapshot) (destroyList []Snapshot) {
 
-	reCandidates := partitionSnapList(snaps, func(snapshot Snapshot) bool {
-		return p.re.MatchString(snapshot.GetName())
+	snaps = filterSnapList(snaps, func(snapshot Snapshot) bool {
+		return p.re.MatchString(snapshot.Name())
 	})
-	if len(reCandidates.Remove) == 0 {
-		return reCandidates
+	if len(snaps) == 0 {
+		return nil
 	}
 
 	// Build adaptors for retention grid
 	adaptors := make([]retentiongrid.Entry, 0)
 	for i := range snaps {
-		adaptors = append(adaptors, retentionGridAdaptor{reCandidates.Remove[i]})
+		adaptors = append(adaptors, retentionGridAdaptor{snaps[i]})
 	}
 
 	// determine 'now' edge
@@ -95,17 +93,12 @@ func (p *KeepGrid) KeepRule(snaps []Snapshot) PruneSnapshotsResult {
 	now := adaptors[len(adaptors)-1].Date()
 
 	// Evaluate retention grid
-	keepa, removea := p.retentionGrid.FitEntries(now, adaptors)
+	_, removea := p.retentionGrid.FitEntries(now, adaptors)
 
 	// Revert adaptors
-	destroyList := make([]Snapshot, len(removea))
+	destroyList = make([]Snapshot, len(removea))
 	for i := range removea {
 		destroyList[i] = removea[i].(retentionGridAdaptor).Snapshot
 	}
-	for _, a := range keepa {
-		reCandidates.Keep = append(reCandidates.Keep, a.(retentionGridAdaptor))
-	}
-	reCandidates.Remove = destroyList
-
-	return reCandidates
+	return destroyList
 }

pruning/keep_helpers.go

@@ -1,13 +1,10 @@
 package pruning
 
-func partitionSnapList(snaps []Snapshot, remove func(Snapshot) bool) (r PruneSnapshotsResult) {
-	r.Keep = make([]Snapshot, 0, len(snaps))
-	r.Remove = make([]Snapshot, 0, len(snaps))
+func filterSnapList(snaps []Snapshot, predicate func(Snapshot) bool) []Snapshot {
+	r := make([]Snapshot, 0, len(snaps))
 	for i := range snaps {
-		if remove(snaps[i]) {
-			r.Remove = append(r.Remove, snaps[i])
-		} else {
-			r.Keep = append(r.Keep, snaps[i])
+		if predicate(snaps[i]) {
+			r = append(r, snaps[i])
 		}
 	}
 	return r

pruning/keep_last_n.go

@@ -17,17 +17,17 @@ func NewKeepLastN(n int) (*KeepLastN, error) {
 	return &KeepLastN{n}, nil
 }
 
-func (k KeepLastN) KeepRule(snaps []Snapshot) PruneSnapshotsResult {
+func (k KeepLastN) KeepRule(snaps []Snapshot) (destroyList []Snapshot) {
 
 	if k.n > len(snaps) {
-		return PruneSnapshotsResult{Keep: snaps}
+		return []Snapshot{}
 	}
 
 	res := shallowCopySnapList(snaps)
 
 	sort.Slice(res, func(i, j int) bool {
-		return res[i].GetCreateTXG() > res[j].GetCreateTXG()
+		return res[i].Date().After(res[j].Date())
 	})
 
-	return PruneSnapshotsResult{Remove: res[k.n:], Keep: res[:k.n]}
+	return res[k.n:]
 }

pruning/keep_not_replicated.go

@@ -2,8 +2,8 @@ package pruning
 
 type KeepNotReplicated struct{}
 
-func (*KeepNotReplicated) KeepRule(snaps []Snapshot) PruneSnapshotsResult {
-	return partitionSnapList(snaps, func(snapshot Snapshot) bool {
+func (*KeepNotReplicated) KeepRule(snaps []Snapshot) (destroyList []Snapshot) {
+	return filterSnapList(snaps, func(snapshot Snapshot) bool {
 		return snapshot.Replicated()
 	})
 }

pruning/keep_regex.go

@@ -27,12 +27,12 @@ func MustKeepRegex(expr string, negate bool) *KeepRegex {
 	return k
 }
 
-func (k *KeepRegex) KeepRule(snaps []Snapshot) PruneSnapshotsResult {
-	return partitionSnapList(snaps, func(s Snapshot) bool {
+func (k *KeepRegex) KeepRule(snaps []Snapshot) []Snapshot {
+	return filterSnapList(snaps, func(s Snapshot) bool {
 		if k.negate {
-			return k.expr.FindStringIndex(s.GetName()) != nil
+			return k.expr.FindStringIndex(s.Name()) != nil
 		} else {
-			return k.expr.FindStringIndex(s.GetName()) == nil
+			return k.expr.FindStringIndex(s.Name()) == nil
 		}
 	})
 }