Merge pull request #2709 from haircommander/journald

Add libpod journald logging

8 files changed, 219 insertions, 7 deletions

diff --git a/libpod/container.go b/libpod/container.go
index c07f4c78d..c8ab42fc3 100644
--- a/libpod/container.go
+++ b/libpod/container.go
@@ -51,6 +51,15 @@ const CgroupfsDefaultCgroupParent = "/libpod_parent"
 // manager in libpod
 const SystemdDefaultCgroupParent = "machine.slice"
 
+// JournaldLogging is the string conmon expects to specify journald logging
+const JournaldLogging = "journald"
+
+// KubernetesLogging is the string conmon expects when specifying to use the kubernetes logging format
+const KubernetesLogging = "k8s-file"
+
+// JSONLogging is the string conmon expects when specifying to use the json logging format
+const JSONLogging = "json-file"
+
 // DefaultWaitInterval is the default interval between container status checks
 // while waiting.
 const DefaultWaitInterval = 250 * time.Millisecond
@@ -368,6 +377,8 @@ type ContainerConfig struct {
 	CgroupParent string `json:"cgroupParent"`
 	// LogPath log location
 	LogPath string `json:"logPath"`
+	// LogDriver driver for logs
+	LogDriver string `json:"logDriver"`
 	// File containing the conmon PID
 	ConmonPidFile string `json:"conmonPidFile,omitempty"`
 	// RestartPolicy indicates what action the container will take upon
@@ -775,6 +786,11 @@ func (c *Container) RestartRetries() uint {
 	return c.config.RestartRetries
 }
 
+// LogDriver returns the log driver for this container
+func (c *Container) LogDriver() string {
+	return c.config.LogDriver
+}
+
 // RuntimeName returns the name of the runtime
 func (c *Container) RuntimeName() string {
 	return c.runtime.ociRuntime.name
diff --git a/libpod/container_log.go b/libpod/container_log.go
index e998ad316..374e5a1fc 100644
--- a/libpod/container_log.go
+++ b/libpod/container_log.go
@@ -19,6 +19,13 @@ const (
 	// zeroes are not trimmed, taken from
 	// https://github.com/golang/go/issues/19635
 	logTimeFormat = "2006-01-02T15:04:05.000000000Z07:00"
+
+	// partialLogType signifies a log line that exceeded the buffer
+	// length and needed to spill into a new line
+	partialLogType = "P"
+
+	// fullLogType signifies a log line is full
+	fullLogType = "F"
 )
 
 // LogOptions is the options you can use for logs
@@ -53,6 +60,15 @@ func (r *Runtime) Log(containers []*Container, options *LogOptions, logChannel c
 
 // ReadLog reads a containers log based on the input options and returns loglines over a channel
 func (c *Container) ReadLog(options *LogOptions, logChannel chan *LogLine) error {
+	// TODO Skip sending logs until journald logs can be read
+	// TODO make this not a magic string
+	if c.LogDriver() == JournaldLogging {
+		return c.readFromJournal(options, logChannel)
+	}
+	return c.readFromLogFile(options, logChannel)
+}
+
+func (c *Container) readFromLogFile(options *LogOptions, logChannel chan *LogLine) error {
 	t, tailLog, err := getLogFile(c.LogPath(), options)
 	if err != nil {
 		// If the log file does not exist, this is not fatal.
@@ -191,7 +207,7 @@ func newLogLine(line string) (*LogLine, error) {
 	if len(splitLine) < 4 {
 		return nil, errors.Errorf("'%s' is not a valid container log line", line)
 	}
-	logTime, err := time.Parse(time.RFC3339Nano, splitLine[0])
+	logTime, err := time.Parse(logTimeFormat, splitLine[0])
 	if err != nil {
 		return nil, errors.Wrapf(err, "unable to convert time %s from container log", splitLine[0])
 	}
@@ -206,7 +222,7 @@ func newLogLine(line string) (*LogLine, error) {
 
 // Partial returns a bool if the log line is a partial log type
 func (l *LogLine) Partial() bool {
-	if l.ParseLogType == "P" {
+	if l.ParseLogType == partialLogType {
 		return true
 	}
 	return false
diff --git a/libpod/container_log_linux.go b/libpod/container_log_linux.go
new file mode 100644
index 000000000..e549673a6
--- /dev/null
+++ b/libpod/container_log_linux.go
@@ -0,0 +1,143 @@
+//+build linux
+//+build systemd
+
+package libpod
+
+import (
+	"fmt"
+	"io"
+	"strings"
+	"time"
+
+	journal "github.com/coreos/go-systemd/sdjournal"
+	"github.com/pkg/errors"
+	"github.com/sirupsen/logrus"
+)
+
+const (
+	// journaldLogOut is the journald priority signifying stdout
+	journaldLogOut = "6"
+
+	// journaldLogErr is the journald priority signifying stderr
+	journaldLogErr = "3"
+
+	// bufLen is the length of the buffer to read from a k8s-file
+	// formatted log line
+	// let's set it as 2k just to be safe if k8s-file format ever changes
+	bufLen = 16384
+)
+
+func (c *Container) readFromJournal(options *LogOptions, logChannel chan *LogLine) error {
+	var config journal.JournalReaderConfig
+	config.NumFromTail = options.Tail
+	config.Formatter = journalFormatter
+	defaultTime := time.Time{}
+	if options.Since != defaultTime {
+		// coreos/go-systemd/sdjournal doesn't correctly handle requests for data in the future
+		// return nothing instead of fasely printing
+		if time.Now().Before(options.Since) {
+			return nil
+		}
+		config.Since = time.Since(options.Since)
+	}
+	config.Matches = append(config.Matches, journal.Match{
+		Field: "CONTAINER_ID_FULL",
+		Value: c.ID(),
+	})
+	options.WaitGroup.Add(1)
+
+	r, err := journal.NewJournalReader(config)
+	if err != nil {
+		return err
+	}
+	if r == nil {
+		return errors.Errorf("journal reader creation failed")
+	}
+	if options.Tail == 0 {
+		r.Rewind()
+	}
+
+	if options.Follow {
+		go func() {
+			follower := FollowBuffer{logChannel}
+			err := r.Follow(nil, follower)
+			if err != nil {
+				logrus.Debugf(err.Error())
+			}
+			r.Close()
+			options.WaitGroup.Done()
+			return
+		}()
+		return nil
+	}
+
+	go func() {
+		bytes := make([]byte, bufLen)
+		// /me complains about no do-while in go
+		ec, err := r.Read(bytes)
+		for ec != 0 && err == nil {
+			// because we are reusing bytes, we need to make
+			// sure the old data doesn't get into the new line
+			bytestr := string(bytes[:ec])
+			logLine, err2 := newLogLine(bytestr)
+			if err2 != nil {
+				logrus.Error(err2)
+				continue
+			}
+			logChannel <- logLine
+			ec, err = r.Read(bytes)
+		}
+		if err != nil && err != io.EOF {
+			logrus.Error(err)
+		}
+		r.Close()
+		options.WaitGroup.Done()
+	}()
+	return nil
+}
+
+func journalFormatter(entry *journal.JournalEntry) (string, error) {
+	usec := entry.RealtimeTimestamp
+	tsString := time.Unix(0, int64(usec)*int64(time.Microsecond)).Format(logTimeFormat)
+	output := fmt.Sprintf("%s ", tsString)
+	priority, ok := entry.Fields["PRIORITY"]
+	if !ok {
+		return "", errors.Errorf("no PRIORITY field present in journal entry")
+	}
+	if priority == journaldLogOut {
+		output += "stdout "
+	} else if priority == journaldLogErr {
+		output += "stderr "
+	} else {
+		return "", errors.Errorf("unexpected PRIORITY field in journal entry")
+	}
+
+	// if CONTAINER_PARTIAL_MESSAGE is defined, the log type is "P"
+	if _, ok := entry.Fields["CONTAINER_PARTIAL_MESSAGE"]; ok {
+		output += fmt.Sprintf("%s ", partialLogType)
+	} else {
+		output += fmt.Sprintf("%s ", fullLogType)
+	}
+
+	// Finally, append the message
+	msg, ok := entry.Fields["MESSAGE"]
+	if !ok {
+		return "", fmt.Errorf("no MESSAGE field present in journal entry")
+	}
+	output += strings.TrimSpace(msg)
+	return output, nil
+}
+
+type FollowBuffer struct {
+	logChannel chan *LogLine
+}
+
+func (f FollowBuffer) Write(p []byte) (int, error) {
+	bytestr := string(p)
+	logLine, err := newLogLine(bytestr)
+	if err != nil {
+		return -1, err
+	}
+	f.logChannel <- logLine
+	return len(p), nil
+}
diff --git a/libpod/container_log_unsupported.go b/libpod/container_log_unsupported.go
new file mode 100644
index 000000000..0ec5740e2
--- /dev/null
+++ b/libpod/container_log_unsupported.go
@@ -0,0 +1,11 @@
+//+build !linux !systemd
+
+package libpod
+
+import (
+	"github.com/pkg/errors"
+)
+
+func (c *Container) readFromJournal(options *LogOptions, logChannel chan *LogLine) error {
+	return errors.Wrapf(ErrOSNotSupported, "Journald logging only enabled with systemd on linux")
+}
diff --git a/libpod/oci.go b/libpod/oci.go
index abc6214b9..7138108c5 100644
--- a/libpod/oci.go
+++ b/libpod/oci.go
@@ -367,8 +367,6 @@ func (r *OCIRuntime) execContainer(c *Container, cmd, capAdd, env []string, tty
 	args := []string{}
 
 	// TODO - should we maintain separate logpaths for exec sessions?
-	args = append(args, "--log", c.LogPath())
-
 	args = append(args, "exec")
 
 	if cwd != "" {
@@ -402,7 +400,7 @@ func (r *OCIRuntime) execContainer(c *Container, cmd, capAdd, env []string, tty
 		args = append(args, "--env", envVar)
 	}
 
-	// Append container ID and command
+	// Append container ID, name and command
 	args = append(args, c.ID())
 	args = append(args, cmd...)
 
diff --git a/libpod/oci_linux.go b/libpod/oci_linux.go
index 1c1e4a203..7c1c18052 100644
--- a/libpod/oci_linux.go
+++ b/libpod/oci_linux.go
@@ -214,10 +214,10 @@ func (r *OCIRuntime) createOCIContainer(ctr *Container, cgroupParent string, res
 	}
 	args = append(args, "-c", ctr.ID())
 	args = append(args, "-u", ctr.ID())
+	args = append(args, "-n", ctr.Name())
 	args = append(args, "-r", r.path)
 	args = append(args, "-b", ctr.bundlePath())
 	args = append(args, "-p", filepath.Join(ctr.state.RunDir, "pidfile"))
-	args = append(args, "-l", ctr.LogPath())
 	args = append(args, "--exit-dir", r.exitsDir)
 	if ctr.config.ConmonPidFile != "" {
 		args = append(args, "--conmon-pidfile", ctr.config.ConmonPidFile)
@@ -237,6 +237,13 @@ func (r *OCIRuntime) createOCIContainer(ctr *Container, cgroupParent string, res
 	if r.logSizeMax >= 0 {
 		args = append(args, "--log-size-max", fmt.Sprintf("%v", r.logSizeMax))
 	}
+
+	logDriver := KubernetesLogging
+	if ctr.LogDriver() != "" {
+		logDriver = ctr.LogDriver()
+	}
+	args = append(args, "-l", fmt.Sprintf("%s:%s", logDriver, ctr.LogPath()))
+
 	if r.noPivot {
 		args = append(args, "--no-pivot")
 	}
diff --git a/libpod/options.go b/libpod/options.go
index 7ec7dfe63..20aa51981 100644
--- a/libpod/options.go
+++ b/libpod/options.go
@@ -979,6 +979,27 @@ func WithStaticIP(ip net.IP) CtrCreateOption {
 	}
 }
 
+// WithLogDriver sets the log driver for the container
+func WithLogDriver(driver string) CtrCreateOption {
+	return func(ctr *Container) error {
+		if ctr.valid {
+			return ErrCtrFinalized
+		}
+		switch driver {
+		case "":
+			return errors.Wrapf(ErrInvalidArg, "log driver must be set")
+		case JournaldLogging, KubernetesLogging, JSONLogging:
+			break
+		default:
+			return errors.Wrapf(ErrInvalidArg, "invalid log driver")
+		}
+
+		ctr.config.LogDriver = driver
+
+		return nil
+	}
+}
+
 // WithLogPath sets the path to the log file.
 func WithLogPath(path string) CtrCreateOption {
 	return func(ctr *Container) error {
diff --git a/libpod/runtime_ctr.go b/libpod/runtime_ctr.go
index cba8bdb1a..0c8d3edab 100644
--- a/libpod/runtime_ctr.go
+++ b/libpod/runtime_ctr.go
@@ -196,7 +196,7 @@ func (r *Runtime) newContainer(ctx context.Context, rSpec *spec.Spec, options ..
 		}
 	}
 
-	if ctr.config.LogPath == "" {
+	if ctr.config.LogPath == "" && ctr.config.LogDriver != JournaldLogging {
 		ctr.config.LogPath = filepath.Join(ctr.config.StaticDir, "ctr.log")
 	}