Go紀錄檔框架zap增強及原始碼解讀

type Config struct {
	// 紀錄檔Level，因為可以動態更改，所以是atomic型別的，畢竟比鎖的效能好
	Level AtomicLevel `json:"level" yaml:"level"`
	// dev模式，啟用後會更改在某些使用情形下的行為，後面原始碼解讀模組會具體看到有什麼作用
	Development bool `json:"development" yaml:"development"`
	// 禁用caller，caller就是會在打的log里加一條屬性，表示這條紀錄檔是在哪裡打的，例如"httpd/svc.go:123"
	DisableCaller bool `json:"disableCaller" yaml:"disableCaller"`
	// 是否要在log里加上呼叫棧，dev模式下只有WarnLevel模式以上有呼叫棧，prod模式下只有ErrorLevel以上有呼叫棧
	DisableStacktrace bool `json:"disableStacktrace" yaml:"disableStacktrace"`
	// 取樣策略，控制打log的速率，也可以做一些其他自定義的操作，不難理解，具體看下面的SamplingConfig
	Sampling *SamplingConfig `json:"sampling" yaml:"sampling"`
	// log格式，自帶的有json和console兩種格式，可以通過使用RegisterEncoder來自定義log格式
	Encoding string `json:"encoding" yaml:"encoding"`
	// log格式具體設定，詳細看下面的EncoderConfig
	EncoderConfig zapcore.EncoderConfig `json:"encoderConfig" yaml:"encoderConfig"`
	// log輸出路徑，看結構表示可以有多個輸出路徑
	OutputPaths []string `json:"outputPaths" yaml:"outputPaths"`
	// 內部錯誤輸出路徑，預設是stderr
	ErrorOutputPaths []string `json:"errorOutputPaths" yaml:"errorOutputPaths"`
	// 每條log都會加上InitialFields裡的內容，顧名思義
	InitialFields map[string]interface{} `json:"initialFields" yaml:"initialFields"`
}

// 取樣策略設定，大致的邏輯是每秒超過Thereafter個相同msg的log會執行自定義的Hook函數(第二個引數為一個標誌，LogDropped)，具體邏輯可以看下面的原始碼解析
type SamplingConfig struct {
	Initial    int                                           `json:"initial" yaml:"initial"`
	Thereafter int                                           `json:"thereafter" yaml:"thereafter"`
	Hook       func(zapcore.Entry, zapcore.SamplingDecision) `json:"-" yaml:"-"`
}
const (
	_numLevels        = _maxLevel - _minLevel + 1
	_countersPerLevel = 4096
)
// 用來記錄紀錄檔打了多少條
type counter struct {
	resetAt atomic.Int64
	counter atomic.Uint64
}
type counters [_numLevels][_countersPerLevel]counter
// 這裡可以看到sampler就是Core外面包了一層Wrapper
type sampler struct {
	Core
	counts            *counters
	tick              time.Duration                // 這裡的tick在初始化Logger的時候已經寫死了是time.Second，也就是1秒
	first, thereafter uint64
	hook              func(Entry, SamplingDecision)
}
// 所有的Core.Check都會先走sampler.Check，然後再走Core.Check
func (s *sampler) Check(ent Entry, ce *CheckedEntry) *CheckedEntry {
	if !s.Enabled(ent.Level) {
		return ce
	}
	if ent.Level >= _minLevel && ent.Level <= _maxLevel {
		// 根據Message獲取counter，也就是這個Message打過幾次紀錄檔了
		counter := s.counts.get(ent.Level, ent.Message)
		// 打一條Message就會記錄一次到counters裡，不過每秒會重置一次counter，具體看IncCheckReset裡的邏輯
		n := counter.IncCheckReset(ent.Time, s.tick)
		// first表示最初的first條紀錄檔呼叫hook時第二個引數傳LogSampled，超過first的紀錄檔，每threrafter條紀錄檔第二個引數傳LogSampled，否則傳LogDropped
		// 假設first是100，thereafter是50，表示同一個Message的log，最初的100條全都會記錄，之後的log在每秒鐘內，每50條記錄一次
		if n > s.first && (s.thereafter == 0 || (n-s.first)%s.thereafter != 0) {
			s.hook(ent, LogDropped)
			return ce
		}
		s.hook(ent, LogSampled)
	}
	return s.Core.Check(ent, ce)
}
// 這裡可能會出現意想不到的情況
// 因為_countersPerLevel寫死了是4096，那麼必然會存在不同的key做完hash後取模會路由到相同的counter裡
// 那麼就會有概率丟棄掉沒有達到丟棄閾值的log
// 假設abc和def的hash值一樣，first是0，thereafter是10，表示每秒鐘每種log每10條才會記錄1次，那麼abc和def這兩種log就會共用同一個counter，這就是問題所在
func (cs *counters) get(lvl Level, key string) *counter {
	i := lvl - _minLevel
	// fnv32a是一個hash函數
	// _countersPerLevel固定是4096
	j := fnv32a(key) % _countersPerLevel
	return &cs[i][j]
}
func (c *counter) IncCheckReset(t time.Time, tick time.Duration) uint64 {
	tn := t.UnixNano()
	resetAfter := c.resetAt.Load()
	if resetAfter > tn {
		return c.counter.Inc()
	}
	c.counter.Store(1)
	newResetAfter := tn + tick.Nanoseconds()
	if !c.resetAt.CAS(resetAfter, newResetAfter) {
		return c.counter.Inc()
	}
	return 1
}

