关于日志:zap源码阅读1

41次阅读

共计 10107 个字符，预计需要花费 26 分钟才能阅读完成。

zap 是 Uber 开发的一个高性能、强类型、分 level 的 go 语言日志库

1. 对象创立

1.1 Logger 构造体

type Logger struct {
    core zapcore.Core  // 外围接口

    development bool
    addCaller   bool 
    onFatal     zapcore.CheckWriteAction // default is WriteThenFatal

    name        string
    errorOutput zapcore.WriteSyncer

    addStack zapcore.LevelEnabler

    callerSkip int

    clock Clock
}

Logger 对象的创立有两种形式:

  1. 通过 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:       _systemClock,
    }
    return log.WithOptions(options...)
}

函数 WithOptions 应用的是 函数式抉择模式, 和 ants 源码浏览（https://segmentfault.com/a/11… 中应用的办法统一。

func (log *Logger) WithOptions(opts ...Option) *Logger {c := log.clone()
    for _, opt := range opts {opt.apply(c)
    }
    return c
}

Option 构造

type Option interface {apply(*Logger)
}

// optionFunc wraps a func so it satisfies the Option interface.
type optionFunc func(*Logger)

func (f optionFunc) apply(log *Logger) {f(log)
}

func WrapCore(f func(zapcore.Core) zapcore.Core) Option {return optionFunc(func(log *Logger) {log.core = f(log.core)
    })
}

func AddStacktrace(lvl zapcore.LevelEnabler) Option {return optionFunc(func(log *Logger) {log.addStack = lvl})
}
  1. 建造者模式 Build 创立
    建造者模式 应用场景：当一个类的结构函数参数个数超过 4 个，而且这些参数有些是可选的参数，思考应用建造者模式。
func (cfg Config) Build(opts ...Option) (*Logger, error) {enc, err := cfg.buildEncoder() // 创立 Encoder 对象, json or console
    if err != nil {return nil, err}
    sink, errSink, err := cfg.openSinks() // 创立失常输入和谬误输入的文件门路
    if err != nil {return nil, err}

    if cfg.Level == (AtomicLevel{}) {return nil, fmt.Errorf("missing Level")
    }

    log := New(zapcore.NewCore(enc, sink, cfg.Level),
        cfg.buildOptions(errSink)...,
    )  // 创立 log 对象
    if len(opts) > 0 {log = log.WithOptions(opts...) // 依据传入的配置重置 log 属性
    }
    return log, nil
}

1.2 Config 构造体解析

type Config struct {

    Level AtomicLevel `json:"level" yaml:"level"` //Level 是用来配置日志级别的，即日志的最低输入级别
    Development bool `json:"development" yaml:"development"` // 这个字段的含意是用来标记是否为开发者模式，在开发者模式下，日志输入的一些行为会和生产环境上不同。DisableCaller bool `json:"disableCaller" yaml:"disableCaller"` // 用来标记是否开启行号和文件名显示性能。DisableStacktrace bool `json:"disableStacktrace" yaml:"disableStacktrace"` // 标记是否开启调用栈追踪能力，即在打印异样日志时，是否打印调用栈。Sampling *SamplingConfig `json:"sampling" yaml:"sampling"` // Sampling 实现了日志的流控性能，或者叫采样配置，次要有两个配置参数，Initial 和 Thereafter，实现的成果是在 1s 的工夫单位内，如果某个日志级别下同样内容的日志输入数量超过了 Initial 的数量，那么超过之后，每隔 Thereafter 的数量，才会再输入一次。是一个对日志输入的爱护性能。Encoding string `json:"encoding" yaml:"encoding"` // json or console 

    EncoderConfig zapcore.EncoderConfig `json:"encoderConfig"  yaml:"encoderConfig"`

    OutputPaths []string `json:"outputPaths" yaml:"outputPaths"` // 日志输入门路

    ErrorOutputPaths []string `json:"errorOutputPaths"  // 谬误日志输入门路
 yaml:"errorOutputPaths"`

    InitialFields map[string]interface{} `json:"initialFields" yaml:"initialFields"` // 初始化 Fields}

EncoderConfig 构造

type EncoderConfig struct {

    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"`
    StacktraceKey string `json:"stacktraceKey" yaml:"stacktraceKey"`
    LineEnding    string `json:"lineEnding" yaml:"lineEnding"` // 配置行分隔符

    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"`
}

1.3 详解 Build

1.3.1 buildEncoder

buildEncoder –> newEncoder –> zapcore.Encoder 接口
newEncoder

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}
    constructor, ok := _encoderNameToConstructor[name] // 返回 Encoder 对象
    if !ok {return nil, fmt.Errorf("no encoder registered for name %q", name)
    }
    return constructor(encoderConfig)
}

