mirror of
https://github.com/zrepl/zrepl.git
synced 2024-11-25 09:54:47 +01:00
10a14a8c50
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
92 lines
1.8 KiB
Go
92 lines
1.8 KiB
Go
package hooks
|
|
|
|
import (
|
|
"bufio"
|
|
"bytes"
|
|
"context"
|
|
"sync"
|
|
|
|
"github.com/zrepl/zrepl/daemon/logging"
|
|
"github.com/zrepl/zrepl/logger"
|
|
"github.com/zrepl/zrepl/util/envconst"
|
|
)
|
|
|
|
type Logger = logger.Logger
|
|
|
|
func GetLogger(ctx context.Context) Logger { return getLogger(ctx) }
|
|
|
|
func getLogger(ctx context.Context) Logger {
|
|
return logging.GetLogger(ctx, logging.SubsysHooks)
|
|
}
|
|
|
|
const MAX_HOOK_LOG_SIZE_DEFAULT int = 1 << 20
|
|
|
|
type logWriter struct {
|
|
/*
|
|
Mutex prevents:
|
|
concurrent writes to buf, scanner in Write([]byte)
|
|
data race on scanner vs Write([]byte)
|
|
and concurrent write to buf (call to buf.Reset())
|
|
in Close()
|
|
|
|
(Also, Close() should generally block until any Write() call completes.)
|
|
*/
|
|
mtx *sync.Mutex
|
|
buf bytes.Buffer
|
|
scanner *bufio.Scanner
|
|
logger Logger
|
|
level logger.Level
|
|
field string
|
|
}
|
|
|
|
func NewLogWriter(mtx *sync.Mutex, logger Logger, level logger.Level, field string) *logWriter {
|
|
w := new(logWriter)
|
|
w.mtx = mtx
|
|
w.scanner = bufio.NewScanner(&w.buf)
|
|
w.logger = logger
|
|
w.level = level
|
|
w.field = field
|
|
return w
|
|
}
|
|
|
|
func (w *logWriter) log(line string) {
|
|
w.logger.WithField(w.field, line).Log(w.level, "hook output")
|
|
}
|
|
|
|
func (w *logWriter) logUnreadBytes() error {
|
|
for w.scanner.Scan() {
|
|
w.log(w.scanner.Text())
|
|
}
|
|
if w.buf.Cap() > envconst.Int("ZREPL_MAX_HOOK_LOG_SIZE", MAX_HOOK_LOG_SIZE_DEFAULT) {
|
|
w.buf.Reset()
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
func (w *logWriter) Write(in []byte) (int, error) {
|
|
w.mtx.Lock()
|
|
defer w.mtx.Unlock()
|
|
|
|
n, bufErr := w.buf.Write(in)
|
|
if bufErr != nil {
|
|
return n, bufErr
|
|
}
|
|
|
|
err := w.logUnreadBytes()
|
|
if err != nil {
|
|
return n, err
|
|
}
|
|
// Always reset the scanner for the next Write
|
|
w.scanner = bufio.NewScanner(&w.buf)
|
|
|
|
return n, nil
|
|
}
|
|
|
|
func (w *logWriter) Close() (err error) {
|
|
w.mtx.Lock()
|
|
defer w.mtx.Unlock()
|
|
|
|
return w.logUnreadBytes()
|
|
}
|