// log格式的詳細設定
type EncoderConfig struct {
	// 設定log內容裡的一些屬性的key
	MessageKey     string `json:"messageKey" yaml:"messageKey"`
	LevelKey       string `json:"levelKey" yaml:"levelKey"`
	TimeKey        string `json:"timeKey" yaml:"timeKey"`
	NameKey        string `json:"nameKey" yaml:"nameKey"`
	CallerKey      string `json:"callerKey" yaml:"callerKey"`
	FunctionKey    string `json:"functionKey" yaml:"functionKey"`
	StacktraceKey  string `json:"stacktraceKey" yaml:"stacktraceKey"`
	// 顧名思義不解釋
	SkipLineEnding bool   `json:"skipLineEnding" yaml:"skipLineEnding"`
	LineEnding     string `json:"lineEnding" yaml:"lineEnding"`
	// Configure the primitive representations of common complex types. For
	// example, some users may want all time.Times serialized as floating-point
	// seconds since epoch, while others may prefer ISO8601 strings.
	// 自定義一些屬性的格式，例如指定Time欄位格式化為2022-05-23 16:16:16
	EncodeLevel    LevelEncoder    `json:"levelEncoder" yaml:"levelEncoder"`
	EncodeTime     TimeEncoder     `json:"timeEncoder" yaml:"timeEncoder"`
	EncodeDuration DurationEncoder `json:"durationEncoder" yaml:"durationEncoder"`
	EncodeCaller   CallerEncoder   `json:"callerEncoder" yaml:"callerEncoder"`
	EncodeName NameEncoder `json:"nameEncoder" yaml:"nameEncoder"`
	// 用於interface型別的encoder，可以自定義，預設為jsonEncoder
	NewReflectedEncoder func(io.Writer) ReflectedEncoder `json:"-" yaml:"-"`
	// console格式的分隔符，預設是tab
	ConsoleSeparator string `json:"consoleSeparator" yaml:"consoleSeparator"`
}

// 通過Config結構體Build出一個Logger
func (cfg Config) Build(opts ...Option) (*Logger, error) {
	// 核心函數buildEncoder
	enc, err := cfg.buildEncoder()
	if err != nil {
		return nil, err
	}
	// 核心函數openSinks
	sink, errSink, err := cfg.openSinks()
	if err != nil {
		return nil, err
	}
	if cfg.Level == (AtomicLevel{}) {
		return nil, fmt.Errorf("missing Level")
	}
	// 核心函數New
	log := New(
		// 核心函數NewCore
		zapcore.NewCore(enc, sink, cfg.Level),
		cfg.buildOptions(errSink)...,
	)
	if len(opts) > 0 {
		log = log.WithOptions(opts...)
	}
	return log, nil
}

