摘要：實現方式是每個掛載目錄使用的插件掃描每個容器日志文件，直接發送給。首先根據容器配置的類別調用返回一個方法類型實質就是從工廠類注冊的插件去查找，具體源碼下文分析。在處理中心統一完成下一步處理。直接接收容器的日志。

目前容器日志有兩種輸出形式：

stdout,stderr 標準輸出

這種形式的日志輸出我們可以直接使用docker logs查看日志， k8s 集群中同樣集群可以使用kubectl logs類似的形式查看日志。

日志文件記錄

這種日志輸出我們無法從以上方法查看日志內容，只能tail日志文件查看。

不論你的業務容器日志如何輸出,都是可以使用統一的日志收集器收集。常見的日志收集方式：

k8s 集群

集群啟動時會在每個機器啟動一個Fluentd agent收集日志然后發送給 Elasticsearch。實現方式是每個agent掛載目錄/var/lib/docker/containers使用fluentd的tail插件掃描每個容器日志文件，直接發送給Elasticsearch。

Fluentd agent起在業務同一個 pod 中共享 volume 然后實現對日志文件的收集發送給Elasticsearch。

docker swarm 集群

swarm 目前暫時沒有提供日志查看機制。但是docker cloud提供了與kubectrl logs類似的機制查看 stdout 的日志。目前還沒有 fluentd 插件直接對服務進行日志收集，暫時考慮直接使用使用跟容器一樣的機制收集。docker service create 支持--log-driver

docker 容器

從 docker1.8 內置了fluentd log driver 。以如下的形式啟動容器，容器 stdout/stderr 日志將發往配置的 fluentd 。如果配置后，docker logs將無法使用。另外默認模式下如果你配置得地址沒有正常服務，容器無法啟動。你也可以使用fluentd-async-connect形式啟動， docker daemon 則能在后臺嘗試連接并緩存日志。

`docker run --log-driver=fluentd --log-opt fluentd-address=myhost.local:24224
`
同樣如果是日志文件，將文件暴露出來直接使用 fluentd 收集。

# /container/container.go:63
type CommonContainer struct {
    StreamConfig *stream.Config
  ...
}
# /container/stream/streams.go:26
type Config struct {
    sync.WaitGroup
    stdout    *broadcaster.Unbuffered
    stderr    *broadcaster.Unbuffered
    stdin     io.ReadCloser
    stdinPipe io.WriteCloser
}

moby源碼來看,每一個container實例都有幾個屬性stdout,stderr,stdin,以及管道stdinPipe(當容器使用-i參數啟動時標準輸入將被運行，daemon將能夠使用此管道向容器內寫入標準輸入).

那么針對如上的實例該如何實現日志收集轉發？

# /container/container.go:312
func (container *Container) StartLogger(cfg containertypes.LogConfig) (logger.Logger, error) {
    c, err := logger.GetLogDriver(cfg.Type)
    if err != nil {
        return nil, fmt.Errorf("Failed to get logging factory: %v", err)
    }
    ctx := logger.Context{
        Config:              cfg.Config,
        ContainerID:         container.ID,
        ContainerName:       container.Name,
        ContainerEntrypoint: container.Path,
        ContainerArgs:       container.Args,
        ContainerImageID:    container.ImageID.String(),
        ContainerImageName:  container.Config.Image,
        ContainerCreated:    container.Created,
        ContainerEnv:        container.Config.Env,
        ContainerLabels:     container.Config.Labels,
        DaemonName:          "docker",
    }

    // Set logging file for "json-logger"
    if cfg.Type == jsonfilelog.Name {
        ctx.LogPath, err = container.GetRootResourcePath(fmt.Sprintf("%s-json.log", container.ID))
        if err != nil {
            return nil, err
        }
    }
    return c(ctx)
}
#/container/container.go:978
func (container *Container) startLogging() error {
    if container.HostConfig.LogConfig.Type == "none" {
        return nil // do not start logging routines
    }

    l, err := container.StartLogger(container.HostConfig.LogConfig)
    if err != nil {
        return fmt.Errorf("Failed to initialize logging driver: %v", err)
    }

    copier := logger.NewCopier(map[string]io.Reader{"stdout": container.StdoutPipe(), "stderr": container.StderrPipe()}, l)
    container.LogCopier = copier
    copier.Run()
    container.LogDriver = l

    // set LogPath field only for json-file logdriver
    if jl, ok := l.(*jsonfilelog.JSONFileLogger); ok {
        container.LogPath = jl.LogPath()
    }

    return nil
}

