moby/daemon/logs.go

package daemon // import "github.com/docker/docker/daemon"

import (
	"context"
	"strconv"
	"time"

	"github.com/containerd/log"
	"github.com/docker/docker/api/types/backend"
	containertypes "github.com/docker/docker/api/types/container"
	timetypes "github.com/docker/docker/api/types/time"
	"github.com/docker/docker/container"
	"github.com/docker/docker/daemon/config"
	"github.com/docker/docker/daemon/logger"
	logcache "github.com/docker/docker/daemon/logger/loggerutils/cache"
	"github.com/docker/docker/errdefs"
	"github.com/pkg/errors"
)

// ContainerLogs copies the container's log channel to the channel provided in
// the config. If ContainerLogs returns an error, no messages have been copied.
// and the channel will be closed without data.
//
// if it returns nil, the config channel will be active and return log
// messages until it runs out or the context is canceled.
func (daemon *Daemon) ContainerLogs(ctx context.Context, containerName string, config *containertypes.LogsOptions) (messages <-chan *backend.LogMessage, isTTY bool, retErr error) {
	lg := log.G(ctx).WithFields(log.Fields{
		"module":    "daemon",
		"method":    "(*Daemon).ContainerLogs",
		"container": containerName,
	})

	if !(config.ShowStdout || config.ShowStderr) {
		return nil, false, errdefs.InvalidParameter(errors.New("You must choose at least one stream"))
	}
	ctr, err := daemon.GetContainer(containerName)
	if err != nil {
		return nil, false, err
	}

	if ctr.RemovalInProgress || ctr.Dead {
		return nil, false, errdefs.Conflict(errors.New("can not get logs from container which is dead or marked for removal"))
	}

	if ctr.HostConfig.LogConfig.Type == "none" {
		return nil, false, logger.ErrReadLogsNotSupported{}
	}

	cLog, cLogCreated, err := daemon.getLogger(ctr)
	if err != nil {
		return nil, false, err
	}
	if cLogCreated {
		defer func() {
			if retErr != nil {
				if err = cLog.Close(); err != nil {
					log.G(ctx).Errorf("Error closing logger: %v", err)
				}
			}
		}()
	}

	logReader, ok := cLog.(logger.LogReader)
	if !ok {
		return nil, false, logger.ErrReadLogsNotSupported{}
	}

	follow := config.Follow && !cLogCreated
	tailLines, err := strconv.Atoi(config.Tail)
	if err != nil {
		tailLines = -1
	}

	var since time.Time
	if config.Since != "" {
		s, n, err := timetypes.ParseTimestamps(config.Since, 0)
		if err != nil {
			return nil, false, err
		}
		since = time.Unix(s, n)
	}

	var until time.Time
	if config.Until != "" && config.Until != "0" {
		s, n, err := timetypes.ParseTimestamps(config.Until, 0)
		if err != nil {
			return nil, false, err
		}
		until = time.Unix(s, n)
	}

	readConfig := logger.ReadConfig{
		Since:  since,
		Until:  until,
		Tail:   tailLines,
		Follow: follow,
	}

	logs := logReader.ReadLogs(readConfig)

	// past this point, we can't possibly return any errors, so we can just
	// start a goroutine and return to tell the caller not to expect errors
	// (if the caller wants to give up on logs, they have to cancel the context)
	// this goroutine functions as a shim between the logger and the caller.
	messageChan := make(chan *backend.LogMessage, 1)
	go func() {
		if cLogCreated {
			defer func() {
				if err = cLog.Close(); err != nil {
					log.G(ctx).Errorf("Error closing logger: %v", err)
				}
			}()
		}
		// signal that the log reader is gone
		defer logs.ConsumerGone()

		// close the messages channel. closing is the only way to signal above
		// that we're doing with logs (other than context cancel i guess).
		defer close(messageChan)

		lg.Debug("begin logs")
		defer lg.Debugf("end logs (%v)", ctx.Err())

		for {
			select {
			// i do not believe as the system is currently designed any error
			// is possible, but we should be prepared to handle it anyway. if
			// we do get an error, copy only the error field to a new object so
			// we don't end up with partial data in the other fields
			case err := <-logs.Err:
				lg.Errorf("Error streaming logs: %v", err)
				select {
				case <-ctx.Done():
				case messageChan <- &backend.LogMessage{Err: err}:
				}
				return
			case <-ctx.Done():
				return
			case msg, ok := <-logs.Msg:
				// there is some kind of pool or ring buffer in the logger that
				// produces these messages, and a possible future optimization
				// might be to use that pool and reuse message objects
				if !ok {
					return
				}
				m := msg.AsLogMessage() // just a pointer conversion, does not copy data

				// there could be a case where the reader stops accepting
				// messages and the context is canceled. we need to check that
				// here, or otherwise we risk blocking forever on the message
				// send.
				select {
				case <-ctx.Done():
					return
				case messageChan <- m:
				}
			}
		}
	}()
	return messageChan, ctr.Config.Tty, nil
}

func (daemon *Daemon) getLogger(container *container.Container) (l logger.Logger, created bool, err error) {
	container.Lock()
	if container.State.Running {
		l = container.LogDriver
	}
	container.Unlock()
	if l == nil {
		created = true
		l, err = container.StartLogger()
	}
	return
}

// mergeAndVerifyLogConfig merges the daemon log config to the container's log config if the container's log driver is not specified.
func (daemon *Daemon) mergeAndVerifyLogConfig(cfg *containertypes.LogConfig) error {
	if cfg.Type == "" {
		cfg.Type = daemon.defaultLogConfig.Type
	}

	if cfg.Config == nil {
		cfg.Config = make(map[string]string)
	}

	if cfg.Type == daemon.defaultLogConfig.Type {
		for k, v := range daemon.defaultLogConfig.Config {
			if _, ok := cfg.Config[k]; !ok {
				cfg.Config[k] = v
			}
		}
	}

	logcache.MergeDefaultLogConfig(cfg.Config, daemon.defaultLogConfig.Config)

	return logger.ValidateLogOpts(cfg.Type, cfg.Config)
}

func defaultLogConfig(cfg *config.Config) (containertypes.LogConfig, error) {
	if len(cfg.LogConfig.Config) > 0 {
		if err := logger.ValidateLogOpts(cfg.LogConfig.Type, cfg.LogConfig.Config); err != nil {
			return containertypes.LogConfig{}, errors.Wrap(err, "failed to set log opts")
		}
	}
	return containertypes.LogConfig{
		Type:   cfg.LogConfig.Type,
		Config: cfg.LogConfig.Config,
	}, nil
}