// 核心函數buildEncoder
func (cfg Config) buildEncoder() (zapcore.Encoder, error) {
	return newEncoder(cfg.Encoding, cfg.EncoderConfig)
}
// _encoderNameToConstructor是一個map[string]constructor，plugin式寫法，可以通過RegisterEncoder函數註冊自定義的Encoder，預設只有console和json
_encoderNameToConstructor = map[string]func(zapcore.EncoderConfig) (zapcore.Encoder, error){
	"console": func(encoderConfig zapcore.EncoderConfig) (zapcore.Encoder, error) {
		return zapcore.NewConsoleEncoder(encoderConfig), nil
	},
	"json": func(encoderConfig zapcore.EncoderConfig) (zapcore.Encoder, error) {
		return zapcore.NewJSONEncoder(encoderConfig), nil
	},
}
func newEncoder(name string, encoderConfig zapcore.EncoderConfig) (zapcore.Encoder, error) {
	if encoderConfig.TimeKey != "" && encoderConfig.EncodeTime == nil {
		return nil, fmt.Errorf("missing EncodeTime in EncoderConfig")
	}
	_encoderMutex.RLock()
	defer _encoderMutex.RUnlock()
	if name == "" {
		return nil, errNoEncoderNameSpecified
	}
	// 通過name，也就是Config.Encoding來決定使用哪種encoder
	constructor, ok := _encoderNameToConstructor[name]
	if !ok {
		return nil, fmt.Errorf("no encoder registered for name %q", name)
	}
	return constructor(encoderConfig)
}
// 這裡只展示jsonEncoder的邏輯，consoleEncoder和jsonEncoder差別不大
func NewJSONEncoder(cfg EncoderConfig) Encoder {
	return newJSONEncoder(cfg, false)
}
func newJSONEncoder(cfg EncoderConfig, spaced bool) *jsonEncoder {
	if cfg.SkipLineEnding {
		cfg.LineEnding = ""
	} else if cfg.LineEnding == "" {
		cfg.LineEnding = DefaultLineEnding
	}
	// If no EncoderConfig.NewReflectedEncoder is provided by the user, then use default
	if cfg.NewReflectedEncoder == nil {
		cfg.NewReflectedEncoder = defaultReflectedEncoder
	}
	return &jsonEncoder{
		EncoderConfig: &cfg,
		// 這個buf是高效能的關鍵之一，使用了簡化的bytesBuffer和sync.Pool，程式碼貼在下面
		buf:           bufferpool.Get(),
		spaced:        spaced,
	}
}

type Buffer struct {
	bs   []byte
	pool Pool
}
type Pool struct {
	p *sync.Pool
}
func NewPool() Pool {
	return Pool{p: &sync.Pool{
		New: func() interface{} {
			return &Buffer{bs: make([]byte, 0, _size)}
		},
	}}
}
// 從Pool裡拿一個Buffer，初始化裡面的[]byte
func (p Pool) Get() *Buffer {
	buf := p.p.Get().(*Buffer)
	buf.Reset()
	// 這裡賦值pool為當前Pool，用於使用完Buffer後把Buffer後放回pool裡，也就是下面的put函數
	buf.pool = p
	return buf
}
func (p Pool) put(buf *Buffer) {
	p.p.Put(buf)
}

