// RAINBOND, Application Management Platform // Copyright (C) 2014-2017 Goodrain Co., Ltd. // This program is free software: you can redistribute it and/or modify // it under the terms of the GNU General Public License as published by // the Free Software Foundation, either version 3 of the License, or // (at your option) any later version. For any non-GPL usage of Rainbond, // one or multiple Commercial Licenses authorized by Goodrain Co., Ltd. // must be obtained first. // This program is distributed in the hope that it will be useful, // but WITHOUT ANY WARRANTY; without even the implied warranty of // MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the // GNU General Public License for more details. // You should have received a copy of the GNU General Public License // along with this program. If not, see . package logger import ( "context" "encoding/json" "fmt" "io" "strings" "sync" "time" "github.com/Sirupsen/logrus" "github.com/docker/docker/api/types" "github.com/docker/docker/api/types/events" "github.com/docker/docker/api/types/filters" "github.com/goodrain/rainbond/cmd/node/option" ) //RFC3339NanoFixed time format var RFC3339NanoFixed = "2006-01-02T15:04:05.000000000Z07:00" //ContainerLogManage container log manage type ContainerLogManage struct { ctx context.Context cancel context.CancelFunc conf *option.Conf cchan chan ContainerEvent containerLogs sync.Map } //CreatContainerLogManage create a container log manage func CreatContainerLogManage(conf *option.Conf) *ContainerLogManage { ctx, cancel := context.WithCancel(context.Background()) return &ContainerLogManage{ ctx: ctx, cancel: cancel, conf: conf, cchan: make(chan ContainerEvent, 100), } } //Start start func (c *ContainerLogManage) Start() error { errchan := make(chan error) go func() { if err := <-errchan; err != nil { logrus.Errorf(err.Error()) } }() go c.handleLogger() go c.listAndWatchContainer(errchan) go c.loollist() logrus.Infof("start container log manage success") return nil } //Stop stop all logger func (c *ContainerLogManage) Stop() { c.containerLogs.Range(func(k, v interface{}) bool { cl := v.(*ContainerLog) cl.Stop() return true }) } func (c *ContainerLogManage) getContainerLogByFile(info types.ContainerJSON) (*LogFile, error) { return nil, nil } func (c *ContainerLogManage) getContainerLogReader(ctx context.Context, containerID string) (io.ReadCloser, io.ReadCloser, error) { stderr, err := c.conf.DockerCli.ContainerLogs(ctx, containerID, types.ContainerLogsOptions{ ShowStderr: true, Since: time.Now().Format(RFC3339NanoFixed), Follow: true, }) if err != nil { return nil, nil, fmt.Errorf("Open container stderr output failure,%s", err.Error()) } stdout, err := c.conf.DockerCli.ContainerLogs(ctx, containerID, types.ContainerLogsOptions{ ShowStdout: true, Since: time.Now().Format(RFC3339NanoFixed), Follow: true, }) if err != nil { return nil, nil, fmt.Errorf("Open container stdout output failure,%s", err.Error()) } return stdout, stderr, nil } func (c *ContainerLogManage) handleLogger() { for { select { case <-c.ctx.Done(): return case cevent := <-c.cchan: switch cevent.Action { case "start": loggerType := cevent.Container.HostConfig.LogConfig.Type if loggerType != "json-file" && loggerType != "syslog" { continue } if logger, ok := c.containerLogs.Load(cevent.Container.ID); ok { clog, okf := logger.(*ContainerLog) if okf { clog.Restart() logrus.Infof("restart copy container log for container %s", cevent.Container.Name) } } else { go func() { retry := 0 for retry < maxJSONDecodeRetry { retry++ reader, err := NewLogFile(cevent.Container.LogPath, 2, false, decodeFunc, 0640, getTailReader) if err != nil { logrus.Errorf("create logger failure %s", err.Error()) time.Sleep(time.Second * 1) continue } clog := createContainerLog(c.ctx, cevent.Container, reader) if err := clog.StartLogging(); err != nil { clog.Stop() if err == ErrNeglectedContainer { return } logrus.Errorf("start copy docker log failure %s", err.Error()) time.Sleep(time.Second * 1) //retry get container inspect info cevent.Container, _ = c.getContainer(cevent.Container.ID) continue } c.containerLogs.Store(cevent.Container.ID, clog) logrus.Infof("start copy container log for container %s", cevent.Container.Name) return } }() } case "stop": if logger, ok := c.containerLogs.Load(cevent.Container.ID); ok { clog, okf := logger.(*ContainerLog) if okf { clog.Stop() logrus.Infof("stop copy container log for container %s", cevent.Container.Name) } } case "die": if logger, ok := c.containerLogs.Load(cevent.Container.ID); ok { clog, okf := logger.(*ContainerLog) if okf { clog.Stop() } c.containerLogs.Delete(cevent.Container.ID) logrus.Infof("remove copy container log for container %s", cevent.Container.Name) } } } } } //ContainerEvent container event type ContainerEvent struct { Action string Container types.ContainerJSON } func (c *ContainerLogManage) cacheContainer(cs ...ContainerEvent) { for _, container := range cs { logrus.Debugf("found a container %s %s", container.Container.Name, container.Action) c.cchan <- container } } func (c *ContainerLogManage) listContainer() []types.Container { lictctx, cancel := context.WithTimeout(c.ctx, time.Second*60) defer cancel() containers, err := c.conf.DockerCli.ContainerList(lictctx, types.ContainerListOptions{}) if err != nil { logrus.Errorf("list containers failure.%s", err.Error()) containers, _ = c.conf.DockerCli.ContainerList(lictctx, types.ContainerListOptions{}) } return containers } func (c *ContainerLogManage) loollist() { ticker := time.NewTicker(time.Minute * 10) for { select { case <-c.ctx.Done(): ticker.Stop() return case <-ticker.C: for _, container := range c.listContainer() { cj, _ := c.getContainer(container.ID) if cj.ContainerJSONBase == nil || cj.HostConfig == nil || cj.HostConfig.LogConfig.Type == "" { continue } loggerType := cj.HostConfig.LogConfig.Type if loggerType != "json-file" && loggerType != "syslog" { continue } if _, exist := c.containerLogs.Load(container.ID); !exist { c.cacheContainer(ContainerEvent{Action: "start", Container: cj}) } } } } } func (c *ContainerLogManage) listAndWatchContainer(errchan chan error) { containers := c.listContainer() for _, con := range containers { container, err := c.getContainer(con.ID) if err != nil { if !strings.Contains(err.Error(), "No such container") { logrus.Errorf("get container detail info failure %s", err.Error()) } // The log path cannot be obtained if the container details cannot be obtained continue } c.cacheContainer(ContainerEvent{Action: "start", Container: container}) } logrus.Info("list containers complete, start watch container") for { if err := c.watchContainer(); err != nil { logrus.Errorf("watch container error %s, will retry", err.Error()) } time.Sleep(time.Second * 1) } } func (c *ContainerLogManage) watchContainer() error { containerFileter := filters.NewArgs() containerFileter.Add("type", "container") eventchan, eventerrchan := c.conf.DockerCli.Events(c.ctx, types.EventsOptions{ Filters: containerFileter, }) for { select { case <-c.ctx.Done(): return nil case err := <-eventerrchan: return err case event, ok := <-eventchan: if !ok { return fmt.Errorf("event chan is closed") } if event.Type == events.ContainerEventType && checkEventAction(event.Action) { container, err := c.getContainer(event.ID) if err != nil { if !strings.Contains(err.Error(), "No such container") { logrus.Errorf("get container detail info failure %s", err.Error()) } break } c.cacheContainer(ContainerEvent{Action: event.Action, Container: container}) } } } } func (c *ContainerLogManage) getContainer(containerID string) (types.ContainerJSON, error) { ctx, cancel := context.WithTimeout(c.ctx, time.Second*5) defer cancel() return c.conf.DockerCli.ContainerInspect(ctx, containerID) } var handleAction = []string{"start", "stop", "die"} func checkEventAction(action string) bool { for _, enable := range handleAction { if enable == action { return true } } return false } func createContainerLog(ctx context.Context, container types.ContainerJSON, reader *LogFile) *ContainerLog { cctx, cancel := context.WithCancel(ctx) return &ContainerLog{ ctx: cctx, cancel: cancel, ContainerJSON: container, reader: reader, } } //ContainerLog container log struct type ContainerLog struct { ctx context.Context cancel context.CancelFunc types.ContainerJSON LogCopier *Copier LogDriver []Logger reader *LogFile since time.Time stoped *bool } //StartLogging start copy log func (container *ContainerLog) StartLogging() error { loggers, err := container.startLogger() if err != nil { if err == ErrNeglectedContainer { logrus.Warnf("find a container %s that do not define rainbond logger.", container.Name) return ErrNeglectedContainer } return fmt.Errorf("failed to initialize logging driver: %v", err) } copier := NewCopier(container.reader, loggers, container.since) container.LogCopier = copier copier.Run() container.LogDriver = loggers return nil } //ContainerLoggerConfig logger config type ContainerLoggerConfig struct { Name string Options map[string]string } func getLoggerConfig(envs []string) []*ContainerLoggerConfig { var configs = make(map[string]*ContainerLoggerConfig) var envMap = make(map[string]string, len(envs)) for _, v := range envs { info := strings.SplitN(v, "=", 2) if len(info) > 1 { envMap[strings.ToLower(info[0])] = info[1] if strings.HasPrefix(info[0], "LOGGER_DRIVER_NAME") { if _, exist := configs[info[1]]; !exist { configs[info[1]] = &ContainerLoggerConfig{ Name: info[1], } } } } } var re []*ContainerLoggerConfig for i, c := range configs { if config, ok := envMap[strings.ToLower("LOGGER_DRIVER_OPT_"+c.Name)]; ok { var options = make(map[string]string) json.Unmarshal([]byte(config), &options) configs[i].Options = options } re = append(re, configs[i]) } return re } //ErrNeglectedContainer not define logger name var ErrNeglectedContainer = fmt.Errorf("Neglected container") // startLogger starts a new logger driver for the container. func (container *ContainerLog) startLogger() ([]Logger, error) { configs := getLoggerConfig(container.Config.Env) var loggers []Logger for _, config := range configs { initDriver, err := GetLogDriver(config.Name) if err != nil { logrus.Warnf("get container log driver failure %s", err.Error()) continue } createTime, _ := time.Parse(RFC3339NanoFixed, container.Created) info := Info{ Config: config.Options, ContainerID: container.ID, ContainerName: container.Name, ContainerEntrypoint: container.Path, ContainerArgs: container.Args, ContainerImageName: container.Config.Image, ContainerCreated: createTime, ContainerEnv: container.Config.Env, ContainerLabels: container.Config.Labels, DaemonName: "docker", } l, err := initDriver(info) if err != nil { logrus.Warnf("init container log driver failure %s", err.Error()) continue } loggers = append(loggers, l) } if len(loggers) == 0 { return nil, ErrNeglectedContainer } return loggers, nil } //Restart restart func (container *ContainerLog) Restart() { if *container.stoped { copier := NewCopier(container.reader, container.LogDriver, container.since) container.LogCopier = copier copier.Run() } } //Stop stop copy container log func (container *ContainerLog) Stop() { if container.LogCopier != nil { container.LogCopier.Close() } container.since = time.Now() var containerLogStop = true container.stoped = &containerLogStop logrus.Debugf("rainbond logger stop for container %s", container.Name) } //Close close func (container *ContainerLog) Close() { if container.LogCopier != nil { container.LogCopier.Close() } container.cancel() logrus.Debugf("rainbond logger close for container %s", container.Name) }