[#307] add package trace, integrate it with logging, and adopt it throughout zrepl

package trace:

- introduce the concept of tasks and spans, tracked as linked list within ctx
    - see package-level docs for an overview of the concepts
    - **main feature 1**: unique stack of task and span IDs
        - makes it easy to follow a series of log entries in concurrent code
    - **main feature 2**: ability to produce a chrome://tracing-compatible trace file
        - either via an env variable or a `zrepl pprof` subcommand
        - this is not a CPU profile, we already have go pprof for that
        - but it is very useful to visually inspect where the
          replication / snapshotter / pruner spends its time
          ( fixes #307 )

usage in package daemon/logging:

- goal: every log entry should have a trace field with the ID stack from package trace

- make `logging.GetLogger(ctx, Subsys)` the authoritative `logger.Logger` factory function
    - the context carries a linked list of injected fields which
      `logging.GetLogger` adds to the logger it returns
    - `logging.GetLogger` also uses package `trace` to get the
      task-and-span-stack and injects it into the returned logger's fields

replication/driver/replication_driver.go

@@ -10,6 +10,7 @@ import (
 	"sync"
 	"time"
 
+	"github.com/zrepl/zrepl/daemon/logging/trace"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
 
@@ -287,6 +288,17 @@ 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()
@@ -294,7 +306,7 @@ func (a *attempt) do(ctx context.Context, prev *attempt) {
 		a.planErr = newTimedError(err, errTime)
 		a.fss = nil
 		a.finishedAt = time.Now()
-		return
+		return nil
 	}
 
 	for _, pfs := range pfss {
@@ -351,7 +363,7 @@ func (a *attempt) do(ctx context.Context, prev *attempt) {
 			a.planErr = newTimedError(errors.New(msg.String()), now)
 			a.fss = nil
 			a.finishedAt = now
-			return
+			return nil
 		}
 		for cur, fss := range prevFSs {
 			if len(fss) > 0 {
@@ -373,6 +385,15 @@ func (a *attempt) do(ctx context.Context, prev *attempt) {
 		}
 	}
 
+	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
@@ -380,6 +401,9 @@ func (a *attempt) do(ctx context.Context, prev *attempt) {
 		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)
 	}
@@ -423,7 +447,7 @@ 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(f, targetDate)()
+		defer pq.WaitReady(ctx, f, targetDate)()
 		psteps, err = f.fs.PlanFS(ctx) // no shadow
 		errTime = time.Now()           // no shadow
 	})
@@ -584,8 +608,10 @@ func (f *fs) do(ctx context.Context, pq *stepQueue, prev *fs) {
 		f.l.DropWhile(func() {
 			// wait for parallel replication
 			targetDate := s.step.TargetDate()
-			defer pq.WaitReady(f, 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
 		})
 

replication/driver/replication_driver_logging.go

@@ -3,23 +3,10 @@ package driver
 import (
 	"context"
 
+	"github.com/zrepl/zrepl/daemon/logging"
 	"github.com/zrepl/zrepl/logger"
 )
 
-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)
+func getLog(ctx context.Context) logger.Logger {
+	return logging.GetLogger(ctx, logging.SubsysReplication)
 }

replication/driver/replication_driver_test.go

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

replication/driver/replication_stepqueue_test.go

@@ -1,6 +1,7 @@
 package driver
 
 import (
+	"context"
 	"fmt"
 	"math"
 	"sort"
@@ -11,18 +12,23 @@ 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("1", time.Unix(9999, 0))()
+		defer q.WaitReady(ctx, "1", time.Unix(9999, 0))()
 		ret := atomic.AddUint32(&ctr, 1)
 		assert.Equal(t, uint32(1), ret)
 		time.Sleep(1 * time.Second)
@@ -34,20 +40,26 @@ 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("2", time.Unix(2, 0))()
+		defer q.WaitReady(ctx, "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("3", time.Unix(3, 0))()
+		defer q.WaitReady(ctx, "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("4", time.Unix(4, 0))()
+		defer q.WaitReady(ctx, "4", time.Unix(4, 0))()
 		ret := atomic.AddUint32(&ctr, 1)
 		assert.Equal(t, uint32(4), ret)
 	}()
@@ -77,6 +89,8 @@ 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
@@ -90,12 +104,14 @@ 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(fs, t)
+				done := q.WaitReady(ctx, fs, t)
 				wakeAt := time.Since(begin)
 				time.Sleep(sleepTimePerStep)
 				done()

replication/logic/replication_logic.go

@@ -9,7 +9,9 @@ import (
 	"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"
@@ -80,6 +82,8 @@ 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
@@ -303,9 +307,11 @@ func (p *Planner) doPlanning(ctx context.Context) ([]*Filesystem, error) {
 
 func (fs *Filesystem) doPlanning(ctx context.Context) ([]*Step, error) {
 
-	log := getLogger(ctx).WithField("filesystem", fs.Path)
+	log := func(ctx context.Context) logger.Logger {
+		return getLogger(ctx).WithField("filesystem", fs.Path)
+	}
 
-	log.Debug("assessing filesystem")
+	log(ctx).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")
@@ -313,14 +319,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.WithError(err).Error("cannot get remote filesystem versions")
+		log(ctx).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.Error(err.Error())
+		log(ctx).Error(err.Error())
 		return nil, err
 	}
 
@@ -328,7 +334,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.WithError(err).Error("receiver error")
+			log(ctx).WithError(err).Error("receiver error")
 			return nil, err
 		}
 		rfsvs = rfsvsres.GetVersions()
@@ -340,17 +346,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.WithField("receiverFS.ResumeToken", resumeTokenRaw).Debug("decode receiver fs resume token")
+		log(ctx).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.WithError(err).Error("cannot decode resume token, aborting")
+			log(ctx).WithError(err).Error("cannot decode resume token, aborting")
 			return nil, err
 		}
-		log.WithField("token", resumeToken).Debug("decode resume token")
+		log(ctx).WithField("token", resumeToken).Debug("decode resume token")
 	}
 
 	// give both sides a hint about how far prior replication attempts got
@@ -369,7 +375,10 @@ func (fs *Filesystem) doPlanning(ctx context.Context) ([]*Step, error) {
 		var wg sync.WaitGroup
 		doHint := func(ep Endpoint, name string) {
 			defer wg.Done()
-			log := log.WithField("to_side", name).
+			ctx, endTask := trace.WithTask(ctx, "hint-mrca-"+name)
+			defer endTask()
+
+			log := log(ctx).WithField("to_side", name).
 				WithField("sender_mrca", sender_mrca.String())
 			log.Debug("hint most recent common ancestor")
 			hint := &pdu.HintMostRecentCommonAncestorReq{
@@ -428,7 +437,7 @@ func (fs *Filesystem) doPlanning(ctx context.Context) ([]*Step, error) {
 			encryptionMatches = true
 		}
 
-		log.WithField("fromVersion", fromVersion).
+		log(ctx).WithField("fromVersion", fromVersion).
 			WithField("toVersion", toVersion).
 			WithField("encryptionMatches", encryptionMatches).
 			Debug("result of resume-token-matching to sender's versions")
@@ -484,11 +493,11 @@ func (fs *Filesystem) doPlanning(ctx context.Context) ([]*Step, error) {
 			var msg string
 			path, msg = resolveConflict(conflict) // no shadowing allowed!
 			if path != nil {
-				log.WithField("conflict", conflict).Info("conflict")
-				log.WithField("resolution", msg).Info("automatically resolved")
+				log(ctx).WithField("conflict", conflict).Info("conflict")
+				log(ctx).WithField("resolution", msg).Info("automatically resolved")
 			} else {
-				log.WithField("conflict", conflict).Error("conflict")
-				log.WithField("problem", msg).Error("cannot resolve conflict")
+				log(ctx).WithField("conflict", conflict).Error("conflict")
+				log(ctx).WithField("problem", msg).Error("cannot resolve conflict")
 			}
 		}
 		if len(path) == 0 {
@@ -522,37 +531,35 @@ func (fs *Filesystem) doPlanning(ctx context.Context) ([]*Step, error) {
 	}
 
 	if len(steps) == 0 {
-		log.Info("planning determined that no replication steps are required")
+		log(ctx).Info("planning determined that no replication steps are required")
 	}
 
-	log.Debug("compute send size estimate")
+	log(ctx).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 {
-		wg.Add(1)
-		go func(step *Step) {
-			defer wg.Done()
-
+		step := step // local copy that is moved into the closure
+		fanOutAdd(func(ctx context.Context) {
 			// 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(fanOutCtx)
+			guard, err := fs.sizeEstimateRequestSem.Acquire(ctx)
 			if err != nil {
 				fanOutCancel()
 				return
 			}
 			defer guard.Release()
 
-			err = step.updateSizeEstimate(fanOutCtx)
+			err = step.updateSizeEstimate(ctx)
 			if err != nil {
-				log.WithError(err).WithField("step", step).Error("error computing size estimate")
+				log(ctx).WithError(err).WithField("step", step).Error("error computing size estimate")
 				fanOutCancel()
 			}
 			errs <- err
-		}(step)
+		})
 	}
-	wg.Wait()
+	fanOutWait()
 	close(errs)
 	var significantErr error = nil
 	for err := range errs {
@@ -566,7 +573,7 @@ func (fs *Filesystem) doPlanning(ctx context.Context) ([]*Step, error) {
 		return nil, significantErr
 	}
 
-	log.Debug("filesystem planning finished")
+	log(ctx).Debug("filesystem planning finished")
 	return steps, nil
 }
 

replication/logic/replication_logic_context.go

@@ -3,26 +3,10 @@ package logic
 import (
 	"context"
 
+	"github.com/zrepl/zrepl/daemon/logging"
 	"github.com/zrepl/zrepl/logger"
 )
 
-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
+func getLogger(ctx context.Context) logger.Logger {
+	return logging.GetLogger(ctx, logging.SubsysReplication)
 }