// 核心函數openSinks
func (cfg Config) openSinks() (zapcore.WriteSyncer, zapcore.WriteSyncer, error) {
	sink, closeOut, err := Open(cfg.OutputPaths...)
	if err != nil {
		return nil, nil, err
	}
	errSink, _, err := Open(cfg.ErrorOutputPaths...)
	if err != nil {
		closeOut()
		return nil, nil, err
	}
	return sink, errSink, nil
}
func Open(paths ...string) (zapcore.WriteSyncer, func(), error) {
	writers, close, err := open(paths)
	if err != nil {
		return nil, nil, err
	}
	writer := CombineWriteSyncers(writers...)
	return writer, close, nil
}
func open(paths []string) ([]zapcore.WriteSyncer, func(), error) {
	writers := make([]zapcore.WriteSyncer, 0, len(paths))
	closers := make([]io.Closer, 0, len(paths))
	close := func() {
		for _, c := range closers {
			c.Close()
		}
	}
	var openErr error
	for _, path := range paths {
		// 核心函數newSink
		sink, err := newSink(path)
		if err != nil {
			openErr = multierr.Append(openErr, fmt.Errorf("couldn't open sink %q: %v", path, err))
			continue
		}
		writers = append(writers, sink)
		closers = append(closers, sink)
	}
	if openErr != nil {
		close()
		return writers, nil, openErr
	}
	return writers, close, nil
}
// 這裡也是plugin式寫法，可以通過RegisterSink來自定義sink，比如自定義一個支援http協定的sink，在文章的尾部會實現一個自定義的sink
_sinkFactories = map[string]func(*url.URL) (Sink, error){
	schemeFile: newFileSink,
}
func newSink(rawURL string) (Sink, error) {
	// 通過rawURL判斷初始化哪種sink，實際上zap只支援file，看上面的_sinkFactories
	u, err := url.Parse(rawURL)
	if err != nil {
		return nil, fmt.Errorf("can't parse %q as a URL: %v", rawURL, err)
	}
	// 如果url是類似於/var/abc.log這種的字串，那麼經過Parse後的u.Scheme就是""，然後會被賦值schemeFile
	// 如果url是類似於http://127.0.0.1:1234這種的字串，那麼經過Parse後的u.Scheme就是"http"，不過zap本身不支援http，可以自定義一個支援http的sink
	if u.Scheme == "" {
		u.Scheme = schemeFile
	}
	_sinkMutex.RLock()
	factory, ok := _sinkFactories[u.Scheme]
	_sinkMutex.RUnlock()
	if !ok {
		return nil, &errSinkNotFound{u.Scheme}
	}
	return factory(u)
}
// 這裡的sink實際上就是一個*File
func newFileSink(u *url.URL) (Sink, error) {
	// ...
	switch u.Path {
	case "stdout":
		return nopCloserSink{os.Stdout}, nil
	case "stderr":
		return nopCloserSink{os.Stderr}, nil
	}
	return os.OpenFile(u.Path, os.O_WRONLY|os.O_APPEND|os.O_CREATE, 0666)
}

// 核心函數NewCore
func NewCore(enc Encoder, ws WriteSyncer, enab LevelEnabler) Core {
	return &ioCore{
		LevelEnabler: enab,
		enc:          enc,
		out:          ws,
	}
}
// 核心函數New
func New(core zapcore.Core, options ...Option) *Logger {
	if core == nil {
		return NewNop()
	}
	log := &Logger{
		core:        core,
		errorOutput: zapcore.Lock(os.Stderr),
		addStack:    zapcore.FatalLevel + 1,
		clock:       zapcore.DefaultClock,
	}
	return log.WithOptions(options...)
}

l, _ := zap.NewProduction()
l.Error("Message Content", zap.String("tagA", "tagAValue"))

func (log *Logger) Error(msg string, fields ...Field) {
	// 核心函數 check
	if ce := log.check(ErrorLevel, msg); ce != nil {
		// 核心函數 Write
		ce.Write(fields...)
	}
}

// 核心函數check，實際邏輯就是檢查了下Level要不要打log，順便新增了呼叫棧和caller
func (log *Logger) check(lvl zapcore.Level, msg string) *zapcore.CheckedEntry {
	// 跳過當前這個check函數以及呼叫check的Error/Info/Fatal等函數
	const callerSkipOffset = 2
	// 檢查level
	if lvl < zapcore.DPanicLevel && !log.core.Enabled(lvl) {
		return nil
	}
	ent := zapcore.Entry{
		LoggerName: log.name,
		Time:       log.clock.Now(),
		Level:      lvl,
		Message:    msg,
	}
	// 核心函數 ioCore.Check
	ce := log.core.Check(ent, nil)
	willWrite := ce != nil
	// ...
	if !willWrite {
		return ce
	}
	// 新增stacktrace和caller相關
	// ...
	return ce
}

// 實際就是把core新增到了CheckedEntry裡，在後續的CheckedEntry.Write裡會被呼叫
func (c *ioCore) Check(ent Entry, ce *CheckedEntry) *CheckedEntry {
	if c.Enabled(ent.Level) {
		return ce.AddCore(ent, c)
	}
	return ce
}
func (ce *CheckedEntry) AddCore(ent Entry, core Core) *CheckedEntry {
	if ce == nil {
		// getCheckedEntry使用了sync.Pool
		ce = getCheckedEntry()
		ce.Entry = ent
	}
	ce.cores = append(ce.cores, core)
	return ce
}