map 中存储函数对象

    _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
        },
    }

NewConsoleEncoder 创立的 consoleEncoder 构造体的底层构造是 jsonEncoder, 只是 consoleEncoder 的spaced 为true, NewJSONEncoder 的 spaced 为false.

jsonEncoder 构造体：

type jsonEncoder struct {
    *EncoderConfig
    buf            *buffer.Buffer
    spaced         bool // include spaces after colons and commas
    openNamespaces int

    // for encoding generic values by reflection
    reflectBuf *buffer.Buffer
    reflectEnc *json.Encoder
}

func newJSONEncoder(cfg EncoderConfig, spaced bool) *jsonEncoder {
    return &jsonEncoder{
        EncoderConfig: &cfg,
        buf:           bufferpool.Get(),
        spaced:        spaced,
    }
}

Buffer 构造体


const _size = 1024 // by default, create 1 KiB buffers

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)} // 默认会创立 1k 的 buffer
        },
    }}
}
1.3.2 openSinks

创立 output, 和 errorput 文件
调用程序: Open–>open–> newSink

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() // 下文中的 close 在这里被调用
        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 {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
}

func newSink(rawURL string) (Sink, error) {u, err := url.Parse(rawURL) // 因为门路还能够是 http 的 url，失常的 path 赋值到 u.Path 上。if err != nil {return nil, fmt.Errorf("can't parse %q as a URL: %v", rawURL, err)
    }
    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)
}

_sinkFactories 是个 map 构造，存储的是回调函数。

    _sinkFactories = map[string]func(*url.URL) (Sink, error){schemeFile: newFileSink,}
    
func newFileSink(u *url.URL) (Sink, error) {
    if u.User != nil {return nil, fmt.Errorf("user and password not allowed with file URLs: got %v", u)
    }
    if u.Fragment != "" {return nil, fmt.Errorf("fragments not allowed with file URLs: got %v", u)
    }
    if u.RawQuery != "" {return nil, fmt.Errorf("query parameters not allowed with file URLs: got %v", u)
    }
    // Error messages are better if we check hostname and port separately.
    if u.Port() != "" {return nil, fmt.Errorf("ports not allowed with file URLs: got %v", u)
    }
    if hn := u.Hostname(); hn != ""&& hn !="localhost" {return nil, fmt.Errorf("file URLs must leave host empty or use localhost: got %v", u)
    }
    fmt.Println("u.Path:", u.Path)
    switch u.Path {
    case "stdout":
        return nopCloserSink{os.Stdout}, nil // stdout
    case "stderr":
        return nopCloserSink{os.Stderr}, nil 
    }
    // 带有具体的输入文件名
    return os.OpenFile(u.Path, os.O_WRONLY|os.O_APPEND|os.O_CREATE, 0666)
}

2. 如何应用

以 Info 为例:

func (log *Logger) Info(msg string, fields ...Field) {if ce := log.check(InfoLevel, msg); ce != nil {ce.Write(fields...)
    }
}

check 函数

