daemon/logger/loggertest/logreader.go
package loggertest // import "github.com/docker/docker/daemon/logger/loggertest"
import (
"fmt"
"runtime"
"strings"
"sync"
"testing"
"time"
"github.com/google/go-cmp/cmp"
"github.com/google/go-cmp/cmp/cmpopts"
"gotest.tools/v3/assert"
is "gotest.tools/v3/assert/cmp"
"gotest.tools/v3/assert/opt"
"github.com/docker/docker/api/types/backend"
"github.com/docker/docker/daemon/logger"
)
type syncer interface {
// Sync commits the current logs to stable storage such that the most
// recently-logged message can be immediately read back by a LogReader.
Sync() error
}
func syncLogger(t *testing.T, l logger.Logger) {
if sl, ok := l.(syncer); ok {
assert.NilError(t, sl.Sync())
}
}
// Reader tests that a logger.LogReader implementation behaves as it should.
type Reader struct {
// Factory returns a function which constructs loggers for the container
// specified in info. Each call to the returned function must yield a
// distinct logger instance which can read back logs written by earlier
// instances.
Factory func(*testing.T, logger.Info) func(*testing.T) logger.Logger
}
var compareLog cmp.Options = []cmp.Option{
// Not all log drivers can round-trip timestamps at full nanosecond
// precision.
opt.TimeWithThreshold(time.Millisecond),
// The json-log driver does not round-trip PLogMetaData and API users do
// not expect it.
cmpopts.IgnoreFields(logger.Message{}, "PLogMetaData"),
cmp.Transformer("string", func(b []byte) string { return string(b) }),
}
// TestTail tests the behavior of the LogReader's tail implementation.
func (tr Reader) TestTail(t *testing.T) {
t.Run("Live", func(t *testing.T) { tr.testTail(t, true) })
t.Run("LiveEmpty", func(t *testing.T) { tr.testTailEmptyLogs(t, true) })
t.Run("Stopped", func(t *testing.T) { tr.testTail(t, false) })
t.Run("StoppedEmpty", func(t *testing.T) { tr.testTailEmptyLogs(t, false) })
}
func makeTestMessages() []*logger.Message {
return []*logger.Message{
{Source: "stdout", Timestamp: time.Now().Add(-1 * 30 * time.Minute), Line: []byte("a message")},
{Source: "stdout", Timestamp: time.Now().Add(-1 * 20 * time.Minute), Line: []byte("another message"), PLogMetaData: &backend.PartialLogMetaData{ID: "aaaaaaaa", Ordinal: 1, Last: true}},
{Source: "stderr", Timestamp: time.Now().Add(-1 * 15 * time.Minute), Line: []byte("to be..."), PLogMetaData: &backend.PartialLogMetaData{ID: "bbbbbbbb", Ordinal: 1}},
{Source: "stderr", Timestamp: time.Now().Add(-1 * 15 * time.Minute), Line: []byte("continued"), PLogMetaData: &backend.PartialLogMetaData{ID: "bbbbbbbb", Ordinal: 2, Last: true}},
{Source: "stderr", Timestamp: time.Now().Add(-1 * 10 * time.Minute), Line: []byte("a really long message " + strings.Repeat("a", 4096))},
{Source: "stderr", Timestamp: time.Now().Add(-1 * 10 * time.Minute), Line: []byte("just one more message")},
{Source: "stdout", Timestamp: time.Now().Add(-1 * 90 * time.Minute), Line: []byte("someone adjusted the clock")},
}
}
func (tr Reader) testTail(t *testing.T, live bool) {
t.Parallel()
factory := tr.Factory(t, logger.Info{
ContainerID: "tailtest0000",
ContainerName: "logtail",
})
l := factory(t)
if live {
defer func() { assert.NilError(t, l.Close()) }()
}
mm := makeTestMessages()
expected := logMessages(t, l, mm)
if !live {
// Simulate reading from a stopped container.
assert.NilError(t, l.Close())
l = factory(t)
defer func() { assert.NilError(t, l.Close()) }()
}
lr := l.(logger.LogReader)
t.Run("Exact", func(t *testing.T) {
t.Parallel()
lw := lr.ReadLogs(logger.ReadConfig{Tail: len(mm)})
defer lw.ConsumerGone()
assert.DeepEqual(t, readAll(t, lw), expected, compareLog)
})
t.Run("LessThanAvailable", func(t *testing.T) {
t.Parallel()
lw := lr.ReadLogs(logger.ReadConfig{Tail: 2})
defer lw.ConsumerGone()
assert.DeepEqual(t, readAll(t, lw), expected[len(mm)-2:], compareLog)
})
t.Run("MoreThanAvailable", func(t *testing.T) {
t.Parallel()
lw := lr.ReadLogs(logger.ReadConfig{Tail: 100})
defer lw.ConsumerGone()
assert.DeepEqual(t, readAll(t, lw), expected, compareLog)
})
t.Run("All", func(t *testing.T) {
t.Parallel()
lw := lr.ReadLogs(logger.ReadConfig{Tail: -1})
defer lw.ConsumerGone()
assert.DeepEqual(t, readAll(t, lw), expected, compareLog)
})
t.Run("Since", func(t *testing.T) {
t.Parallel()
lw := lr.ReadLogs(logger.ReadConfig{Tail: -1, Since: mm[1].Timestamp.Truncate(time.Millisecond)})
defer lw.ConsumerGone()
assert.DeepEqual(t, readAll(t, lw), expected[1:], compareLog)
})
t.Run("MoreThanSince", func(t *testing.T) {
t.Parallel()
lw := lr.ReadLogs(logger.ReadConfig{Tail: len(mm), Since: mm[1].Timestamp.Truncate(time.Millisecond)})
defer lw.ConsumerGone()
assert.DeepEqual(t, readAll(t, lw), expected[1:], compareLog)
})
t.Run("LessThanSince", func(t *testing.T) {
t.Parallel()
lw := lr.ReadLogs(logger.ReadConfig{Tail: len(mm) - 2, Since: mm[1].Timestamp.Truncate(time.Millisecond)})
defer lw.ConsumerGone()
assert.DeepEqual(t, readAll(t, lw), expected[2:], compareLog)
})
t.Run("Until", func(t *testing.T) {
t.Parallel()
lw := lr.ReadLogs(logger.ReadConfig{Tail: -1, Until: mm[2].Timestamp.Add(-time.Millisecond)})
defer lw.ConsumerGone()
assert.DeepEqual(t, readAll(t, lw), expected[:2], compareLog)
})
t.Run("SinceAndUntil", func(t *testing.T) {
t.Parallel()
lw := lr.ReadLogs(logger.ReadConfig{Tail: -1, Since: mm[1].Timestamp.Truncate(time.Millisecond), Until: mm[1].Timestamp.Add(time.Millisecond)})
defer lw.ConsumerGone()
assert.DeepEqual(t, readAll(t, lw), expected[1:2], compareLog)
})
}
func (tr Reader) testTailEmptyLogs(t *testing.T, live bool) {
t.Parallel()
factory := tr.Factory(t, logger.Info{
ContainerID: "tailemptytest",
ContainerName: "logtail",
})
l := factory(t)
if !live {
assert.NilError(t, l.Close())
l = factory(t)
}
defer func() { assert.NilError(t, l.Close()) }()
for _, tt := range []struct {
name string
cfg logger.ReadConfig
}{
{name: "Zero", cfg: logger.ReadConfig{}},
{name: "All", cfg: logger.ReadConfig{Tail: -1}},
{name: "Tail", cfg: logger.ReadConfig{Tail: 42}},
{name: "Since", cfg: logger.ReadConfig{Since: time.Unix(1, 0)}},
{name: "Until", cfg: logger.ReadConfig{Until: time.Date(2100, time.January, 1, 1, 1, 1, 0, time.UTC)}},
{name: "SinceAndUntil", cfg: logger.ReadConfig{Since: time.Unix(1, 0), Until: time.Date(2100, time.January, 1, 1, 1, 1, 0, time.UTC)}},
} {
tt := tt
t.Run(tt.name, func(t *testing.T) {
t.Parallel()
lw := l.(logger.LogReader).ReadLogs(logger.ReadConfig{})
defer lw.ConsumerGone()
assert.DeepEqual(t, readAll(t, lw), ([]*logger.Message)(nil), cmpopts.EquateEmpty())
})
}
}
// TestFollow tests the LogReader's follow implementation.
//
// The LogReader is expected to be able to follow an arbitrary number of
// messages at a high rate with no dropped messages.
func (tr Reader) TestFollow(t *testing.T) {
// Reader sends all logs and closes after logger is closed
// - Starting from empty log (like run)
for i, tail := range []int{-1, 0, 1, 42} {
i, tail := i, tail
t.Run(fmt.Sprintf("FromEmptyLog/Tail=%d", tail), func(t *testing.T) {
t.Parallel()
l := tr.Factory(t, logger.Info{
ContainerID: fmt.Sprintf("followstart%d", i),
ContainerName: fmt.Sprintf("logloglog%d", i),
})(t)
lw := l.(logger.LogReader).ReadLogs(logger.ReadConfig{Tail: tail, Follow: true})
defer lw.ConsumerGone()
doneReading := make(chan struct{})
var logs []*logger.Message
go func() {
defer close(doneReading)
logs = readAll(t, lw)
}()
mm := makeTestMessages()
expected := logMessages(t, l, mm)
assert.NilError(t, l.Close())
<-doneReading
assert.DeepEqual(t, logs, expected, compareLog)
})
}
t.Run("AttachMidStream", func(t *testing.T) {
t.Parallel()
l := tr.Factory(t, logger.Info{
ContainerID: "followmiddle",
ContainerName: "logloglog",
})(t)
mm := makeTestMessages()
expected := logMessages(t, l, mm[0:1])
lw := l.(logger.LogReader).ReadLogs(logger.ReadConfig{Tail: -1, Follow: true})
defer lw.ConsumerGone()
doneReading := make(chan struct{})
var logs []*logger.Message
go func() {
defer close(doneReading)
logs = readAll(t, lw)
}()
expected = append(expected, logMessages(t, l, mm[1:])...)
assert.NilError(t, l.Close())
<-doneReading
assert.DeepEqual(t, logs, expected, compareLog)
})
t.Run("Since", func(t *testing.T) {
t.Parallel()
l := tr.Factory(t, logger.Info{
ContainerID: "followsince0",
ContainerName: "logloglog",
})(t)
mm := makeTestMessages()
lw := l.(logger.LogReader).ReadLogs(logger.ReadConfig{Tail: -1, Follow: true, Since: mm[2].Timestamp.Truncate(time.Millisecond)})
defer lw.ConsumerGone()
doneReading := make(chan struct{})
var logs []*logger.Message
go func() {
defer close(doneReading)
logs = readAll(t, lw)
}()
expected := logMessages(t, l, mm)[2:]
assert.NilError(t, l.Close())
<-doneReading
assert.DeepEqual(t, logs, expected, compareLog)
})
t.Run("Until", func(t *testing.T) {
t.Parallel()
l := tr.Factory(t, logger.Info{
ContainerID: "followuntil0",
ContainerName: "logloglog",
})(t)
mm := makeTestMessages()
lw := l.(logger.LogReader).ReadLogs(logger.ReadConfig{Tail: -1, Follow: true, Until: mm[2].Timestamp.Add(-time.Millisecond)})
defer lw.ConsumerGone()
doneReading := make(chan struct{})
var logs []*logger.Message
go func() {
defer close(doneReading)
logs = readAll(t, lw)
}()
expected := logMessages(t, l, mm)[:2]
defer assert.NilError(t, l.Close()) // Reading should end before the logger is closed.
<-doneReading
assert.DeepEqual(t, logs, expected, compareLog)
})
t.Run("SinceAndUntil", func(t *testing.T) {
t.Parallel()
l := tr.Factory(t, logger.Info{
ContainerID: "followbounded",
ContainerName: "logloglog",
})(t)
mm := makeTestMessages()
lw := l.(logger.LogReader).ReadLogs(logger.ReadConfig{Tail: -1, Follow: true, Since: mm[1].Timestamp.Add(-time.Millisecond), Until: mm[2].Timestamp.Add(-time.Millisecond)})
defer lw.ConsumerGone()
doneReading := make(chan struct{})
var logs []*logger.Message
go func() {
defer close(doneReading)
logs = readAll(t, lw)
}()
expected := logMessages(t, l, mm)[1:2]
defer assert.NilError(t, l.Close()) // Reading should end before the logger is closed.
<-doneReading
assert.DeepEqual(t, logs, expected, compareLog)
})
t.Run("Tail=0", func(t *testing.T) {
t.Parallel()
l := tr.Factory(t, logger.Info{
ContainerID: "followtail00",
ContainerName: "logloglog",
})(t)
mm := makeTestMessages()
logMessages(t, l, mm[0:2])
syncLogger(t, l)
lw := l.(logger.LogReader).ReadLogs(logger.ReadConfig{Tail: 0, Follow: true})
defer lw.ConsumerGone()
doneReading := make(chan struct{})
var logs []*logger.Message
go func() {
defer close(doneReading)
logs = readAll(t, lw)
}()
expected := logMessages(t, l, mm[2:])
assert.NilError(t, l.Close())
<-doneReading
assert.DeepEqual(t, logs, expected, compareLog)
})
t.Run("Tail>0", func(t *testing.T) {
t.Parallel()
l := tr.Factory(t, logger.Info{
ContainerID: "followtail00",
ContainerName: "logloglog",
})(t)
mm := makeTestMessages()
expected := logMessages(t, l, mm[0:2])[1:]
syncLogger(t, l)
lw := l.(logger.LogReader).ReadLogs(logger.ReadConfig{Tail: 1, Follow: true})
defer lw.ConsumerGone()
doneReading := make(chan struct{})
var logs []*logger.Message
go func() {
defer close(doneReading)
logs = readAll(t, lw)
}()
expected = append(expected, logMessages(t, l, mm[2:])...)
assert.NilError(t, l.Close())
<-doneReading
assert.DeepEqual(t, logs, expected, compareLog)
})
t.Run("MultipleStarts", func(t *testing.T) {
t.Parallel()
factory := tr.Factory(t, logger.Info{
ContainerID: "startrestart",
ContainerName: "startmeup",
})
mm := makeTestMessages()
l := factory(t)
expected := logMessages(t, l, mm[:3])
assert.NilError(t, l.Close())
l = factory(t)
lw := l.(logger.LogReader).ReadLogs(logger.ReadConfig{Tail: -1, Follow: true})
defer lw.ConsumerGone()
doneReading := make(chan struct{})
var logs []*logger.Message
go func() {
defer close(doneReading)
logs = readAll(t, lw)
}()
expected = append(expected, logMessages(t, l, mm[3:])...)
assert.NilError(t, l.Close())
<-doneReading
assert.DeepEqual(t, logs, expected, compareLog)
})
t.Run("Concurrent", tr.TestConcurrent)
}
// TestConcurrent tests the Logger and its LogReader implementation for
// race conditions when logging from multiple goroutines concurrently.
func (tr Reader) TestConcurrent(t *testing.T) {
t.Parallel()
l := tr.Factory(t, logger.Info{
ContainerID: "logconcurrent0",
ContainerName: "logconcurrent123",
})(t)
// Split test messages
stderrMessages := []*logger.Message{}
stdoutMessages := []*logger.Message{}
for _, m := range makeTestMessages() {
if m.Source == "stdout" {
stdoutMessages = append(stdoutMessages, m)
} else if m.Source == "stderr" {
stderrMessages = append(stderrMessages, m)
}
}
// Follow all logs
lw := l.(logger.LogReader).ReadLogs(logger.ReadConfig{Follow: true, Tail: -1})
defer lw.ConsumerGone()
// Log concurrently from two sources and close log
wg := &sync.WaitGroup{}
logAll := func(msgs []*logger.Message) {
defer wg.Done()
for _, m := range msgs {
assert.Check(t, l.Log(copyLogMessage(m)), "failed to log message %+v", m)
}
}
closed := make(chan struct{})
wg.Add(2)
go logAll(stdoutMessages)
go logAll(stderrMessages)
go func() {
defer close(closed)
defer l.Close()
wg.Wait()
}()
defer func() {
// Make sure log gets closed before we return
// so the temporary dir can be deleted
select {
case <-time.After(10 * time.Second):
t.Fatal("timed out waiting for logger to close")
case <-closed:
}
}()
// Check if the message count, order and content is equal to what was logged
for {
l := readMessage(t, lw)
if l == nil {
break
}
var messages *[]*logger.Message
if l.Source == "stdout" {
messages = &stdoutMessages
} else if l.Source == "stderr" {
messages = &stderrMessages
} else {
t.Fatalf("Corrupted message.Source = %q", l.Source)
}
expectedMsg := transformToExpected((*messages)[0])
assert.DeepEqual(t, *expectedMsg, *l, compareLog)
*messages = (*messages)[1:]
}
assert.Check(t, is.Len(stdoutMessages, 0), "expected stdout messages were not read")
assert.Check(t, is.Len(stderrMessages, 0), "expected stderr messages were not read")
}
// logMessages logs messages to l and returns a slice of messages as would be
// expected to be read back. The message values are not modified and the
// returned slice of messages are deep-copied.
func logMessages(t *testing.T, l logger.Logger, messages []*logger.Message) []*logger.Message {
t.Helper()
var expected []*logger.Message
for _, m := range messages {
// Copy the log message because the underlying log writer resets
// the log message and returns it to a buffer pool.
assert.NilError(t, l.Log(copyLogMessage(m)))
runtime.Gosched()
expect := transformToExpected(m)
expected = append(expected, expect)
}
return expected
}
// Existing API consumers expect a newline to be appended to
// messages other than nonterminal partials as that matches the
// existing behavior of the json-file log driver.
func transformToExpected(m *logger.Message) *logger.Message {
// Copy the log message again so as not to mutate the input.
copy := copyLogMessage(m)
if m.PLogMetaData == nil || m.PLogMetaData.Last {
copy.Line = append(copy.Line, '\n')
}
return copy
}
func copyLogMessage(src *logger.Message) *logger.Message {
dst := logger.NewMessage()
dst.Source = src.Source
dst.Timestamp = src.Timestamp
dst.Attrs = src.Attrs
dst.Err = src.Err
dst.Line = append(dst.Line, src.Line...)
if src.PLogMetaData != nil {
lmd := *src.PLogMetaData
dst.PLogMetaData = &lmd
}
return dst
}
func readMessage(t *testing.T, lw *logger.LogWatcher) *logger.Message {
t.Helper()
timeout := time.NewTimer(5 * time.Second)
defer timeout.Stop()
select {
case <-timeout.C:
t.Error("timed out waiting for message")
return nil
case err, open := <-lw.Err:
t.Errorf("unexpected receive on lw.Err: err=%v, open=%v", err, open)
return nil
case msg, open := <-lw.Msg:
if !open {
select {
case err, open := <-lw.Err:
t.Errorf("unexpected receive on lw.Err with closed lw.Msg: err=%v, open=%v", err, open)
default:
}
return nil
}
assert.Assert(t, msg != nil)
t.Logf("[%v] %s: %s", msg.Timestamp, msg.Source, msg.Line)
return msg
}
}
func readAll(t *testing.T, lw *logger.LogWatcher) []*logger.Message {
t.Helper()
var msgs []*logger.Message
for {
m := readMessage(t, lw)
if m == nil {
return msgs
}
msgs = append(msgs, m)
}
}