// 核心函數 Write
func (ce *CheckedEntry) Write(fields ...Field) {
	// ...
	var err error
	// 實際就是呼叫了Core.Write
	for i := range ce.cores {
		err = multierr.Append(err, ce.cores[i].Write(ce.Entry, fields))
	}
	if err != nil && ce.ErrorOutput != nil {
		fmt.Fprintf(ce.ErrorOutput, "%v write error: %vn", ce.Time, err)
		ce.ErrorOutput.Sync()
	}
	should, msg := ce.should, ce.Message
	// 把CheckedEntry放回到pool裡
	putCheckedEntry(ce)
	// ...
}
func (c *ioCore) Write(ent Entry, fields []Field) error {
	// 首先Encode，高效能的核心就在EncodeEntry裡
	buf, err := c.enc.EncodeEntry(ent, fields)
	if err != nil {
		return err
	}
	// 然後Write，out就是sink
	_, err = c.out.Write(buf.Bytes())
	// 然後把buf放回到pool裡
	buf.Free()
	if err != nil {
		return err
	}
	if ent.Level > ErrorLevel {
		// Since we may be crashing the program, sync the output. Ignore Sync
		// errors, pending a clean solution to issue #370.
		c.Sync()
	}
	return nil
}

// zap並沒有使用類似marshalJson的方法來encode，而是使用了拼接字串的方式手動拼出了一個json字串，這種方式的效能比marshalJson的效能要好很多
// 裡面的具體邏輯很簡單，就是append一個key，append一個value
func (enc *jsonEncoder) EncodeEntry(ent Entry, fields []Field) (*buffer.Buffer, error) {
	final := enc.clone()
	final.buf.AppendByte('{')
	if final.LevelKey != "" && final.EncodeLevel != nil {
		final.addKey(final.LevelKey)
		cur := final.buf.Len()
		final.EncodeLevel(ent.Level, final)
		if cur == final.buf.Len() {
			// User-supplied EncodeLevel was a no-op. Fall back to strings to keep
			// output JSON valid.
			final.AppendString(ent.Level.String())
		}
	}
	if final.TimeKey != "" {
		final.AddTime(final.TimeKey, ent.Time)
	}
	if ent.LoggerName != "" && final.NameKey != "" {
		final.addKey(final.NameKey)
		cur := final.buf.Len()
		nameEncoder := final.EncodeName
		// if no name encoder provided, fall back to FullNameEncoder for backwards
		// compatibility
		if nameEncoder == nil {
			nameEncoder = FullNameEncoder
		}
		nameEncoder(ent.LoggerName, final)
		if cur == final.buf.Len() {
			// User-supplied EncodeName was a no-op. Fall back to strings to
			// keep output JSON valid.
			final.AppendString(ent.LoggerName)
		}
	}
	if ent.Caller.Defined {
		if final.CallerKey != "" {
			final.addKey(final.CallerKey)
			cur := final.buf.Len()
			final.EncodeCaller(ent.Caller, final)
			if cur == final.buf.Len() {
				// User-supplied EncodeCaller was a no-op. Fall back to strings to
				// keep output JSON valid.
				final.AppendString(ent.Caller.String())
			}
		}
		if final.FunctionKey != "" {
			final.addKey(final.FunctionKey)
			final.AppendString(ent.Caller.Function)
		}
	}
	if final.MessageKey != "" {
		final.addKey(enc.MessageKey)
		final.AppendString(ent.Message)
	}
	if enc.buf.Len() > 0 {
		final.addElementSeparator()
		final.buf.Write(enc.buf.Bytes())
	}
	addFields(final, fields)
	final.closeOpenNamespaces()
	if ent.Stack != "" && final.StacktraceKey != "" {
		final.AddString(final.StacktraceKey, ent.Stack)
	}
	final.buf.AppendByte('}')
	final.buf.AppendString(final.LineEnding)
	ret := final.buf
	putJSONEncoder(final)
	return ret, nil
}

lc := zap.NewProductionConfig()
lc.Encoding = "console"
lc.Sampling.Initial = 1    // 當Initial為1時，第二條紀錄檔不會列印出來，改為大於1時第二條紀錄檔才會列印出來
lc.Sampling.Thereafter = 10
l, _ := lc.Build()
l.Info("abc")
l.Info("yTI")
l.Info("def")

