1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
|
//+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()
}
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
timestamp := time.Unix(0, int64(usec)*int64(time.Microsecond))
output := timestamp.Format(logTimeFormat) + " "
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
}
|