第一個方法是為container查找log-driver。首先根據容器配置的log-driver類別調用：logger.GetLogDriver(cfg.Type)返回一個方法類型：

/daemon/logger/factory.go:9
type Creator func(Context) (Logger, error)

實質就是從工廠類注冊的logdriver插件去查找，具體源碼下文分析。獲取到c方法后構建調用參數具體就是容器的一些信息。然后使用調用c方法返回driver。driver是個接口類型，我們看看有哪些方法：

# /daemon/logger/logger.go:61
type Logger interface {
    Log(*Message) error
    Name() string
    Close() error
}

很簡單的三個方法，也很容易理解,Log()發送日志消息到driver,Close()進行關閉操作（根據不同實現）。
也就是說我們自己實現一個logdriver，只需要實現如上三個方法，然后注冊到logger工廠類中即可。下面我們來看/daemon/logger/factory.go

第二個方法就是處理日志了，獲取到日志driver,在創建一個Copier,顧名思義就是復制日志，分別從stdout 和stderr復制到logger driver。下面看看具體關鍵實現：

#/daemon/logger/copir.go:41
func (c *Copier) copySrc(name string, src io.Reader) {
    defer c.copyJobs.Done()
    reader := bufio.NewReader(src)

    for {
        select {
        case <-c.closed:
            return
        default:
            line, err := reader.ReadBytes("
")
            line = bytes.TrimSuffix(line, []byte{"
"})

            // ReadBytes can return full or partial output even when it failed.
            // e.g. it can return a full entry and EOF.
            if err == nil || len(line) > 0 {
                if logErr := c.dst.Log(&Message{Line: line, Source: name, Timestamp: time.Now().UTC()}); logErr != nil {
                    logrus.Errorf("Failed to log msg %q for logger %s: %s", line, c.dst.Name(), logErr)
                }
            }

            if err != nil {
                if err != io.EOF {
                    logrus.Errorf("Error scanning log stream: %s", err)
                }
                return
            }
        }
    }
}

每讀取一行數據，構建一個消息，調用logdriver的log方法發送到driver處理。

日志driver注冊器

位于/daemon/logger/factory.go的源碼實現即時日志driver的注冊器，其中幾個重要的方法（上文已經提到一個）:

# /daemon/logger/factory.go:21
func (lf *logdriverFactory) register(name string, c Creator) error {
    if lf.driverRegistered(name) {
        return fmt.Errorf("logger: log driver named "%s" is already registered", name)
    }

    lf.m.Lock()
    lf.registry[name] = c
    lf.m.Unlock()
    return nil
}
# /daemon/logger/factory.go:39
func (lf *logdriverFactory) registerLogOptValidator(name string, l LogOptValidator) error {
    lf.m.Lock()
    defer lf.m.Unlock()

    if _, ok := lf.optValidator[name]; ok {
        return fmt.Errorf("logger: log validator named "%s" is already registered", name)
    }
    lf.optValidator[name] = l
    return nil
}

看起來很簡單，就是將一個Creator方法類型添加到一個map結構中，將LogOptValidator添加到另一個map這里注意加鎖的操作。

#/daemon/logger/factory.go:13
type LogOptValidator func(cfg map[string]string) error

這個主要是驗證driver的參數 ,dockerd和docker啟動參數中有：--log-opt

使用自己實現的 zeroMQ-driver 直接將容器日志通過 0MQ 發到日志統一處理中心。在處理中心統一完成下一步處理。如果平臺用戶需要將日志向外輸出或者直接對接平臺內日志分析應用，我們的處理是在應用 pod 中啟動日志收集插件容器（封裝擴展的 fluentd ）,根據用戶的需要配置日志出口，實現應用級日志收集。容器日志首先是由 docker-daemon 收集到，再根據容器 log-driver 配置進行相應操作，也就是說如果你的宿主機網絡與容器網絡不通(k8s 集群)，日志從宿主機到 pod 中的收集容器只有兩種方式：走外層網絡，文件掛載。 我們采用文件掛載方式。
以zmq-driver為例講講我們怎么實現自己的driver。直接接收容器的日志。

//定義一個struct,這里包含一個zmq套接字
type ZmqLogger struct {
    writer      *zmq.Socket
    containerId string
    tenantId    string
    serviceId   string
    felock      sync.Mutex
}
//定義init方法調用logger注冊器的方法注冊當前driver
//和參數驗證方法。
func init() {
    if err := logger.RegisterLogDriver(name, New); err != nil {
        logrus.Fatal(err)
    }
    if err := logger.RegisterLogOptValidator(name, ValidateLogOpt); err != nil {
        logrus.Fatal(err)
    }
}
//實現一個上文提到的Creator方法注冊logdriver.
//這里新建一個zmq套接字構建一個實例
func New(ctx logger.Context) (logger.Logger, error) {
    zmqaddress := ctx.Config[zmqAddress]

    puber, err := zmq.NewSocket(zmq.PUB)
    if err != nil {
        return nil, err
    }
    var (
        env       = make(map[string]string)
        tenantId  string
        serviceId string
    )
    for _, pair := range ctx.ContainerEnv {
        p := strings.SplitN(pair, "=", 2)
        //logrus.Errorf("ContainerEnv pair: %s", pair)
        if len(p) == 2 {
            key := p[0]
            value := p[1]
            env[key] = value
        }
    }
    tenantId = env["TENANT_ID"]
    serviceId = env["SERVICE_ID"]

    if tenantId == "" {
        tenantId = "default"
    }

    if serviceId == "" {
        serviceId = "default"
    }

    puber.Connect(zmqaddress)

    return &ZmqLogger{
        writer:      puber,
        containerId: ctx.ID(),
        tenantId:    tenantId,
        serviceId:   serviceId,
        felock:      sync.Mutex{},
    }, nil
}
//實現Log方法，這里使用zmq socket發送日志消息
//這里必須注意，zmq socket是線程不安全的，我們知道
//本方法可能被兩個線程(復制stdout和膚質stderr)調用//必須使用鎖保證線程安全。否則會發生錯誤。
func (s *ZmqLogger) Log(msg *logger.Message) error {
    s.felock.Lock()
    defer s.felock.Unlock()
    s.writer.Send(s.tenantId, zmq.SNDMORE)
    s.writer.Send(s.serviceId, zmq.SNDMORE)
    if msg.Source == "stderr" {
        s.writer.Send(s.containerId+": "+string(msg.Line), zmq.DONTWAIT)
    } else {
        s.writer.Send(s.containerId+": "+string(msg.Line), zmq.DONTWAIT)
    }
    return nil
}
//實現Close方法，這里用來關閉zmq socket。
//同樣注意線程安全，調用此方法的是容器關閉協程。
func (s *ZmqLogger) Close() error {
    s.felock.Lock()
    defer s.felock.Unlock()
    if s.writer != nil {
        return s.writer.Close()
    }
    return nil
}

func (s *ZmqLogger) Name() string {
    return name
}
//驗證參數的方法，我們使用參數傳入zmq pub的地址。
func ValidateLogOpt(cfg map[string]string) error {
    for key := range cfg {
        switch key {
        case zmqAddress:
        default:
            return fmt.Errorf("unknown log opt "%s" for %s log driver", key, name)
        }
    }
    if cfg[zmqAddress] == "" {
        return fmt.Errorf("must specify a value for log opt "%s"", zmqAddress)
    }
    return nil
}