replication/driver: fs.debug() helper that automatically prefixes with fs name

refs #140
refs #302

Makefile

@@ -66,7 +66,6 @@ wrapup-and-checksum:
 		-acf $(NOARCH_TARBALL) \
 		$(ARTIFACTDIR)/docs/html \
 		$(ARTIFACTDIR)/bash_completion \
-		$(ARTIFACTDIR)/_zrepl.zsh_completion \
 		$(ARTIFACTDIR)/go_env.txt \
 		dist \
 		config/samples
@@ -161,7 +160,7 @@ platformtest: # do not track dependency on platformtest-bin to allow build of pl
 		$(ZREPL_PLATFORMTEST_ARGS)
 
 ##################### NOARCH #####################
-.PHONY: noarch $(ARTIFACTDIR)/bash_completion $(ARTIFACTDIR)/_zrepl.zsh_completion $(ARTIFACTDIR)/go_env.txt docs docs-clean
+.PHONY: noarch $(ARTIFACTDIR)/bash_completion $(ARTIFACTDIR)/go_env.txt docs docs-clean
 
 
 $(ARTIFACTDIR):
@@ -169,16 +168,12 @@ $(ARTIFACTDIR):
 $(ARTIFACTDIR)/docs: $(ARTIFACTDIR)
 	mkdir -p "$@"
 
-noarch: $(ARTIFACTDIR)/bash_completion $(ARTIFACTDIR)/_zrepl.zsh_completion $(ARTIFACTDIR)/go_env.txt docs
+noarch: $(ARTIFACTDIR)/bash_completion $(ARTIFACTDIR)/go_env.txt docs
 	# pass
 
 $(ARTIFACTDIR)/bash_completion:
 	$(MAKE) zrepl-bin GOOS=$(GOHOSTOS) GOARCH=$(GOHOSTARCH)
-	artifacts/zrepl-$(GOHOSTOS)-$(GOHOSTARCH) gencompletion bash "$@"
-
-$(ARTIFACTDIR)/_zrepl.zsh_completion:
-	$(MAKE) zrepl-bin GOOS=$(GOHOSTOS) GOARCH=$(GOHOSTARCH)
-	artifacts/zrepl-$(GOHOSTOS)-$(GOHOSTARCH) gencompletion zsh "$@"
+	artifacts/zrepl-$(GOHOSTOS)-$(GOHOSTARCH) bashcomp "$@"
 
 $(ARTIFACTDIR)/go_env.txt:
 	$(GO_ENV_VARS) $(GO) env > $@

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"
 )
@@ -21,55 +19,29 @@ var rootCmd = &cobra.Command{
 	Short: "One-stop ZFS replication solution",
 }
 
