From fc9dbdf44944097b543c9a19c1e53504a4f77930 Mon Sep 17 00:00:00 2001 From: Christian Schwarz Date: Sat, 25 Apr 2020 12:39:59 +0200 Subject: [PATCH] [WIP] factor out trace functionality into separate package and add Go docs --- cli/cli.go | 4 +- daemon/daemon.go | 3 +- daemon/hooks/hooks_test.go | 3 +- daemon/job/active.go | 14 +- daemon/job/passive.go | 10 +- daemon/job/snapjob.go | 10 +- daemon/logging/build_logging.go | 10 +- .../logging/logging_activity_trace_debug.go | 15 -- daemon/logging/logging_context.go | 4 +- .../{logging_span.go => trace/trace.go} | 251 +++++++++--------- .../trace_chrometrace.go} | 10 +- daemon/logging/trace/trace_context.go | 27 ++ daemon/logging/trace/trace_convenience.go | 74 ++++++ .../trace_convenience_test.go} | 4 +- daemon/logging/trace/trace_debug.go | 15 ++ daemon/logging/trace/trace_genID.go | 47 ++++ .../trace_unique_concurrent_task_namer.go} | 2 +- ...race_unique_concurrent_task_namer_test.go} | 2 +- daemon/pprof.go | 4 +- daemon/snapper/snapper.go | 3 +- endpoint/endpoint.go | 48 ++-- platformtest/harness/harness.go | 3 +- replication/driver/replication_driver.go | 10 +- replication/driver/replication_driver_test.go | 4 +- replication/driver/replication_stepqueue.go | 4 +- .../driver/replication_stepqueue_test.go | 17 +- replication/logic/replication_logic.go | 8 +- rpc/rpc_client.go | 20 +- util/semaphore/semaphore.go | 5 +- util/semaphore/semaphore_test.go | 7 +- zfs/zfscmd/zfscmd.go | 8 +- 31 files changed, 392 insertions(+), 254 deletions(-) delete mode 100644 daemon/logging/logging_activity_trace_debug.go rename daemon/logging/{logging_span.go => trace/trace.go} (52%) rename daemon/logging/{logging_activity_trace.go => trace/trace_chrometrace.go} (95%) create mode 100644 daemon/logging/trace/trace_context.go create mode 100644 daemon/logging/trace/trace_convenience.go rename daemon/logging/{logging_span_test.go => trace/trace_convenience_test.go} (74%) create mode 100644 daemon/logging/trace/trace_debug.go create mode 100644 daemon/logging/trace/trace_genID.go rename daemon/logging/{logging_span_unique_concurrent_task_namer.go => trace/trace_unique_concurrent_task_namer.go} (98%) rename daemon/logging/{logging_span_unique_concurrent_task_namer_test.go => trace/trace_unique_concurrent_task_namer_test.go} (98%) diff --git a/cli/cli.go b/cli/cli.go index dceb563..4e6feb9 100644 --- a/cli/cli.go +++ b/cli/cli.go @@ -7,9 +7,9 @@ import ( "github.com/spf13/cobra" "github.com/spf13/pflag" + "github.com/zrepl/zrepl/daemon/logging/trace" "github.com/zrepl/zrepl/config" - "github.com/zrepl/zrepl/daemon/logging" ) var rootArgs struct { @@ -99,7 +99,7 @@ func (s *Subcommand) Config() *config.Config { func (s *Subcommand) run(cmd *cobra.Command, args []string) { s.tryParseConfig() ctx := context.Background() - endTask := logging.WithTaskFromStackUpdateCtx(&ctx) + endTask := trace.WithTaskFromStackUpdateCtx(&ctx) defer endTask() err := s.Run(ctx, s, args) endTask() diff --git a/daemon/daemon.go b/daemon/daemon.go index 0d85617..ba43af0 100644 --- a/daemon/daemon.go +++ b/daemon/daemon.go @@ -12,6 +12,7 @@ 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" @@ -84,7 +85,7 @@ func Run(ctx context.Context, conf *config.Config) error { // register global (=non job-local) metrics zfscmd.RegisterMetrics(prometheus.DefaultRegisterer) - logging.RegisterMetrics(prometheus.DefaultRegisterer) + trace.RegisterMetrics(prometheus.DefaultRegisterer) log.Info("starting daemon") diff --git a/daemon/hooks/hooks_test.go b/daemon/hooks/hooks_test.go index 4891bec..0a25b05 100644 --- a/daemon/hooks/hooks_test.go +++ b/daemon/hooks/hooks_test.go @@ -10,6 +10,7 @@ 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" @@ -70,7 +71,7 @@ func curry(f comparisonAssertionFunc, expected interface{}, right bool) (ret val } func TestHooks(t *testing.T) { - ctx, end := logging.WithTaskFromStack(context.Background()) + ctx, end := trace.WithTaskFromStack(context.Background()) defer end() testFSName := "testpool/testdataset" diff --git a/daemon/job/active.go b/daemon/job/active.go index 25dedc3..46f1d65 100644 --- a/daemon/job/active.go +++ b/daemon/job/active.go @@ -9,12 +9,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" @@ -377,7 +377,7 @@ func (j *ActiveSide) SenderConfig() *endpoint.SenderConfig { } func (j *ActiveSide) Run(ctx context.Context) { - ctx, endTask := logging.WithTaskAndSpan(ctx, "active-side-job", j.Name()) + ctx, endTask := trace.WithTaskAndSpan(ctx, "active-side-job", j.Name()) defer endTask() log := GetLogger(ctx) @@ -386,7 +386,7 @@ func (j *ActiveSide) Run(ctx context.Context) { periodicDone := make(chan struct{}) ctx, cancel := context.WithCancel(ctx) defer cancel() - periodicCtx, endTask := logging.WithTask(ctx, "periodic") + periodicCtx, endTask := trace.WithTask(ctx, "periodic") defer endTask() go j.mode.RunPeriodic(periodicCtx, periodicDone) @@ -404,7 +404,7 @@ outer: case <-periodicDone: } invocationCount++ - invocationCtx, endSpan := logging.WithSpan(ctx, fmt.Sprintf("invocation-%d", invocationCount)) + invocationCtx, endSpan := trace.WithSpan(ctx, fmt.Sprintf("invocation-%d", invocationCount)) j.do(invocationCtx) endSpan() } @@ -435,7 +435,7 @@ func (j *ActiveSide) do(ctx context.Context) { return default: } - ctx, endSpan := logging.WithSpan(ctx, "replication") + ctx, endSpan := trace.WithSpan(ctx, "replication") ctx, repCancel := context.WithCancel(ctx) var repWait driver.WaitFunc j.updateTasks(func(tasks *activeSideTasks) { @@ -459,7 +459,7 @@ func (j *ActiveSide) do(ctx context.Context) { return default: } - ctx, endSpan := logging.WithSpan(ctx, "prune_sender") + 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) @@ -478,7 +478,7 @@ func (j *ActiveSide) do(ctx context.Context) { return default: } - ctx, endSpan := logging.WithSpan(ctx, "prune_recever") + 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) diff --git a/daemon/job/passive.go b/daemon/job/passive.go index 7970d56..3d3f79c 100644 --- a/daemon/job/passive.go +++ b/daemon/job/passive.go @@ -6,6 +6,7 @@ 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" @@ -164,12 +165,12 @@ func (j *PassiveSide) SenderConfig() *endpoint.SenderConfig { func (*PassiveSide) RegisterMetrics(registerer prometheus.Registerer) {} func (j *PassiveSide) Run(ctx context.Context) { - ctx, endTask := logging.WithTaskAndSpan(ctx, "passive-side-job", j.Name()) + ctx, endTask := trace.WithTaskAndSpan(ctx, "passive-side-job", j.Name()) defer endTask() log := GetLogger(ctx) defer log.Info("job exiting") { - ctx, endTask := logging.WithTask(ctx, "periodic") // shadowing + ctx, endTask := trace.WithTask(ctx, "periodic") // shadowing defer endTask() ctx, cancel := context.WithCancel(ctx) defer cancel() @@ -182,10 +183,11 @@ func (j *PassiveSide) Run(ctx context.Context) { } ctxInterceptor := func(handlerCtx context.Context, info rpc.HandlerContextInterceptorData, handler func(ctx context.Context)) { - // the handlerCtx is clean => need to inherit logging config from job 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 := logging.WithTaskAndSpan(handlerCtx, "handler", fmt.Sprintf("job=%q client=%q method=%q", j.Name(), info.ClientIdentity(), info.FullMethod())) + handlerCtx, endTask := trace.WithTaskAndSpan(handlerCtx, "handler", fmt.Sprintf("job=%q client=%q method=%q", j.Name(), info.ClientIdentity(), info.FullMethod())) defer endTask() handler(handlerCtx) } diff --git a/daemon/job/snapjob.go b/daemon/job/snapjob.go index 75780cd..3ce7ee4 100644 --- a/daemon/job/snapjob.go +++ b/daemon/job/snapjob.go @@ -7,11 +7,11 @@ 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" "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,7 +90,7 @@ 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 := logging.WithTaskAndSpan(ctx, "snap-job", j.Name()) + ctx, endTask := trace.WithTaskAndSpan(ctx, "snap-job", j.Name()) defer endTask() log := GetLogger(ctx) @@ -99,7 +99,7 @@ func (j *SnapJob) Run(ctx context.Context) { periodicDone := make(chan struct{}) ctx, cancel := context.WithCancel(ctx) defer cancel() - periodicCtx, endTask := logging.WithTask(ctx, "snapshotting") + periodicCtx, endTask := trace.WithTask(ctx, "snapshotting") defer endTask() go j.snapper.Run(periodicCtx, periodicDone) @@ -117,7 +117,7 @@ outer: } invocationCount++ - invocationCtx, endSpan := logging.WithSpan(ctx, fmt.Sprintf("invocation-%d", invocationCount)) + invocationCtx, endSpan := trace.WithSpan(ctx, fmt.Sprintf("invocation-%d", invocationCount)) j.doPrune(invocationCtx) endSpan() } @@ -167,7 +167,7 @@ func (h alwaysUpToDateReplicationCursorHistory) ListFilesystems(ctx context.Cont } func (j *SnapJob) doPrune(ctx context.Context) { - ctx, endSpan := logging.WithSpan(ctx, "snap-job-do-prune") + ctx, endSpan := trace.WithSpan(ctx, "snap-job-do-prune") defer endSpan() log := GetLogger(ctx) sender := endpoint.NewSender(endpoint.SenderConfig{ diff --git a/daemon/logging/build_logging.go b/daemon/logging/build_logging.go index b875770..f34fcd6 100644 --- a/daemon/logging/build_logging.go +++ b/daemon/logging/build_logging.go @@ -9,8 +9,8 @@ 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/logger" "github.com/zrepl/zrepl/tlsconf" ) @@ -157,13 +157,7 @@ func getLoggerImpl(ctx context.Context, subsys Subsystem, panicIfEnded bool) log l = l.WithField(SubsysField, subsys) - if nI := ctx.Value(contextKeyTraceNode); nI != nil { - n := nI.(*traceNode) - _, spanStack := currentTaskNameAndSpanStack(n) - l = l.WithField(SpanField, spanStack) - } else { - l = l.WithField(SpanField, "NOSPAN") - } + l = l.WithField(SpanField, trace.GetSpanStackOrDefault(ctx, "NOSPAN")) fields := make(logger.Fields) iterInjectedFields(ctx, func(field string, value interface{}) { diff --git a/daemon/logging/logging_activity_trace_debug.go b/daemon/logging/logging_activity_trace_debug.go deleted file mode 100644 index ed67c23..0000000 --- a/daemon/logging/logging_activity_trace_debug.go +++ /dev/null @@ -1,15 +0,0 @@ -package logging - -import ( - "fmt" - "os" -) - -const chrometraceEnableDebug = false - -func chrometraceDebug(format string, args ...interface{}) { - if !chrometraceEnableDebug { - return - } - fmt.Fprintf(os.Stderr, format+"\n", args...) -} diff --git a/daemon/logging/logging_context.go b/daemon/logging/logging_context.go index da37373..3ba6dee 100644 --- a/daemon/logging/logging_context.go +++ b/daemon/logging/logging_context.go @@ -5,13 +5,11 @@ import "context" type contextKey int const ( - contextKeyTraceNode contextKey = 1 + iota - contextKeyLoggers + contextKeyLoggers contextKey = 1 + iota contextKeyInjectedField ) var contextKeys = []contextKey{ - contextKeyTraceNode, contextKeyLoggers, contextKeyInjectedField, } diff --git a/daemon/logging/logging_span.go b/daemon/logging/trace/trace.go similarity index 52% rename from daemon/logging/logging_span.go rename to daemon/logging/trace/trace.go index 4774761..6db9281 100644 --- a/daemon/logging/logging_span.go +++ b/daemon/logging/trace/trace.go @@ -1,20 +1,103 @@ -package logging +// 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" - "encoding/base64" "fmt" - "math/rand" - "os" - "runtime" "strings" - "sync" "sync/atomic" "time" "github.com/prometheus/client_golang/prometheus" - - "github.com/zrepl/zrepl/util/envconst" ) var metrics struct { @@ -25,13 +108,13 @@ var metrics struct { func init() { metrics.activeTasks = prometheus.NewGauge(prometheus.GaugeOpts{ Namespace: "zrepl", - Subsystem: "logging", + Subsystem: "trace", Name: "active_tasks", Help: "number of active (tracing-level) tasks in the daemon", }) metrics.uniqueConcurrentTaskNameBitvecLength = prometheus.NewGaugeVec(prometheus.GaugeOpts{ Namespace: "zrepl", - Subsystem: "logging", + 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"}) @@ -42,75 +125,27 @@ func RegisterMetrics(r prometheus.Registerer) { r.MustRegister(metrics.uniqueConcurrentTaskNameBitvecLength) } -type DoneFunc func() - -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 "" -} - -// 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 func()) { - 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 -} - var taskNamer = newUniqueTaskNamer(metrics.uniqueConcurrentTaskNameBitvecLength) +type traceNode struct { + id string + annotation string + parentTask *traceNode + activeChildTasks int32 // only for task nodes, insignificant for span nodes + parentSpan *traceNode + hasActiveChildSpan int32 + + startedAt time.Time + endedAt time.Time + ended int32 +} + +// 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() + // Start a new root task or create a child task of an existing task. // // This is required when starting a new goroutine and @@ -137,7 +172,7 @@ func WithTask(ctx context.Context, taskName string) (context.Context, DoneFunc) taskName, taskNameDone := taskNamer.UniqueConcurrentTaskName(taskName) this := &traceNode{ - id: newTraceNodeId(), + id: genID(), annotation: taskName, parentTask: parentTask, activeChildTasks: 0, @@ -185,9 +220,8 @@ func WithTask(ctx context.Context, taskName string) (context.Context, DoneFunc) return ctx, endTaskFunc } -// ctx must have an active task (see WithTask) -// -// spans must nest (stack-like), otherwise, this function panics +// 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) @@ -203,7 +237,7 @@ func WithSpan(ctx context.Context, annotation string) (context.Context, DoneFunc } this := &traceNode{ - id: newTraceNodeId(), + id: genID(), annotation: annotation, parentTask: parentTask, parentSpan: parentSpan, @@ -233,19 +267,6 @@ func WithSpan(ctx context.Context, annotation string) (context.Context, DoneFunc return ctx, endTaskFunc } -type traceNode struct { - id string - annotation string - parentTask *traceNode - activeChildTasks int32 // only for task nodes, insignificant for span nodes - parentSpan *traceNode - hasActiveChildSpan int32 - - startedAt time.Time - endedAt time.Time - ended int32 -} - func currentTaskNameAndSpanStack(this *traceNode) (taskName string, spanIdStack string) { task := this.parentTask @@ -279,38 +300,12 @@ func currentTaskNameAndSpanStack(this *traceNode) (taskName string, spanIdStack return task.annotation, spanIdStack } -var traceNodeIdPRNG = rand.New(rand.NewSource(1)) - -func init() { - traceNodeIdPRNG.Seed(time.Now().UnixNano()) - traceNodeIdPRNG.Seed(int64(os.Getpid())) -} - -var traceNodeIdBytes = envconst.Int("ZREPL_LOGGING_TRACE_ID_BYTES", 3) - -func init() { - if traceNodeIdBytes < 1 { - panic("trace node id byte length must be at least 1") +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 } } - -func newTraceNodeId() string { - var out strings.Builder - enc := base64.NewEncoder(base64.RawStdEncoding, &out) - buf := make([]byte, traceNodeIdBytes) - for i := 0; i < len(buf); { - n, err := traceNodeIdPRNG.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() -} diff --git a/daemon/logging/logging_activity_trace.go b/daemon/logging/trace/trace_chrometrace.go similarity index 95% rename from daemon/logging/logging_activity_trace.go rename to daemon/logging/trace/trace_chrometrace.go index 708ab15..72c655f 100644 --- a/daemon/logging/logging_activity_trace.go +++ b/daemon/logging/trace/trace_chrometrace.go @@ -1,4 +1,4 @@ -package logging +package trace // The functions in this file are concerned with the generation // of trace files based on the information from WithTask and WithSpan. @@ -134,7 +134,7 @@ func init() { chrometraceConsumers.write = make(chan []byte) go func() { kickConsumer := func(c chrometraceConsumerRegistration, err error) { - chrometraceDebug("chrometrace kicking consumer %#v after error %v", c, err) + debug("chrometrace kicking consumer %#v after error %v", c, err) select { case c.errored <- err: default: @@ -145,7 +145,7 @@ func init() { for { select { case reg := <-chrometraceConsumers.register: - chrometraceDebug("registered chrometrace consumer %#v", reg) + debug("registered chrometrace consumer %#v", reg) chrometraceConsumers.consumers[reg] = true n, err := reg.w.Write([]byte("[\n")) if err != nil { @@ -156,12 +156,12 @@ func init() { // successfully registered case buf := <-chrometraceConsumers.write: - chrometraceDebug("chrometrace write request: %s", string(buf)) + 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) - chrometraceDebug("chrometrace wrote n=%v bytes to consumer %#v", n, c) + debug("chrometrace wrote n=%v bytes to consumer %#v", n, c) if err != nil { kickConsumer(c, err) } diff --git a/daemon/logging/trace/trace_context.go b/daemon/logging/trace/trace_context.go new file mode 100644 index 0000000..d04c308 --- /dev/null +++ b/daemon/logging/trace/trace_context.go @@ -0,0 +1,27 @@ +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 +} diff --git a/daemon/logging/trace/trace_convenience.go b/daemon/logging/trace/trace_convenience.go new file mode 100644 index 0000000..f14ce53 --- /dev/null +++ b/daemon/logging/trace/trace_convenience.go @@ -0,0 +1,74 @@ +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 "" +} diff --git a/daemon/logging/logging_span_test.go b/daemon/logging/trace/trace_convenience_test.go similarity index 74% rename from daemon/logging/logging_span_test.go rename to daemon/logging/trace/trace_convenience_test.go index e96e256..3f27bfd 100644 --- a/daemon/logging/logging_span_test.go +++ b/daemon/logging/trace/trace_convenience_test.go @@ -1,4 +1,4 @@ -package logging +package trace import ( "testing" @@ -12,5 +12,5 @@ func TestGetCallerOrPanic(t *testing.T) { } ret := withStackFromCtxMock() // zrepl prefix is stripped - assert.Equal(t, "daemon/logging.TestGetCallerOrPanic", ret) + assert.Equal(t, "daemon/logging/trace.TestGetCallerOrPanic", ret) } diff --git a/daemon/logging/trace/trace_debug.go b/daemon/logging/trace/trace_debug.go new file mode 100644 index 0000000..773e3e9 --- /dev/null +++ b/daemon/logging/trace/trace_debug.go @@ -0,0 +1,15 @@ +package trace + +import ( + "fmt" + "os" +) + +const debugEnabled = false + +func debug(format string, args ...interface{}) { + if !debugEnabled { + return + } + fmt.Fprintf(os.Stderr, format+"\n", args...) +} diff --git a/daemon/logging/trace/trace_genID.go b/daemon/logging/trace/trace_genID.go new file mode 100644 index 0000000..0fcd043 --- /dev/null +++ b/daemon/logging/trace/trace_genID.go @@ -0,0 +1,47 @@ +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() +} diff --git a/daemon/logging/logging_span_unique_concurrent_task_namer.go b/daemon/logging/trace/trace_unique_concurrent_task_namer.go similarity index 98% rename from daemon/logging/logging_span_unique_concurrent_task_namer.go rename to daemon/logging/trace/trace_unique_concurrent_task_namer.go index 8cfd734..630cdcc 100644 --- a/daemon/logging/logging_span_unique_concurrent_task_namer.go +++ b/daemon/logging/trace/trace_unique_concurrent_task_namer.go @@ -1,4 +1,4 @@ -package logging +package trace import ( "fmt" diff --git a/daemon/logging/logging_span_unique_concurrent_task_namer_test.go b/daemon/logging/trace/trace_unique_concurrent_task_namer_test.go similarity index 98% rename from daemon/logging/logging_span_unique_concurrent_task_namer_test.go rename to daemon/logging/trace/trace_unique_concurrent_task_namer_test.go index c5ffbf6..7b56a39 100644 --- a/daemon/logging/logging_span_unique_concurrent_task_namer_test.go +++ b/daemon/logging/trace/trace_unique_concurrent_task_namer_test.go @@ -1,4 +1,4 @@ -package logging +package trace import ( "fmt" diff --git a/daemon/pprof.go b/daemon/pprof.go index 0b80605..208bb09 100644 --- a/daemon/pprof.go +++ b/daemon/pprof.go @@ -8,10 +8,10 @@ import ( "net" "net/http/pprof" + "github.com/zrepl/zrepl/daemon/logging/trace" "golang.org/x/net/websocket" "github.com/zrepl/zrepl/daemon/job" - "github.com/zrepl/zrepl/daemon/logging" ) type pprofServer struct { @@ -67,7 +67,7 @@ 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(logging.ChrometraceClientWebsocketHandler)) + mux.Handle("/debug/zrepl/activity-trace", websocket.Handler(trace.ChrometraceClientWebsocketHandler)) go func() { err := http.Serve(s.listener, mux) if ctx.Err() != nil { diff --git a/daemon/snapper/snapper.go b/daemon/snapper/snapper.go index 233da85..35433d5 100644 --- a/daemon/snapper/snapper.go +++ b/daemon/snapper/snapper.go @@ -8,6 +8,7 @@ import ( "time" "github.com/pkg/errors" + "github.com/zrepl/zrepl/daemon/logging/trace" "github.com/zrepl/zrepl/config" "github.com/zrepl/zrepl/daemon/filters" @@ -133,7 +134,7 @@ func PeriodicFromConfig(g *config.Global, fsf *filters.DatasetMapFilter, in *con } func (s *Snapper) Run(ctx context.Context, snapshotsTaken chan<- struct{}) { - defer logging.WithSpanFromStackUpdateCtx(&ctx)() + defer trace.WithSpanFromStackUpdateCtx(&ctx)() getLogger(ctx).Debug("start") defer getLogger(ctx).Debug("stop") diff --git a/endpoint/endpoint.go b/endpoint/endpoint.go index 0528f33..98f0575 100644 --- a/endpoint/endpoint.go +++ b/endpoint/endpoint.go @@ -10,8 +10,8 @@ import ( "sync" "github.com/pkg/errors" + "github.com/zrepl/zrepl/daemon/logging/trace" - "github.com/zrepl/zrepl/daemon/logging" "github.com/zrepl/zrepl/replication/logic/pdu" "github.com/zrepl/zrepl/util/chainedio" "github.com/zrepl/zrepl/util/chainlock" @@ -74,7 +74,7 @@ func (s *Sender) filterCheckFS(fs string) (*zfs.DatasetPath, error) { } func (s *Sender) ListFilesystems(ctx context.Context, r *pdu.ListFilesystemReq) (*pdu.ListFilesystemRes, error) { - defer logging.WithSpanFromStackUpdateCtx(&ctx)() + defer trace.WithSpanFromStackUpdateCtx(&ctx)() fss, err := zfs.ZFSListMapping(ctx, s.FSFilter) if err != nil { @@ -98,7 +98,7 @@ func (s *Sender) ListFilesystems(ctx context.Context, r *pdu.ListFilesystemReq) } func (s *Sender) ListFilesystemVersions(ctx context.Context, r *pdu.ListFilesystemVersionsReq) (*pdu.ListFilesystemVersionsRes, error) { - defer logging.WithSpanFromStackUpdateCtx(&ctx)() + defer trace.WithSpanFromStackUpdateCtx(&ctx)() lp, err := s.filterCheckFS(r.GetFilesystem()) if err != nil { @@ -118,7 +118,7 @@ func (s *Sender) ListFilesystemVersions(ctx context.Context, r *pdu.ListFilesyst } func (p *Sender) HintMostRecentCommonAncestor(ctx context.Context, r *pdu.HintMostRecentCommonAncestorReq) (*pdu.HintMostRecentCommonAncestorRes, error) { - defer logging.WithSpanFromStackUpdateCtx(&ctx)() + defer trace.WithSpanFromStackUpdateCtx(&ctx)() fsp, err := p.filterCheckFS(r.GetFilesystem()) if err != nil { @@ -251,7 +251,7 @@ func sendArgsFromPDUAndValidateExistsAndGetVersion(ctx context.Context, fs strin } func (s *Sender) Send(ctx context.Context, r *pdu.SendReq) (*pdu.SendRes, io.ReadCloser, error) { - defer logging.WithSpanFromStackUpdateCtx(&ctx)() + defer trace.WithSpanFromStackUpdateCtx(&ctx)() _, err := s.filterCheckFS(r.Filesystem) if err != nil { @@ -357,7 +357,7 @@ func (s *Sender) Send(ctx context.Context, r *pdu.SendReq) (*pdu.SendRes, io.Rea } func (p *Sender) SendCompleted(ctx context.Context, r *pdu.SendCompletedReq) (*pdu.SendCompletedRes, error) { - defer logging.WithSpanFromStackUpdateCtx(&ctx)() + defer trace.WithSpanFromStackUpdateCtx(&ctx)() orig := r.GetOriginalReq() // may be nil, always use proto getters fsp, err := p.filterCheckFS(orig.GetFilesystem()) @@ -412,7 +412,7 @@ func (p *Sender) SendCompleted(ctx context.Context, r *pdu.SendCompletedReq) (*p wg.Add(2) go func() { defer wg.Done() - ctx, endTask := logging.WithTask(ctx, "release-step-hold-to") + ctx, endTask := trace.WithTask(ctx, "release-step-hold-to") defer endTask() log(ctx).Debug("release step-hold of or step-bookmark on `to`") @@ -426,7 +426,7 @@ func (p *Sender) SendCompleted(ctx context.Context, r *pdu.SendCompletedReq) (*p }() go func() { defer wg.Done() - ctx, endTask := logging.WithTask(ctx, "release-step-hold-from") + ctx, endTask := trace.WithTask(ctx, "release-step-hold-from") defer endTask() if from == nil { @@ -458,7 +458,7 @@ 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 logging.WithSpanFromStackUpdateCtx(&ctx)() + defer trace.WithSpanFromStackUpdateCtx(&ctx)() dp, err := p.filterCheckFS(req.Filesystem) if err != nil { @@ -468,7 +468,7 @@ func (p *Sender) DestroySnapshots(ctx context.Context, req *pdu.DestroySnapshots } func (p *Sender) Ping(ctx context.Context, req *pdu.PingReq) (*pdu.PingRes, error) { - defer logging.WithSpanFromStackUpdateCtx(&ctx)() + defer trace.WithSpanFromStackUpdateCtx(&ctx)() res := pdu.PingRes{ Echo: req.GetMessage(), @@ -477,19 +477,19 @@ 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 logging.WithSpanFromStackUpdateCtx(&ctx)() + defer trace.WithSpanFromStackUpdateCtx(&ctx)() return p.Ping(ctx, req) } func (p *Sender) WaitForConnectivity(ctx context.Context) error { - defer logging.WithSpanFromStackUpdateCtx(&ctx)() + defer trace.WithSpanFromStackUpdateCtx(&ctx)() return nil } func (p *Sender) ReplicationCursor(ctx context.Context, req *pdu.ReplicationCursorReq) (*pdu.ReplicationCursorRes, error) { - defer logging.WithSpanFromStackUpdateCtx(&ctx)() + defer trace.WithSpanFromStackUpdateCtx(&ctx)() dp, err := p.filterCheckFS(req.Filesystem) if err != nil { @@ -621,7 +621,7 @@ func (f subroot) MapToLocal(fs string) (*zfs.DatasetPath, error) { } func (s *Receiver) ListFilesystems(ctx context.Context, req *pdu.ListFilesystemReq) (*pdu.ListFilesystemRes, error) { - defer logging.WithSpanFromStackUpdateCtx(&ctx)() + defer trace.WithSpanFromStackUpdateCtx(&ctx)() root := s.clientRootFromCtx(ctx) filtered, err := zfs.ZFSListMapping(ctx, subroot{root}) @@ -673,7 +673,7 @@ func (s *Receiver) ListFilesystems(ctx context.Context, req *pdu.ListFilesystemR } func (s *Receiver) ListFilesystemVersions(ctx context.Context, req *pdu.ListFilesystemVersionsReq) (*pdu.ListFilesystemVersionsRes, error) { - defer logging.WithSpanFromStackUpdateCtx(&ctx)() + defer trace.WithSpanFromStackUpdateCtx(&ctx)() root := s.clientRootFromCtx(ctx) lp, err := subroot{root}.MapToLocal(req.GetFilesystem()) @@ -696,7 +696,7 @@ func (s *Receiver) ListFilesystemVersions(ctx context.Context, req *pdu.ListFile } func (s *Receiver) Ping(ctx context.Context, req *pdu.PingReq) (*pdu.PingRes, error) { - defer logging.WithSpanFromStackUpdateCtx(&ctx)() + defer trace.WithSpanFromStackUpdateCtx(&ctx)() res := pdu.PingRes{ Echo: req.GetMessage(), @@ -705,29 +705,29 @@ 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 logging.WithSpanFromStackUpdateCtx(&ctx)() + defer trace.WithSpanFromStackUpdateCtx(&ctx)() return s.Ping(ctx, req) } func (s *Receiver) WaitForConnectivity(ctx context.Context) error { - defer logging.WithSpanFromStackUpdateCtx(&ctx)() + defer trace.WithSpanFromStackUpdateCtx(&ctx)() return nil } func (s *Receiver) ReplicationCursor(ctx context.Context, _ *pdu.ReplicationCursorReq) (*pdu.ReplicationCursorRes, error) { - defer logging.WithSpanFromStackUpdateCtx(&ctx)() + defer trace.WithSpanFromStackUpdateCtx(&ctx)() 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 logging.WithSpanFromStackUpdateCtx(&ctx)() + defer trace.WithSpanFromStackUpdateCtx(&ctx)() return nil, nil, fmt.Errorf("receiver does not implement Send()") } var maxConcurrentZFSRecvSemaphore = semaphore.New(envconst.Int64("ZREPL_ENDPOINT_MAX_CONCURRENT_RECV", 10)) func (s *Receiver) Receive(ctx context.Context, req *pdu.ReceiveReq, receive io.ReadCloser) (*pdu.ReceiveRes, error) { - defer logging.WithSpanFromStackUpdateCtx(&ctx)() + defer trace.WithSpanFromStackUpdateCtx(&ctx)() getLogger(ctx).Debug("incoming Receive") defer receive.Close() @@ -957,7 +957,7 @@ 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 logging.WithSpanFromStackUpdateCtx(&ctx)() + defer trace.WithSpanFromStackUpdateCtx(&ctx)() root := s.clientRootFromCtx(ctx) lp, err := subroot{root}.MapToLocal(req.Filesystem) @@ -968,7 +968,7 @@ func (s *Receiver) DestroySnapshots(ctx context.Context, req *pdu.DestroySnapsho } func (p *Receiver) HintMostRecentCommonAncestor(ctx context.Context, r *pdu.HintMostRecentCommonAncestorReq) (*pdu.HintMostRecentCommonAncestorRes, error) { - defer logging.WithSpanFromStackUpdateCtx(&ctx)() + 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. // @@ -978,7 +978,7 @@ func (p *Receiver) HintMostRecentCommonAncestor(ctx context.Context, r *pdu.Hint } func (p *Receiver) SendCompleted(ctx context.Context, _ *pdu.SendCompletedReq) (*pdu.SendCompletedRes, error) { - defer logging.WithSpanFromStackUpdateCtx(&ctx)() + defer trace.WithSpanFromStackUpdateCtx(&ctx)() return &pdu.SendCompletedRes{}, nil } diff --git a/platformtest/harness/harness.go b/platformtest/harness/harness.go index e927969..2452fa0 100644 --- a/platformtest/harness/harness.go +++ b/platformtest/harness/harness.go @@ -11,6 +11,7 @@ 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,7 +70,7 @@ func doMain() error { panic(err) } ctx := context.Background() - defer logging.WithTaskFromStackUpdateCtx(&ctx)() + defer trace.WithTaskFromStackUpdateCtx(&ctx)() ctx = platformtest.WithLogger(ctx, logger) ex := platformtest.NewEx(logger) diff --git a/replication/driver/replication_driver.go b/replication/driver/replication_driver.go index cf4f3f2..96654c7 100644 --- a/replication/driver/replication_driver.go +++ b/replication/driver/replication_driver.go @@ -10,10 +10,10 @@ import ( "sync" "time" + "github.com/zrepl/zrepl/daemon/logging/trace" "google.golang.org/grpc/codes" "google.golang.org/grpc/status" - "github.com/zrepl/zrepl/daemon/logging" "github.com/zrepl/zrepl/replication/report" "github.com/zrepl/zrepl/util/chainlock" "github.com/zrepl/zrepl/util/envconst" @@ -297,7 +297,7 @@ func (a *attempt) do(ctx context.Context, prev *attempt) { // 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 := logging.WithSpan(ctx, "plan") + ctx, endSpan := trace.WithSpan(ctx, "plan") defer endSpan() pfss, err := a.planner.Plan(ctx) errTime := time.Now() @@ -389,7 +389,7 @@ func (a *attempt) doGlobalPlanning(ctx context.Context, prev *attempt) map[*fs]* } func (a *attempt) doFilesystems(ctx context.Context, prevs map[*fs]*fs) { - ctx, endSpan := logging.WithSpan(ctx, "do-repl") + ctx, endSpan := trace.WithSpan(ctx, "do-repl") defer endSpan() defer a.l.Lock().Unlock() @@ -402,7 +402,7 @@ func (a *attempt) doFilesystems(ctx context.Context, prevs map[*fs]*fs) { go func(f *fs) { defer fssesDone.Done() // avoid explosion of tasks with name f.report().Info.Name - ctx, endTask := logging.WithTaskAndSpan(ctx, "repl-fs", f.report().Info.Name) + ctx, endTask := trace.WithTaskAndSpan(ctx, "repl-fs", f.report().Info.Name) defer endTask() f.do(ctx, stepQueue, prevs[f]) }(f) @@ -610,7 +610,7 @@ func (f *fs) do(ctx context.Context, pq *stepQueue, prev *fs) { targetDate := s.step.TargetDate() defer pq.WaitReady(ctx, f, targetDate)() // do the step - ctx, endSpan := logging.WithSpan(ctx, fmt.Sprintf("%#v", s.step.ReportInfo())) + ctx, endSpan := trace.WithSpan(ctx, fmt.Sprintf("%#v", s.step.ReportInfo())) defer endSpan() err, errTime = s.step.Step(ctx), time.Now() // no shadow }) diff --git a/replication/driver/replication_driver_test.go b/replication/driver/replication_driver_test.go index deecc1e..257d93b 100644 --- a/replication/driver/replication_driver_test.go +++ b/replication/driver/replication_driver_test.go @@ -10,8 +10,8 @@ import ( "time" "github.com/stretchr/testify/require" + "github.com/zrepl/zrepl/daemon/logging/trace" - "github.com/zrepl/zrepl/daemon/logging" "github.com/zrepl/zrepl/replication/report" "github.com/stretchr/testify/assert" @@ -150,7 +150,7 @@ func (f *mockStep) ReportInfo() *report.StepInfo { func TestReplication(t *testing.T) { ctx := context.Background() - defer logging.WithTaskFromStackUpdateCtx(&ctx)() + defer trace.WithTaskFromStackUpdateCtx(&ctx)() mp := &mockPlanner{} getReport, wait := Do(ctx, mp) diff --git a/replication/driver/replication_stepqueue.go b/replication/driver/replication_stepqueue.go index 30a89c3..bc8b066 100644 --- a/replication/driver/replication_stepqueue.go +++ b/replication/driver/replication_stepqueue.go @@ -5,7 +5,7 @@ import ( "context" "time" - "github.com/zrepl/zrepl/daemon/logging" + "github.com/zrepl/zrepl/daemon/logging/trace" "github.com/zrepl/zrepl/util/chainlock" ) @@ -158,7 +158,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 logging.WithSpanFromStackUpdateCtx(&ctx)() + defer trace.WithSpanFromStackUpdateCtx(&ctx)() if targetDate.IsZero() { panic("targetDate of zero is reserved for marking Done") } diff --git a/replication/driver/replication_stepqueue_test.go b/replication/driver/replication_stepqueue_test.go index b587652..f79c163 100644 --- a/replication/driver/replication_stepqueue_test.go +++ b/replication/driver/replication_stepqueue_test.go @@ -12,21 +12,20 @@ import ( "github.com/montanaflynn/stats" "github.com/stretchr/testify/assert" - - "github.com/zrepl/zrepl/daemon/logging" + "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 := logging.WithTaskFromStack(context.Background()) + ctx, end := trace.WithTaskFromStack(context.Background()) defer end() var ctr uint32 q := newStepQueue() var wg sync.WaitGroup wg.Add(4) go func() { - ctx, end := logging.WithTaskFromStack(ctx) + ctx, end := trace.WithTaskFromStack(ctx) defer end() defer wg.Done() defer q.WaitReady(ctx, "1", time.Unix(9999, 0))() @@ -41,7 +40,7 @@ func TestPqNotconcurrent(t *testing.T) { // while "1" is still running, queue in "2", "3" and "4" go func() { - ctx, end := logging.WithTaskFromStack(ctx) + ctx, end := trace.WithTaskFromStack(ctx) defer end() defer wg.Done() defer q.WaitReady(ctx, "2", time.Unix(2, 0))() @@ -49,7 +48,7 @@ func TestPqNotconcurrent(t *testing.T) { assert.Equal(t, uint32(2), ret) }() go func() { - ctx, end := logging.WithTaskFromStack(ctx) + ctx, end := trace.WithTaskFromStack(ctx) defer end() defer wg.Done() defer q.WaitReady(ctx, "3", time.Unix(3, 0))() @@ -57,7 +56,7 @@ func TestPqNotconcurrent(t *testing.T) { assert.Equal(t, uint32(3), ret) }() go func() { - ctx, end := logging.WithTaskFromStack(ctx) + ctx, end := trace.WithTaskFromStack(ctx) defer end() defer wg.Done() defer q.WaitReady(ctx, "4", time.Unix(4, 0))() @@ -90,7 +89,7 @@ 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 := logging.WithTaskFromStack(context.Background()) + ctx, end := trace.WithTaskFromStack(context.Background()) defer end() q := newStepQueue() @@ -105,7 +104,7 @@ func TestPqConcurrent(t *testing.T) { records := make(chan []record, filesystems) for fs := 0; fs < filesystems; fs++ { go func(fs int) { - ctx, end := logging.WithTaskFromStack(ctx) + ctx, end := trace.WithTaskFromStack(ctx) defer end() defer wg.Done() recs := make([]record, 0) diff --git a/replication/logic/replication_logic.go b/replication/logic/replication_logic.go index 6c4151c..3b2548a 100644 --- a/replication/logic/replication_logic.go +++ b/replication/logic/replication_logic.go @@ -9,8 +9,8 @@ import ( "time" "github.com/prometheus/client_golang/prometheus" + "github.com/zrepl/zrepl/daemon/logging/trace" - "github.com/zrepl/zrepl/daemon/logging" "github.com/zrepl/zrepl/logger" "github.com/zrepl/zrepl/replication/driver" . "github.com/zrepl/zrepl/replication/logic/diff" @@ -82,7 +82,7 @@ func (p *Planner) WaitForConnectivity(ctx context.Context) error { var wg sync.WaitGroup doPing := func(endpoint Endpoint, errOut *error) { defer wg.Done() - ctx, endTask := logging.WithTaskFromStack(ctx) + ctx, endTask := trace.WithTaskFromStack(ctx) defer endTask() err := endpoint.WaitForConnectivity(ctx) if err != nil { @@ -375,7 +375,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 := logging.WithTask(ctx, "hint-mrca-"+name) + ctx, endTask := trace.WithTask(ctx, "hint-mrca-"+name) defer endTask() log := log(ctx).WithField("to_side", name). @@ -537,7 +537,7 @@ func (fs *Filesystem) doPlanning(ctx context.Context) ([]*Step, error) { log(ctx).Debug("compute send size estimate") errs := make(chan error, len(steps)) fanOutCtx, fanOutCancel := context.WithCancel(ctx) - _, fanOutAdd, fanOutWait := logging.WithTaskGroup(fanOutCtx, "compute-size-estimate") + _, fanOutAdd, fanOutWait := trace.WithTaskGroup(fanOutCtx, "compute-size-estimate") defer fanOutCancel() for _, step := range steps { step := step // local copy that is moved into the closure diff --git a/rpc/rpc_client.go b/rpc/rpc_client.go index 6f9df73..e3b76d4 100644 --- a/rpc/rpc_client.go +++ b/rpc/rpc_client.go @@ -10,11 +10,11 @@ import ( "sync/atomic" "time" + "github.com/zrepl/zrepl/daemon/logging/trace" "google.golang.org/grpc" "github.com/google/uuid" - "github.com/zrepl/zrepl/daemon/logging" "github.com/zrepl/zrepl/replication/logic" "github.com/zrepl/zrepl/replication/logic/pdu" "github.com/zrepl/zrepl/rpc/dataconn" @@ -84,7 +84,7 @@ 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 := logging.WithSpan(ctx, "rpc.client.Send") + ctx, endSpan := trace.WithSpan(ctx, "rpc.client.Send") defer endSpan() // TODO the returned sendStream may return a read error created by the remote side @@ -101,56 +101,56 @@ func (c *Client) Send(ctx context.Context, r *pdu.SendReq) (*pdu.SendRes, io.Rea } func (c *Client) Receive(ctx context.Context, req *pdu.ReceiveReq, stream io.ReadCloser) (*pdu.ReceiveRes, error) { - ctx, endSpan := logging.WithSpan(ctx, "rpc.client.Receive") + ctx, endSpan := trace.WithSpan(ctx, "rpc.client.Receive") defer endSpan() return c.dataClient.ReqRecv(ctx, req, stream) } func (c *Client) ListFilesystems(ctx context.Context, in *pdu.ListFilesystemReq) (*pdu.ListFilesystemRes, error) { - ctx, endSpan := logging.WithSpan(ctx, "rpc.client.ListFilesystems") + 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 := logging.WithSpan(ctx, "rpc.client.ListFilesystemVersions") + 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 := logging.WithSpan(ctx, "rpc.client.DestroySnapshots") + 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 := logging.WithSpan(ctx, "rpc.client.ReplicationCursor") + 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 := logging.WithSpan(ctx, "rpc.client.SendCompleted") + 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 := logging.WithSpan(ctx, "rpc.client.HintMostRecentCommonAncestor") + 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 := logging.WithSpan(ctx, "rpc.client.WaitForConnectivity") + ctx, endSpan := trace.WithSpan(ctx, "rpc.client.WaitForConnectivity") defer endSpan() ctx, cancel := context.WithCancel(ctx) diff --git a/util/semaphore/semaphore.go b/util/semaphore/semaphore.go index e754386..8fefaf7 100644 --- a/util/semaphore/semaphore.go +++ b/util/semaphore/semaphore.go @@ -3,9 +3,8 @@ package semaphore import ( "context" + "github.com/zrepl/zrepl/daemon/logging/trace" wsemaphore "golang.org/x/sync/semaphore" - - "github.com/zrepl/zrepl/daemon/logging" ) type S struct { @@ -23,7 +22,7 @@ type AcquireGuard struct { // The returned AcquireGuard is not goroutine-safe. func (s *S) Acquire(ctx context.Context) (*AcquireGuard, error) { - defer logging.WithSpanFromStackUpdateCtx(&ctx)() + defer trace.WithSpanFromStackUpdateCtx(&ctx)() if err := s.ws.Acquire(ctx, 1); err != nil { return nil, err } else if err := ctx.Err(); err != nil { diff --git a/util/semaphore/semaphore_test.go b/util/semaphore/semaphore_test.go index ede637f..017a86c 100644 --- a/util/semaphore/semaphore_test.go +++ b/util/semaphore/semaphore_test.go @@ -9,8 +9,7 @@ import ( "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" - - "github.com/zrepl/zrepl/daemon/logging" + "github.com/zrepl/zrepl/daemon/logging/trace" ) func TestSemaphore(t *testing.T) { @@ -27,13 +26,13 @@ func TestSemaphore(t *testing.T) { } ctx := context.Background() - defer logging.WithTaskFromStackUpdateCtx(&ctx)() + defer trace.WithTaskFromStackUpdateCtx(&ctx)() var wg sync.WaitGroup wg.Add(numGoroutines) for i := 0; i < numGoroutines; i++ { go func() { - ctx, end := logging.WithTaskFromStack(ctx) + ctx, end := trace.WithTaskFromStack(ctx) defer end() defer wg.Done() res, err := sem.Acquire(ctx) diff --git a/zfs/zfscmd/zfscmd.go b/zfs/zfscmd/zfscmd.go index 11de3ce..9834fa2 100644 --- a/zfs/zfscmd/zfscmd.go +++ b/zfs/zfscmd/zfscmd.go @@ -14,7 +14,7 @@ import ( "sync" "time" - "github.com/zrepl/zrepl/daemon/logging" + "github.com/zrepl/zrepl/daemon/logging/trace" "github.com/zrepl/zrepl/util/circlog" ) @@ -23,7 +23,7 @@ type Cmd struct { ctx context.Context mtx sync.RWMutex startedAt, waitStartedAt, waitReturnedAt time.Time - waitReturnEndSpanCb logging.DoneFunc + waitReturnEndSpanCb trace.DoneFunc } func CommandContext(ctx context.Context, name string, arg ...string) *Cmd { @@ -104,9 +104,9 @@ func (c *Cmd) Wait() (err error) { func (c *Cmd) startPre(newTask bool) { if newTask { // avoid explosion of tasks with name c.String() - c.ctx, c.waitReturnEndSpanCb = logging.WithTaskAndSpan(c.ctx, "zfscmd", c.String()) + c.ctx, c.waitReturnEndSpanCb = trace.WithTaskAndSpan(c.ctx, "zfscmd", c.String()) } else { - c.ctx, c.waitReturnEndSpanCb = logging.WithSpan(c.ctx, c.String()) + c.ctx, c.waitReturnEndSpanCb = trace.WithSpan(c.ctx, c.String()) } startPreLogging(c, time.Now()) }