func (log *Logger) check(lvl zapcore.Level, msg string) *zapcore.CheckedEntry {
    // check must always be called directly by a method in the Logger interface
    // (e.g., Check, Info, Fatal).
    const callerSkipOffset = 2

    // Check the level first to reduce the cost of disabled log calls.
    // Since Panic and higher may exit, we skip the optimization for those levels.
    if lvl < zapcore.DPanicLevel && !log.core.Enabled(lvl) {return nil}

    // Create basic checked entry thru the core; this will be non-nil if the
    // log message will actually be written somewhere.
    ent := zapcore.Entry{
        LoggerName: log.name,
        Time:       log.clock.Now(),
        Level:      lvl,
        Message:    msg,
    }
    ce := log.core.Check(ent, nil) // 满足达到日志等级才新建 CheckedEntry. 该 checkedEntry 应用 sync.Pool 创立。willWrite := ce != nil

    // Set up any required terminal behavior.
    switch ent.Level {
    case zapcore.PanicLevel:
        ce = ce.Should(ent, zapcore.WriteThenPanic)
    case zapcore.FatalLevel:
        onFatal := log.onFatal
        // Noop is the default value for CheckWriteAction, and it leads to
        // continued execution after a Fatal which is unexpected.
        if onFatal == zapcore.WriteThenNoop {onFatal = zapcore.WriteThenFatal}
        ce = ce.Should(ent, onFatal)
    case zapcore.DPanicLevel:
        if log.development {ce = ce.Should(ent, zapcore.WriteThenPanic)
        }
    }

    // Only do further annotation if we're going to write this message; checked
    // entries that exist only for terminal behavior don't benefit from
    // annotation.
    if !willWrite {return ce}
     // 其余省略
}

Write 函数

func (ce *CheckedEntry) Write(fields ...Field) {
    if ce == nil {return}

    if ce.dirty {
        if ce.ErrorOutput != nil {
            // Make a best effort to detect unsafe re-use of this CheckedEntry.
            // If the entry is dirty, log an internal error; because the
            // CheckedEntry is being used after it was returned to the pool,
            // the message may be an amalgamation from multiple call sites.
            fmt.Fprintf(ce.ErrorOutput, "%v Unsafe CheckedEntry re-use near Entry %+v.\n", ce.Time, ce.Entry)
            ce.ErrorOutput.Sync()}
        return
    }
    ce.dirty = true

    var err error
    for i := range ce.cores {err = multierr.Append(err, ce.cores[i].Write(ce.Entry, fields)) // 外围，写入格式化数据
    }
    if ce.ErrorOutput != nil {
        if err != nil {fmt.Fprintf(ce.ErrorOutput, "%v write error: %v\n", ce.Time, err)
            ce.ErrorOutput.Sync()}
    }

    should, msg := ce.should, ce.Message
    putCheckedEntry(ce)

    switch should {
    case WriteThenPanic:
        panic(msg)
    case WriteThenFatal:
        exit.Exit()
    case WriteThenGoexit:
        runtime.Goexit()}
}

外围写数据 writeContext


func (c *ioCore) Write(ent Entry, fields []Field) error {buf, err := c.enc.EncodeEntry(ent, fields) // 把数据写入缓存
    if err != nil {return err}
    _, err = c.out.Write(buf.Bytes()) // 写入数据到文件
    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
}

func (c consoleEncoder) writeContext(line *buffer.Buffer, extra []Field) {context := c.jsonEncoder.Clone().(*jsonEncoder)
    defer func() {
        // putJSONEncoder assumes the buffer is still used, but we write out the buffer so
        // we can free it.
        context.buf.Free()
        putJSONEncoder(context)
    }()

    addFields(context, extra) // 把 Fields 写入到 buffer 中
    context.closeOpenNamespaces()
    if context.buf.Len() == 0 {return}

    c.addSeparatorIfNecessary(line)
    line.AppendByte('{')
    line.Write(context.buf.Bytes())
    line.AppendByte('}')
}

参考: 1. https://zhuanlan.zhihu.com/p/…

正文完
日志
发表至： 日志
2021-04-26
 0
关于日志:三款日志管理工具横向对比Splunk-vs-Sumo-Logic-vs-Logstash
关于日志:日志脱敏之后无法根据信息快速定位怎么办
关于日志:日志分析常规操作
Choerodon如何进行日志收集与告警