log: add FilterCore and filter TiDB log (#862) (#895)

Signed-off-by: ti-srebot <ti-srebot@pingcap.com>

Author: ti-srebot

cmd/br/cmd.go

@@ -11,10 +11,9 @@ import (
 	"sync/atomic"
 	"time"
 
-	tidbutils "github.com/pingcap/tidb-tools/pkg/utils"
-
 	"github.com/pingcap/errors"
 	"github.com/pingcap/log"
+	tidbutils "github.com/pingcap/tidb-tools/pkg/utils"
 	"github.com/pingcap/tidb/util/logutil"
 	"github.com/sirupsen/logrus"
 	"github.com/spf13/cobra"

cmd/tidb-lightning/main.go

@@ -83,7 +83,7 @@ func main() {
 			if err := cfg.LoadFromGlobal(globalCfg); err != nil {
 				return err
 			}
-			return app.RunOnce(context.Background(), cfg, nil, nil)
+			return app.RunOnce(context.Background(), cfg, nil)
 		}
 	}()
 

pkg/lightning/lightning.go

@@ -178,7 +178,7 @@ func (l *Lightning) goServe(statusAddr string, realAddrWriter io.Writer) error {
 //   use a default glue later.
 // - for lightning as a library, taskCtx could be a meaningful context that get canceled outside, and glue could be a
 //   caller implemented glue.
-func (l *Lightning) RunOnce(taskCtx context.Context, taskCfg *config.Config, glue glue.Glue, replaceLogger *zap.Logger) error {
+func (l *Lightning) RunOnce(taskCtx context.Context, taskCfg *config.Config, glue glue.Glue) error {
 	if err := taskCfg.Adjust(taskCtx); err != nil {
 		return err
 	}
@@ -188,9 +188,6 @@ func (l *Lightning) RunOnce(taskCtx context.Context, taskCfg *config.Config, glu
 		taskCfg.TaskID = int64(val.(int))
 	})
 
-	if replaceLogger != nil {
-		log.SetAppLogger(replaceLogger)
-	}
 	return l.run(taskCtx, taskCfg, glue)
 }
 

pkg/lightning/lightning_test.go

@@ -64,7 +64,7 @@ func (s *lightningSuite) TestRun(c *C) {
 	cfg := config.NewConfig()
 	err := cfg.LoadFromGlobal(globalConfig)
 	c.Assert(err, IsNil)
-	err = lightning.RunOnce(context.Background(), cfg, nil, nil)
+	err = lightning.RunOnce(context.Background(), cfg, nil)
 	c.Assert(err, ErrorMatches, ".*mydumper dir does not exist")
 
 	path, _ := filepath.Abs(".")
@@ -357,7 +357,7 @@ func (s *lightningServerSuite) TestHTTPAPIOutsideServerMode(c *C) {
 	err := cfg.LoadFromGlobal(s.lightning.globalCfg)
 	c.Assert(err, IsNil)
 	go func() {
-		errCh <- s.lightning.RunOnce(s.lightning.ctx, cfg, nil, nil)
+		errCh <- s.lightning.RunOnce(s.lightning.ctx, cfg, nil)
 	}()
 	time.Sleep(100 * time.Millisecond)
 

pkg/lightning/log/filter.go

@@ -0,0 +1,57 @@
+// Copyright 2021 PingCAP, Inc. Licensed under Apache-2.0.
+
+package log
+
+import (
+	"strings"
+
+	"go.uber.org/zap/zapcore"
+)
+
+var _ zapcore.Core = (*FilterCore)(nil)
+
+// FilterCore is a zapcore.Core implementation, it filters log by path-qualified
+// package name.
+type FilterCore struct {
+	zapcore.Core
+	filters []string
+}
+
+// NewFilterCore returns a FilterCore.
+//
+// Example, filter TiDB's log, `NewFilterCore(core, "github.com/pingcap/tidb/")`.
+// Note, must set AddCaller() to the logger.
+func NewFilterCore(core zapcore.Core, filteredPackages ...string) *FilterCore {
+	return &FilterCore{
+		Core:    core,
+		filters: filteredPackages,
+	}
+}
+
+// With adds structured context to the Core.
+func (f *FilterCore) With(fields []zapcore.Field) zapcore.Core {
+	return &FilterCore{
+		Core:    f.Core.With(fields),
+		filters: f.filters,
+	}
+}
+
+// Check overrides wrapper core.Check and adds itself to zapcore.CheckedEntry.
+func (f *FilterCore) Check(entry zapcore.Entry, ce *zapcore.CheckedEntry) *zapcore.CheckedEntry {
+	if f.Enabled(entry.Level) {
+		return ce.AddCore(entry, f)
+	}
+	return ce
+}
+
+// Write filters entry by checking if entry's Caller.Function matches filtered
+// package path.
+func (f *FilterCore) Write(entry zapcore.Entry, fields []zapcore.Field) error {
+	for i := range f.filters {
+		// Caller.Function is a package path-qualified function name.
+		if strings.Contains(entry.Caller.Function, f.filters[i]) {
+			return nil
+		}
+	}
+	return f.Core.Write(entry, fields)
+}

pkg/lightning/log/filter_test.go

@@ -0,0 +1,96 @@
+// Copyright 2021 PingCAP, Inc. Licensed under Apache-2.0.
+
+package log_test
+
+import (
+	"regexp"
+	"strings"
+
+	. "github.com/pingcap/check"
+	"go.uber.org/zap"
+	"go.uber.org/zap/zapcore"
+
+	"github.com/pingcap/br/pkg/lightning/log"
+)
+
+var _ = Suite(&testFilterSuite{})
+
+type testFilterSuite struct{}
+
+func (s *testFilterSuite) TestFilter(c *C) {
+	logger, buffer := log.MakeTestLogger()
+	logger.Warn("the message", zap.Int("number", 123456), zap.Ints("array", []int{7, 8, 9}))
+	c.Assert(
+		buffer.Stripped(), Equals,
+		`{"$lvl":"WARN","$msg":"the message","number":123456,"array":[7,8,9]}`,
+	)
+
+	logger, buffer = log.MakeTestLogger(zap.WrapCore(func(c zapcore.Core) zapcore.Core {
+		return log.NewFilterCore(c, "github.com/pingcap/br/")
+	}), zap.AddCaller())
+	logger.Warn("the message", zap.Int("number", 123456), zap.Ints("array", []int{7, 8, 9}))
+	c.Assert(buffer.Stripped(), HasLen, 0)
+
+	logger, buffer = log.MakeTestLogger(zap.WrapCore(func(c zapcore.Core) zapcore.Core {
+		return log.NewFilterCore(c, "github.com/pingcap/tidb/").With([]zap.Field{zap.String("a", "b")})
+	}), zap.AddCaller())
+	logger.Warn("the message", zap.Int("number", 123456), zap.Ints("array", []int{7, 8, 9}))
+	c.Assert(
+		buffer.Stripped(), Equals,
+		`{"$lvl":"WARN","$msg":"the message","a":"b","number":123456,"array":[7,8,9]}`,
+	)
+
+	logger, buffer = log.MakeTestLogger(zap.WrapCore(func(c zapcore.Core) zapcore.Core {
+		return log.NewFilterCore(c, "github.com/pingcap/br/").With([]zap.Field{zap.String("a", "b")})
+	}), zap.AddCaller())
+	logger.Warn("the message", zap.Int("number", 123456), zap.Ints("array", []int{7, 8, 9}))
+	c.Assert(buffer.Stripped(), HasLen, 0)
+
+	// Fields won't trigger filter.
+	logger, buffer = log.MakeTestLogger(zap.WrapCore(func(c zapcore.Core) zapcore.Core {
+		return log.NewFilterCore(c, "github.com/pingcap/check/").With([]zap.Field{zap.String("a", "b")})
+	}), zap.AddCaller())
+	logger.Warn("the message", zap.String("stack", "github.com/pingcap/check/"))
+	c.Assert(
+		buffer.Stripped(), Equals,
+		`{"$lvl":"WARN","$msg":"the message","a":"b","stack":"github.com/pingcap/check/"}`,
+	)
+}
+
+// PASS: filter_test.go:82: testFilterSuite.BenchmarkFilterRegexMatchString 1000000               1163 ns/op
+// PASS: filter_test.go:64: testFilterSuite.BenchmarkFilterStringsContains  10000000               159 ns/op
+//
+// Run `go test github.com/pingcap/br/pkg/lightning/log -check.b -test.v` to get benchmark result.
+func (s *testFilterSuite) BenchmarkFilterStringsContains(c *C) {
+	c.ResetTimer()
+
+	inputs := []string{
+		"github.com/pingcap/tidb/some/package/path",
+		"github.com/tikv/pd/some/package/path",
+		"github.com/pingcap/br/some/package/path",
+	}
+	filters := []string{"github.com/pingcap/tidb/", "github.com/tikv/pd/"}
+	for i := 0; i < c.N; i++ {
+		for i := range inputs {
+			for j := range filters {
+				_ = strings.Contains(inputs[i], filters[j])
+			}
+		}
+	}
+}
+
+func (s *testFilterSuite) BenchmarkFilterRegexMatchString(c *C) {
+	c.ResetTimer()
+
+	inputs := []string{
+		"github.com/pingcap/tidb/some/package/path",
+		"github.com/tikv/pd/some/package/path",
+		"github.com/pingcap/br/some/package/path",
+	}
+	filters := regexp.MustCompile(`github.com/(pingcap/tidb|tikv/pd)/`)
+	for i := 0; i < c.N; i++ {
+		for i := range inputs {
+			_ = filters.MatchString(inputs[i])
+		}
+	}
+}

pkg/lightning/log/log.go

@@ -78,8 +78,13 @@ func InitLogger(cfg *Config, tidbLoglevel string) error {
 	logutil.InitLogger(&logutil.LogConfig{Config: pclog.Config{Level: tidbLoglevel}})
 
 	logCfg := &pclog.Config{
-		Level: cfg.Level,
+		Level:         cfg.Level,
+		DisableCaller: false, // FilterCore requires zap.AddCaller.
 	}
+	filterTiDBLog := zap.WrapCore(func(core zapcore.Core) zapcore.Core {
+		// Filter logs from TiDB and PD.
+		return NewFilterCore(core, "github.com/pingcap/tidb/", "github.com/tikv/pd/")
+	})
 	if len(cfg.File) > 0 {
 		logCfg.File = pclog.FileLogConfig{
 			Filename:   cfg.File,
@@ -88,10 +93,11 @@ func InitLogger(cfg *Config, tidbLoglevel string) error {
 			MaxBackups: cfg.FileMaxBackups,
 		}
 	}
-	logger, props, err := pclog.InitLogger(logCfg)
+	logger, props, err := pclog.InitLogger(logCfg, filterTiDBLog)
 	if err != nil {
 		return err
 	}
+	pclog.ReplaceGlobals(logger, props)
 
 	// Do not log stack traces at all, as we'll get the stack trace from the
 	// error itself.
@@ -106,11 +112,6 @@ func L() Logger {
 	return appLogger
 }
 
-// SetAppLogger replaces the default logger in this package to given one
-func SetAppLogger(l *zap.Logger) {
-	appLogger = Logger{l.WithOptions(zap.AddStacktrace(zap.DPanicLevel))}
-}
-
 // Level returns the current global log level.
 func Level() zapcore.Level {
 	return appLevel.Level()

pkg/lightning/log/testlogger.go

@@ -20,7 +20,7 @@ import (
 )
 
 // MakeTestLogger creates a Logger instance which produces JSON logs.
-func MakeTestLogger() (Logger, *zaptest.Buffer) {
+func MakeTestLogger(opts ...zap.Option) (Logger, *zaptest.Buffer) {
 	buffer := new(zaptest.Buffer)
 	logger := zap.New(zapcore.NewCore(
 		zapcore.NewJSONEncoder(zapcore.EncoderConfig{
@@ -31,6 +31,6 @@ func MakeTestLogger() (Logger, *zaptest.Buffer) {
 		}),
 		buffer,
 		zap.DebugLevel,
-	))
+	), opts...)
 	return Logger{Logger: logger}, buffer
 }