-func init() {
-	rootCmd.PersistentFlags().StringVar(&rootArgs.configPath, "config", "", "config file path")
-}
-
-var genCompletionCmd = &cobra.Command{
-	Use:   "gencompletion",
-	Short: "generate shell auto-completions",
-}
-
-type completionCmdInfo struct {
-	genFunc func(outpath string) error
-	help    string
-}
-
-var completionCmdMap = map[string]completionCmdInfo{
-	"zsh": {
-		rootCmd.GenZshCompletionFile,
-		"  save to file `_zrepl` in your zsh's $fpath",
-	},
-	"bash": {
-		rootCmd.GenBashCompletionFile,
-		"  save to a path and source that path in your .bashrc",
+var bashcompCmd = &cobra.Command{
+	Use:   "bashcomp path/to/out/file",
+	Short: "generate bash completions",
+	Run: func(cmd *cobra.Command, args []string) {
+		if len(args) != 1 {
+			fmt.Fprintf(os.Stderr, "specify exactly one positional agument\n")
+			err := cmd.Usage()
+			if err != nil {
+				panic(err)
+			}
+			os.Exit(1)
+		}
+		if err := rootCmd.GenBashCompletionFile(args[0]); err != nil {
+			fmt.Fprintf(os.Stderr, "error generating bash completion: %s", err)
+			os.Exit(1)
+		}
 	},
+	Hidden: true,
 }
 
 func init() {
-	for sh, info := range completionCmdMap {
-		sh, info := sh, info
-		genCompletionCmd.AddCommand(&cobra.Command{
-			Use:     fmt.Sprintf("%s path/to/out/file", sh),
-			Short:   fmt.Sprintf("generate %s completions", sh),
-			Example: info.help,
-			Run: func(cmd *cobra.Command, args []string) {
-				if len(args) != 1 {
-					fmt.Fprintf(os.Stderr, "specify exactly one positional agument\n")
-					err := cmd.Usage()
-					if err != nil {
-						panic(err)
-					}
-					os.Exit(1)
-				}
-				if err := info.genFunc(args[0]); err != nil {
-					fmt.Fprintf(os.Stderr, "error generating %s completion: %s", sh, err)
-					os.Exit(1)
-				}
-			},
-		})
-	}
-	rootCmd.AddCommand(genCompletionCmd)
+	rootCmd.PersistentFlags().StringVar(&rootArgs.configPath, "config", "", "config file path")
+	rootCmd.AddCommand(bashcompCmd)
 }
 
 type Subcommand struct {
@@ -77,7 +49,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 +70,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/hold_create_step_hold.go

@@ -12,28 +12,28 @@ import (
 	"github.com/zrepl/zrepl/zfs"
 )
 
-var zabsCreateStepHoldFlags struct {
+var holdsCreateStepHoldFlags struct {
 	target string
 	jobid  JobIDFlag
 }
 
-var zabsCmdCreateStepHold = &cli.Subcommand{
+var holdsCmdCreateStepHold = &cli.Subcommand{
 	Use:             "step",
-	Run:             doZabsCreateStep,
+	Run:             doHoldsCreateStep,
 	NoRequireConfig: true,
 	Short:           `create a step hold or bookmark`,
 	SetupFlags: func(f *pflag.FlagSet) {
-		f.StringVarP(&zabsCreateStepHoldFlags.target, "target", "t", "", "snapshot to be held / bookmark to be held")
-		f.VarP(&zabsCreateStepHoldFlags.jobid, "jobid", "j", "jobid for which the hold is installed")
+		f.StringVarP(&holdsCreateStepHoldFlags.target, "target", "t", "", "snapshot to be held / bookmark to be held")
+		f.VarP(&holdsCreateStepHoldFlags.jobid, "jobid", "j", "jobid for which the hold is installed")
 	},
 }
 
-func doZabsCreateStep(ctx context.Context, sc *cli.Subcommand, args []string) error {
+func doHoldsCreateStep(sc *cli.Subcommand, args []string) error {
 	if len(args) > 0 {
 		return errors.New("subcommand takes no arguments")
 	}
 
-	f := &zabsCreateStepHoldFlags
+	f := &holdsCreateStepHoldFlags
 
 	fs, _, _, err := zfs.DecomposeVersionString(f.target)
 	if err != nil {
@@ -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/holds.go

@@ -14,15 +14,15 @@ import (
 )
 
 var (
-	ZFSAbstractionsCmd = &cli.Subcommand{
-		Use:   "zfs-abstraction",
-		Short: "manage abstractions that zrepl builds on top of ZFS",
+	HoldsCmd = &cli.Subcommand{
+		Use:   "holds",
+		Short: "manage holds & step bookmarks",
 		SetupSubcommands: func() []*cli.Subcommand {
 			return []*cli.Subcommand{
-				zabsCmdList,
-				zabsCmdReleaseAll,
-				zabsCmdReleaseStale,
-				zabsCmdCreate,
+				holdsCmdList,
+				holdsCmdReleaseAll,
+				holdsCmdReleaseStale,
+				holdsCmdCreate,
 			}
 		},
 	}
@@ -30,7 +30,7 @@ var (
 
 // a common set of CLI flags that map to the fields of an
 // endpoint.ListZFSHoldsAndBookmarksQuery
-type zabsFilterFlags struct {
+type holdsFilterFlags struct {
 	Filesystems FilesystemsFilterFlag
 	Job         JobIDFlag
 	Types       AbstractionTypesFlag
@@ -38,7 +38,7 @@ type zabsFilterFlags struct {
 }
 
 // produce a query from the CLI flags
-func (f zabsFilterFlags) Query() (endpoint.ListZFSHoldsAndBookmarksQuery, error) {
+func (f holdsFilterFlags) Query() (endpoint.ListZFSHoldsAndBookmarksQuery, error) {
 	q := endpoint.ListZFSHoldsAndBookmarksQuery{
 		FS:          f.Filesystems.FlagValue(),
 		What:        f.Types.FlagValue(),
@@ -48,7 +48,7 @@ func (f zabsFilterFlags) Query() (endpoint.ListZFSHoldsAndBookmarksQuery, error)
 	return q, q.Validate()
 }
 
-func (f *zabsFilterFlags) registerZabsFilterFlags(s *pflag.FlagSet, verb string) {
+func (f *holdsFilterFlags) registerHoldsFilterFlags(s *pflag.FlagSet, verb string) {
 	// Note: the default value is defined in the .FlagValue methods
 	s.Var(&f.Filesystems, "fs", fmt.Sprintf("only %s holds on the specified filesystem [default: all filesystems] [comma-separated list of <dataset-pattern>:<ok|!> pairs]", verb))
 	s.Var(&f.Job, "job", fmt.Sprintf("only %s holds created by the specified job [default: any job]", verb))

client/holds_create.go

@@ -0,0 +1,14 @@
+package client
+
+import "github.com/zrepl/zrepl/cli"
+
+var holdsCmdCreate = &cli.Subcommand{
+	Use:             "create",
+	NoRequireConfig: true,
+	Short:           `create zrepl-managed holds and boomkmarks (for debugging & development only!)`,
+	SetupSubcommands: func() []*cli.Subcommand {
+		return []*cli.Subcommand{
+			holdsCmdCreateStepHold,
+		}
+	},
+}

client/holds_list.go

@@ -16,30 +16,31 @@ import (
 	"github.com/zrepl/zrepl/util/chainlock"
 )
 
-var zabsListFlags struct {
-	Filter zabsFilterFlags
+var holdsListFlags struct {
+	Filter holdsFilterFlags
 	Json   bool
 }
 
-var zabsCmdList = &cli.Subcommand{
+var holdsCmdList = &cli.Subcommand{
 	Use:             "list",
-	Short:           `list zrepl ZFS abstractions`,
-	Run:             doZabsList,
+	Run:             doHoldsList,
 	NoRequireConfig: true,
+	Short:           "list holds and bookmarks",
 	SetupFlags: func(f *pflag.FlagSet) {
-		zabsListFlags.Filter.registerZabsFilterFlags(f, "list")
-		f.BoolVar(&zabsListFlags.Json, "json", false, "emit JSON")
+		holdsListFlags.Filter.registerHoldsFilterFlags(f, "list")
+		f.BoolVar(&holdsListFlags.Json, "json", false, "emit JSON")
 	},
 }
 
-func doZabsList(ctx context.Context, sc *cli.Subcommand, args []string) error {
+func doHoldsList(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")
 	}
 
-	q, err := zabsListFlags.Filter.Query()
+	q, err := holdsListFlags.Filter.Query()
 	if err != nil {
 		return errors.Wrap(err, "invalid filter specification on command line")
 	}
@@ -61,7 +62,7 @@ func doZabsList(ctx context.Context, sc *cli.Subcommand, args []string) error {
 		for a := range abstractions {
 			func() {
 				defer line.Lock().Unlock()
-				if zabsListFlags.Json {
+				if holdsListFlags.Json {
 					enc.SetIndent("", "  ")
 					if err := enc.Encode(abstractions); err != nil {
 						panic(err)

client/holds_release.go

@@ -15,42 +15,43 @@ import (
 )
 
 // shared between release-all and release-step
-var zabsReleaseFlags struct {
-	Filter zabsFilterFlags
+var holdsReleaseFlags struct {
+	Filter holdsFilterFlags
 	Json   bool
 	DryRun bool
 }
 
-func registerZabsReleaseFlags(s *pflag.FlagSet) {
-	zabsReleaseFlags.Filter.registerZabsFilterFlags(s, "release")
-	s.BoolVar(&zabsReleaseFlags.Json, "json", false, "emit json instead of pretty-printed")
-	s.BoolVar(&zabsReleaseFlags.DryRun, "dry-run", false, "do a dry-run")
+func registerHoldsReleaseFlags(s *pflag.FlagSet) {
+	holdsReleaseFlags.Filter.registerHoldsFilterFlags(s, "release")
+	s.BoolVar(&holdsReleaseFlags.Json, "json", false, "emit json instead of pretty-printed")
+	s.BoolVar(&holdsReleaseFlags.DryRun, "dry-run", false, "do a dry-run")
 }
 
-var zabsCmdReleaseAll = &cli.Subcommand{
+var holdsCmdReleaseAll = &cli.Subcommand{
 	Use:             "release-all",
-	Run:             doZabsReleaseAll,
+	Run:             doHoldsReleaseAll,
 	NoRequireConfig: true,
-	Short:           `(DANGEROUS) release ALL zrepl ZFS abstractions (mostly useful for uninstalling zrepl completely or for "de-zrepl-ing" a filesystem)`,
-	SetupFlags:      registerZabsReleaseFlags,
+	Short:           `(DANGEROUS) release all zrepl-managed holds and bookmarks, mostly useful for uninstalling zrepl`,
+	SetupFlags:      registerHoldsReleaseFlags,
 }
 
-var zabsCmdReleaseStale = &cli.Subcommand{
+var holdsCmdReleaseStale = &cli.Subcommand{
 	Use:             "release-stale",
-	Run:             doZabsReleaseStale,
+	Run:             doHoldsReleaseStale,
 	NoRequireConfig: true,
-	Short:           `release stale zrepl ZFS abstractions (useful if zrepl has a bug and does not do it by itself)`,
-	SetupFlags:      registerZabsReleaseFlags,
+	Short:           `release stale zrepl-managed holds and boomkarks (useful if zrepl has a bug and doesn't do it by itself)`,
+	SetupFlags:      registerHoldsReleaseFlags,
 }
 
-func doZabsReleaseAll(ctx context.Context, sc *cli.Subcommand, args []string) error {
+func doHoldsReleaseAll(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")
 	}
 
-	q, err := zabsReleaseFlags.Filter.Query()
+	q, err := holdsReleaseFlags.Filter.Query()
 	if err != nil {
 		return errors.Wrap(err, "invalid filter specification on command line")
 	}
@@ -64,18 +65,19 @@ func doZabsReleaseAll(ctx context.Context, sc *cli.Subcommand, args []string) er
 		// proceed anyways with rest of abstractions
 	}
 
-	return doZabsRelease_Common(ctx, abstractions)
+	return doHoldsRelease_Common(ctx, abstractions)
 }
 
-func doZabsReleaseStale(ctx context.Context, sc *cli.Subcommand, args []string) error {
+func doHoldsReleaseStale(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")
 	}
 
-	q, err := zabsReleaseFlags.Filter.Query()
+	q, err := holdsReleaseFlags.Filter.Query()
 	if err != nil {
 		return errors.Wrap(err, "invalid filter specification on command line")
 	}
@@ -85,13 +87,13 @@ func doZabsReleaseStale(ctx context.Context, sc *cli.Subcommand, args []string)
 		return err // context clear by invocation of command
 	}
 
-	return doZabsRelease_Common(ctx, stalenessInfo.Stale)
+	return doHoldsRelease_Common(ctx, stalenessInfo.Stale)
 }
 
-func doZabsRelease_Common(ctx context.Context, destroy []endpoint.Abstraction) error {
+func doHoldsRelease_Common(ctx context.Context, destroy []endpoint.Abstraction) error {
 
-	if zabsReleaseFlags.DryRun {
-		if zabsReleaseFlags.Json {
+	if holdsReleaseFlags.DryRun {
+		if holdsReleaseFlags.Json {
 			m, err := json.MarshalIndent(destroy, "", "  ")
 			if err != nil {
 				panic(err)
@@ -119,7 +121,7 @@ func doZabsRelease_Common(ctx context.Context, destroy []endpoint.Abstraction) e
 
 	for res := range outcome {
 		hadErr = hadErr || res.DestroyErr != nil
-		if zabsReleaseFlags.Json {
+		if holdsReleaseFlags.Json {
 			err := enc.Encode(res)
 			if err != nil {
 				colorErr.Fprintf(os.Stderr, "cannot marshal there were errors in destroying the abstractions")

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

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.go

@@ -1,14 +0,0 @@
-package client
-
-import "github.com/zrepl/zrepl/cli"
-
-var zabsCmdCreate = &cli.Subcommand{
-	Use:             "create",
-	NoRequireConfig: true,
-	Short:           `create zrepl ZFS abstractions (mostly useful for debugging & development, users should not need to use this command)`,
-	SetupSubcommands: func() []*cli.Subcommand {
-		return []*cli.Subcommand{
-			zabsCmdCreateStepHold,
-		}
-	},
-}

daemon/daemon.go

@@ -12,7 +12,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/job"
@@ -24,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)
@@ -39,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 {
@@ -49,14 +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))
-
 	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
@@ -85,7 +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)
 
 	log.Info("starting daemon")
 
@@ -100,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
 }
@@ -124,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
 }
@@ -203,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))
@@ -219,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)
@@ -228,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() {
@@ -377,18 +376,15 @@ func (j *ActiveSide) SenderConfig() *endpoint.SenderConfig {
 }
 
 func (j *ActiveSide) Run(ctx context.Context) {
-	ctx, endTask := trace.WithTaskAndSpan(ctx, "active-side-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, "periodic")
-	defer endTask()
-	go j.mode.RunPeriodic(periodicCtx, periodicDone)
+	go j.mode.RunPeriodic(ctx, periodicDone)
 
 	invocationCount := 0
 outer:
@@ -404,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)
@@ -435,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()
 	}
 
 	{
@@ -459,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 {
@@ -478,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/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"
@@ -75,97 +83,28 @@ const (
 	SubsysZFSCmd       Subsystem = "zfs.cmd"
 )
 
-var AllSubsystems = []Subsystem{
-	SubsysMeta,
-	SubsysJob,
-	SubsysReplication,
-	SubsysEndpoint,
-	SubsysPruning,
-	SubsysSnapshot,
-	SubsysHooks,
-	SubsysTransport,
-	SubsysTransportMux,
-	SubsysRPC,
-	SubsysRPCControl,
-	SubsysRPCData,
-	SubsysZFSCmd,
+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, "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,355 +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
-	uniqueConcurrentTaskNameBitvecLength *prometheus.GaugeVec
-}
-
-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",
-	})
-	metrics.uniqueConcurrentTaskNameBitvecLength = prometheus.NewGaugeVec(prometheus.GaugeOpts{
-		Namespace: "zrepl",
-		Subsystem: "trace",
-		Name:      "unique_concurrent_task_name_bitvec_length",
-		Help:      "length of the bitvec used to find unique names for concurrent tasks",
-	}, []string{"task_name"})
-}
-
-func RegisterMetrics(r prometheus.Registerer) {
-	r.MustRegister(metrics.activeTasks)
-	r.MustRegister(metrics.uniqueConcurrentTaskNameBitvecLength)
-}
-
-var taskNamer = newUniqueTaskNamer(metrics.uniqueConcurrentTaskNameBitvecLength)
-
-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
-}
-
-// 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")
-var ErrParentTaskAlreadyEnded = fmt.Errorf("create task: parent task already ended")
-
-// 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
-		}
-	}
-
-	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 this.parentTask != nil {
-		this.parentTask.mtx.HoldWhile(func() {
-			if !this.parentTask.endedAt.IsZero() {
-				panic(ErrParentTaskAlreadyEnded)
-			}
-			this.parentTask.activeChildTasks++
-		})
-	}
-
-	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.activeChildSpan))
-			}
-
-			// 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)
-
-	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)
-	}
-
-	return ctx, endTaskFunc
-}
-
-func currentTaskNameAndSpanStack(this *traceNode) (taskName string, spanIdStack string) {
-
-	task := this.parentTask
-	if this.parentSpan == nil {
-		task = this
-	}
-
-	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, tasks[i].id)
-	}
-
-	var spanIdsRev []string
-	for i := len(spansInTask) - 1; i >= 0; i-- {
-		spanIdsRev = append(spanIdsRev, spansInTask[i].id)
-	}
-
-	taskStack := strings.Join(taskIdsRev, "$")
-	spanIdStack = fmt.Sprintf("%s$%s", taskStack, strings.Join(spanIdsRev, "."))
-
-	return task.annotation, spanIdStack
-}
-
-func GetSpanStackOrDefault(ctx context.Context, def string) string {
-	if nI := ctx.Value(contextKeyTraceNode); nI != nil {
-		n := nI.(*traceNode)
-		_, spanStack := currentTaskNameAndSpanStack(n)
-		return spanStack
-	} else {
-		return def
-	}
-}

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, _ := currentTaskNameAndSpanStack(s)
-	chrometraceWrite(chrometraceEvent{
-		Name:                      s.annotation,
-		Phase:                     "B",
-		TimestampUnixMicroseconds: s.startedAt.UnixNano() / 1000,
-		Pid:                       chrometracePID,
-		Tid:                       taskName,
-	})
-}
-
-func chrometraceEndSpan(s *traceNode) {
-	taskName, _ := currentTaskNameAndSpanStack(s)
-	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, _ := currentTaskNameAndSpanStack(s.parentTask)
-	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, _ := currentTaskNameAndSpanStack(s)
-	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,172 +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 TestCannotReuseEndedTask(t *testing.T) {
-	root, end := WithTask(context.Background(), "root")
-	end()
-	require.PanicsWithValue(t, ErrParentTaskAlreadyEnded, func() { WithTask(root, "child-after-parent-ended") })
-}
-
-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()
-}

daemon/logging/trace/trace_unique_concurrent_task_namer.go

@@ -1,61 +0,0 @@
-package trace
-
-import (
-	"fmt"
-	"strings"
-	"sync"
-
-	"github.com/prometheus/client_golang/prometheus"
-	"github.com/willf/bitset"
-)
-
-type uniqueConcurrentTaskNamer struct {
-	mtx    sync.Mutex
-	active map[string]*bitset.BitSet
-
-	bitvecLengthGauge *prometheus.GaugeVec
-}
-
-// bitvecLengthGauge may be nil
-func newUniqueTaskNamer(bitvecLengthGauge *prometheus.GaugeVec) *uniqueConcurrentTaskNamer {
-	return &uniqueConcurrentTaskNamer{
-		active:            make(map[string]*bitset.BitSet),
-		bitvecLengthGauge: bitvecLengthGauge,
-	}
-}
-
-// 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()
-
-	if namer.bitvecLengthGauge != nil {
-		namer.bitvecLengthGauge.WithLabelValues(name).Set(float64(act.Len()))
-	}
-
-	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(nil)
-
-	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,9 +8,6 @@ import (
 	"net"
 	"net/http/pprof"
 
-	"github.com/zrepl/zrepl/daemon/logging/trace"
-	"golang.org/x/net/websocket"
-
 	"github.com/zrepl/zrepl/daemon/job"
 )
 
@@ -67,7 +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("/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/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"
@@ -36,13 +35,17 @@ 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 {
@@ -135,7 +138,7 @@ func NewPrunerFactory(in config.PruningSenderReceiver, promPruneSecs *prometheus
 func (f *PrunerFactory) BuildSenderPruner(ctx context.Context, target Target, receiver History) *Pruner {
 	p := &Pruner{
 		args: args{
-			context.WithValue(ctx, contextKeyPruneSide, "sender"),
+			WithLogger(ctx, GetLogger(ctx).WithField("prune_side", "sender")),
 			target,
 			receiver,
 			f.senderRules,
@@ -151,7 +154,7 @@ func (f *PrunerFactory) BuildSenderPruner(ctx context.Context, target Target, re
 func (f *PrunerFactory) BuildReceiverPruner(ctx context.Context, target Target, receiver History) *Pruner {
 	p := &Pruner{
 		args: args{
-			context.WithValue(ctx, contextKeyPruneSide, "receiver"),
+			WithLogger(ctx, GetLogger(ctx).WithField("prune_side", "receiver")),
 			target,
 			receiver,
 			f.receiverRules,
@@ -167,7 +170,7 @@ func (f *PrunerFactory) BuildReceiverPruner(ctx context.Context, target Target,
 func (f *LocalPrunerFactory) BuildLocalPruner(ctx context.Context, target Target, receiver History) *Pruner {
 	p := &Pruner{
 		args: args{
-			context.WithValue(ctx, contextKeyPruneSide, "local"),
+			ctx,
 			target,
 			receiver,
 			f.keepRules,

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(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/changelog.rst

@@ -62,9 +62,7 @@ Actual changelog:
 * |bugfix| |docs| snapshotting: clarify sync-up behavior and warn about filesystems
   that will not be snapshotted until the sync-up phase is over
 * |docs| Document new replication features in the :ref:`config overview <overview-how-replication-works>` and :repomasterlink:`replication/design.md`.
-* |feature| documented subcommand to generate ``bash`` and ``zsh`` completions
 * **[MAINTAINER NOTICE]** New platform tests in this version, please make sure you run them for your distro!
-* **[MAINTAINER NOTICE]** Please add the shell completions to the zrepl packages.
 
 0.2.1
 -----

docs/usage.rst

@@ -38,8 +38,8 @@ CLI Overview
     * - ``zrepl migrate``
       - | perform on-disk state / ZFS property migrations
         | (see :ref:`changelog <changelog>` for details)
-    * - ``zrepl zfs-abstractions``
-      - list and remove zrepl's abstractions on top of ZFS, e.g. holds and step bookmarks (see :ref:`overview <replication-cursor-and-last-received-hold>` )
+    * - ``zrepl holds``
+      - list and remove holds and step bookmarks created by zrepl (see :ref:`overview <replication-cursor-and-last-received-hold>` )
 
 .. _usage-zrepl-daemon:
 

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

@@ -2,18 +2,14 @@
 package endpoint
 
 import (
-	"bytes"
 	"context"
 	"fmt"
-	"io"
 	"path"
 	"sync"
 
 	"github.com/pkg/errors"
-	"github.com/zrepl/zrepl/daemon/logging/trace"
 
 	"github.com/zrepl/zrepl/replication/logic/pdu"
-	"github.com/zrepl/zrepl/util/chainedio"
 	"github.com/zrepl/zrepl/util/chainlock"
 	"github.com/zrepl/zrepl/util/envconst"
 	"github.com/zrepl/zrepl/util/semaphore"
@@ -74,8 +70,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,13 +92,11 @@ 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
 	}
-	fsvs, err := zfs.ZFSListFilesystemVersions(ctx, lp, zfs.ListFilesystemVersionsOptions{})
+	fsvs, err := zfs.ZFSListFilesystemVersions(lp, zfs.ListFilesystemVersionsOptions{})
 	if err != nil {
 		return nil, err
 	}
@@ -118,7 +110,6 @@ func (s *Sender) ListFilesystemVersions(ctx context.Context, r *pdu.ListFilesyst
 }
 
 func (p *Sender) HintMostRecentCommonAncestor(ctx context.Context, r *pdu.HintMostRecentCommonAncestorReq) (*pdu.HintMostRecentCommonAncestorRes, error) {
-	defer trace.WithSpanFromStackUpdateCtx(&ctx)()
 
 	fsp, err := p.filterCheckFS(r.GetFilesystem())
 	if err != nil {
@@ -250,8 +241,7 @@ func sendArgsFromPDUAndValidateExistsAndGetVersion(ctx context.Context, fs strin
 	return version, nil
 }
 
-func (s *Sender) Send(ctx context.Context, r *pdu.SendReq) (*pdu.SendRes, io.ReadCloser, error) {
-	defer trace.WithSpanFromStackUpdateCtx(&ctx)()
+func (s *Sender) Send(ctx context.Context, r *pdu.SendReq) (*pdu.SendRes, zfs.StreamCopier, error) {
 
 	_, err := s.filterCheckFS(r.Filesystem)
 	if err != nil {
@@ -349,15 +339,14 @@ func (s *Sender) Send(ctx context.Context, r *pdu.SendReq) (*pdu.SendRes, io.Rea
 
 	// step holds & replication cursor released / moved forward in s.SendCompleted => s.moveCursorAndReleaseSendHolds
 
-	sendStream, err := zfs.ZFSSend(ctx, sendArgs)
+	streamCopier, err := zfs.ZFSSend(ctx, sendArgs)
 	if err != nil {
 		return nil, nil, errors.Wrap(err, "zfs send failed")
 	}
-	return res, sendStream, nil
+	return res, streamCopier, 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())
@@ -379,30 +368,27 @@ 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)
 	}
 
-	log(ctx).Debug("move replication cursor to most recent common version")
+	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 release the hold if we can't move the cursor!
 			return &pdu.SendCompletedRes{}, err
 		}
 	} else {
-		log(ctx).Info("successfully moved replication cursor")
+		log.Info("successfully moved replication cursor")
 	}
 
 	// kick off releasing of step holds / bookmarks
@@ -412,27 +398,21 @@ func (p *Sender) SendCompleted(ctx context.Context, r *pdu.SendCompletedReq) (*p
 	wg.Add(2)
 	go func() {
 		defer wg.Done()
-		ctx, endTask := trace.WithTask(ctx, "release-step-hold-to")
-		defer endTask()
-
-		log(ctx).Debug("release step-hold of or step-bookmark on `to`")
+		log.Debug("release step-hold of or step-bookmark on `to`")
 		err = ReleaseStep(ctx, fs, to, p.jobId)
 		if err != nil {
-			log(ctx).WithError(err).Error("cannot release step-holds on or destroy step-bookmark of `to`")
+			log.WithError(err).Error("cannot release step-holds on or destroy step-bookmark of `to`")
 		} else {
-			log(ctx).Info("successfully released step-holds on or destroyed step-bookmark of `to`")
+			log.Info("successfully released step-holds on or destroyed step-bookmark of `to`")
 		}
 
 	}()
 	go func() {
 		defer wg.Done()
-		ctx, endTask := trace.WithTask(ctx, "release-step-hold-from")
-		defer endTask()
-
 		if from == nil {
 			return
 		}
-		log(ctx).Debug("release step-hold of or step-bookmark on `from`")
+		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 {
@@ -441,15 +421,15 @@ func (p *Sender) SendCompleted(ctx context.Context, r *pdu.SendCompletedReq) (*p
 				// In that case, nonexistence of `from` is not an error, otherwise it is.
 				for _, c := range destroyedCursors {
 					if c.GetFullPath() == dne.Path {
-						log(ctx).Info("`from` was a replication cursor and has already been destroyed")
+						log.Info("`from` was a replication cursor and has already been destroyed")
 						return
 					}
 				}
 				// fallthrough
 			}
-			log(ctx).WithError(err).Error("cannot release step-holds on or destroy step-bookmark of `from`")
+			log.WithError(err).Error("cannot release step-holds on or destroy step-bookmark of `from`")
 		} else {
-			log(ctx).Info("successfully released step-holds on or destroyed step-bookmark of `from`")
+			log.Info("successfully released step-holds on or destroyed step-bookmark of `from`")
 		}
 	}()
 	wg.Wait()
@@ -458,8 +438,6 @@ func (p *Sender) SendCompleted(ctx context.Context, r *pdu.SendCompletedReq) (*p
 }
 
 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
@@ -468,8 +446,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(),
 	}
@@ -477,20 +453,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
@@ -506,7 +476,7 @@ func (p *Sender) ReplicationCursor(ctx context.Context, req *pdu.ReplicationCurs
 	return &pdu.ReplicationCursorRes{Result: &pdu.ReplicationCursorRes_Guid{Guid: cursor.Guid}}, nil
 }
 
-func (p *Sender) Receive(ctx context.Context, r *pdu.ReceiveReq, _ io.ReadCloser) (*pdu.ReceiveRes, error) {
+func (p *Sender) Receive(ctx context.Context, r *pdu.ReceiveReq, receive zfs.StreamCopier) (*pdu.ReceiveRes, error) {
 	return nil, fmt.Errorf("sender does not implement Receive()")
 }
 
@@ -621,15 +591,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 {
@@ -680,8 +641,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 {
@@ -689,7 +648,7 @@ func (s *Receiver) ListFilesystemVersions(ctx context.Context, req *pdu.ListFile
 	}
 	// TODO share following code with sender
 
-	fsvs, err := zfs.ZFSListFilesystemVersions(ctx, lp, zfs.ListFilesystemVersionsOptions{})
+	fsvs, err := zfs.ZFSListFilesystemVersions(lp, zfs.ListFilesystemVersionsOptions{})
 	if err != nil {
 		return nil, err
 	}
@@ -703,8 +662,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(),
 	}
@@ -712,30 +669,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)()
+func (s *Receiver) Send(ctx context.Context, req *pdu.SendReq) (*pdu.SendRes, zfs.StreamCopier, error) {
 	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)()
-
+func (s *Receiver) Receive(ctx context.Context, req *pdu.ReceiveReq, receive zfs.StreamCopier) (*pdu.ReceiveRes, error) {
 	getLogger(ctx).Debug("incoming Receive")
 	defer receive.Close()
 
@@ -814,30 +765,21 @@ func (s *Receiver) Receive(ctx context.Context, req *pdu.ReceiveReq, receive io.
 		return nil, visitErr
 	}
 
-	log := getLogger(ctx).WithField("proto_fs", req.GetFilesystem()).WithField("local_fs", lp.ToString())
-
 	// determine whether we need to rollback the filesystem / change its placeholder state
 	var clearPlaceholderProperty bool
 	var recvOpts zfs.RecvOptions
 	ph, err := zfs.ZFSGetFilesystemPlaceholderState(ctx, lp)
-	if err != nil {
-		return nil, errors.Wrap(err, "cannot get placeholder state")
-	}
-	log.WithField("placeholder_state", fmt.Sprintf("%#v", ph)).Debug("placeholder state")
-	if ph.FSExists && ph.IsPlaceholder {
+	if err == nil && ph.FSExists && ph.IsPlaceholder {
 		recvOpts.RollbackAndForceRecv = true
 		clearPlaceholderProperty = true
 	}
-
 	if clearPlaceholderProperty {
-		log.Info("clearing placeholder property")
 		if err := zfs.ZFSSetPlaceholder(ctx, lp, false); err != nil {
 			return nil, fmt.Errorf("cannot clear placeholder property for forced receive: %s", err)
 		}
 	}
 
 	if req.ClearResumeToken && ph.FSExists {
-		log.Info("clearing resume token")
 		if err := zfs.ZFSRecvClearResumeToken(ctx, lp.ToString()); err != nil {
 			return nil, errors.Wrap(err, "cannot clear resume token")
 		}
@@ -848,7 +790,7 @@ func (s *Receiver) Receive(ctx context.Context, req *pdu.ReceiveReq, receive io.
 		return nil, errors.Wrap(err, "cannot determine whether we can use resumable send & recv")
 	}
 
-	log.Debug("acquire concurrent recv semaphore")
+	getLogger(ctx).Debug("acquire concurrent recv semaphore")
 	// TODO use try-acquire and fail with resource-exhaustion rpc status
 	// => would require handling on the client-side
 	// => this is a dataconn endpoint, doesn't have the status code semantics of gRPC
@@ -858,89 +800,14 @@ 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)
-	if _, err := io.Copy(&peek, io.LimitReader(receive, MaxPeek)); err != nil {
-		log.WithError(err).Error("cannot read peek-buffer from send stream")
-	}
-	var peekCopy bytes.Buffer
-	if n, err := peekCopy.Write(peek.Bytes()); err != nil || n != peek.Len() {
-		panic(peek.Len())
-	}
-
-	log.WithField("opts", fmt.Sprintf("%#v", recvOpts)).Debug("start receive command")
+	getLogger(ctx).WithField("opts", fmt.Sprintf("%#v", recvOpts)).Debug("start receive command")
 
 	snapFullPath := to.FullPath(lp.ToString())
-	if err := zfs.ZFSRecv(ctx, lp.ToString(), to, chainedio.NewChainedReader(&peek, receive), recvOpts); err != nil {
-
-		// best-effort rollback of placeholder state if the recv didn't start
-		_, resumableStatePresent := err.(*zfs.RecvFailedWithResumeTokenErr)
-		disablePlaceholderRestoration := envconst.Bool("ZREPL_ENDPOINT_DISABLE_PLACEHOLDER_RESTORATION", false)
-		placeholderRestored := !ph.IsPlaceholder
-		if !disablePlaceholderRestoration && !resumableStatePresent && recvOpts.RollbackAndForceRecv && ph.FSExists && ph.IsPlaceholder && clearPlaceholderProperty {
-			log.Info("restoring placeholder property")
-			if phErr := zfs.ZFSSetPlaceholder(ctx, lp, true); phErr != nil {
-				log.WithError(phErr).Error("cannot restore placeholder property after failed receive, subsequent replications will likely fail with a different error")
-				// fallthrough
-			} else {
-				placeholderRestored = true
-			}
-			// fallthrough
-		}
-
-		// deal with failing initial encrypted send & recv
-		if _, ok := err.(*zfs.RecvDestroyOrOverwriteEncryptedErr); ok && ph.IsPlaceholder && placeholderRestored {
-			msg := `cannot automatically replace placeholder filesystem with incoming send stream - please see receive-side log for details`
-			err := errors.New(msg)
-			log.Error(msg)
-
-			log.Error(`zrepl creates placeholder filesystems on the receiving side of a replication to match the sending side's dataset hierarchy`)
-			log.Error(`zrepl uses zfs receive -F to replace those placeholders with incoming full sends`)
-			log.Error(`OpenZFS native encryption prohibits zfs receive -F for encrypted filesystems`)
-			log.Error(`the current zrepl placeholder filesystem concept is thus incompatible with OpenZFS native encryption`)
-
-			tempStartFullRecvFS := lp.Copy().ToString() + ".zrepl.initial-recv"
-			tempStartFullRecvFSDP, dpErr := zfs.NewDatasetPath(tempStartFullRecvFS)
-			if dpErr != nil {
-				log.WithError(dpErr).Error("cannot determine temporary filesystem name for initial encrypted recv workaround")
-				return nil, err // yes, err, not dpErr
-			}
-
-			log := log.WithField("temp_recv_fs", tempStartFullRecvFS)
-			log.Error(`as a workaround, zrepl will now attempt to re-receive the beginning of the stream into a temporary filesystem temp_recv_fs`)
-			log.Error(`if that step succeeds: shut down zrepl and use 'zfs rename' to swap temp_recv_fs with local_fs, then restart zrepl`)
-			log.Error(`replication will then resume using resumable send+recv`)
-
-			tempPH, phErr := zfs.ZFSGetFilesystemPlaceholderState(ctx, tempStartFullRecvFSDP)
-			if phErr != nil {
-				log.WithError(phErr).Error("cannot determine placeholder state of temp_recv_fs")
-				return nil, err // yes, err, not dpErr
-			}
-			if tempPH.FSExists {
-				log.Error("temp_recv_fs already exists, assuming a (partial) initial recv to that filesystem has already been done")
-				return nil, err
-			}
-
-			recvOpts.RollbackAndForceRecv = false
-			recvOpts.SavePartialRecvState = true
-			rerecvErr := zfs.ZFSRecv(ctx, tempStartFullRecvFS, to, chainedio.NewChainedReader(&peekCopy), recvOpts)
-			if _, isResumable := rerecvErr.(*zfs.RecvFailedWithResumeTokenErr); rerecvErr == nil || isResumable {
-				log.Error("completed re-receive into temporary filesystem temp_recv_fs, now shut down zrepl and use zfs rename to swap temp_recv_fs with local_fs")
-			} else {
-				log.WithError(rerecvErr).Error("failed to receive the beginning of the stream into temporary filesystem temp_recv_fs")
-				log.Error("we advise you to collect the error log and current configuration, open an issue on GitHub, and revert to your previous configuration in the meantime")
-			}
-
-			log.Error(`if you would like to see improvements to this situation, please open an issue on GitHub`)
-			return nil, err
-		}
-
-		log.
+	if err := zfs.ZFSRecv(ctx, lp.ToString(), to, receive, recvOpts); err != nil {
+		getLogger(ctx).
 			WithError(err).
 			WithField("opts", fmt.Sprintf("%#v", recvOpts)).
 			Error("zfs receive failed")
-
 		return nil, err
 	}
 
@@ -948,13 +815,13 @@ func (s *Receiver) Receive(ctx context.Context, req *pdu.ReceiveReq, receive io.
 	toRecvd, err := to.ValidateExistsAndGetVersion(ctx, lp.ToString())
 	if err != nil {
 		msg := "receive request's `To` version does not match what we received in the stream"
-		log.WithError(err).WithField("snap", snapFullPath).Error(msg)
-		log.Error("aborting recv request, but keeping received snapshot for inspection")
+		getLogger(ctx).WithError(err).WithField("snap", snapFullPath).Error(msg)
+		getLogger(ctx).Error("aborting recv request, but keeping received snapshot for inspection")
 		return nil, errors.Wrap(err, msg)
 	}
 
 	if s.conf.UpdateLastReceivedHold {
-		log.Debug("move last-received-hold")
+		getLogger(ctx).Debug("move last-received-hold")
 		if err := MoveLastReceivedHold(ctx, lp.ToString(), toRecvd, s.conf.JobID); err != nil {
 			return nil, errors.Wrap(err, "cannot move last-received-hold")
 		}
@@ -964,8 +831,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 {
@@ -975,7 +840,6 @@ func (s *Receiver) DestroySnapshots(ctx context.Context, req *pdu.DestroySnapsho
 }
 
 func (p *Receiver) HintMostRecentCommonAncestor(ctx context.Context, r *pdu.HintMostRecentCommonAncestorReq) (*pdu.HintMostRecentCommonAncestorRes, error) {
-	defer trace.WithSpanFromStackUpdateCtx(&ctx)()
 	// we don't move last-received-hold as part of this hint
 	// because that wouldn't give us any benefit wrt resumability.
 	//
@@ -984,9 +848,7 @@ func (p *Receiver) HintMostRecentCommonAncestor(ctx context.Context, r *pdu.Hint
 	return &pdu.HintMostRecentCommonAncestorRes{}, nil
 }
 
-func (p *Receiver) SendCompleted(ctx context.Context, _ *pdu.SendCompletedReq) (*pdu.SendCompletedRes, error) {
-	defer trace.WithSpanFromStackUpdateCtx(&ctx)()
-
+func (p *Receiver) SendCompleted(context.Context, *pdu.SendCompletedReq) (*pdu.SendCompletedRes, error) {
 	return &pdu.SendCompletedRes{}, nil
 }
 
@@ -1008,7 +870,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_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"
@@ -468,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"))
 }
@@ -530,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())
@@ -549,7 +547,7 @@ func ListAbstractionsStreamed(ctx context.Context, query ListZFSHoldsAndBookmark
 					defer g.Release()
 					listAbstractionsImplFS(ctx, fss[i], &query, emitAbstraction, errCb)
 				}()
-			})
+			}(i)
 		}
 	}()
 
@@ -575,7 +573,7 @@ func listAbstractionsImplFS(ctx context.Context, fs string, query *ListZFSHoldsA
 			whatTypes[zfs.Snapshot] = true
 		}
 	}
-	fsvs, err := zfs.ZFSListFilesystemVersions(ctx, fsp, zfs.ListFilesystemVersionsOptions{
+	fsvs, err := zfs.ZFSListFilesystemVersions(fsp, zfs.ListFilesystemVersionsOptions{
 		Types: whatTypes,
 	})
 	if err != nil {
@@ -733,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]
 

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,12 +23,10 @@ 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

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=
@@ -281,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=
@@ -372,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=

main.go

@@ -17,7 +17,7 @@ func init() {
 	cli.AddSubcommand(client.PprofCmd)
 	cli.AddSubcommand(client.TestCmd)
 	cli.AddSubcommand(client.MigrateCmd)
-	cli.AddSubcommand(client.ZFSAbstractionsCmd)
+	cli.AddSubcommand(client.HoldsCmd)
 }
 
 func main() {

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"
@@ -69,9 +68,7 @@ func doMain() error {
 		logger.Error(err.Error())
 		panic(err)
 	}
-	ctx := context.Background()
-	defer trace.WithTaskFromStackUpdateCtx(&ctx)()
-	ctx = platformtest.WithLogger(ctx, logger)
+	ctx := platformtest.WithLogger(context.Background(), logger)
 	ex := platformtest.NewEx(logger)
 
 	type invocation struct {

platformtest/tests/batchDestroy.go

@@ -32,7 +32,7 @@ func BatchDestroy(ctx *platformtest.Context) {
 			Name:       "2",
 		},
 	}
-	zfs.ZFSDestroyFilesystemVersions(ctx, reqs)
+	zfs.ZFSDestroyFilesystemVersions(reqs)
 	if *reqs[0].ErrOut != nil {
 		panic("expecting no error")
 	}

platformtest/tests/helpers.go

@@ -136,7 +136,7 @@ func makeResumeSituation(ctx *platformtest.Context, src dummySnapshotSituation,
 		return situation
 	}
 
-	limitedCopier := limitio.ReadCloser(copier, src.dummyDataLen/2)
+	limitedCopier := zfs.NewReadCloserCopier(limitio.ReadCloser(copier, src.dummyDataLen/2))
 	defer limitedCopier.Close()
 
 	require.NotNil(ctx, sendArgs.To)

platformtest/tests/listFilesystemVersions.go

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

platformtest/tests/sendArgsValidation.go

@@ -35,7 +35,7 @@ func SendArgsValidationEncryptedSendOfUnencryptedDatasetForbidden(ctx *platformt
 		ResumeToken: "",
 	}.Validate(ctx)
 
-	var stream *zfs.SendStream
+	var stream *zfs.ReadCloserCopier
 	if err == nil {
 		stream, err = zfs.ZFSSend(ctx, sendArgs) // no shadow
 		if err == nil {

replication/driver/replication_driver.go

@@ -10,7 +10,6 @@ import (
 	"sync"
 	"time"
 
-	"github.com/zrepl/zrepl/daemon/logging/trace"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
 
@@ -147,12 +146,6 @@ type fs struct {
 
 	l *chainlock.L
 
-	// ordering relationship that must be maintained for initial replication
-	initialRepOrd struct {
-		parents, children []*fs
-		parentDidUpdate   chan struct{}
-	}
-
 	planning struct {
 		done bool
 		err  *timedError
@@ -288,17 +281,6 @@ func Do(ctx context.Context, planner Planner) (ReportFunc, WaitFunc) {
 }
 
 func (a *attempt) do(ctx context.Context, prev *attempt) {
-	prevs := a.doGlobalPlanning(ctx, prev)
-	if prevs == nil {
-		return
-	}
-	a.doFilesystems(ctx, prevs)
-}
-
-// if no error occurs, returns a map that maps this attempt's a.fss to `prev`'s a.fss
-func (a *attempt) doGlobalPlanning(ctx context.Context, prev *attempt) map[*fs]*fs {
-	ctx, endSpan := trace.WithSpan(ctx, "plan")
-	defer endSpan()
 	pfss, err := a.planner.Plan(ctx)
 	errTime := time.Now()
 	defer a.l.Lock().Unlock()
@@ -306,7 +288,7 @@ func (a *attempt) doGlobalPlanning(ctx context.Context, prev *attempt) map[*fs]*
 		a.planErr = newTimedError(err, errTime)
 		a.fss = nil
 		a.finishedAt = time.Now()
-		return nil
+		return
 	}
 
 	for _, pfs := range pfss {
@@ -314,7 +296,6 @@ func (a *attempt) doGlobalPlanning(ctx context.Context, prev *attempt) map[*fs]*
 			fs: pfs,
 			l:  a.l,
 		}
-		fs.initialRepOrd.parentDidUpdate = make(chan struct{}, 1)
 		a.fss = append(a.fss, fs)
 	}
 
@@ -363,7 +344,7 @@ func (a *attempt) doGlobalPlanning(ctx context.Context, prev *attempt) map[*fs]*
 			a.planErr = newTimedError(errors.New(msg.String()), now)
 			a.fss = nil
 			a.finishedAt = now
-			return nil
+			return
 		}
 		for cur, fss := range prevFSs {
 			if len(fss) > 0 {
@@ -373,27 +354,6 @@ func (a *attempt) doGlobalPlanning(ctx context.Context, prev *attempt) map[*fs]*
 	}
 	// invariant: prevs contains an entry for each unambiguous correspondence
 
-	// build up parent-child relationship (FIXME (O(n^2), but who's going to have that many filesystems...))
-	for _, f1 := range a.fss {
-		fs1 := f1.fs.ReportInfo().Name
-		for _, f2 := range a.fss {
-			fs2 := f2.fs.ReportInfo().Name
-			if strings.HasPrefix(fs1, fs2) && fs1 != fs2 {
-				f1.initialRepOrd.parents = append(f1.initialRepOrd.parents, f2)
-				f2.initialRepOrd.children = append(f2.initialRepOrd.children, f1)
-			}
-		}
-	}
-
-	return prevs
-}
-
-func (a *attempt) doFilesystems(ctx context.Context, prevs map[*fs]*fs) {
-	ctx, endSpan := trace.WithSpan(ctx, "do-repl")
-	defer endSpan()
-
-	defer a.l.Lock().Unlock()
-
 	stepQueue := newStepQueue()
 	defer stepQueue.Start(envconst.Int("ZREPL_REPLICATION_EXPERIMENTAL_REPLICATION_CONCURRENCY", 1))() // TODO parallel replication
 	var fssesDone sync.WaitGroup
@@ -401,9 +361,6 @@ func (a *attempt) doFilesystems(ctx context.Context, prevs map[*fs]*fs) {
 		fssesDone.Add(1)
 		go func(f *fs) {
 			defer fssesDone.Done()
-			// avoid explosion of tasks with name f.report().Info.Name
-			ctx, endTask := trace.WithTaskAndSpan(ctx, "repl-fs", f.report().Info.Name)
-			defer endTask()
 			f.do(ctx, stepQueue, prevs[f])
 		}(f)
 	}
@@ -417,27 +374,9 @@ func (f *fs) debug(format string, args ...interface{}) {
 	debugPrefix("fs=%s", f.fs.ReportInfo().Name)(format, args...)
 }
 
-// wake up children that watch for f.{planning.{err,done},planned.{step,stepErr}}
-func (f *fs) initialRepOrdWakeupChildren() {
-	var children []string
-	for _, c := range f.initialRepOrd.children {
-		// no locking required, c.fs does not change
-		children = append(children, c.fs.ReportInfo().Name)
-	}
-	f.debug("wakeup children %s", children)
-	for _, child := range f.initialRepOrd.children {
-		select {
-		// no locking required, child.initialRepOrd does not change
-		case child.initialRepOrd.parentDidUpdate <- struct{}{}:
-		default:
-		}
-	}
-}
-
 func (f *fs) do(ctx context.Context, pq *stepQueue, prev *fs) {
 
 	defer f.l.Lock().Unlock()
-	defer f.initialRepOrdWakeupChildren()
 
 	// get planned steps from replication logic
 	var psteps []Step
@@ -447,10 +386,11 @@ func (f *fs) do(ctx context.Context, pq *stepQueue, prev *fs) {
 		// TODO hacky
 		// choose target time that is earlier than any snapshot, so fs planning is always prioritized
 		targetDate := time.Unix(0, 0)
-		defer pq.WaitReady(ctx, f, targetDate)()
+		defer pq.WaitReady(f, targetDate)()
 		psteps, err = f.fs.PlanFS(ctx) // no shadow
 		errTime = time.Now()           // no shadow
 	})
+	f.planning.done = true
 	if err != nil {
 		f.planning.err = newTimedError(err, errTime)
 		return
@@ -462,8 +402,6 @@ func (f *fs) do(ctx context.Context, pq *stepQueue, prev *fs) {
 		}
 		f.planned.steps = append(f.planned.steps, step)
 	}
-	// we're not done planning yet, f.planned.steps might still be changed by next block
-	// => don't set f.planning.done just yet
 	f.debug("initial len(fs.planned.steps) = %d", len(f.planned.steps))
 
 	// for not-first attempts, only allow fs.planned.steps
@@ -518,112 +456,24 @@ func (f *fs) do(ctx context.Context, pq *stepQueue, prev *fs) {
 	}
 	f.debug("post-prev-merge len(fs.planned.steps) = %d", len(f.planned.steps))
 
-	// now we are done planning (f.planned.steps won't change from now on)
-	f.planning.done = true
-
-	// wait for parents' initial replication
-	var parents []string
-	for _, p := range f.initialRepOrd.parents {
-		parents = append(parents, p.fs.ReportInfo().Name)
-	}
-	f.debug("wait for parents %s", parents)
-	for {
-		var initialReplicatingParentsWithErrors []string
-		allParentsPresentOnReceiver := true
-		f.l.DropWhile(func() {
-			for _, p := range f.initialRepOrd.parents {
-				p.l.HoldWhile(func() {
-					// (get the preconditions that allow us to inspect p.planned)
-					parentHasPlanningDone := p.planning.done && p.planning.err == nil
-					if !parentHasPlanningDone {
-						// if the parent couldn't be planned, we cannot know whether it needs initial replication
-						// or incremental replication => be conservative and assume it was initial replication
-						allParentsPresentOnReceiver = false
-						if p.planning.err != nil {
-							initialReplicatingParentsWithErrors = append(initialReplicatingParentsWithErrors, p.fs.ReportInfo().Name)
-						}
-						return
-					}
-					// now allowed to inspect p.planned
-
-					// if there are no steps to be done, the filesystem must exist on the receiving side
-					// (otherwise we'd replicate it, and there would be a step for that)
-					// (FIXME hardcoded initial replication policy, assuming the policy will always do _some_ initial replication)
-					parentHasNoSteps := len(p.planned.steps) == 0
-
-					// OR if it has completed at least one step
-					// (remember that .step points to the next step to be done)
-					// (TODO technically, we could make this step ready in the moment the recv-side
-					//  dataset exists, i.e. after the first few megabytes of transferred data, but we'd have to ask the receiver for that -> poll ListFilesystems RPC)
-					parentHasTakenAtLeastOneSuccessfulStep := !parentHasNoSteps && p.planned.step >= 1
-
-					parentFirstStepIsIncremental := // no need to lock for .report() because step.l == it's fs.l
-						len(p.planned.steps) > 0 && p.planned.steps[0].report().IsIncremental()
-
-					f.debug("parentHasNoSteps=%v parentFirstStepIsIncremental=%v parentHasTakenAtLeastOneSuccessfulStep=%v",
-						parentHasNoSteps, parentFirstStepIsIncremental, parentHasTakenAtLeastOneSuccessfulStep)
-
-					parentPresentOnReceiver := parentHasNoSteps || parentFirstStepIsIncremental || parentHasTakenAtLeastOneSuccessfulStep
-
-					allParentsPresentOnReceiver = allParentsPresentOnReceiver && parentPresentOnReceiver // no shadow
-
-					if !parentPresentOnReceiver && p.planned.stepErr != nil {
-						initialReplicatingParentsWithErrors = append(initialReplicatingParentsWithErrors, p.fs.ReportInfo().Name)
-					}
-
-				})
-			}
-		})
-
-		if len(initialReplicatingParentsWithErrors) > 0 {
-			f.planned.stepErr = newTimedError(fmt.Errorf("parent(s) failed during initial replication: %s", initialReplicatingParentsWithErrors), time.Now())
-			return
-		}
-
-		if allParentsPresentOnReceiver {
-			break // good to go
-		}
-
-		// wait for wakeups from parents, then check again
-		// lock must not be held while waiting in order for reporting to work
-		f.l.DropWhile(func() {
-			select {
-			case <-ctx.Done():
-				f.planned.stepErr = newTimedError(ctx.Err(), time.Now())
-				return
-			case <-f.initialRepOrd.parentDidUpdate:
-				// loop
-			}
-		})
-		if f.planned.stepErr != nil {
-			return
-		}
-	}
-
-	f.debug("all parents ready, start replication %s", parents)
-
-	// do our steps
 	for i, s := range f.planned.steps {
+		var (
+			err     error
+			errTime time.Time
+		)
 		// lock must not be held while executing step in order for reporting to work
 		f.l.DropWhile(func() {
-			// wait for parallel replication
 			targetDate := s.step.TargetDate()
-			defer pq.WaitReady(ctx, f, targetDate)()
-			// do the step
-			ctx, endSpan := trace.WithSpan(ctx, fmt.Sprintf("%#v", s.step.ReportInfo()))
-			defer endSpan()
-			err, errTime = s.step.Step(ctx), time.Now() // no shadow
+			defer pq.WaitReady(f, targetDate)()
+			err = s.step.Step(ctx) // no shadow
+			errTime = time.Now()   // no shadow
 		})
-
 		if err != nil {
 			f.planned.stepErr = newTimedError(err, errTime)
 			break
 		}
 		f.planned.step = i + 1 // fs.planned.step must be == len(fs.planned.steps) if all went OK
-
-		f.initialRepOrdWakeupChildren()
 	}
-
 }
 
 // caller must hold lock l

replication/driver/replication_driver_debug.go

@@ -26,6 +26,6 @@ type debugFunc func(format string, args ...interface{})
 func debugPrefix(prefixFormat string, prefixFormatArgs ...interface{}) debugFunc {
 	prefix := fmt.Sprintf(prefixFormat, prefixFormatArgs...)
 	return func(format string, args ...interface{}) {
-		debug("%s: %s", prefix, fmt.Sprintf(format, args...))
+		debug("%s: %s", prefix, fmt.Sprintf(format, args))
 	}
 }

replication/driver/replication_driver_logging.go

@@ -3,10 +3,23 @@ package driver
 import (
 	"context"
 
-	"github.com/zrepl/zrepl/daemon/logging"
 	"github.com/zrepl/zrepl/logger"
 )
 
-func getLog(ctx context.Context) logger.Logger {
-	return logging.GetLogger(ctx, logging.SubsysReplication)
+type Logger = logger.Logger
+
+type contextKey int
+
+const contextKeyLogger contextKey = iota + 1
+
+func getLog(ctx context.Context) Logger {
+	l, ok := ctx.Value(contextKeyLogger).(Logger)
+	if !ok {
+		l = logger.NewNullLogger()
+	}
+	return l
+}
+
+func WithLogger(ctx context.Context, log Logger) context.Context {
+	return context.WithValue(ctx, contextKeyLogger, log)
 }

replication/driver/replication_driver_test.go

@@ -10,7 +10,6 @@ import (
 	"time"
 
 	"github.com/stretchr/testify/require"
-	"github.com/zrepl/zrepl/daemon/logging/trace"
 
 	"github.com/zrepl/zrepl/replication/report"
 
@@ -150,7 +149,6 @@ func (f *mockStep) ReportInfo() *report.StepInfo {
 func TestReplication(t *testing.T) {
 
 	ctx := context.Background()
-	defer trace.WithTaskFromStackUpdateCtx(&ctx)()
 
 	mp := &mockPlanner{}
 	getReport, wait := Do(ctx, mp)

replication/driver/replication_stepqueue.go

@@ -2,10 +2,8 @@ package driver
 
 import (
 	"container/heap"
-	"context"
 	"time"
 
-	"github.com/zrepl/zrepl/daemon/logging/trace"
 	"github.com/zrepl/zrepl/util/chainlock"
 )
 
@@ -157,8 +155,7 @@ func (q *stepQueue) sendAndWaitForWakeup(ident interface{}, targetDate time.Time
 }
 
 // Wait for the ident with targetDate to be selected to run.
-func (q *stepQueue) WaitReady(ctx context.Context, ident interface{}, targetDate time.Time) StepCompletedFunc {
-	defer trace.WithSpanFromStackUpdateCtx(&ctx)()
+func (q *stepQueue) WaitReady(ident interface{}, targetDate time.Time) StepCompletedFunc {
 	if targetDate.IsZero() {
 		panic("targetDate of zero is reserved for marking Done")
 	}

replication/driver/replication_stepqueue_test.go

@@ -1,7 +1,6 @@
 package driver
 
 import (
-	"context"
 	"fmt"
 	"math"
 	"sort"
@@ -12,23 +11,18 @@ import (
 
 	"github.com/montanaflynn/stats"
 	"github.com/stretchr/testify/assert"
-	"github.com/zrepl/zrepl/daemon/logging/trace"
 )
 
 // FIXME: this test relies on timing and is thus rather flaky
 // (relies on scheduler responsiveness of < 500ms)
 func TestPqNotconcurrent(t *testing.T) {
-	ctx, end := trace.WithTaskFromStack(context.Background())
-	defer end()
 	var ctr uint32
 	q := newStepQueue()
 	var wg sync.WaitGroup
 	wg.Add(4)
 	go func() {
-		ctx, end := trace.WithTaskFromStack(ctx)
-		defer end()
 		defer wg.Done()
-		defer q.WaitReady(ctx, "1", time.Unix(9999, 0))()
+		defer q.WaitReady("1", time.Unix(9999, 0))()
 		ret := atomic.AddUint32(&ctr, 1)
 		assert.Equal(t, uint32(1), ret)
 		time.Sleep(1 * time.Second)
@@ -40,26 +34,20 @@ func TestPqNotconcurrent(t *testing.T) {
 
 	// while "1" is still running, queue in "2", "3" and "4"
 	go func() {
-		ctx, end := trace.WithTaskFromStack(ctx)
-		defer end()
 		defer wg.Done()
-		defer q.WaitReady(ctx, "2", time.Unix(2, 0))()
+		defer q.WaitReady("2", time.Unix(2, 0))()
 		ret := atomic.AddUint32(&ctr, 1)
 		assert.Equal(t, uint32(2), ret)
 	}()
 	go func() {
-		ctx, end := trace.WithTaskFromStack(ctx)
-		defer end()
 		defer wg.Done()
-		defer q.WaitReady(ctx, "3", time.Unix(3, 0))()
+		defer q.WaitReady("3", time.Unix(3, 0))()
 		ret := atomic.AddUint32(&ctr, 1)
 		assert.Equal(t, uint32(3), ret)
 	}()
 	go func() {
-		ctx, end := trace.WithTaskFromStack(ctx)
-		defer end()
 		defer wg.Done()
-		defer q.WaitReady(ctx, "4", time.Unix(4, 0))()
+		defer q.WaitReady("4", time.Unix(4, 0))()
 		ret := atomic.AddUint32(&ctr, 1)
 		assert.Equal(t, uint32(4), ret)
 	}()
@@ -89,8 +77,6 @@ func (r record) String() string {
 // Hence, perform some statistics on the wakeup times and assert that the mean wakeup
 // times for each step are close together.
 func TestPqConcurrent(t *testing.T) {
-	ctx, end := trace.WithTaskFromStack(context.Background())
-	defer end()
 
 	q := newStepQueue()
 	var wg sync.WaitGroup
@@ -104,14 +90,12 @@ func TestPqConcurrent(t *testing.T) {
 	records := make(chan []record, filesystems)
 	for fs := 0; fs < filesystems; fs++ {
 		go func(fs int) {
-			ctx, end := trace.WithTaskFromStack(ctx)
-			defer end()
 			defer wg.Done()
 			recs := make([]record, 0)
 			for step := 0; step < stepsPerFS; step++ {
 				pos := atomic.AddUint32(&globalCtr, 1)
 				t := time.Unix(int64(step), 0)
-				done := q.WaitReady(ctx, fs, t)
+				done := q.WaitReady(fs, t)
 				wakeAt := time.Since(begin)
 				time.Sleep(sleepTimePerStep)
 				done()

replication/logic/replication_logic.go

@@ -4,14 +4,11 @@ import (
 	"context"
 	"errors"
 	"fmt"
-	"io"
 	"sync"
 	"time"
 
 	"github.com/prometheus/client_golang/prometheus"
-	"github.com/zrepl/zrepl/daemon/logging/trace"
 
-	"github.com/zrepl/zrepl/logger"
 	"github.com/zrepl/zrepl/replication/driver"
 	. "github.com/zrepl/zrepl/replication/logic/diff"
 	"github.com/zrepl/zrepl/replication/logic/pdu"
@@ -41,7 +38,7 @@ type Sender interface {
 	// If a non-nil io.ReadCloser is returned, it is guaranteed to be closed before
 	// any next call to the parent github.com/zrepl/zrepl/replication.Endpoint.
 	// If the send request is for dry run the io.ReadCloser will be nil
-	Send(ctx context.Context, r *pdu.SendReq) (*pdu.SendRes, io.ReadCloser, error)
+	Send(ctx context.Context, r *pdu.SendReq) (*pdu.SendRes, zfs.StreamCopier, error)
 	SendCompleted(ctx context.Context, r *pdu.SendCompletedReq) (*pdu.SendCompletedRes, error)
 	ReplicationCursor(ctx context.Context, req *pdu.ReplicationCursorReq) (*pdu.ReplicationCursorRes, error)
 }
@@ -50,7 +47,7 @@ type Receiver interface {
 	Endpoint
 	// Receive sends r and sendStream (the latter containing a ZFS send stream)
 	// to the parent github.com/zrepl/zrepl/replication.Endpoint.
-	Receive(ctx context.Context, req *pdu.ReceiveReq, receive io.ReadCloser) (*pdu.ReceiveRes, error)
+	Receive(ctx context.Context, req *pdu.ReceiveReq, receive zfs.StreamCopier) (*pdu.ReceiveRes, error)
 }
 
 type PlannerPolicy struct {
@@ -82,8 +79,6 @@ func (p *Planner) WaitForConnectivity(ctx context.Context) error {
 	var wg sync.WaitGroup
 	doPing := func(endpoint Endpoint, errOut *error) {
 		defer wg.Done()
-		ctx, endTask := trace.WithTaskFromStack(ctx)
-		defer endTask()
 		err := endpoint.WaitForConnectivity(ctx)
 		if err != nil {
 			*errOut = err
@@ -167,7 +162,7 @@ type Step struct {
 
 	// byteCounter is nil initially, and set later in Step.doReplication
 	// => concurrent read of that pointer from Step.ReportInfo must be protected
-	byteCounter    bytecounter.ReadCloser
+	byteCounter    bytecounter.StreamCopier
 	byteCounterMtx chainlock.L
 }
 
@@ -307,11 +302,9 @@ func (p *Planner) doPlanning(ctx context.Context) ([]*Filesystem, error) {
 
 func (fs *Filesystem) doPlanning(ctx context.Context) ([]*Step, error) {
 
-	log := func(ctx context.Context) logger.Logger {
-		return getLogger(ctx).WithField("filesystem", fs.Path)
-	}
+	log := getLogger(ctx).WithField("filesystem", fs.Path)
 
-	log(ctx).Debug("assessing filesystem")
+	log.Debug("assessing filesystem")
 
 	if fs.policy.EncryptedSend == True && !fs.senderFS.GetIsEncrypted() {
 		return nil, fmt.Errorf("sender filesystem is not encrypted but policy mandates encrypted send")
@@ -319,14 +312,14 @@ func (fs *Filesystem) doPlanning(ctx context.Context) ([]*Step, error) {
 
 	sfsvsres, err := fs.sender.ListFilesystemVersions(ctx, &pdu.ListFilesystemVersionsReq{Filesystem: fs.Path})
 	if err != nil {
-		log(ctx).WithError(err).Error("cannot get remote filesystem versions")
+		log.WithError(err).Error("cannot get remote filesystem versions")
 		return nil, err
 	}
 	sfsvs := sfsvsres.GetVersions()
 
 	if len(sfsvs) < 1 {
 		err := errors.New("sender does not have any versions")
-		log(ctx).Error(err.Error())
+		log.Error(err.Error())
 		return nil, err
 	}
 
@@ -334,7 +327,7 @@ func (fs *Filesystem) doPlanning(ctx context.Context) ([]*Step, error) {
 	if fs.receiverFS != nil && !fs.receiverFS.GetIsPlaceholder() {
 		rfsvsres, err := fs.receiver.ListFilesystemVersions(ctx, &pdu.ListFilesystemVersionsReq{Filesystem: fs.Path})
 		if err != nil {
-			log(ctx).WithError(err).Error("receiver error")
+			log.WithError(err).Error("receiver error")
 			return nil, err
 		}
 		rfsvs = rfsvsres.GetVersions()
@@ -346,17 +339,17 @@ func (fs *Filesystem) doPlanning(ctx context.Context) ([]*Step, error) {
 	var resumeTokenRaw string
 	if fs.receiverFS != nil && fs.receiverFS.ResumeToken != "" {
 		resumeTokenRaw = fs.receiverFS.ResumeToken // shadow
-		log(ctx).WithField("receiverFS.ResumeToken", resumeTokenRaw).Debug("decode receiver fs resume token")
+		log.WithField("receiverFS.ResumeToken", resumeTokenRaw).Debug("decode receiver fs resume token")
 		resumeToken, err = zfs.ParseResumeToken(ctx, resumeTokenRaw) // shadow
 		if err != nil {
 			// TODO in theory, we could do replication without resume token, but that would mean that
 			// we need to discard the resumable state on the receiver's side.
 			// Would be easy by setting UsedResumeToken=false in the RecvReq ...
 			// FIXME / CHECK semantics UsedResumeToken if SendReq.ResumeToken == ""
-			log(ctx).WithError(err).Error("cannot decode resume token, aborting")
+			log.WithError(err).Error("cannot decode resume token, aborting")
 			return nil, err
 		}
-		log(ctx).WithField("token", resumeToken).Debug("decode resume token")
+		log.WithField("token", resumeToken).Debug("decode resume token")
 	}
 
 	// give both sides a hint about how far prior replication attempts got
@@ -375,10 +368,7 @@ func (fs *Filesystem) doPlanning(ctx context.Context) ([]*Step, error) {
 		var wg sync.WaitGroup
 		doHint := func(ep Endpoint, name string) {
 			defer wg.Done()
-			ctx, endTask := trace.WithTask(ctx, "hint-mrca-"+name)
-			defer endTask()
-
-			log := log(ctx).WithField("to_side", name).
+			log := log.WithField("to_side", name).
 				WithField("sender_mrca", sender_mrca.String())
 			log.Debug("hint most recent common ancestor")
 			hint := &pdu.HintMostRecentCommonAncestorReq{
@@ -437,7 +427,7 @@ func (fs *Filesystem) doPlanning(ctx context.Context) ([]*Step, error) {
 			encryptionMatches = true
 		}
 
-		log(ctx).WithField("fromVersion", fromVersion).
+		log.WithField("fromVersion", fromVersion).
 			WithField("toVersion", toVersion).
 			WithField("encryptionMatches", encryptionMatches).
 			Debug("result of resume-token-matching to sender's versions")
@@ -493,11 +483,11 @@ func (fs *Filesystem) doPlanning(ctx context.Context) ([]*Step, error) {
 			var msg string
 			path, msg = resolveConflict(conflict) // no shadowing allowed!
 			if path != nil {
-				log(ctx).WithField("conflict", conflict).Info("conflict")
-				log(ctx).WithField("resolution", msg).Info("automatically resolved")
+				log.WithField("conflict", conflict).Info("conflict")
+				log.WithField("resolution", msg).Info("automatically resolved")
 			} else {
-				log(ctx).WithField("conflict", conflict).Error("conflict")
-				log(ctx).WithField("problem", msg).Error("cannot resolve conflict")
+				log.WithField("conflict", conflict).Error("conflict")
+				log.WithField("problem", msg).Error("cannot resolve conflict")
 			}
 		}
 		if len(path) == 0 {
@@ -531,35 +521,37 @@ func (fs *Filesystem) doPlanning(ctx context.Context) ([]*Step, error) {
 	}
 
 	if len(steps) == 0 {
-		log(ctx).Info("planning determined that no replication steps are required")
+		log.Info("planning determined that no replication steps are required")
 	}
 
-	log(ctx).Debug("compute send size estimate")
+	log.Debug("compute send size estimate")
 	errs := make(chan error, len(steps))
+	var wg sync.WaitGroup
 	fanOutCtx, fanOutCancel := context.WithCancel(ctx)
-	_, fanOutAdd, fanOutWait := trace.WithTaskGroup(fanOutCtx, "compute-size-estimate")
 	defer fanOutCancel()
 	for _, step := range steps {
-		step := step // local copy that is moved into the closure
-		fanOutAdd(func(ctx context.Context) {
+		wg.Add(1)
+		go func(step *Step) {
+			defer wg.Done()
+
 			// TODO instead of the semaphore, rely on resource-exhaustion signaled by the remote endpoint to limit size-estimate requests
 			// Send is handled over rpc/dataconn ATM, which doesn't support the resource exhaustion status codes that gRPC defines
-			guard, err := fs.sizeEstimateRequestSem.Acquire(ctx)
+			guard, err := fs.sizeEstimateRequestSem.Acquire(fanOutCtx)
 			if err != nil {
 				fanOutCancel()
 				return
 			}
 			defer guard.Release()
 
-			err = step.updateSizeEstimate(ctx)
+			err = step.updateSizeEstimate(fanOutCtx)
 			if err != nil {
-				log(ctx).WithError(err).WithField("step", step).Error("error computing size estimate")
+				log.WithError(err).WithField("step", step).Error("error computing size estimate")
 				fanOutCancel()
 			}
 			errs <- err
-		})
+		}(step)
 	}
-	fanOutWait()
+	wg.Wait()
 	close(errs)
 	var significantErr error = nil
 	for err := range errs {
@@ -573,7 +565,7 @@ func (fs *Filesystem) doPlanning(ctx context.Context) ([]*Step, error) {
 		return nil, significantErr
 	}
 
-	log(ctx).Debug("filesystem planning finished")
+	log.Debug("filesystem planning finished")
 	return steps, nil
 }
 
@@ -610,23 +602,23 @@ func (s *Step) doReplication(ctx context.Context) error {
 
 	fs := s.parent.Path
 
-	log := getLogger(ctx).WithField("filesystem", fs)
+	log := getLogger(ctx)
 	sr := s.buildSendRequest(false)
 
 	log.Debug("initiate send request")
-	sres, stream, err := s.sender.Send(ctx, sr)
+	sres, sstreamCopier, err := s.sender.Send(ctx, sr)
 	if err != nil {
 		log.WithError(err).Error("send request failed")
 		return err
 	}
-	if stream == nil {
+	if sstreamCopier == nil {
 		err := errors.New("send request did not return a stream, broken endpoint implementation")
 		return err
 	}
-	defer stream.Close()
+	defer sstreamCopier.Close()
 
 	// Install a byte counter to track progress + for status report
-	byteCountingStream := bytecounter.NewReadCloser(stream)
+	byteCountingStream := bytecounter.NewStreamCopier(sstreamCopier)
 	s.byteCounterMtx.Lock()
 	s.byteCounter = byteCountingStream
 	s.byteCounterMtx.Unlock()

replication/logic/replication_logic_context.go

@@ -3,10 +3,26 @@ package logic
 import (
 	"context"
 
-	"github.com/zrepl/zrepl/daemon/logging"
 	"github.com/zrepl/zrepl/logger"
 )
 
-func getLogger(ctx context.Context) logger.Logger {
-	return logging.GetLogger(ctx, logging.SubsysReplication)
+type contextKey int
+
+const (
+	contextKeyLog contextKey = iota
+)
+
+type Logger = logger.Logger
+
+func WithLogger(ctx context.Context, l Logger) context.Context {
+	ctx = context.WithValue(ctx, contextKeyLog, l)
+	return ctx
+}
+
+func getLogger(ctx context.Context) Logger {
+	l, ok := ctx.Value(contextKeyLog).(Logger)
+	if !ok {
+		l = logger.NewNullLogger()
+	}
+	return l
 }

rpc/dataconn/dataconn_client.go

@@ -4,7 +4,6 @@ import (
 	"bytes"
 	"context"
 	"fmt"
-	"io"
 	"strings"
 
 	"github.com/golang/protobuf/proto"
@@ -12,6 +11,7 @@ import (
 	"github.com/zrepl/zrepl/replication/logic/pdu"
 	"github.com/zrepl/zrepl/rpc/dataconn/stream"
 	"github.com/zrepl/zrepl/transport"
+	"github.com/zrepl/zrepl/zfs"
 )
 
 type Client struct {
@@ -26,7 +26,7 @@ func NewClient(connecter transport.Connecter, log Logger) *Client {
 	}
 }
 
-func (c *Client) send(ctx context.Context, conn *stream.Conn, endpoint string, req proto.Message, stream io.ReadCloser) error {
+func (c *Client) send(ctx context.Context, conn *stream.Conn, endpoint string, req proto.Message, streamCopier zfs.StreamCopier) error {
 
 	var buf bytes.Buffer
 	_, memErr := buf.WriteString(endpoint)
@@ -46,8 +46,8 @@ func (c *Client) send(ctx context.Context, conn *stream.Conn, endpoint string, r
 		return err
 	}
 
-	if stream != nil {
-		return conn.SendStream(ctx, stream, ZFSStream)
+	if streamCopier != nil {
+		return conn.SendStream(ctx, streamCopier, ZFSStream)
 	} else {
 		return nil
 	}
@@ -109,7 +109,7 @@ func (c *Client) putWire(conn *stream.Conn) {
 	}
 }
 
-func (c *Client) ReqSend(ctx context.Context, req *pdu.SendReq) (*pdu.SendRes, io.ReadCloser, error) {
+func (c *Client) ReqSend(ctx context.Context, req *pdu.SendReq) (*pdu.SendRes, zfs.StreamCopier, error) {
 	conn, err := c.getWire(ctx)
 	if err != nil {
 		return nil, nil, err
@@ -130,19 +130,17 @@ func (c *Client) ReqSend(ctx context.Context, req *pdu.SendReq) (*pdu.SendRes, i
 		return nil, nil, err
 	}
 
-	var stream io.ReadCloser
+	var copier zfs.StreamCopier = nil
 	if !req.DryRun {
 		putWireOnReturn = false
-		stream, err = conn.ReadStream(ZFSStream, true) // no shadow
-		if err != nil {
-			return nil, nil, err
-		}
+		copier = &streamCopier{streamConn: conn, closeStreamOnClose: true}
 	}
 
-	return &res, stream, nil
+	return &res, copier, nil
 }
 
-func (c *Client) ReqRecv(ctx context.Context, req *pdu.ReceiveReq, stream io.ReadCloser) (*pdu.ReceiveRes, error) {
+func (c *Client) ReqRecv(ctx context.Context, req *pdu.ReceiveReq, streamCopier zfs.StreamCopier) (*pdu.ReceiveRes, error) {
+
 	defer c.log.Debug("ReqRecv returns")
 	conn, err := c.getWire(ctx)
 	if err != nil {
@@ -168,7 +166,7 @@ func (c *Client) ReqRecv(ctx context.Context, req *pdu.ReceiveReq, stream io.Rea
 
 	sendErrChan := make(chan error)
 	go func() {
-		if err := c.send(ctx, conn, EndpointRecv, req, stream); err != nil {
+		if err := c.send(ctx, conn, EndpointRecv, req, streamCopier); err != nil {
 			sendErrChan <- err
 		} else {
 			sendErrChan <- nil

rpc/dataconn/dataconn_server.go

@@ -4,8 +4,6 @@ import (
 	"bytes"
 	"context"
 	"fmt"
-	"io"
-	"sync"
 
 	"github.com/golang/protobuf/proto"
 
@@ -13,6 +11,7 @@ import (
 	"github.com/zrepl/zrepl/replication/logic/pdu"
 	"github.com/zrepl/zrepl/rpc/dataconn/stream"
 	"github.com/zrepl/zrepl/transport"
+	"github.com/zrepl/zrepl/zfs"
 )
 
 // WireInterceptor has a chance to exchange the context and connection on each client connection.
@@ -22,44 +21,27 @@ type WireInterceptor func(ctx context.Context, rawConn *transport.AuthConn) (con
 type Handler interface {
 	// Send handles a SendRequest.
 	// The returned io.ReadCloser is allowed to be nil, for example if the requested Send is a dry-run.
-	Send(ctx context.Context, r *pdu.SendReq) (*pdu.SendRes, io.ReadCloser, error)
+	Send(ctx context.Context, r *pdu.SendReq) (*pdu.SendRes, zfs.StreamCopier, error)
 	// Receive handles a ReceiveRequest.
 	// It is guaranteed that Server calls Receive with a stream that holds the IdleConnTimeout
 	// configured in ServerConfig.Shared.IdleConnTimeout.
-	Receive(ctx context.Context, r *pdu.ReceiveReq, receive io.ReadCloser) (*pdu.ReceiveRes, error)
+	Receive(ctx context.Context, r *pdu.ReceiveReq, receive zfs.StreamCopier) (*pdu.ReceiveRes, error)
 	// PingDataconn handles a PingReq
 	PingDataconn(ctx context.Context, r *pdu.PingReq) (*pdu.PingRes, error)
 }
 
 type Logger = logger.Logger
 
-type ContextInterceptorData interface {
-	FullMethod() string
-	ClientIdentity() string
-}
-
-type ContextInterceptor = func(ctx context.Context, data ContextInterceptorData, handler func(ctx context.Context))
-
 type Server struct {
 	h   Handler
 	wi  WireInterceptor
-	ci  ContextInterceptor
 	log Logger
 }
 
-var noopContextInteceptor = func(ctx context.Context, _ ContextInterceptorData, handler func(context.Context)) {
-	handler(ctx)
-}
-
-// wi and ci may be nil
-func NewServer(wi WireInterceptor, ci ContextInterceptor, logger Logger, handler Handler) *Server {
-	if ci == nil {
-		ci = noopContextInteceptor
-	}
+func NewServer(wi WireInterceptor, logger Logger, handler Handler) *Server {
 	return &Server{
 		h:   handler,
 		wi:  wi,
-		ci:  ci,
 		log: logger,
 	}
 }
@@ -68,26 +50,16 @@ func NewServer(wi WireInterceptor, ci ContextInterceptor, logger Logger, handler
 // No accept errors are returned: they are logged to the Logger passed
 // to the constructor.
 func (s *Server) Serve(ctx context.Context, l transport.AuthenticatedListener) {
-	var wg sync.WaitGroup
-	defer wg.Wait()
 
-	ctx, cancel := context.WithCancel(ctx)
-	defer cancel()
-
-	wg.Add(1)
 	go func() {
-		defer wg.Done()
 		<-ctx.Done()
-		s.log.Debug("context done, closing listener")
+		s.log.Debug("context done")
 		if err := l.Close(); err != nil {
 			s.log.WithError(err).Error("cannot close listener")
 		}
 	}()
 	conns := make(chan *transport.AuthConn)
-	wg.Add(1)
 	go func() {
-		defer wg.Done()
-		defer close(conns)
 		for {
 			conn, err := l.Accept(ctx)
 			if err != nil {
@@ -102,22 +74,10 @@ func (s *Server) Serve(ctx context.Context, l transport.AuthenticatedListener) {
 		}
 	}()
 	for conn := range conns {
-		wg.Add(1)
-		go func(conn *transport.AuthConn) {
-			defer wg.Done()
-			s.serveConn(conn)
-		}(conn)
+		go s.serveConn(conn)
 	}
 }
 
-type contextInterceptorData struct {
-	fullMethod     string
-	clientIdentity string
-}
-
-func (d contextInterceptorData) FullMethod() string     { return d.fullMethod }
-func (d contextInterceptorData) ClientIdentity() string { return d.clientIdentity }
-
 func (s *Server) serveConn(nc *transport.AuthConn) {
 	s.log.Debug("serveConn begin")
 	defer s.log.Debug("serveConn done")
@@ -142,17 +102,6 @@ func (s *Server) serveConn(nc *transport.AuthConn) {
 	}
 	endpoint := string(header)
 
-	data := contextInterceptorData{
-		fullMethod:     endpoint,
-		clientIdentity: nc.ClientIdentity(),
-	}
-	s.ci(ctx, data, func(ctx context.Context) {
-		s.serveConnRequest(ctx, endpoint, c)
-	})
-}
-
-func (s *Server) serveConnRequest(ctx context.Context, endpoint string, c *stream.Conn) {
-
 	reqStructured, err := c.ReadStreamedMessage(ctx, RequestStructuredMaxSize, ReqStructured)
 	if err != nil {
 		s.log.WithError(err).Error("error reading structured part")
@@ -162,7 +111,7 @@ func (s *Server) serveConnRequest(ctx context.Context, endpoint string, c *strea
 	s.log.WithField("endpoint", endpoint).Debug("calling handler")
 
 	var res proto.Message
-	var sendStream io.ReadCloser
+	var sendStream zfs.StreamCopier
 	var handlerErr error
 	switch endpoint {
 	case EndpointSend:
@@ -178,12 +127,7 @@ func (s *Server) serveConnRequest(ctx context.Context, endpoint string, c *strea
 			s.log.WithError(err).Error("cannot unmarshal receive request")
 			return
 		}
-		stream, err := c.ReadStream(ZFSStream, false)
-		if err != nil {
-			s.log.WithError(err).Error("cannot open stream in receive request")
-			return
-		}
-		res, handlerErr = s.h.Receive(ctx, &req, stream) // SHADOWING
+		res, handlerErr = s.h.Receive(ctx, &req, &streamCopier{streamConn: c, closeStreamOnClose: false}) // SHADOWING
 	case EndpointPing:
 		var req pdu.PingReq
 		if err := proto.Unmarshal(reqStructured, &req); err != nil {

rpc/dataconn/dataconn_shared.go

@@ -1,7 +1,12 @@
 package dataconn
 
 import (
+	"io"
+	"sync"
 	"time"
+
+	"github.com/zrepl/zrepl/rpc/dataconn/stream"
+	"github.com/zrepl/zrepl/zfs"
 )
 
 const (
@@ -34,3 +39,33 @@ const (
 	responseHeaderHandlerOk          = "HANDLER OK\n"
 	responseHeaderHandlerErrorPrefix = "HANDLER ERROR:\n"
 )
+
+type streamCopier struct {
+	mtx                sync.Mutex
+	used               bool
+	streamConn         *stream.Conn
+	closeStreamOnClose bool
+}
+
+// WriteStreamTo implements zfs.StreamCopier
+func (s *streamCopier) WriteStreamTo(w io.Writer) zfs.StreamCopierError {
+	s.mtx.Lock()
+	defer s.mtx.Unlock()
+	if s.used {
+		panic("streamCopier used multiple times")
+	}
+	s.used = true
+	return s.streamConn.ReadStreamInto(w, ZFSStream)
+}
+
+// Close implements zfs.StreamCopier
+func (s *streamCopier) Close() error {
+	// only record the close here, what we do actually depends on whether
+	// the streamCopier is instantiated server-side or client-side
+	s.mtx.Lock()
+	defer s.mtx.Unlock()
+	if s.closeStreamOnClose {
+		return s.streamConn.Close()
+	}
+	return nil
+}

rpc/dataconn/microbenchmark/microbenchmark.go

@@ -29,6 +29,7 @@ import (
 	"github.com/zrepl/zrepl/rpc/dataconn/timeoutconn"
 	"github.com/zrepl/zrepl/transport"
 	"github.com/zrepl/zrepl/util/devnoop"
+	"github.com/zrepl/zrepl/zfs"
 )
 
 func orDie(err error) {
@@ -41,9 +42,23 @@ type readerStreamCopier struct{ io.Reader }
 
 func (readerStreamCopier) Close() error { return nil }
 
+type readerStreamCopierErr struct {
+	error
+}
+
+func (readerStreamCopierErr) IsReadError() bool  { return false }
+func (readerStreamCopierErr) IsWriteError() bool { return true }
+
+func (c readerStreamCopier) WriteStreamTo(w io.Writer) zfs.StreamCopierError {
+	var buf [1 << 21]byte
+	_, err := io.CopyBuffer(w, c.Reader, buf[:])
+	// always assume write error
+	return readerStreamCopierErr{err}
+}
+
 type devNullHandler struct{}
 
-func (devNullHandler) Send(ctx context.Context, r *pdu.SendReq) (*pdu.SendRes, io.ReadCloser, error) {
+func (devNullHandler) Send(ctx context.Context, r *pdu.SendReq) (*pdu.SendRes, zfs.StreamCopier, error) {
 	var res pdu.SendRes
 	if args.devnoopReader {
 		return &res, readerStreamCopier{devnoop.Get()}, nil
@@ -52,12 +67,12 @@ func (devNullHandler) Send(ctx context.Context, r *pdu.SendReq) (*pdu.SendRes, i
 	}
 }
 
-func (devNullHandler) Receive(ctx context.Context, r *pdu.ReceiveReq, stream io.ReadCloser) (*pdu.ReceiveRes, error) {
+func (devNullHandler) Receive(ctx context.Context, r *pdu.ReceiveReq, stream zfs.StreamCopier) (*pdu.ReceiveRes, error) {
 	var out io.Writer = os.Stdout
 	if args.devnoopWriter {
 		out = devnoop.Get()
 	}
-	_, err := io.Copy(out, stream)
+	err := stream.WriteStreamTo(out)
 	var res pdu.ReceiveRes
 	return &res, err
 }
@@ -112,7 +127,7 @@ func server() {
 	orDie(err)
 	l := tcpListener{nl.(*net.TCPListener), "fakeclientidentity"}
 
-	srv := dataconn.NewServer(nil, nil, logger.NewStderrDebugLogger(), devNullHandler{})
+	srv := dataconn.NewServer(nil, logger.NewStderrDebugLogger(), devNullHandler{})
 
 	ctx := context.Background()
 
@@ -157,7 +172,7 @@ func client() {
 		req := pdu.SendReq{}
 		_, stream, err := client.ReqSend(ctx, &req)
 		orDie(err)
-		_, err = io.Copy(os.Stdout, stream)
+		err = stream.WriteStreamTo(os.Stdout)
 		orDie(err)
 	case "recv":
 		var r io.Reader = os.Stdin

rpc/dataconn/stream/stream.go

@@ -7,7 +7,6 @@ import (
 	"io"
 	"net"
 	"strings"
-	"sync"
 	"sync/atomic"
 	"unicode/utf8"
 
@@ -15,6 +14,7 @@ import (
 	"github.com/zrepl/zrepl/rpc/dataconn/base2bufpool"
 	"github.com/zrepl/zrepl/rpc/dataconn/frameconn"
 	"github.com/zrepl/zrepl/rpc/dataconn/heartbeatconn"
+	"github.com/zrepl/zrepl/zfs"
 )
 
 type Logger = logger.Logger
@@ -81,13 +81,9 @@ func doWriteStream(ctx context.Context, c *heartbeatconn.Conn, stream io.Reader,
 		err error
 	}
 
-	var wg sync.WaitGroup
-	defer wg.Wait()
 	reads := make(chan read, 5)
 	var stopReading uint32
-	wg.Add(1)
 	go func() {
-		defer wg.Done()
 		defer close(reads)
 		for atomic.LoadUint32(&stopReading) == 0 {
 			buffer := bufpool.Get(1 << FramePayloadShift)
@@ -202,7 +198,7 @@ func (e ReadStreamError) Temporary() bool {
 	return false
 }
 
-var _ net.Error = &ReadStreamError{}
+var _ zfs.StreamCopierError = &ReadStreamError{}
 
 func (e ReadStreamError) IsReadError() bool {
 	return e.Kind != ReadStreamErrorKindWrite

rpc/dataconn/stream/stream_conn.go

@@ -14,6 +14,7 @@ import (
 
 	"github.com/zrepl/zrepl/rpc/dataconn/heartbeatconn"
 	"github.com/zrepl/zrepl/rpc/dataconn/timeoutconn"
+	"github.com/zrepl/zrepl/zfs"
 )
 
 type Conn struct {
@@ -39,7 +40,15 @@ type Conn struct {
 
 var readMessageSentinel = fmt.Errorf("read stream complete")
 
-var errWriteStreamToErrorUnknownState = fmt.Errorf("dataconn read stream: connection is in unknown state")
+type writeStreamToErrorUnknownState struct{}
+
+func (e writeStreamToErrorUnknownState) Error() string {
+	return "dataconn read stream: connection is in unknown state"
+}
+
+func (e writeStreamToErrorUnknownState) IsReadError() bool { return true }
+
+func (e writeStreamToErrorUnknownState) IsWriteError() bool { return false }
 
 func Wrap(nc timeoutconn.Wire, sendHeartbeatInterval, peerTimeout time.Duration) *Conn {
 	hc := heartbeatconn.Wrap(nc, sendHeartbeatInterval, peerTimeout)
@@ -114,28 +123,14 @@ func (c *Conn) ReadStreamedMessage(ctx context.Context, maxSize uint32, frameTyp
 	}
 }
 
-type StreamReader struct {
-	*io.PipeReader
-	conn             *Conn
-	closeConnOnClose bool
-}
-
-func (r *StreamReader) Close() error {
-	err := r.PipeReader.Close()
-	if r.closeConnOnClose {
-		r.conn.Close() // TODO error logging
-	}
-	return err
-}
-
 // WriteStreamTo reads a stream from Conn and writes it to w.
-func (c *Conn) ReadStream(frameType uint32, closeConnOnClose bool) (_ *StreamReader, err error) {
+func (c *Conn) ReadStreamInto(w io.Writer, frameType uint32) (err zfs.StreamCopierError) {
 
 	// if we are closed while writing, return that as an error
 	if closeGuard, cse := c.closeState.RWEntry(); cse != nil {
-		return nil, cse
+		return cse
 	} else {
-		defer func(err *error) {
+		defer func(err *zfs.StreamCopierError) {
 			if closed := closeGuard.RWExit(); closed != nil {
 				*err = closed
 			}
@@ -143,23 +138,18 @@ func (c *Conn) ReadStream(frameType uint32, closeConnOnClose bool) (_ *StreamRea
 	}
 
 	c.readMtx.Lock()
+	defer c.readMtx.Unlock()
 	if !c.readClean {
-		return nil, errWriteStreamToErrorUnknownState
+		return writeStreamToErrorUnknownState{}
 	}
+	var rse *ReadStreamError = readStream(c.frameReads, c.hc, w, frameType)
+	c.readClean = isConnCleanAfterRead(rse)
 
-	r, w := io.Pipe()
-	go func() {
-		defer c.readMtx.Unlock()
-		var err *ReadStreamError = readStream(c.frameReads, c.hc, w, frameType)
-		if err != nil {
-			_ = w.CloseWithError(err) // doc guarantees that error will always be nil
-		} else {
-			w.Close()
-		}
-		c.readClean = isConnCleanAfterRead(err)
-	}()
-
-	return &StreamReader{PipeReader: r, conn: c, closeConnOnClose: closeConnOnClose}, nil
+	// https://golang.org/doc/faq#nil_error
+	if rse == nil {
+		return nil
+	}
+	return rse
 }
 
 func (c *Conn) WriteStreamedMessage(ctx context.Context, buf io.Reader, frameType uint32) (err error) {
@@ -188,7 +178,7 @@ func (c *Conn) WriteStreamedMessage(ctx context.Context, buf io.Reader, frameTyp
 	return errConn
 }
 
-func (c *Conn) SendStream(ctx context.Context, stream io.ReadCloser, frameType uint32) (err error) {
+func (c *Conn) SendStream(ctx context.Context, src zfs.StreamCopier, frameType uint32) (err error) {
 
 	// if we are closed while reading, return that as an error
 	if closeGuard, cse := c.closeState.RWEntry(); cse != nil {
@@ -207,17 +197,49 @@ func (c *Conn) SendStream(ctx context.Context, stream io.ReadCloser, frameType u
 		return fmt.Errorf("dataconn send stream: connection is in unknown state")
 	}
 
-	errStream, errConn := writeStream(ctx, c.hc, stream, frameType)
-
-	c.writeClean = isConnCleanAfterWrite(errConn) // TODO correct?
-
-	if errStream != nil {
-		return errStream
-	} else if errConn != nil {
-		return errConn
+	// avoid io.Pipe if zfs.StreamCopier is an io.Reader
+	var r io.Reader
+	var w *io.PipeWriter
+	streamCopierErrChan := make(chan zfs.StreamCopierError, 1)
+	if reader, ok := src.(io.Reader); ok {
+		r = reader
+		streamCopierErrChan <- nil
+		close(streamCopierErrChan)
+	} else {
+		r, w = io.Pipe()
+		go func() {
+			streamCopierErrChan <- src.WriteStreamTo(w)
+			w.Close()
+		}()
+	}
+
+	type writeStreamRes struct {
+		errStream, errConn error
+	}
+	writeStreamErrChan := make(chan writeStreamRes, 1)
+	go func() {
+		var res writeStreamRes
+		res.errStream, res.errConn = writeStream(ctx, c.hc, r, frameType)
+		if w != nil {
+			_ = w.CloseWithError(res.errStream) // always returns nil
+		}
+		writeStreamErrChan <- res
+	}()
+
+	writeRes := <-writeStreamErrChan
+	streamCopierErr := <-streamCopierErrChan
+	c.writeClean = isConnCleanAfterWrite(writeRes.errConn) // TODO correct?
+	if streamCopierErr != nil && streamCopierErr.IsReadError() {
+		return streamCopierErr // something on our side is bad
+	} else {
+		if writeRes.errStream != nil {
+			return writeRes.errStream
+		} else if writeRes.errConn != nil {
+			return writeRes.errConn
+		}
+		// TODO combined error?
+		return streamCopierErr
 	}
-	// TODO combined error?
-	return nil
 }
 
 type closeState struct {
@@ -226,13 +248,17 @@ type closeState struct {
 
 type closeStateErrConnectionClosed struct{}
 
+var _ zfs.StreamCopierError = (*closeStateErrConnectionClosed)(nil)
+var _ error = (*closeStateErrConnectionClosed)(nil)
 var _ net.Error = (*closeStateErrConnectionClosed)(nil)
 
 func (e *closeStateErrConnectionClosed) Error() string {
 	return "connection closed"
 }
-func (e *closeStateErrConnectionClosed) Timeout() bool   { return false }
-func (e *closeStateErrConnectionClosed) Temporary() bool { return false }
+func (e *closeStateErrConnectionClosed) IsReadError() bool  { return true }
+func (e *closeStateErrConnectionClosed) IsWriteError() bool { return true }
+func (e *closeStateErrConnectionClosed) Timeout() bool      { return false }
+func (e *closeStateErrConnectionClosed) Temporary() bool    { return false }
 
 func (s *closeState) CloseEntry() error {
 	firstCloser := atomic.AddUint32(&s.closeCount, 1) == 1
@@ -247,7 +273,7 @@ type closeStateEntry struct {
 	entryCount uint32
 }
 
-func (s *closeState) RWEntry() (e *closeStateEntry, err net.Error) {
+func (s *closeState) RWEntry() (e *closeStateEntry, err zfs.StreamCopierError) {
 	entry := &closeStateEntry{s, atomic.LoadUint32(&s.closeCount)}
 	if entry.entryCount > 0 {
 		return nil, &closeStateErrConnectionClosed{}
@@ -255,7 +281,7 @@ func (s *closeState) RWEntry() (e *closeStateEntry, err net.Error) {
 	return entry, nil
 }
 
-func (e *closeStateEntry) RWExit() net.Error {
+func (e *closeStateEntry) RWExit() zfs.StreamCopierError {
 	if atomic.LoadUint32(&e.entryCount) == e.entryCount {
 		// no calls to Close() while running rw operation
 		return nil

rpc/grpcclientidentity/authlistener_grpc_adaptor.go

@@ -99,23 +99,10 @@ func (*transportCredentials) OverrideServerName(string) error {
 	panic("not implemented")
 }
 
-type ContextInterceptorData interface {
-	FullMethod() string
-	ClientIdentity() string
-}
+type ContextInterceptor = func(ctx context.Context) context.Context
 
-type contextInterceptorData struct {
-	fullMethod     string
-	clientIdentity string
-}
-
-func (d contextInterceptorData) FullMethod() string     { return d.fullMethod }
-func (d contextInterceptorData) ClientIdentity() string { return d.clientIdentity }
-
-type Interceptor = func(ctx context.Context, data ContextInterceptorData, handler func(ctx context.Context))
-
-func NewInterceptors(logger Logger, clientIdentityKey interface{}, interceptor Interceptor) (unary grpc.UnaryServerInterceptor, stream grpc.StreamServerInterceptor) {
-	unary = func(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (interface{}, error) {
+func NewInterceptors(logger Logger, clientIdentityKey interface{}, ctxInterceptor ContextInterceptor) (unary grpc.UnaryServerInterceptor, stream grpc.StreamServerInterceptor) {
+	unary = func(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (resp interface{}, err error) {
 		logger.WithField("fullMethod", info.FullMethod).Debug("request")
 		p, ok := peer.FromContext(ctx)
 		if !ok {
@@ -128,18 +115,10 @@ func NewInterceptors(logger Logger, clientIdentityKey interface{}, interceptor I
 		}
 		logger.WithField("peer_client_identity", a.clientIdentity).Debug("peer client identity")
 		ctx = context.WithValue(ctx, clientIdentityKey, a.clientIdentity)
-		data := contextInterceptorData{
-			fullMethod:     info.FullMethod,
-			clientIdentity: a.clientIdentity,
+		if ctxInterceptor != nil {
+			ctx = ctxInterceptor(ctx)
 		}
-		var (
-			resp interface{}
-			err  error
-		)
-		interceptor(ctx, data, func(ctx context.Context) {
-			resp, err = handler(ctx, req) // no-shadow
-		})
-		return resp, err
+		return handler(ctx, req)
 	}
 	stream = func(srv interface{}, ss grpc.ServerStream, info *grpc.StreamServerInfo, handler grpc.StreamHandler) error {
 		panic("unimplemented")

rpc/grpcclientidentity/grpchelper/authlistener_grpc_adaptor_wrapper.go

@@ -50,7 +50,7 @@ func ClientConn(cn transport.Connecter, log Logger) *grpc.ClientConn {
 }
 
 // NewServer is a convenience interface around the TransportCredentials and Interceptors interface.
-func NewServer(authListener transport.AuthenticatedListener, clientIdentityKey interface{}, logger grpcclientidentity.Logger, ctxInterceptor grpcclientidentity.Interceptor) (srv *grpc.Server, serve func() error) {
+func NewServer(authListener transport.AuthenticatedListener, clientIdentityKey interface{}, logger grpcclientidentity.Logger, ctxInterceptor grpcclientidentity.ContextInterceptor) (srv *grpc.Server, serve func() error) {
 	ka := grpc.KeepaliveParams(keepalive.ServerParameters{
 		Time:    StartKeepalivesAfterInactivityDuration,
 		Timeout: KeepalivePeerTimeout,

rpc/netadaptor/authlistener_netlistener_adaptor.go

@@ -33,12 +33,8 @@ import (
 
 type Logger = logger.Logger
 
-type acceptRes struct {
-	conn *transport.AuthConn
-	err  error
-}
 type acceptReq struct {
-	callback chan acceptRes
+	callback chan net.Conn
 }
 
 type Listener struct {
@@ -68,19 +64,10 @@ func New(authListener transport.AuthenticatedListener, l Logger) *Listener {
 // The returned net.Conn is guaranteed to be *transport.AuthConn, i.e., the type of connection
 // returned by the wrapped transport.AuthenticatedListener.
 func (a Listener) Accept() (net.Conn, error) {
-	req := acceptReq{make(chan acceptRes, 1)}
-
-	select {
-	case a.accepts <- req:
-	case <-a.stop:
-		return nil, fmt.Errorf("already closed") // TODO net.Error
-	}
-
-	res, ok := <-req.callback
-	if !ok {
-		return nil, fmt.Errorf("already closed") // TODO net.Error
-	}
-	return res.conn, res.err
+	req := acceptReq{make(chan net.Conn, 1)}
+	a.accepts <- req
+	conn := <-req.callback
+	return conn, nil
 }
 
 func (a Listener) handleAccept() {
@@ -90,9 +77,18 @@ func (a Listener) handleAccept() {
 			a.logger.Debug("handleAccept stop accepting")
 			return
 		case req := <-a.accepts:
-			a.logger.Debug("accept authListener")
-			authConn, err := a.al.Accept(context.Background())
-			req.callback <- acceptRes{authConn, err}
+			for {
+				a.logger.Debug("accept authListener")
+				authConn, err := a.al.Accept(context.Background())
+				if err != nil {
+					a.logger.WithError(err).Error("accept error")
+					continue
+				}
+				a.logger.WithField("type", fmt.Sprintf("%T", authConn)).
+					Debug("accept complete")
+				req.callback <- authConn
+				break
+			}
 		}
 	}
 }

rpc/rpc_client.go

@@ -4,13 +4,11 @@ import (
 	"context"
 	"errors"
 	"fmt"
-	"io"
 	"net"
 	"sync"
 	"sync/atomic"
 	"time"
 
-	"github.com/zrepl/zrepl/daemon/logging/trace"
 	"google.golang.org/grpc"
 
 	"github.com/google/uuid"
@@ -22,6 +20,7 @@ import (
 	"github.com/zrepl/zrepl/rpc/versionhandshake"
 	"github.com/zrepl/zrepl/transport"
 	"github.com/zrepl/zrepl/util/envconst"
+	"github.com/zrepl/zrepl/zfs"
 )
 
 // Client implements the active side of a replication setup.
@@ -83,76 +82,49 @@ func (c *Client) Close() {
 
 // callers must ensure that the returned io.ReadCloser is closed
 // TODO expose dataClient interface to the outside world
-func (c *Client) Send(ctx context.Context, r *pdu.SendReq) (*pdu.SendRes, io.ReadCloser, error) {
-	ctx, endSpan := trace.WithSpan(ctx, "rpc.client.Send")
-	defer endSpan()
-
+func (c *Client) Send(ctx context.Context, r *pdu.SendReq) (*pdu.SendRes, zfs.StreamCopier, error) {
 	// TODO the returned sendStream may return a read error created by the remote side
-	res, stream, err := c.dataClient.ReqSend(ctx, r)
+	res, streamCopier, err := c.dataClient.ReqSend(ctx, r)
 	if err != nil {
 		return nil, nil, err
 	}
-	if stream == nil {
+	if streamCopier == nil {
 		return res, nil, nil
 	}
 
-	return res, stream, nil
+	return res, streamCopier, nil
 
 }
 
-func (c *Client) Receive(ctx context.Context, req *pdu.ReceiveReq, stream io.ReadCloser) (*pdu.ReceiveRes, error) {
-	ctx, endSpan := trace.WithSpan(ctx, "rpc.client.Receive")
-	defer endSpan()
-
-	return c.dataClient.ReqRecv(ctx, req, stream)
+func (c *Client) Receive(ctx context.Context, req *pdu.ReceiveReq, streamCopier zfs.StreamCopier) (*pdu.ReceiveRes, error) {
+	return c.dataClient.ReqRecv(ctx, req, streamCopier)
 }
 
 func (c *Client) ListFilesystems(ctx context.Context, in *pdu.ListFilesystemReq) (*pdu.ListFilesystemRes, error) {
-	ctx, endSpan := trace.WithSpan(ctx, "rpc.client.ListFilesystems")
-	defer endSpan()
-
 	return c.controlClient.ListFilesystems(ctx, in)
 }
 
 func (c *Client) ListFilesystemVersions(ctx context.Context, in *pdu.ListFilesystemVersionsReq) (*pdu.ListFilesystemVersionsRes, error) {
-	ctx, endSpan := trace.WithSpan(ctx, "rpc.client.ListFilesystemVersions")
-	defer endSpan()
-
 	return c.controlClient.ListFilesystemVersions(ctx, in)
 }
 
 func (c *Client) DestroySnapshots(ctx context.Context, in *pdu.DestroySnapshotsReq) (*pdu.DestroySnapshotsRes, error) {
-	ctx, endSpan := trace.WithSpan(ctx, "rpc.client.DestroySnapshots")
-	defer endSpan()
-
 	return c.controlClient.DestroySnapshots(ctx, in)
 }
 
 func (c *Client) ReplicationCursor(ctx context.Context, in *pdu.ReplicationCursorReq) (*pdu.ReplicationCursorRes, error) {
-	ctx, endSpan := trace.WithSpan(ctx, "rpc.client.ReplicationCursor")
-	defer endSpan()
-
 	return c.controlClient.ReplicationCursor(ctx, in)
 }
 
 func (c *Client) SendCompleted(ctx context.Context, in *pdu.SendCompletedReq) (*pdu.SendCompletedRes, error) {
-	ctx, endSpan := trace.WithSpan(ctx, "rpc.client.SendCompleted")
-	defer endSpan()
-
 	return c.controlClient.SendCompleted(ctx, in)
 }
 
 func (c *Client) HintMostRecentCommonAncestor(ctx context.Context, in *pdu.HintMostRecentCommonAncestorReq) (*pdu.HintMostRecentCommonAncestorRes, error) {
-	ctx, endSpan := trace.WithSpan(ctx, "rpc.client.HintMostRecentCommonAncestor")
-	defer endSpan()
-
 	return c.controlClient.HintMostRecentCommonAncestor(ctx, in)
 }
 
 func (c *Client) WaitForConnectivity(ctx context.Context) error {
-	ctx, endSpan := trace.WithSpan(ctx, "rpc.client.WaitForConnectivity")
-	defer endSpan()
-
 	ctx, cancel := context.WithCancel(ctx)
 	defer cancel()
 	msg := uuid.New().String()

rpc/rpc_logging.go

@@ -3,12 +3,17 @@ package rpc
 import (
 	"context"
 
-	"github.com/zrepl/zrepl/daemon/logging"
 	"github.com/zrepl/zrepl/logger"
 )
 
 type Logger = logger.Logger
 
+type contextKey int
+
+const (
+	contextKeyLoggers contextKey = iota
+)
+
 /// All fields must be non-nil
 type Loggers struct {
 	General Logger
@@ -16,10 +21,11 @@ type Loggers struct {
 	Data    Logger
 }
 
-func GetLoggersOrPanic(ctx context.Context) Loggers {
-	return Loggers{
-		General: logging.GetLogger(ctx, logging.SubsysRPC),
-		Control: logging.GetLogger(ctx, logging.SubsysRPCControl),
-		Data:    logging.GetLogger(ctx, logging.SubsysRPCData),
-	}
+func WithLoggers(ctx context.Context, loggers Loggers) context.Context {
+	ctx = context.WithValue(ctx, contextKeyLoggers, loggers)
+	return ctx
+}
+
+func GetLoggersOrPanic(ctx context.Context) Loggers {
+	return ctx.Value(contextKeyLoggers).(Loggers)
 }

rpc/rpc_server.go

@@ -7,7 +7,6 @@ import (
 	"github.com/zrepl/zrepl/endpoint"
 	"github.com/zrepl/zrepl/replication/logic/pdu"
 	"github.com/zrepl/zrepl/rpc/dataconn"
-	"github.com/zrepl/zrepl/rpc/grpcclientidentity"
 	"github.com/zrepl/zrepl/rpc/grpcclientidentity/grpchelper"
 	"github.com/zrepl/zrepl/rpc/versionhandshake"
 	"github.com/zrepl/zrepl/transport"
@@ -31,20 +30,7 @@ type Server struct {
 	dataServerServe    serveFunc
 }
 
-type HandlerContextInterceptorData interface {
-	FullMethod() string
-	ClientIdentity() string
-}
-
-type interceptorData struct {
-	prefixMethod string
-	wrapped      HandlerContextInterceptorData
-}
-
-func (d interceptorData) ClientIdentity() string { return d.wrapped.ClientIdentity() }
-func (d interceptorData) FullMethod() string     { return d.prefixMethod + d.wrapped.FullMethod() }
-
-type HandlerContextInterceptor func(ctx context.Context, data HandlerContextInterceptorData, handler func(ctx context.Context))
+type HandlerContextInterceptor func(ctx context.Context) context.Context
 
 // config must be valid (use its Validate function).
 func NewServer(handler Handler, loggers Loggers, ctxInterceptor HandlerContextInterceptor) *Server {
@@ -52,10 +38,7 @@ func NewServer(handler Handler, loggers Loggers, ctxInterceptor HandlerContextIn
 	// setup control server
 	controlServerServe := func(ctx context.Context, controlListener transport.AuthenticatedListener, errOut chan<- error) {
 
-		var controlCtxInterceptor grpcclientidentity.Interceptor = func(ctx context.Context, data grpcclientidentity.ContextInterceptorData, handler func(ctx context.Context)) {
-			ctxInterceptor(ctx, interceptorData{"control://", data}, handler)
-		}
-		controlServer, serve := grpchelper.NewServer(controlListener, endpoint.ClientIdentityKey, loggers.Control, controlCtxInterceptor)
+		controlServer, serve := grpchelper.NewServer(controlListener, endpoint.ClientIdentityKey, loggers.Control, ctxInterceptor)
 		pdu.RegisterReplicationServer(controlServer, handler)
 
 		// give time for graceful stop until deadline expires, then hard stop
@@ -64,9 +47,8 @@ func NewServer(handler Handler, loggers Loggers, ctxInterceptor HandlerContextIn
 			if dl, ok := ctx.Deadline(); ok {
 				go time.AfterFunc(dl.Sub(dl), controlServer.Stop)
 			}
-			loggers.Control.Debug("gracefully shutting down control server")
+			loggers.Control.Debug("shutting down control server")
 			controlServer.GracefulStop()
-			loggers.Control.Debug("gracdeful shut down of control server complete")
 		}()
 
 		errOut <- serve()
@@ -76,12 +58,12 @@ func NewServer(handler Handler, loggers Loggers, ctxInterceptor HandlerContextIn
 	dataServerClientIdentitySetter := func(ctx context.Context, wire *transport.AuthConn) (context.Context, *transport.AuthConn) {
 		ci := wire.ClientIdentity()
 		ctx = context.WithValue(ctx, endpoint.ClientIdentityKey, ci)
+		if ctxInterceptor != nil {
+			ctx = ctxInterceptor(ctx) // SHADOWING
+		}
 		return ctx, wire
 	}
-	var dataCtxInterceptor dataconn.ContextInterceptor = func(ctx context.Context, data dataconn.ContextInterceptorData, handler func(ctx context.Context)) {
-		ctxInterceptor(ctx, interceptorData{"data://", data}, handler)
-	}
-	dataServer := dataconn.NewServer(dataServerClientIdentitySetter, dataCtxInterceptor, loggers.Data, handler)
+	dataServer := dataconn.NewServer(dataServerClientIdentitySetter, loggers.Data, handler)
 	dataServerServe := func(ctx context.Context, dataListener transport.AuthenticatedListener, errOut chan<- error) {
 		dataServer.Serve(ctx, dataListener)
 		errOut <- nil // TODO bad design of dataServer?
@@ -102,8 +84,6 @@ func NewServer(handler Handler, loggers Loggers, ctxInterceptor HandlerContextIn
 // Serve never returns an error, it logs them to the Server's logger.
 func (s *Server) Serve(ctx context.Context, l transport.AuthenticatedListener) {
 	ctx, cancel := context.WithCancel(ctx)
-	defer cancel()
-	defer s.logger.Debug("rpc.(*Server).Serve done")
 
 	l = versionhandshake.Listener(l, envconst.Duration("ZREPL_RPC_SERVER_VERSIONHANDSHAKE_TIMEOUT", 10*time.Second))
 

rpc/transportmux/transportmux.go

@@ -7,23 +7,33 @@ package transportmux
 
 import (
 	"context"
-	"sync/atomic"
-	"syscall"
 
 	"fmt"
 	"io"
 	"net"
 	"time"
 
-	"github.com/zrepl/zrepl/daemon/logging"
 	"github.com/zrepl/zrepl/logger"
 	"github.com/zrepl/zrepl/transport"
 )
 
+type contextKey int
+
+const (
+	contextKeyLog contextKey = 1 + iota
+)
+
 type Logger = logger.Logger
 
+func WithLogger(ctx context.Context, log Logger) context.Context {
+	return context.WithValue(ctx, contextKeyLog, log)
+}
+
 func getLog(ctx context.Context) Logger {
-	return logging.GetLogger(ctx, logging.SubsysTransportMux)
+	if l, ok := ctx.Value(contextKeyLog).(Logger); ok {
+		return l
+	}
+	return logger.NewNullLogger()
 }
 
 type acceptRes struct {
@@ -32,31 +42,12 @@ type acceptRes struct {
 }
 
 type demuxListener struct {
-	closed int32
-	conns  chan acceptRes
-}
-
-var ErrClosed = &net.OpError{
-	Op:     "accept",
-	Net:    "demux",
-	Source: nil,
-	Addr:   nil,
-	Err:    syscall.EINVAL,
+	conns chan acceptRes
 }
 
 func (l *demuxListener) Accept(ctx context.Context) (*transport.AuthConn, error) {
-	if atomic.LoadInt32(&l.closed) != 0 {
-		return nil, ErrClosed
-	}
-	select {
-	case r, ok := <-l.conns:
-		if !ok {
-			return nil, ErrClosed
-		}
-		return r.conn, r.err
-	case <-ctx.Done():
-		return nil, ctx.Err()
-	}
+	res := <-l.conns
+	return res.conn, res.err
 }
 
 type demuxAddr struct{}
@@ -68,10 +59,7 @@ func (l *demuxListener) Addr() net.Addr {
 	return demuxAddr{}
 }
 
-func (l *demuxListener) Close() error {
-	atomic.StoreInt32(&l.closed, 1)
-	return nil
-}
+func (l *demuxListener) Close() error { return nil } // TODO
 
 // Exact length of a label in bytes (0-byte padded if it is shorter).
 // This is a protocol constant, changing it breaks the wire protocol.
@@ -102,10 +90,7 @@ func Demux(ctx context.Context, rawListener transport.AuthenticatedListener, lab
 		if _, ok := padded[labelPadded]; ok {
 			return nil, fmt.Errorf("duplicate label %q", label)
 		}
-		dl := &demuxListener{
-			closed: 0,
-			conns:  make(chan acceptRes, 1),
-		}
+		dl := &demuxListener{make(chan acceptRes)}
 		padded[labelPadded] = dl
 		ret[label] = dl
 	}
@@ -118,37 +103,10 @@ func Demux(ctx context.Context, rawListener transport.AuthenticatedListener, lab
 		if err := rawListener.Close(); err != nil {
 			getLog(ctx).WithError(err).Error("error closing listener")
 		}
-
-		drainConns := func(ch chan acceptRes) {
-			for c := range ch {
-				if c.conn != nil {
-					if err := c.conn.Close(); err != nil {
-						getLog(ctx).WithError(err).Error("error closing connection while draining after listener was closed")
-					}
-				}
-			}
-		}
-		for _, dl := range ret {
-			atomic.StoreInt32(&dl.(*demuxListener).closed, 1)
-			drainConns(dl.(*demuxListener).conns)
-		}
 	}()
 
 	go func() {
-		defer func() {
-			for _, dl := range ret {
-				close(dl.(*demuxListener).conns)
-			}
-		}()
-
 		for {
-			select {
-			case <-ctx.Done():
-				getLog(ctx).WithError(ctx.Err()).Info("stop accepting new connections after context done")
-				return
-			default:
-			}
-
 			rawConn, err := rawListener.Accept(ctx)
 			if err != nil {
 				if ctx.Err() != nil {
@@ -189,6 +147,7 @@ func Demux(ctx context.Context, rawListener transport.AuthenticatedListener, lab
 			if err != nil {
 				getLog(ctx).WithError(err).Error("cannot reset deadline")
 			}
+			// blocking is intentional
 			demuxListener.conns <- acceptRes{conn: rawConn, err: nil}
 		}
 	}()

transport/tcp/serve_tcp.go

@@ -65,12 +65,6 @@ type TCPAuthListener struct {
 }
 
 func (f *TCPAuthListener) Accept(ctx context.Context) (*transport.AuthConn, error) {
-	ctx, cancel := context.WithCancel(ctx)
-	defer cancel()
-	go func() {
-		<-ctx.Done()
-		cancel()
-	}()
 	nc, err := f.TCPListener.AcceptTCP()
 	if err != nil {
 		return nil, err

transport/transport.go

@@ -8,7 +8,6 @@ import (
 	"net"
 	"syscall"
 
-	"github.com/zrepl/zrepl/daemon/logging"
 	"github.com/zrepl/zrepl/logger"
 	"github.com/zrepl/zrepl/rpc/dataconn/timeoutconn"
 	"github.com/zrepl/zrepl/zfs"
@@ -67,8 +66,19 @@ func ValidateClientIdentity(in string) (err error) {
 	return nil
 }
 
+type contextKey int
+
+const contextKeyLog contextKey = 0
+
 type Logger = logger.Logger
 
-func GetLogger(ctx context.Context) Logger {
-	return logging.GetLogger(ctx, logging.SubsysTransport)
+func WithLogger(ctx context.Context, log Logger) context.Context {
+	return context.WithValue(ctx, contextKeyLog, log)
+}
+
+func GetLogger(ctx context.Context) Logger {
+	if log, ok := ctx.Value(contextKeyLog).(Logger); ok {
+		return log
+	}
+	return logger.NewNullLogger()
 }

util/bytecounter/bytecounter_readcloser.go

@@ -1,39 +0,0 @@
-package bytecounter
-
-import (
-	"io"
-	"sync/atomic"
-)
-
-// ReadCloser wraps an io.ReadCloser, reimplementing
-// its interface and counting the bytes written to during copying.
-type ReadCloser interface {
-	io.ReadCloser
-	Count() int64
-}
-
-// NewReadCloser wraps rc.
-func NewReadCloser(rc io.ReadCloser) ReadCloser {
-	return &readCloser{rc, 0}
-}
-
-type readCloser struct {
-	rc    io.ReadCloser
-	count int64
-}
-
-func (r *readCloser) Count() int64 {
-	return atomic.LoadInt64(&r.count)
-}
-
-var _ io.ReadCloser = &readCloser{}
-
-func (r *readCloser) Close() error {
-	return r.rc.Close()
-}
-
-func (r *readCloser) Read(p []byte) (int, error) {
-	n, err := r.rc.Read(p)
-	atomic.AddInt64(&r.count, int64(n))
-	return n, err
-}

util/bytecounter/bytecounter_reader.go

@@ -0,0 +1,49 @@
+package bytecounter
+
+import (
+	"io"
+	"sync/atomic"
+	"time"
+)
+
+type ByteCounterReader struct {
+	reader io.ReadCloser
+
+	// called & accessed synchronously during Read, no external access
+	cb       func(full int64)
+	cbEvery  time.Duration
+	lastCbAt time.Time
+
+	// set atomically because it may be read by multiple threads
+	bytes int64
+}
+
+func NewByteCounterReader(reader io.ReadCloser) *ByteCounterReader {
+	return &ByteCounterReader{
+		reader: reader,
+	}
+}
+
+func (b *ByteCounterReader) SetCallback(every time.Duration, cb func(full int64)) {
+	b.cbEvery = every
+	b.cb = cb
+}
+
+func (b *ByteCounterReader) Close() error {
+	return b.reader.Close()
+}
+
+func (b *ByteCounterReader) Read(p []byte) (n int, err error) {
+	n, err = b.reader.Read(p)
+	full := atomic.AddInt64(&b.bytes, int64(n))
+	now := time.Now()
+	if b.cb != nil && now.Sub(b.lastCbAt) > b.cbEvery {
+		b.cb(full)
+		b.lastCbAt = now
+	}
+	return n, err
+}
+
+func (b *ByteCounterReader) Bytes() int64 {
+	return atomic.LoadInt64(&b.bytes)
+}

util/bytecounter/bytecounter_streamcopier.go

@@ -0,0 +1,71 @@
+package bytecounter
+
+import (
+	"io"
+	"sync/atomic"
+
+	"github.com/zrepl/zrepl/zfs"
+)
+
+// StreamCopier wraps a zfs.StreamCopier, reimplementing
+// its interface and counting the bytes written to during copying.
+type StreamCopier interface {
+	zfs.StreamCopier
+	Count() int64
+}
+
+// NewStreamCopier wraps sc into a StreamCopier.
+// If sc is io.Reader, it is guaranteed that the returned StreamCopier
+// implements that interface, too.
+func NewStreamCopier(sc zfs.StreamCopier) StreamCopier {
+	bsc := &streamCopier{sc, 0}
+	if scr, ok := sc.(io.Reader); ok {
+		return streamCopierAndReader{bsc, scr}
+	} else {
+		return bsc
+	}
+}
+
+type streamCopier struct {
+	sc    zfs.StreamCopier
+	count int64
+}
+
+// proxy writer used by streamCopier
+type streamCopierWriter struct {
+	parent *streamCopier
+	w      io.Writer
+}
+
+func (w streamCopierWriter) Write(p []byte) (n int, err error) {
+	n, err = w.w.Write(p)
+	atomic.AddInt64(&w.parent.count, int64(n))
+	return
+}
+
+func (s *streamCopier) Count() int64 {
+	return atomic.LoadInt64(&s.count)
+}
+
+var _ zfs.StreamCopier = &streamCopier{}
+
+func (s streamCopier) Close() error {
+	return s.sc.Close()
+}
+
+func (s *streamCopier) WriteStreamTo(w io.Writer) zfs.StreamCopierError {
+	ww := streamCopierWriter{s, w}
+	return s.sc.WriteStreamTo(ww)
+}
+
+// a streamCopier whose underlying sc is an io.Reader
+type streamCopierAndReader struct {
+	*streamCopier
+	asReader io.Reader
+}
+
+func (scr streamCopierAndReader) Read(p []byte) (int, error) {
+	n, err := scr.asReader.Read(p)
+	atomic.AddInt64(&scr.streamCopier.count, int64(n))
+	return n, err
+}

util/bytecounter/bytecounter_streamcopier_test.go

@@ -0,0 +1,39 @@
+package bytecounter
+
+import (
+	"io"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+
+	"github.com/zrepl/zrepl/zfs"
+)
+
+type mockStreamCopierAndReader struct {
+	zfs.StreamCopier // to satisfy interface
+	reads            int
+}
+
+func (r *mockStreamCopierAndReader) Read(p []byte) (int, error) {
+	r.reads++
+	return len(p), nil
+}
+
+var _ io.Reader = &mockStreamCopierAndReader{}
+
+func TestNewStreamCopierReexportsReader(t *testing.T) {
+	mock := &mockStreamCopierAndReader{}
+	x := NewStreamCopier(mock)
+
+	r, ok := x.(io.Reader)
+	if !ok {
+		t.Fatalf("%T does not implement io.Reader, hence reader cannout have been wrapped", x)
+	}
+
+	var buf [23]byte
+	n, err := r.Read(buf[:])
+	assert.True(t, mock.reads == 1)
+	assert.True(t, n == len(buf))
+	assert.NoError(t, err)
+	assert.True(t, x.Count() == 23)
+}

util/chainedio/chainedio_readcloser.go

@@ -1,43 +0,0 @@
-package chainedio
-
-import "io"
-
-type ChainedReadCloser struct {
-	readers   []io.Reader
-	curReader int
-}
-
-func NewChainedReader(reader ...io.Reader) *ChainedReadCloser {
-	return &ChainedReadCloser{
-		readers:   reader,
-		curReader: 0,
-	}
-}
-
-func (c *ChainedReadCloser) Read(buf []byte) (n int, err error) {
-
-	n = 0
-
-	for c.curReader < len(c.readers) {
-		n, err = c.readers[c.curReader].Read(buf)
-		if err == io.EOF {
-			c.curReader++
-			continue
-		}
-		break
-	}
-	if c.curReader == len(c.readers) {
-		err = io.EOF // actually, there was no gap
-	}
-
-	return
-}
-
-func (c *ChainedReadCloser) Close() error {
-	for _, r := range c.readers {
-		if c, ok := r.(io.Closer); ok {
-			c.Close() // TODO debug log error?
-		}
-	}
-	return nil
-}

util/chainedio/chainedio_reader.go

@@ -0,0 +1,34 @@
+package chainedio
+
+import "io"
+
+type ChainedReader struct {
+	Readers   []io.Reader
+	curReader int
+}
+
+func NewChainedReader(reader ...io.Reader) *ChainedReader {
+	return &ChainedReader{
+		Readers:   reader,
+		curReader: 0,
+	}
+}
+
+func (c *ChainedReader) Read(buf []byte) (n int, err error) {
+
+	n = 0
+
+	for c.curReader < len(c.Readers) {
+		n, err = c.Readers[c.curReader].Read(buf)
+		if err == io.EOF {
+			c.curReader++
+			continue
+		}
+		break
+	}
+	if c.curReader == len(c.Readers) {
+		err = io.EOF // actually, there was no gap
+	}
+
+	return
+}

util/chainlock/chainlock.go

@@ -40,8 +40,3 @@ func (l *L) DropWhile(f func()) {
 	defer l.Unlock().Lock()
 	f()
 }
-
-func (l *L) HoldWhile(f func()) {
-	defer l.Lock().Unlock()
-	f()
-}

util/semaphore/semaphore.go

@@ -3,7 +3,6 @@ package semaphore
 import (
 	"context"
 
-	"github.com/zrepl/zrepl/daemon/logging/trace"
 	wsemaphore "golang.org/x/sync/semaphore"
 )
 
@@ -22,7 +21,6 @@ type AcquireGuard struct {
 
 // The returned AcquireGuard is not goroutine-safe.
 func (s *S) Acquire(ctx context.Context) (*AcquireGuard, error) {
-	defer trace.WithSpanFromStackUpdateCtx(&ctx)()
 	if err := s.ws.Acquire(ctx, 1); err != nil {
 		return nil, err
 	} else if err := ctx.Err(); err != nil {

util/semaphore/semaphore_test.go

@@ -9,7 +9,6 @@ import (
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
-	"github.com/zrepl/zrepl/daemon/logging/trace"
 )
 
 func TestSemaphore(t *testing.T) {
@@ -25,17 +24,12 @@ func TestSemaphore(t *testing.T) {
 		beforeT, afterT uint32
 	}
 
-	ctx := context.Background()
-	defer trace.WithTaskFromStackUpdateCtx(&ctx)()
-
 	var wg sync.WaitGroup
 	wg.Add(numGoroutines)
 	for i := 0; i < numGoroutines; i++ {
 		go func() {
-			ctx, end := trace.WithTaskFromStack(ctx)
-			defer end()
 			defer wg.Done()
-			res, err := sem.Acquire(ctx)
+			res, err := sem.Acquire(context.Background())
 			require.NoError(t, err)
 			defer res.Release()
 			if time.Since(begin) > sleepTime {

zfs/versions.go

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

zfs/versions_destroy.go

@@ -38,8 +38,8 @@ func (o *DestroySnapOp) String() string {
 	return fmt.Sprintf("destroy operation %s@%s", o.Filesystem, o.Name)
 }
 
-func ZFSDestroyFilesystemVersions(ctx context.Context, reqs []*DestroySnapOp) {
-	doDestroy(ctx, reqs, destroyerSingleton)
+func ZFSDestroyFilesystemVersions(reqs []*DestroySnapOp) {
+	doDestroy(context.TODO(), reqs, destroyerSingleton)
 }
 
 func setDestroySnapOpErr(b []*DestroySnapOp, err error) {

zfs/zfs.go

@@ -354,6 +354,63 @@ func (a ZFSSendArgsUnvalidated) buildCommonSendArgs() ([]string, error) {
 	return args, nil
 }
 
+type ReadCloserCopier struct {
+	recorder readErrRecorder
+}
+
+type readErrRecorder struct {
+	io.ReadCloser
+	readErr error
+}
+
+type sendStreamCopierError struct {
+	isReadErr bool // if false, it's a write error
+	err       error
+}
+
+func (e sendStreamCopierError) Error() string {
+	if e.isReadErr {
+		return fmt.Sprintf("stream: read error: %s", e.err)
+	} else {
+		return fmt.Sprintf("stream: writer error: %s", e.err)
+	}
+}
+
+func (e sendStreamCopierError) IsReadError() bool  { return e.isReadErr }
+func (e sendStreamCopierError) IsWriteError() bool { return !e.isReadErr }
+
+func (r *readErrRecorder) Read(p []byte) (n int, err error) {
+	n, err = r.ReadCloser.Read(p)
+	r.readErr = err
+	return n, err
+}
+
+func NewReadCloserCopier(stream io.ReadCloser) *ReadCloserCopier {
+	return &ReadCloserCopier{recorder: readErrRecorder{stream, nil}}
+}
+
+func (c *ReadCloserCopier) WriteStreamTo(w io.Writer) StreamCopierError {
+	debug("sendStreamCopier.WriteStreamTo: begin")
+	_, err := io.Copy(w, &c.recorder)
+	debug("sendStreamCopier.WriteStreamTo: copy done")
+	if err != nil {
+		if c.recorder.readErr != nil {
+			return sendStreamCopierError{isReadErr: true, err: c.recorder.readErr}
+		} else {
+			return sendStreamCopierError{isReadErr: false, err: err}
+		}
+	}
+	return nil
+}
+
+func (c *ReadCloserCopier) Read(p []byte) (n int, err error) {
+	return c.recorder.Read(p)
+}
+
+func (c *ReadCloserCopier) Close() error {
+	return c.recorder.ReadCloser.Close()
+}
+
 func pipeWithCapacityHint(capacity int) (r, w *os.File, err error) {
 	if capacity <= 0 {
 		panic(fmt.Sprintf("capacity must be positive %v", capacity))
@@ -366,7 +423,7 @@ func pipeWithCapacityHint(capacity int) (r, w *os.File, err error) {
 	return stdoutReader, stdoutWriter, nil
 }
 
-type SendStream struct {
+type sendStream struct {
 	cmd  *zfscmd.Cmd
 	kill context.CancelFunc
 
@@ -376,7 +433,7 @@ type SendStream struct {
 	opErr        error
 }
 
-func (s *SendStream) Read(p []byte) (n int, err error) {
+func (s *sendStream) Read(p []byte) (n int, err error) {
 	s.closeMtx.Lock()
 	opErr := s.opErr
 	s.closeMtx.Unlock()
@@ -397,12 +454,12 @@ func (s *SendStream) Read(p []byte) (n int, err error) {
 	return n, err
 }
 
-func (s *SendStream) Close() error {
+func (s *sendStream) Close() error {
 	debug("sendStream: close called")
 	return s.killAndWait(nil)
 }
 
-func (s *SendStream) killAndWait(precedingReadErr error) error {
+func (s *sendStream) killAndWait(precedingReadErr error) error {
 
 	debug("sendStream: killAndWait enter")
 	defer debug("sendStream: killAndWait leave")
@@ -773,7 +830,7 @@ var ErrEncryptedSendNotSupported = fmt.Errorf("raw sends which are required for
 // (if from is "" a full ZFS send is done)
 //
 // Returns ErrEncryptedSendNotSupported if encrypted send is requested but not supported by CLI
-func ZFSSend(ctx context.Context, sendArgs ZFSSendArgsValidated) (*SendStream, error) {
+func ZFSSend(ctx context.Context, sendArgs ZFSSendArgsValidated) (*ReadCloserCopier, error) {
 
 	args := make([]string, 0)
 	args = append(args, "send")
@@ -822,14 +879,14 @@ func ZFSSend(ctx context.Context, sendArgs ZFSSendArgsValidated) (*SendStream, e
 	// close our writing-end of the pipe so that we don't wait for ourselves when reading from the reading  end
 	stdoutWriter.Close()
 
-	stream := &SendStream{
+	stream := &sendStream{
 		cmd:          cmd,
 		kill:         cancel,
 		stdoutReader: stdoutReader,
 		stderrBuf:    stderrBuf,
 	}
 
-	return stream, nil
+	return NewReadCloserCopier(stream), nil
 }
 
 type DrySendType string
@@ -968,6 +1025,24 @@ func ZFSSendDry(ctx context.Context, sendArgs ZFSSendArgsValidated) (_ *DrySendI
 	return &si, nil
 }
 
+type StreamCopierError interface {
+	error
+	IsReadError() bool
+	IsWriteError() bool
+}
+
+type StreamCopier interface {
+	// WriteStreamTo writes the stream represented by this StreamCopier
+	// to the given io.Writer.
+	WriteStreamTo(w io.Writer) StreamCopierError
+	// Close must be called as soon as it is clear that no more data will
+	// be read from the StreamCopier.
+	// If StreamCopier gets its data from a connection, it might hold
+	// a lock on the connection until Close is called. Only closing ensures
+	// that the connection can be used afterwards.
+	Close() error
+}
+
 type RecvOptions struct {
 	// Rollback to the oldest snapshot, destroy it, then perform `recv -F`.
 	// Note that this doesn't change property values, i.e. an existing local property value will be kept.
@@ -992,9 +1067,7 @@ func (e *ErrRecvResumeNotSupported) Error() string {
 	return buf.String()
 }
 
-const RecvStderrBufSiz = 1 << 15
-
-func ZFSRecv(ctx context.Context, fs string, v *ZFSSendArgVersion, stream io.ReadCloser, opts RecvOptions) (err error) {
+func ZFSRecv(ctx context.Context, fs string, v *ZFSSendArgVersion, streamCopier StreamCopier, opts RecvOptions) (err error) {
 
 	if err := v.ValidateInMemory(fs); err != nil {
 		return errors.Wrap(err, "invalid version")
@@ -1011,7 +1084,7 @@ func ZFSRecv(ctx context.Context, fs string, v *ZFSSendArgVersion, stream io.Rea
 	if opts.RollbackAndForceRecv {
 		// destroy all snapshots before `recv -F` because `recv -F`
 		// does not perform a rollback unless `send -R` was used (which we assume hasn't been the case)
-		snaps, err := ZFSListFilesystemVersions(ctx, fsdp, ListFilesystemVersionsOptions{
+		snaps, err := ZFSListFilesystemVersions(fsdp, ListFilesystemVersionsOptions{
 			Types: Snapshots,
 		})
 		if err != nil {
@@ -1061,7 +1134,7 @@ func ZFSRecv(ctx context.Context, fs string, v *ZFSSendArgVersion, stream io.Rea
 	//   cannot receive new filesystem stream: invalid backup stream
 	stdout := bytes.NewBuffer(make([]byte, 0, 1024))
 
-	stderr := bytes.NewBuffer(make([]byte, 0, RecvStderrBufSiz))
+	stderr := bytes.NewBuffer(make([]byte, 0, 1024))
 
 	stdin, stdinWriter, err := pipeWithCapacityHint(ZFSRecvPipeCapacityHint)
 	if err != nil {
@@ -1089,10 +1162,9 @@ func ZFSRecv(ctx context.Context, fs string, v *ZFSSendArgVersion, stream io.Rea
 
 	debug("started")
 
-	copierErrChan := make(chan error)
+	copierErrChan := make(chan StreamCopierError)
 	go func() {
-		_, err := io.Copy(stdinWriter, stream)
-		copierErrChan <- err
+		copierErrChan <- streamCopier.WriteStreamTo(stdinWriter)
 		stdinWriter.Close()
 	}()
 	waitErrChan := make(chan error)
@@ -1101,10 +1173,6 @@ func ZFSRecv(ctx context.Context, fs string, v *ZFSSendArgVersion, stream io.Rea
 		if err = cmd.Wait(); err != nil {
 			if rtErr := tryRecvErrorWithResumeToken(ctx, stderr.String()); rtErr != nil {
 				waitErrChan <- rtErr
-			} else if owErr := tryRecvDestroyOrOverwriteEncryptedErr(stderr.Bytes()); owErr != nil {
-				waitErrChan <- owErr
-			} else if readErr := tryRecvCannotReadFromStreamErr(stderr.Bytes()); readErr != nil {
-				waitErrChan <- readErr
 			} else {
 				waitErrChan <- &ZFSError{
 					Stderr:  stderr.Bytes(),
@@ -1115,23 +1183,22 @@ func ZFSRecv(ctx context.Context, fs string, v *ZFSSendArgVersion, stream io.Rea
 		}
 	}()
 
+	// streamCopier always fails before or simultaneously with Wait
+	// thus receive from it first
 	copierErr := <-copierErrChan
 	debug("copierErr: %T %s", copierErr, copierErr)
 	if copierErr != nil {
-		debug("killing zfs recv command after copierErr")
 		cancelCmd()
 	}
 
 	waitErr := <-waitErrChan
 	debug("waitErr: %T %s", waitErr, waitErr)
-
 	if copierErr == nil && waitErr == nil {
 		return nil
-	} else if _, isReadErr := waitErr.(*RecvCannotReadFromStreamErr); isReadErr {
-		return copierErr // likely network error reading from stream
-	} else {
-		return waitErr // almost always more interesting info. NOTE: do not wrap!
+	} else if waitErr != nil && (copierErr == nil || copierErr.IsWriteError()) {
+		return waitErr // has more interesting info in that case
 	}
+	return copierErr // if it's not a write error, the copier error is more interesting
 }
 
 type RecvFailedWithResumeTokenErr struct {
@@ -1161,43 +1228,6 @@ func (e *RecvFailedWithResumeTokenErr) Error() string {
 	return fmt.Sprintf("receive failed, resume token available: %s\n%#v", e.ResumeTokenRaw, e.ResumeTokenParsed)
 }
 
-type RecvDestroyOrOverwriteEncryptedErr struct {
-	Msg string
-}
-
-func (e *RecvDestroyOrOverwriteEncryptedErr) Error() string {
-	return e.Msg
-}
-
-var recvDestroyOrOverwriteEncryptedErrRe = regexp.MustCompile(`^(cannot receive new filesystem stream: zfs receive -F cannot be used to destroy an encrypted filesystem or overwrite an unencrypted one with an encrypted one)`)
-
-func tryRecvDestroyOrOverwriteEncryptedErr(stderr []byte) *RecvDestroyOrOverwriteEncryptedErr {
-	debug("tryRecvDestroyOrOverwriteEncryptedErr: %v", stderr)
-	m := recvDestroyOrOverwriteEncryptedErrRe.FindSubmatch(stderr)
-	if m == nil {
-		return nil
-	}
-	return &RecvDestroyOrOverwriteEncryptedErr{Msg: string(m[1])}
-}
-
-type RecvCannotReadFromStreamErr struct {
-	Msg string
-}
-
-func (e *RecvCannotReadFromStreamErr) Error() string {
-	return e.Msg
-}
-
-var reRecvCannotReadFromStreamErr = regexp.MustCompile(`^(cannot receive: failed to read from stream)$`)
-
-func tryRecvCannotReadFromStreamErr(stderr []byte) *RecvCannotReadFromStreamErr {
-	m := reRecvCannotReadFromStreamErr.FindSubmatch(stderr)
-	if m == nil {
-		return nil
-	}
-	return &RecvCannotReadFromStreamErr{Msg: string(m[1])}
-}
-
 type ClearResumeTokenError struct {
 	ZFSOutput []byte
 	CmdError  error

zfs/zfs_test.go

@@ -2,14 +2,11 @@ package zfs
 
 import (
 	"context"
-	"strings"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
-	"github.com/stretchr/testify/require"
 )
 
-// FIXME make this a platformtest
 func TestZFSListHandlesProducesZFSErrorOnNonZeroExit(t *testing.T) {
 	t.SkipNow() // FIXME ZFS_BINARY does not work if tests run in parallel
 
@@ -262,12 +259,3 @@ size	10518512
 		})
 	}
 }
-
-func TestTryRecvDestroyOrOverwriteEncryptedErr(t *testing.T) {
-	msg := "cannot receive new filesystem stream: zfs receive -F cannot be used to destroy an encrypted filesystem or overwrite an unencrypted one with an encrypted one\n"
-	assert.GreaterOrEqual(t, RecvStderrBufSiz, len(msg))
-
-	err := tryRecvDestroyOrOverwriteEncryptedErr([]byte(msg))
-	require.NotNil(t, err)
-	assert.EqualError(t, err, strings.TrimSpace(msg))
-}

zfs/zfscmd/zfscmd-logging-scraper/zfscmd_logging_scraper.go

@@ -23,7 +23,7 @@ type RuntimeLine struct {
 	Error                        string
 }
 
-var humanFormatterLineRE = regexp.MustCompile(`^(\[[^\]]+\]){2}\[zfs.cmd\]\[[^\]]+\]:\s+command\s+exited\s+(with|without)\s+error\s+(.+)`)
+var humanFormatterLineRE = regexp.MustCompile(`^(\[[^\]]+\]){2}\[zfs.cmd\]:\s+command\s+exited\s+(with|without)\s+error\s+(.+)`)
 
 func parseSecs(s string) (time.Duration, error) {
 	d, err := time.ParseDuration(s + "s")

zfs/zfscmd/zfscmd-logging-scraper/zfscmd_logging_scrapter_test.go

@@ -30,7 +30,7 @@ func TestParseHumanFormatter(t *testing.T) {
 	tcs := []testCase{
 		{
 			Name:  "human-formatter-noerror",
-			Input: `2020-04-04T00:00:05+02:00 [DEBG][jobname][zfs.cmd][task$stack$span.stack]: command exited without error usertime_s="0.008445" cmd="zfs list -H -p -o name -r -t filesystem,volume" systemtime_s="0.033783" invocation="84" total_time_s="0.037828619"`,
+			Input: `2020-04-04T00:00:05+02:00 [DEBG][jobname][zfs.cmd]: command exited without error usertime_s="0.008445" cmd="zfs list -H -p -o name -r -t filesystem,volume" systemtime_s="0.033783" invocation="84" total_time_s="0.037828619"`,
 			Expect: &RuntimeLine{
 				Cmd:       "zfs list -H -p -o name -r -t filesystem,volume",
 				TotalTime: secs("0.037828619"),
@@ -42,7 +42,7 @@ func TestParseHumanFormatter(t *testing.T) {
 		},
 		{
 			Name:  "human-formatter-witherror",
-			Input: `2020-04-04T00:00:05+02:00 [DEBG][jobname][zfs.cmd][task$stack$span.stack]: command exited with error usertime_s="0.008445" cmd="zfs list -H -p -o name -r -t filesystem,volume" systemtime_s="0.033783" invocation="84" total_time_s="0.037828619" err="some error"`,
+			Input: `2020-04-04T00:00:05+02:00 [DEBG][jobname][zfs.cmd]: command exited with error usertime_s="0.008445" cmd="zfs list -H -p -o name -r -t filesystem,volume" systemtime_s="0.033783" invocation="84" total_time_s="0.037828619" err="some error"`,
 			Expect: &RuntimeLine{
 				Cmd:       "zfs list -H -p -o name -r -t filesystem,volume",
 				TotalTime: secs("0.037828619"),
@@ -54,7 +54,7 @@ func TestParseHumanFormatter(t *testing.T) {
 		},
 		{
 			Name:  "from graylog",
-			Input: `2020-04-04T00:00:05+02:00 [DEBG][csnas][zfs.cmd][task$stack$span.stack]:  command  exited  without  error  usertime_s="0"  cmd="zfs  send  -i  zroot/ezjail/synapse-12@zrepl_20200329_095518_000  zroot/ezjail/synapse-12@zrepl_20200329_102454_000"  total_time_s="0.101598591"  invocation="85"  systemtime_s="0.041581"`,
+			Input: `2020-04-04T00:00:05+02:00 [DEBG][csnas][zfs.cmd]:  command  exited  without  error  usertime_s="0"  cmd="zfs  send  -i  zroot/ezjail/synapse-12@zrepl_20200329_095518_000  zroot/ezjail/synapse-12@zrepl_20200329_102454_000"  total_time_s="0.101598591"  invocation="85"  systemtime_s="0.041581"`,
 			Expect: &RuntimeLine{
 				Cmd:       "zfs  send  -i  zroot/ezjail/synapse-12@zrepl_20200329_095518_000  zroot/ezjail/synapse-12@zrepl_20200329_102454_000",
 				TotalTime: secs("0.101598591"),

zfs/zfscmd/zfscmd.go

@@ -14,16 +14,14 @@ import (
 	"sync"
 	"time"
 
-	"github.com/zrepl/zrepl/daemon/logging/trace"
 	"github.com/zrepl/zrepl/util/circlog"
 )
 
 type Cmd struct {
-	cmd                                      *exec.Cmd
-	ctx                                      context.Context
-	mtx                                      sync.RWMutex
-	startedAt, waitStartedAt, waitReturnedAt time.Time
-	waitReturnEndSpanCb                      trace.DoneFunc
+	cmd                       *exec.Cmd
+	ctx                       context.Context
+	mtx                       sync.RWMutex
+	startedAt, waitReturnedAt time.Time
 }
 
 func CommandContext(ctx context.Context, name string, arg ...string) *Cmd {
@@ -33,7 +31,7 @@ func CommandContext(ctx context.Context, name string, arg ...string) *Cmd {
 
 // err.(*exec.ExitError).Stderr will NOT be set
 func (c *Cmd) CombinedOutput() (o []byte, err error) {
-	c.startPre(false)
+	c.startPre()
 	c.startPost(nil)
 	c.waitPre()
 	o, err = c.cmd.CombinedOutput()
@@ -43,7 +41,7 @@ func (c *Cmd) CombinedOutput() (o []byte, err error) {
 
 // err.(*exec.ExitError).Stderr will be set
 func (c *Cmd) Output() (o []byte, err error) {
-	c.startPre(false)
+	c.startPre()
 	c.startPost(nil)
 	c.waitPre()
 	o, err = c.cmd.Output()
@@ -80,7 +78,7 @@ func (c *Cmd) log() Logger {
 }
 
 func (c *Cmd) Start() (err error) {
-	c.startPre(true)
+	c.startPre()
 	err = c.cmd.Start()
 	c.startPost(err)
 	return err
@@ -97,17 +95,15 @@ func (c *Cmd) Process() *os.Process {
 func (c *Cmd) Wait() (err error) {
 	c.waitPre()
 	err = c.cmd.Wait()
+	if !c.waitReturnedAt.IsZero() {
+		// ignore duplicate waits
+		return err
+	}
 	c.waitPost(err)
 	return err
 }
 
-func (c *Cmd) startPre(newTask bool) {
-	if newTask {
-		// avoid explosion of tasks with name c.String()
-		c.ctx, c.waitReturnEndSpanCb = trace.WithTaskAndSpan(c.ctx, "zfscmd", c.String())
-	} else {
-		c.ctx, c.waitReturnEndSpanCb = trace.WithSpan(c.ctx, c.String())
-	}
+func (c *Cmd) startPre() {
 	startPreLogging(c, time.Now())
 }
 
@@ -123,68 +119,19 @@ func (c *Cmd) startPost(err error) {
 }
 
 func (c *Cmd) waitPre() {
-	now := time.Now()
-
-	// ignore duplicate waits
-	c.mtx.Lock()
-	// ignore duplicate waits
-	if !c.waitStartedAt.IsZero() {
-		c.mtx.Unlock()
-		return
-	}
-	c.waitStartedAt = now
-	c.mtx.Unlock()
-
-	waitPreLogging(c, now)
-}
-
-type usage struct {
-	total_secs, system_secs, user_secs float64
+	waitPreLogging(c, time.Now())
 }
 
 func (c *Cmd) waitPost(err error) {
 	now := time.Now()
 
 	c.mtx.Lock()
-	// ignore duplicate waits
-	if !c.waitReturnedAt.IsZero() {
-		c.mtx.Unlock()
-		return
-	}
 	c.waitReturnedAt = now
 	c.mtx.Unlock()
 
-	// build usage
-	var u usage
-	{
-		var s *os.ProcessState
-		if err == nil {
-			s = c.cmd.ProcessState
-		} else if ee, ok := err.(*exec.ExitError); ok {
-			s = ee.ProcessState
-		}
-
-		if s == nil {
-			u = usage{
-				total_secs:  c.Runtime().Seconds(),
-				system_secs: -1,
-				user_secs:   -1,
-			}
-		} else {
-			u = usage{
-				total_secs:  c.Runtime().Seconds(),
-				system_secs: s.SystemTime().Seconds(),
-				user_secs:   s.UserTime().Seconds(),
-			}
-		}
-	}
-
-	waitPostReport(c, u, now)
-	waitPostLogging(c, u, err, now)
-	waitPostPrometheus(c, u, err, now)
-
-	// must be last because c.ctx might be used by other waitPost calls
-	c.waitReturnEndSpanCb()
+	waitPostReport(c, now)
+	waitPostLogging(c, err, now)
+	waitPostPrometheus(c, err, now)
 }
 
 // returns 0 if the command did not yet finish

zfs/zfscmd/zfscmd_context.go

@@ -3,14 +3,14 @@ package zfscmd
 import (
 	"context"
 
-	"github.com/zrepl/zrepl/daemon/logging"
 	"github.com/zrepl/zrepl/logger"
 )
 
 type contextKey int
 
 const (
-	contextKeyJobID contextKey = 1 + iota
+	contextKeyLogger contextKey = iota
+	contextKeyJobID
 )
 
 type Logger = logger.Logger
@@ -27,6 +27,13 @@ func getJobIDOrDefault(ctx context.Context, def string) string {
 	return ret
 }
 
-func getLogger(ctx context.Context) Logger {
-	return logging.GetLogger(ctx, logging.SubsysZFSCmd)
+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()
 }

zfs/zfscmd/zfscmd_logging.go

@@ -27,12 +27,11 @@ func waitPreLogging(c *Cmd, now time.Time) {
 	c.log().Debug("start waiting")
 }
 
-func waitPostLogging(c *Cmd, u usage, err error, now time.Time) {
-
+func waitPostLogging(c *Cmd, err error, now time.Time) {
 	log := c.log().
-		WithField("total_time_s", u.total_secs).
-		WithField("systemtime_s", u.system_secs).
-		WithField("usertime_s", u.user_secs)
+		WithField("total_time_s", c.Runtime().Seconds()).
+		WithField("systemtime_s", c.cmd.ProcessState.SystemTime().Seconds()).
+		WithField("usertime_s", c.cmd.ProcessState.UserTime().Seconds())
 
 	if err == nil {
 		log.Info("command exited without error")

zfs/zfscmd/zfscmd_platform_test.go

@@ -1,19 +1,13 @@
 package zfscmd
 
 import (
-	"bufio"
 	"bytes"
-	"context"
 	"io"
-	"os"
 	"os/exec"
-	"strings"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
-
-	"github.com/zrepl/zrepl/util/circlog"
 )
 
 const testBin = "./zfscmd_platform_test.bash"
@@ -64,62 +58,3 @@ func TestCmdStderrBehaviorStdoutPipe(t *testing.T) {
 	require.True(t, ok)
 	require.Empty(t, ee.Stderr) // !!!!! probably not what one would expect if we only redirect stdout
 }
-
-func TestCmdProcessState(t *testing.T) {
-
-	ctx, cancel := context.WithCancel(context.Background())
-	defer cancel()
-	cmd := exec.CommandContext(ctx, "bash", "-c", "echo running; sleep 3600")
-	stdout, err := cmd.StdoutPipe()
-	require.NoError(t, err)
-	err = cmd.Start()
-	require.NoError(t, err)
-
-	r := bufio.NewReader(stdout)
-	line, err := r.ReadString('\n')
-	require.NoError(t, err)
-	require.Equal(t, "running\n", line)
-
-	// we know it's running and sleeping
-	cancel()
-	err = cmd.Wait()
-	t.Logf("wait err %T\n%s", err, err)
-	require.Error(t, err)
-	ee, ok := err.(*exec.ExitError)
-	require.True(t, ok)
-	require.NotNil(t, ee.ProcessState)
-	require.Contains(t, ee.Error(), "killed")
-}
-
-func TestSigpipe(t *testing.T) {
-	ctx, cancel := context.WithCancel(context.Background())
-	defer cancel()
-	cmd := CommandContext(ctx, "bash", "-c", "sleep 5; echo invalid input; exit 23")
-	r, w, err := os.Pipe()
-	require.NoError(t, err)
-	output := circlog.MustNewCircularLog(1 << 20)
-	cmd.SetStdio(Stdio{
-		Stdin:  r,
-		Stdout: output,
-		Stderr: output,
-	})
-	err = cmd.Start()
-	require.NoError(t, err)
-	err = r.Close()
-	require.NoError(t, err)
-
-	// the script doesn't read stdin, but this input is almost certainly smaller than the pipe buffer
-	const LargerThanPipeBuffer = 1 << 21
-	_, err = io.Copy(w, bytes.NewBuffer(bytes.Repeat([]byte("i"), LargerThanPipeBuffer)))
-	// => io.Copy is going to block because the pipe buffer is full and the
-	//    script is not reading from it
-	// => the script is going to exit after 5s
-	// => we should expect a broken pipe error from the copier's perspective
-	t.Logf("copy err = %T: %s", err, err)
-	require.NotNil(t, err)
-	require.True(t, strings.Contains(err.Error(), "broken pipe"))
-
-	err = cmd.Wait()
-	require.EqualError(t, err, "exit status 23")
-
-}

zfs/zfscmd/zfscmd_prometheus.go

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

zfs/zfscmd/zfscmd_report.go

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