func init() {
	// 這裡註冊http sink
	err := zap.RegisterSink("http", httpSink)
	if err != nil {
		fmt.Println("Register http sink fail", err)
	}
}
func httpSink(url *url.URL) (zap.Sink, error) {
	return &Http{
		// httpc是我封裝的httpClient，沒什麼別的邏輯，直接當成http.Client就好
		httpc: httpc.New(httpc.NewConfig(), context.Background()),
		url:   url,
	}, nil
}
type Http struct {
	httpc *httpc.HttpC
	url   *url.URL
}
// 主要邏輯就是Write
func (h *Http) Write(p []byte) (n int, err error) {
	// 初始化request
	req, err := http.NewRequest("POST", h.url.String(), bytes.NewReader(p))
	if err != nil {
		return 0, err
	}
	// 執行http請求
	resp, err := h.httpc.Do(req)
	if err != nil {
		return 0, err
	}
	defer resp.Body.Close()
	// 獲取response
	respBody, err := ioutil.ReadAll(resp.Body)
	if err != nil {
		return 0, err
	}
	if resp.StatusCode != http.StatusNoContent && resp.StatusCode != http.StatusOK {
		return 0, errors.New(util.Bytes2String(respBody))
	}
	return len(p), nil
}
// 可以搞個內建的queue或者[]log，在Sync函數裡用來做批次傳送提升效能，這裡只是簡單的實現，所以Sync沒什麼邏輯
func (h *Http) Sync() error {
	return nil
}
func (h *Http) Close() error {
	return h.httpc.Close()
}

package main
import (
	"go.uber.org/zap"
)
var l *zap.Logger
func test_a() {
	test_b()
}
func test_b() {
	test_c()
}
func test_c() {
	l.Error("err content")
}
func main() {
	l, _ = zap.NewDevelopment()
	test_a()
}

2022-05-23T23:16:36.598+0800    ERROR   gtil/main.go:20 err content
main.test_c
        D:/workspace/code/go/gtil/main.go:20
main.test_b
        D:/workspace/code/go/gtil/main.go:16
main.test_a
        D:/workspace/code/go/gtil/main.go:12
main.main
        D:/workspace/code/go/gtil/main.go:25
runtime.main
        D:/workspace/env/scoop/apps/go/current/src/runtime/proc.go:250

func test_a() error {
	return test_b()
}
func test_b() error {
	return test_c()
}
func test_c() error {
	// 底層的函數出現error應該return，而不是打log
	return errors.new("do test_c fail")
}
func main() {
	l, _ = zap.NewProduction()
	err := test_a()
	if err != nil {
		l.Error("main error", zap.Error(err))
	}
}

2022-05-23T23:16:54.955+0800    ERROR   gtil/main.go:27 main error      {"error": "do test_c fail"}
main.main
        D:/workspace/code/go/gtil/main.go:27
runtime.main
        D:/workspace/env/scoop/apps/go/current/src/runtime/proc.go:250

func test_a() error {
	err := test_b()
	if err != nil {
		return errors.Wrap(err, "do test_a fail")
	}
	return nil
}
func test_b() error {
	err := test_c()
	if err != nil {
		return errors.Wrap(err, "do test_b fail")
	}
	return nil
}
func test_c() error {
	return errors.New("do test_c fail")
}
func main() {
	l, _ = zap.NewDevelopment()
	err := test_a()
	if err != nil {
		l.Error("main error", zap.Error(err))
	}
}

func NewErrStackCore(c zapcore.Core) zapcore.Core {
    return &errStackCore{c}
}
type errStackCore struct {
    zapcore.Core
}
func (c *errStackCore) With(fields []zapcore.Field) zapcore.Core {
    return &errStackCore{
        c.Core.With(fields),
    }
}
func (c *errStackCore) Write(ent zapcore.Entry, fields []zapcore.Field) error {
    // 判斷fields裡有沒有error欄位
    if !hasStackedErr(fields) {
        return c.Core.Write(ent, fields)
    }
    // 這裡是重點，從fields裡取出error欄位，把內容放到ent.Stack裡，邏輯就是這樣，具體程式碼就不給出了
    ent.Stack, fields = getStacks(fields)
    return c.Core.Write(ent, fields)
}
func (c *errStackCore) Check(ent zapcore.Entry, ce *zapcore.CheckedEntry) *zapcore.CheckedEntry {
    return c.Core.Check(ent, ce)
}