add tracing package (forgot to commit)

README.md

@@ -1,9 +1,8 @@
 [![GitHub license](https://img.shields.io/github/license/zrepl/zrepl.svg)](https://github.com/zrepl/zrepl/blob/master/LICENSE)
 [![Language: Go](https://img.shields.io/badge/language-Go-6ad7e5.svg)](https://golang.org/)
 [![User Docs](https://img.shields.io/badge/docs-web-blue.svg)](https://zrepl.github.io)
-[![Donate via Patreon](https://img.shields.io/endpoint.svg?url=https%3A%2F%2Fshieldsio-patreon.herokuapp.com%2Fzrepl%2Fpledges&style=flat&color=yellow)](https://www.patreon.com/zrepl)
-[![Donate via Liberapay](https://img.shields.io/liberapay/receives/zrepl.svg?logo=liberapay)](https://liberapay.com/zrepl/donate)
 [![Donate via PayPal](https://img.shields.io/badge/donate-paypal-yellow.svg)](https://www.paypal.com/cgi-bin/webscr?cmd=_s-xclick&hosted_button_id=R5QSXJVYHGX96)
+[![Donate via Liberapay](https://img.shields.io/liberapay/receives/zrepl.svg?logo=liberapay)](https://liberapay.com/zrepl/donate)
 [![Twitter](https://img.shields.io/twitter/url/https/github.com/zrepl/zrepl.svg?style=social)](https://twitter.com/intent/tweet?text=Wow:&url=https%3A%2F%2Fgithub.com%2Fzrepl%2Fzrepl)
 
 # zrepl
@@ -24,7 +23,6 @@ zrepl is a one-stop ZFS backup & replication solution.
    If so, think of an expressive configuration example.
 2. Think of at least one use case that generalizes from your concrete application.
 3. Open an issue on GitHub with example conf & use case attached.
-4. **Optional**: [Post a bounty](https://www.bountysource.com/teams/zrepl) on the issue, or [contact Christian Schwarz](https://cschwarz.com) for contract work.
 
 The above does not apply if you already implemented everything.
 Check out the *Coding Workflow* section below for details.
@@ -48,8 +46,11 @@ zrepl is written in [Go](https://golang.org) and uses [Go modules](https://githu
 The documentation is written in [ReStructured Text](http://docutils.sourceforge.net/rst.html) using the [Sphinx](https://www.sphinx-doc.org) framework.
 
 To get started, run `./lazy.sh devsetup` to easily install build dependencies and read `docs/installation.rst -> Compiling from Source`.
-`lazy.sh` uses `python3-pip` to fetch the build dependencies for the docs - you might want to use a [venv](https://docs.python.org/3/library/venv.html).
-If you just want to install the Go dependencies, run `./lazy.sh godep`.
+
+### Overall Architecture
+
+The application architecture is documented as part of the user docs in the *Implementation* section (`docs/content/impl`).
+Make sure to develop an understanding how zrepl is typically used by studying the user docs first.
 
 ### Project Structure
 
@@ -61,15 +62,14 @@ If you just want to install the Go dependencies, run `./lazy.sh godep`.
 │   └── samples
 ├── daemon                  # the implementation of `zrepl daemon` subcommand
 │   ├── filters
-│   ├── hooks               # snapshot hooks
 │   ├── job                 # job implementations
 │   ├── logging             # logging outlets + formatters
 │   ├── nethelpers
 │   ├── prometheus
 │   ├── pruner              # pruner implementation
 │   ├── snapper             # snapshotter implementation
-├── dist                    # supplemental material for users & package maintainers
 ├── docs                    # sphinx-based documentation
+├── dist                    # supplemental material for users & package maintainers
 │   ├── **/*.rst            # documentation in reStructuredText
 │   ├── sphinxconf
 │   │   └── conf.py         # sphinx config (see commit 445a280 why its not in docs/)
@@ -78,7 +78,6 @@ If you just want to install the Go dependencies, run `./lazy.sh godep`.
 │   └── public_git          # checkout of zrepl.github.io managed by above shell script
 ├── endpoint                # implementation of replication endpoints (=> package replication)
 ├── logger                  # our own logger package
-├── platformtest            # test suite for our zfs abstractions (error classification, etc)
 ├── pruning                 # pruning rules (the logic, not the actual execution)
 │   └── retentiongrid
 ├── replication
@@ -93,13 +92,14 @@ If you just want to install the Go dependencies, run `./lazy.sh godep`.
 │   ├── transportmux        # TCP connecter and listener used to split control & data traffic
 │   └── versionhandshake    # replication protocol version handshake perfomed on newly established connections
 ├── tlsconf                 # abstraction for Go TLS server + client config
-├── transport               # transport implementations
+├── transport               # transports implementation
 │   ├── fromconfig
 │   ├── local
 │   ├── ssh
 │   ├── tcp
 │   └── tls
 ├── util
+├── vendor                  # managed by dep
 ├── version                 # abstraction for versions (filled during build by Makefile)
 └── zfs                     # zfs(8) wrappers
 ```
@@ -134,15 +134,3 @@ There will not be a big refactoring (an attempt was made, but it's destroying to
 
 However, new contributions & patches should fix naming without further notice in the commit message.
 
-### RPC debugging
-
-Optionally, there are various RPC-related environment varibles, that if set to something != `""` will produce additional debug output on stderr:
-
-https://github.com/zrepl/zrepl/blob/master/rpc/rpc_debug.go#L11
-
-https://github.com/zrepl/zrepl/blob/master/rpc/dataconn/dataconn_debug.go#L11
-
-https://github.com/zrepl/zrepl/blob/master/rpc/dataconn/stream/stream_debug.go#L11
-
-https://github.com/zrepl/zrepl/blob/master/rpc/dataconn/heartbeatconn/heartbeatconn_debug.go#L11
-

client/testcmd.go

@@ -60,8 +60,6 @@ func runTestFilterCmd(subcommand *cli.Subcommand, args []string) error {
 		confFilter = j.Filesystems
 	case *config.PushJob:
 		confFilter = j.Filesystems
-	case *config.SnapJob:
-		confFilter = j.Filesystems
 	default:
 		return fmt.Errorf("job type %T does not have filesystems filter", j)
 	}

daemon/control.go

@@ -147,8 +147,8 @@ func (j *controlJob) Run(ctx context.Context) {
 	server := http.Server{
 		Handler: mux,
 		// control socket is local, 1s timeout should be more than sufficient, even on a loaded system
-		WriteTimeout: envconst.Duration("ZREPL_DAEMON_CONTROL_WRITE_TIMEOUT", 1*time.Second),
-		ReadTimeout:  envconst.Duration("ZREPL_DAEMON_CONTROL_READ_TIMEOUT", 1*time.Second),
+		WriteTimeout: 1 * time.Second,
+		ReadTimeout:  1 * time.Second,
 	}
 
 outer:

daemon/job/passive.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 
+	"github.com/golang/protobuf/proto"
 	"github.com/pkg/errors"
 	"github.com/prometheus/client_golang/prometheus"
 
@@ -180,11 +181,39 @@ func (j *PassiveSide) Run(ctx context.Context) {
 	}
 
 	ctxInterceptor := func(handlerCtx context.Context) context.Context {
-		return logging.WithSubsystemLoggers(handlerCtx, log)
+		reqLog := log.WithField(logging.ReqIDField, rpc.MustGetRequestID(handlerCtx))
+		return logging.WithSubsystemLoggers(handlerCtx, reqLog)
+	}
+
+	// ctxInterceptor already injected request loggers (with request id)
+	pre := func(ctx context.Context, method string, req interface{}) {
+		l := endpoint.GetLogger(ctx).WithField("method", method).WithField("reqT", fmt.Sprintf("%T", req))
+		if p, ok := req.(proto.Message); ok {
+			l = l.WithField("req", proto.CompactTextString(p))
+		}
+		ci, ok := ctx.Value(endpoint.ClientIdentityKey).(string)
+		if !ok {
+			// like endpoint, we must assume that rpc.Server sets it
+			panic(method)
+		}
+		l = l.WithField("clientIdentity", ci)
+		l.Debug("incoming")
+	}
+	post := func(ctx context.Context, res interface{}, err error) {
+		l := endpoint.GetLogger(ctx).
+			WithField("responseT", fmt.Sprintf("%T", res)).
+			WithField("errT", fmt.Sprintf("%T", err))
+		if s, ok := res.(fmt.Stringer); ok {
+			l = l.WithField("response", s.String())
+		}
+		if err != nil {
+			l = l.WithError(err)
+		}
+		l.Debug("handler done")
 	}
 
 	rpcLoggers := rpc.GetLoggersOrPanic(ctx) // WithSubsystemLoggers above
-	server := rpc.NewServer(handler, rpcLoggers, ctxInterceptor)
+	server := rpc.NewServer(handler, rpcLoggers, ctxInterceptor, pre, post)
 
 	listener, err := j.listen()
 	if err != nil {

daemon/logging/build_logging.go

@@ -96,6 +96,11 @@ func WithSubsystemLoggers(ctx context.Context, log logger.Logger) context.Contex
 			Control: log.WithField(SubsysField, SubsysRPCControl),
 			Data:    log.WithField(SubsysField, SubsysRPCData),
 		},
+		// rpc.Loggers{
+		// 	General: logger.NewNullLogger(),
+		// 	Control: logger.NewNullLogger(),
+		// 	Data:    logger.NewNullLogger(),
+		// },
 	)
 	return ctx
 }

daemon/logging/logging_formatters.go

@@ -22,6 +22,7 @@ const (
 const (
 	JobField    string = "job"
 	SubsysField string = "subsystem"
+	ReqIDField  string = "reqid"
 )
 
 type MetadataFlags int64
@@ -85,7 +86,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}
+	prefixFields := []string{JobField, SubsysField, ReqIDField}
 	prefixed := make(map[string]bool, len(prefixFields)+2)
 	for _, field := range prefixFields {
 		val, ok := e.Fields[field]

docs/installation.rst

@@ -101,9 +101,9 @@ and serves as a reference for build dependencies and procedure:
 
 ::
 
-    git clone https://github.com/zrepl/zrepl.git && \
-    cd zrepl && \
-    sudo docker build -t zrepl_build -f build.Dockerfile . && \
+    git clone https://github.com/zrepl/zrepl.git
+    cd zrepl
+    sudo docker build -t zrepl_build -f build.Dockerfile .
     sudo docker run -it --rm \
         -v "${PWD}:/src" \
         --user "$(id -u):$(id -g)" \

endpoint/context.go

@@ -19,6 +19,10 @@ func WithLogger(ctx context.Context, log Logger) context.Context {
 	return context.WithValue(ctx, contextKeyLogger, log)
 }
 
+func GetLogger(ctx context.Context) Logger {
+	return getLogger(ctx)
+}
+
 func getLogger(ctx context.Context) Logger {
 	if l, ok := ctx.Value(contextKeyLogger).(Logger); ok {
 		return l

endpoint/endpoint.go

@@ -159,6 +159,7 @@ func sendArgsFromPDUAndValidateExists(ctx context.Context, fs string, fsv *pdu.F
 }
 
 func (s *Sender) Send(ctx context.Context, r *pdu.SendReq) (*pdu.SendRes, zfs.StreamCopier, error) {
+	getLogger(ctx).WithField("req", r.String()).Debug("incoming Send request")
 
 	_, err := s.filterCheckFS(r.Filesystem)
 	if err != nil {
@@ -259,6 +260,7 @@ func (s *Sender) Send(ctx context.Context, r *pdu.SendReq) (*pdu.SendRes, zfs.St
 }
 
 func (p *Sender) SendCompleted(ctx context.Context, r *pdu.SendCompletedReq) (*pdu.SendCompletedRes, error) {
+	getLogger(ctx).WithField("req", r.String()).Debug("incoming SendCompleted request")
 	orig := r.GetOriginalReq() // may be nil, always use proto getters
 	fs := orig.GetFilesystem()
 
@@ -510,7 +512,7 @@ func (s *Receiver) ListFilesystems(ctx context.Context, req *pdu.ListFilesystemR
 	// present filesystem without the root_fs prefix
 	fss := make([]*pdu.Filesystem, 0, len(filtered))
 	for _, a := range filtered {
-		l := getLogger(ctx).WithField("fs", a)
+		l := getLogger(ctx).WithField("fs", a.ToString())
 		ph, err := zfs.ZFSGetFilesystemPlaceholderState(a)
 		if err != nil {
 			l.WithError(err).Error("error getting placeholder state")
@@ -597,7 +599,7 @@ func (s *Receiver) Send(ctx context.Context, req *pdu.SendReq) (*pdu.SendRes, zf
 var maxConcurrentZFSRecvSemaphore = semaphore.New(envconst.Int64("ZREPL_ENDPOINT_MAX_CONCURRENT_RECV", 10))
 
 func (s *Receiver) Receive(ctx context.Context, req *pdu.ReceiveReq, receive zfs.StreamCopier) (*pdu.ReceiveRes, error) {
-	getLogger(ctx).Debug("incoming Receive")
+	getLogger(ctx).WithField("rr", req.String()).Debug("incoming Receive")
 	defer receive.Close()
 
 	root := s.clientRootFromCtx(ctx)

go.mod

@@ -7,6 +7,7 @@ require (
 	github.com/gdamore/tcell v1.2.0
 	github.com/go-logfmt/logfmt v0.4.0
 	github.com/go-sql-driver/mysql v1.4.1-0.20190907122137-b2c03bcae3d4
+	github.com/gogo/protobuf v1.1.1
 	github.com/golang/protobuf v1.3.2
 	github.com/google/uuid v1.1.1
 	github.com/jinzhu/copier v0.0.0-20170922082739-db4671f3a9b8

go.sum

@@ -68,6 +68,7 @@ github.com/go-toolsmith/strparse v1.0.0/go.mod h1:YI2nUKP9YGZnL/L1/DLFBfixrcjslW
 github.com/go-toolsmith/typep v0.0.0-20181030061450-d63dc7650676/go.mod h1:JSQCQMUPdRlMZFswiq3TGpNp1GMktqkR2Ns5AIQkATU=
 github.com/go-toolsmith/typep v1.0.0/go.mod h1:JSQCQMUPdRlMZFswiq3TGpNp1GMktqkR2Ns5AIQkATU=
 github.com/gobwas/glob v0.2.3/go.mod h1:d3Ez4x06l9bZtSvzIay5+Yzi0fmZzPgnTbPcKjJAkT8=
+github.com/gogo/protobuf v1.1.1 h1:72R+M5VuhED/KujmZVcIquuo8mBgX4oVda//DQb3PXo=
 github.com/gogo/protobuf v1.1.1/go.mod h1:r8qH/GZQm5c6nD/R0oafs1akxWv10x8SbQlK7atdtwQ=
 github.com/gogo/protobuf v1.2.1/go.mod h1:hp+jE20tsWTFYpLwKvXlhS1hjn+gTNwPg2I6zVXpSg4=
 github.com/golang/glog v0.0.0-20160126235308-23def4e6c14b h1:VKtxabqXZkF25pY9ekfRL6a582T4P37/31XEstQ5p58=

replication/driver/replication_driver.go

@@ -5,6 +5,7 @@ import (
 	"errors"
 	"fmt"
 	"net"
+	"os"
 	"sort"
 	"strings"
 	"sync"
@@ -14,6 +15,7 @@ import (
 	"google.golang.org/grpc/status"
 
 	"github.com/zrepl/zrepl/replication/report"
+	"github.com/zrepl/zrepl/tracing"
 	"github.com/zrepl/zrepl/util/chainlock"
 	"github.com/zrepl/zrepl/util/envconst"
 )
@@ -206,6 +208,7 @@ func Do(ctx context.Context, planner Planner) (ReportFunc, WaitFunc) {
 			}
 			run.attempts = append(run.attempts, cur)
 			run.l.DropWhile(func() {
+				ctx := tracing.Child(ctx, fmt.Sprintf("attempt#%d", ano)) // shadow
 				cur.do(ctx, prev)
 			})
 			prev = cur
@@ -281,7 +284,7 @@ func Do(ctx context.Context, planner Planner) (ReportFunc, WaitFunc) {
 }
 
 func (a *attempt) do(ctx context.Context, prev *attempt) {
-	pfss, err := a.planner.Plan(ctx)
+	pfss, err := a.planner.Plan(tracing.Child(ctx, "plan"))
 	errTime := time.Now()
 	defer a.l.Lock().Unlock()
 	if err != nil {
@@ -361,7 +364,7 @@ func (a *attempt) do(ctx context.Context, prev *attempt) {
 		fssesDone.Add(1)
 		go func(f *fs) {
 			defer fssesDone.Done()
-			f.do(ctx, stepQueue, prevs[f])
+			f.do(tracing.Child(ctx, f.fs.ReportInfo().Name), stepQueue, prevs[f])
 		}(f)
 	}
 	a.l.DropWhile(func() {
@@ -372,6 +375,8 @@ func (a *attempt) do(ctx context.Context, prev *attempt) {
 
 func (fs *fs) do(ctx context.Context, pq *stepQueue, prev *fs) {
 
+	fmt.Fprintf(os.Stderr, "CHILD STACK: %v\n", tracing.GetStack(ctx))
+
 	defer fs.l.Lock().Unlock()
 
 	// get planned steps from replication logic

replication/logic/replication_logic.go

@@ -262,6 +262,7 @@ func (p *Planner) doPlanning(ctx context.Context) ([]*Filesystem, error) {
 		return nil, err
 	}
 	sfss := slfssres.GetFilesystems()
+	// no progress here since we could run in a live-lock on connectivity issues
 
 	rlfssres, err := p.receiver.ListFilesystems(ctx, &pdu.ListFilesystemReq{})
 	if err != nil {
@@ -605,7 +606,7 @@ func (s *Step) doReplication(ctx context.Context) error {
 	log := getLogger(ctx)
 	sr := s.buildSendRequest(false)
 
-	log.Debug("initiate send request")
+	log.WithField("sr", sr.String()).Debug("initiate send request")
 	sres, sstreamCopier, err := s.sender.Send(ctx, sr)
 	if err != nil {
 		log.WithError(err).Error("send request failed")
@@ -632,7 +633,7 @@ func (s *Step) doReplication(ctx context.Context) error {
 		To:               sr.GetTo(),
 		ClearResumeToken: !sres.UsedResumeToken,
 	}
-	log.Debug("initiate receive request")
+	log.WithField("rr", rr.String()).Debug("initiate receive request")
 	_, err = s.receiver.Receive(ctx, rr, byteCountingStream)
 	if err != nil {
 		log.
@@ -648,10 +649,11 @@ func (s *Step) doReplication(ctx context.Context) error {
 	}
 	log.Debug("receive finished")
 
-	log.Debug("tell sender replication completed")
-	_, err = s.sender.SendCompleted(ctx, &pdu.SendCompletedReq{
+	scr := &pdu.SendCompletedReq{
 		OriginalReq: sr,
-	})
+	}
+	log.WithField("scr", scr.String()).Debug("tell sender replication completed")
+	_, err = s.sender.SendCompleted(ctx, scr)
 	if err != nil {
 		log.WithError(err).Error("error telling sender that replication completed successfully")
 		return err

rpc/dataconn/dataconn_client.go

@@ -75,7 +75,7 @@ func (c *Client) recv(ctx context.Context, conn *stream.Conn, res proto.Message)
 	if err != nil {
 		return err
 	}
-	header := string(headerBuf)
+	header := string(headerBuf) // FIXME
 	if strings.HasPrefix(header, responseHeaderHandlerErrorPrefix) {
 		// FIXME distinguishable error type
 		return &RemoteHandlerError{strings.TrimPrefix(header, responseHeaderHandlerErrorPrefix)}

rpc/dataconn/dataconn_server.go

@@ -14,8 +14,8 @@ import (
 	"github.com/zrepl/zrepl/zfs"
 )
 
-// WireInterceptor has a chance to exchange the context and connection on each client connection.
-type WireInterceptor func(ctx context.Context, rawConn *transport.AuthConn) (context.Context, *transport.AuthConn)
+// ReqInterceptor has a chance to exchange the context and connection on each request.
+type ReqInterceptor func(ctx context.Context, rawConn *transport.AuthConn) (context.Context, *transport.AuthConn)
 
 // Handler implements the functionality that is exposed by Server to the Client.
 type Handler interface {
@@ -30,19 +30,27 @@ type Handler interface {
 	PingDataconn(ctx context.Context, r *pdu.PingReq) (*pdu.PingRes, error)
 }
 
+type PreHandlerInspector = func(ctx context.Context, endpoint string, req interface{})
+
+type PostHandlerInspector = func(ctx context.Context, response interface{}, err error)
+
 type Logger = logger.Logger
 
 type Server struct {
-	h   Handler
-	wi  WireInterceptor
-	log Logger
+	h    Handler
+	ri   ReqInterceptor
+	log  Logger
+	pre  PreHandlerInspector
+	post PostHandlerInspector
 }
 
-func NewServer(wi WireInterceptor, logger Logger, handler Handler) *Server {
+func NewServer(ri ReqInterceptor, logger Logger, pre PreHandlerInspector, handler Handler, post PostHandlerInspector) *Server {
 	return &Server{
 		h:   handler,
-		wi:  wi,
+		ri:  ri,
 		log: logger,
+		pre: pre,
+		post: post,
 	}
 }
 
@@ -83,8 +91,8 @@ func (s *Server) serveConn(nc *transport.AuthConn) {
 	defer s.log.Debug("serveConn done")
 
 	ctx := context.Background()
-	if s.wi != nil {
-		ctx, nc = s.wi(ctx, nc)
+	if s.ri != nil {
+		ctx, nc = s.ri(ctx, nc)
 	}
 
 	c := stream.Wrap(nc, HeartbeatInterval, HeartbeatPeerTimeout)
@@ -100,7 +108,7 @@ func (s *Server) serveConn(nc *transport.AuthConn) {
 		s.log.WithError(err).Error("error reading structured part")
 		return
 	}
-	endpoint := string(header)
+	endpoint := string(header) // FIXME
 
 	reqStructured, err := c.ReadStreamedMessage(ctx, RequestStructuredMaxSize, ReqStructured)
 	if err != nil {
@@ -120,6 +128,7 @@ func (s *Server) serveConn(nc *transport.AuthConn) {
 			s.log.WithError(err).Error("cannot unmarshal send request")
 			return
 		}
+		s.pre(ctx, endpoint, &req)
 		res, sendStream, handlerErr = s.h.Send(ctx, &req) // SHADOWING
 	case EndpointRecv:
 		var req pdu.ReceiveReq
@@ -127,6 +136,7 @@ func (s *Server) serveConn(nc *transport.AuthConn) {
 			s.log.WithError(err).Error("cannot unmarshal receive request")
 			return
 		}
+		s.pre(ctx, endpoint, &req)
 		res, handlerErr = s.h.Receive(ctx, &req, &streamCopier{streamConn: c, closeStreamOnClose: false}) // SHADOWING
 	case EndpointPing:
 		var req pdu.PingReq
@@ -134,6 +144,7 @@ func (s *Server) serveConn(nc *transport.AuthConn) {
 			s.log.WithError(err).Error("cannot unmarshal ping request")
 			return
 		}
+		s.pre(ctx, endpoint, &req)
 		res, handlerErr = s.h.PingDataconn(ctx, &req) // SHADOWING
 	default:
 		s.log.WithField("endpoint", endpoint).Error("unknown endpoint")
@@ -142,6 +153,8 @@ func (s *Server) serveConn(nc *transport.AuthConn) {
 
 	s.log.WithField("endpoint", endpoint).WithField("errType", fmt.Sprintf("%T", handlerErr)).Debug("handler returned")
 
+	s.post(ctx, res, handlerErr)
+
 	// prepare protobuf now to return the protobuf error in the header
 	// if marshaling fails. We consider failed marshaling a handler error
 	var protobuf *bytes.Buffer

rpc/dataconn/timeoutconn/timeoutconn.go

@@ -49,7 +49,7 @@ type Wire interface {
 	// No data that could otherwise be Read is lost as a consequence of this call.
 	// The use case for this API is abortive connection shutdown.
 	// To provide any value over draining Wire using io.Read, an implementation
-	// will likely use out-of-band messaging mechanisms.
+	// will likely use out-of-bounds messaging mechanisms.
 	// TODO WaitForPeerClose() (supported bool, err error)
 }
 

rpc/grpcclientidentity/authlistener_grpc_adaptor.go

@@ -101,7 +101,11 @@ func (*transportCredentials) OverrideServerName(string) error {
 
 type ContextInterceptor = func(ctx context.Context) context.Context
 
-func NewInterceptors(logger Logger, clientIdentityKey interface{}, ctxInterceptor ContextInterceptor) (unary grpc.UnaryServerInterceptor, stream grpc.StreamServerInterceptor) {
+type PreHandlerInspector func(ctx context.Context, endpoint string, req interface{})
+
+type PostHandlerInspector func(ctx context.Context, response interface{}, err error)
+
+func NewInterceptors(logger Logger, clientIdentityKey interface{}, ctxInterceptor ContextInterceptor, pre PreHandlerInspector, post PostHandlerInspector) (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)
@@ -118,7 +122,10 @@ func NewInterceptors(logger Logger, clientIdentityKey interface{}, ctxIntercepto
 		if ctxInterceptor != nil {
 			ctx = ctxInterceptor(ctx)
 		}
-		return handler(ctx, req)
+		pre(ctx, info.FullMethod, req)
+		res, err := handler(ctx, req)
+		post(ctx, res, err)
+		return res, err
 	}
 	stream = func(srv interface{}, ss grpc.ServerStream, info *grpc.StreamServerInfo, handler grpc.StreamHandler) error {
 		panic("unimplemented")

rpc/grpcclientidentity/grpchelper/authlistener_grpc_adaptor_wrapper.go

@@ -4,14 +4,18 @@ package grpchelper
 
 import (
 	"context"
+	"fmt"
+	"strings"
 	"time"
 
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/keepalive"
+	"google.golang.org/grpc/metadata"
 
 	"github.com/zrepl/zrepl/logger"
 	"github.com/zrepl/zrepl/rpc/grpcclientidentity"
 	"github.com/zrepl/zrepl/rpc/netadaptor"
+	"github.com/zrepl/zrepl/tracing"
 	"github.com/zrepl/zrepl/transport"
 )
 
@@ -28,15 +32,40 @@ type Logger = logger.Logger
 
 // ClientConn is an easy-to-use wrapper around the Dialer and TransportCredentials interface
 // to produce a grpc.ClientConn
-func ClientConn(cn transport.Connecter, log Logger) *grpc.ClientConn {
+func ClientConn(cn transport.Connecter, log Logger, genReqID func() string) *grpc.ClientConn {
 	ka := grpc.WithKeepaliveParams(keepalive.ClientParameters{
 		Time:                StartKeepalivesAfterInactivityDuration,
 		Timeout:             KeepalivePeerTimeout,
 		PermitWithoutStream: true,
 	})
+	unaryIntcpt := grpc.WithUnaryInterceptor(func(ctx context.Context, method string, req, reply interface{}, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error {
+
+		traceStack := tracing.GetStack(ctx)
+		if len(traceStack) == 0 {
+			panic("implementation error: expecting trace stack")
+		}
+		reqId := genReqID() + " " + strings.Join(traceStack, " <> ")
+
+		l := log
+		l = l.WithField("reqID", reqId)
+		l = l.WithField("reqT", fmt.Sprintf("%T", req))
+		l = l.WithField("repT", fmt.Sprintf("%T", reply))
+		l = l.WithField("ctxT", fmt.Sprintf("%T", ctx))
+		l.Debug("unary intercepted")
+
+		ctx = metadata.AppendToOutgoingContext(ctx, "zrepl-req-id", reqId)
+
+		err := invoker(ctx, method, req, reply, cc, opts...)
+
+		l = log.WithField("repT", fmt.Sprintf("%T", reply))
+		l = log.WithField("errT", fmt.Sprintf("%T", err))
+		l.Debug("reply intercepted")
+
+		return err
+	})
 	dialerOption := grpc.WithDialer(grpcclientidentity.NewDialer(log, cn))
 	cred := grpc.WithTransportCredentials(grpcclientidentity.NewTransportCredentials(log))
-	cc, err := grpc.DialContext(context.Background(), "doesn't matter done by dialer", dialerOption, cred, ka)
+	cc, err := grpc.DialContext(context.Background(), "doesn't matter done by dialer", dialerOption, cred, ka, unaryIntcpt)
 	if err != nil {
 		log.WithError(err).Error("cannot create gRPC client conn (non-blocking)")
 		// It's ok to panic here: the we call grpc.DialContext without the
@@ -50,7 +79,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.ContextInterceptor) (srv *grpc.Server, serve func() error) {
+func NewServer(authListener transport.AuthenticatedListener, clientIdentityKey interface{}, logger grpcclientidentity.Logger, interceptor grpcclientidentity.ContextInterceptor, pre grpcclientidentity.PreHandlerInspector, post grpcclientidentity.PostHandlerInspector) (srv *grpc.Server, serve func() error) {
 	ka := grpc.KeepaliveParams(keepalive.ServerParameters{
 		Time:    StartKeepalivesAfterInactivityDuration,
 		Timeout: KeepalivePeerTimeout,
@@ -60,8 +89,17 @@ func NewServer(authListener transport.AuthenticatedListener, clientIdentityKey i
 		PermitWithoutStream: true,
 	})
 	tcs := grpcclientidentity.NewTransportCredentials(logger)
-	unary, stream := grpcclientidentity.NewInterceptors(logger, clientIdentityKey, ctxInterceptor)
-	srv = grpc.NewServer(grpc.Creds(tcs), grpc.UnaryInterceptor(unary), grpc.StreamInterceptor(stream), ka, ep)
+	unary, stream := grpcclientidentity.NewInterceptors(logger, clientIdentityKey, interceptor, pre, post)
+	unary2 := func(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (resp interface{}, err error) {
+		md, ok := metadata.FromIncomingContext(ctx)
+		if !ok {
+			logger.Error("no request id in incoming request")
+		} else {
+			logger.Info(fmt.Sprintf("req-id = %v", md.Get("zrepl-req-id")))
+		}
+		return unary(ctx, req, info, handler)
+	}
+	srv = grpc.NewServer(grpc.Creds(tcs), grpc.UnaryInterceptor(unary2), grpc.StreamInterceptor(stream), ka, ep)
 
 	serve = func() error {
 		if err := srv.Serve(netadaptor.New(authListener, logger)); err != nil {

rpc/rpc_client.go

@@ -26,6 +26,7 @@ import (
 // Client implements the active side of a replication setup.
 // It satisfies the Endpoint, Sender and Receiver interface defined by package replication.
 type Client struct {
+	reqIDGen      *requestIDGenerator
 	dataClient    *dataconn.Client
 	controlClient pdu.ReplicationClient // this the grpc client instance, see constructor
 	controlConn   *grpc.ClientConn
@@ -47,10 +48,13 @@ func NewClient(cn transport.Connecter, loggers Loggers) *Client {
 	muxedConnecter := mux(cn)
 
 	c := &Client{
-		loggers: loggers,
-		closed:  make(chan struct{}),
+		loggers:  loggers,
+		closed:   make(chan struct{}),
+		reqIDGen: newRequestIDGenerator(),
 	}
-	grpcConn := grpchelper.ClientConn(muxedConnecter.control, loggers.Control)
+	grpcConn := grpchelper.ClientConn(muxedConnecter.control, loggers.Control, func() string {
+		return c.reqIDGen.newID().String()
+	})
 
 	go func() {
 		ctx, cancel := context.WithCancel(context.Background())

rpc/rpc_requestid.go

@@ -0,0 +1,58 @@
+package rpc
+
+import (
+	"context"
+	"encoding/base64"
+	"strings"
+
+	"github.com/google/uuid"
+)
+
+type requestIDGenerator struct{}
+
+func newRequestIDGenerator() *requestIDGenerator {
+	return &requestIDGenerator{}
+}
+
+type RequestID struct{ s string }
+
+func (r RequestID) String() string { return r.s }
+
+type requestIDContextKey int
+
+const (
+	requestIDContextKeyRequestID requestIDContextKey = 1 + iota
+)
+
+func (g *requestIDGenerator) newID() RequestID {
+	id := uuid.New()
+	var buf strings.Builder
+	enc := base64.NewEncoder(base64.RawStdEncoding, &buf)
+	n, err := enc.Write(id[:])
+	if err != nil {
+		panic(err)
+	} else if n != len(id) {
+		panic(n)
+	}
+	if err := enc.Close(); err != nil {
+		panic(err)
+	}
+	return RequestID{buf.String()}
+}
+
+func (g *requestIDGenerator) inject(ctx context.Context) context.Context {
+	return context.WithValue(ctx, requestIDContextKeyRequestID, g.newID())
+}
+
+func GetRequestID(ctx context.Context) (id RequestID, ok bool) {
+	id, ok = ctx.Value(requestIDContextKeyRequestID).(RequestID)
+	return id, ok
+}
+
+func MustGetRequestID(ctx context.Context) (id RequestID) {
+	id, ok := GetRequestID(ctx)
+	if !ok {
+		panic("calling context expectes request id to bet set")
+	}
+	return id
+}

rpc/rpc_server.go

@@ -7,6 +7,7 @@ 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"
@@ -30,15 +31,35 @@ type Server struct {
 	dataServerServe    serveFunc
 }
 
-type HandlerContextInterceptor func(ctx context.Context) context.Context
+type CtxInterceptor = grpcclientidentity.ContextInterceptor
+
+func chainInterceptors(interceptors []CtxInterceptor) CtxInterceptor {
+	return func(ctx context.Context) (_ context.Context) {
+		for _, i := range interceptors {
+			ctx = i(ctx)
+		}
+		return ctx
+	}
+}
+
+type PreHandlerInspector = func(ctx context.Context, endpoint string, req interface{})
+
+type PostHandlerInspector = func(ctx context.Context, response interface{}, err error)
 
 // config must be valid (use its Validate function).
-func NewServer(handler Handler, loggers Loggers, ctxInterceptor HandlerContextInterceptor) *Server {
+func NewServer(handler Handler, loggers Loggers, ctxInterceptor CtxInterceptor, pre PreHandlerInspector, post PostHandlerInspector) *Server {
+
+	reqIdGen := newRequestIDGenerator()
+
+	ctxInterceptor = chainInterceptors([]CtxInterceptor{
+		reqIdGen.inject,
+		ctxInterceptor,
+	})
 
 	// setup control server
 	controlServerServe := func(ctx context.Context, controlListener transport.AuthenticatedListener, errOut chan<- error) {
 
-		controlServer, serve := grpchelper.NewServer(controlListener, endpoint.ClientIdentityKey, loggers.Control, ctxInterceptor)
+		controlServer, serve := grpchelper.NewServer(controlListener, endpoint.ClientIdentityKey, loggers.Control, ctxInterceptor, pre, post)
 		pdu.RegisterReplicationServer(controlServer, handler)
 
 		// give time for graceful stop until deadline expires, then hard stop
@@ -63,7 +84,7 @@ func NewServer(handler Handler, loggers Loggers, ctxInterceptor HandlerContextIn
 		}
 		return ctx, wire
 	}
-	dataServer := dataconn.NewServer(dataServerClientIdentitySetter, loggers.Data, handler)
+	dataServer := dataconn.NewServer(dataServerClientIdentitySetter, loggers.Data, pre, handler, post)
 	dataServerServe := func(ctx context.Context, dataListener transport.AuthenticatedListener, errOut chan<- error) {
 		dataServer.Serve(ctx, dataListener)
 		errOut <- nil // TODO bad design of dataServer?

tracing/tracing.go

@@ -0,0 +1,53 @@
+package tracing
+
+import "context"
+
+type tracingContextKey int
+
+const (
+	CallerContext tracingContextKey = 1 + iota
+)
+
+type jobSubtree struct {
+	jobid string
+}
+
+type ctx struct {
+	parent *ctx
+	job    *jobSubtree
+	ident  string
+}
+
+var root = &ctx{nil, nil, ""}
+
+func getParentOrRoot(c context.Context) *ctx {
+	parent, ok := c.Value(CallerContext).(*ctx)
+	if !ok {
+		parent = root
+	}
+	return parent
+}
+
+func makeChild(c context.Context, child *ctx) context.Context {
+	if child.parent == nil {
+		panic(child)
+	}
+	return context.WithValue(c, CallerContext, child)
+}
+
+func Child(c context.Context, ident string) context.Context {
+	parent := getParentOrRoot(c)
+	return makeChild(c, &ctx{parent: parent, ident: ident})
+}
+
+func GetStack(c context.Context) (idents []string) {
+	ct, ok := c.Value(CallerContext).(*ctx)
+	if !ok {
+		return idents
+	}
+	for ct.parent != nil {
+		idents = append(idents, ct.ident)
+		ct = ct.parent
+	}
+	return idents
+}

tracing/tracing_test.go

@@ -0,0 +1,21 @@
+package tracing
+
+import (
+	"context"
+	"strings"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestIt(t *testing.T) {
+	ctx := context.Background()
+
+	ctx = Child(ctx, "a")
+	ctx = Child(ctx, "b")
+	ctx = Child(ctx, "c")
+	ctx = Child(ctx, "d")
+
+	assert.Equal(t, "dcba", strings.Join(GetStack(ctx), ""))
+
+}

transport/tls/serve_tls.go

@@ -32,7 +32,7 @@ func TLSListenerFactoryFromConfig(c *config.Global, in *config.TLSServe) (transp
 
 	serverCert, err := tls.LoadX509KeyPair(in.Cert, in.Key)
 	if err != nil {
-		return nil, errors.Wrap(err, "cannot parse cert/key pair")
+		return nil, errors.Wrap(err, "cannot parse cer/key pair")
 	}
 
 	clientCNs := make(map[string]struct{}, len(in.ClientCNs))