test: Add tests for logging

1. Add integration tests for the ContainerLogs API call
Each test handle a distinct case of ContainerLogs output.
- Muxed stream, when container is started without tty
- Single stream, when container is started with tty

2. Add unit test for LogReader suite that tests concurrent logging
It checks that there are no race conditions when logging concurrently
from multiple goroutines.

Co-authored-by: Cory Snider <csnider@mirantis.com>
Signed-off-by: Cory Snider <csnider@mirantis.com>
Signed-off-by: Paweł Gronowski <pawel.gronowski@docker.com>

daemon/logger/loggertest/logreader.go

@@ -3,6 +3,7 @@ package loggertest // import "github.com/docker/docker/daemon/logger/loggertest"
 import (
 	"runtime"
 	"strings"
+	"sync"
 	"testing"
 	"time"
 
@@ -381,6 +382,81 @@ func (tr Reader) TestFollow(t *testing.T) {
 		<-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 {
+			l.Log(copyLogMessage(m))
+		}
+	}
+
+	closed := make(chan struct{})
+	wg.Add(2)
+	go logAll(stdoutMessages)
+	go logAll(stderrMessages)
+	go func() {
+		defer close(closed)
+		defer l.Close()
+		wg.Wait()
+	}()
+
+	// 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.Equal(t, len(stdoutMessages), 0)
+	assert.Equal(t, len(stderrMessages), 0)
+
+	// Make sure log gets closed before we return
+	// so the temporary dir can be deleted
+	<-closed
 }
 
 // logMessages logs messages to l and returns a slice of messages as would be
@@ -395,19 +471,25 @@ func logMessages(t *testing.T, l logger.Logger, messages []*logger.Message) []*l
 		assert.NilError(t, l.Log(copyLogMessage(m)))
 		runtime.Gosched()
 
-		// Copy the log message again so as not to mutate the input.
-		expect := copyLogMessage(m)
-		// 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.
-		if m.PLogMetaData == nil || m.PLogMetaData.Last {
-			expect.Line = append(expect.Line, '\n')
-		}
+		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

integration/container/logs_test.go

@@ -1,14 +1,22 @@
 package container // import "github.com/docker/docker/integration/container"
 
 import (
+	"bytes"
 	"context"
 	"io"
+	"strings"
 	"testing"
+	"time"
 
 	"github.com/docker/docker/api/types"
+	"github.com/docker/docker/daemon/logger/jsonfilelog"
+	"github.com/docker/docker/daemon/logger/local"
 	"github.com/docker/docker/integration/internal/container"
+	"github.com/docker/docker/integration/internal/termtest"
 	"github.com/docker/docker/pkg/stdcopy"
 	"gotest.tools/v3/assert"
+	"gotest.tools/v3/assert/cmp"
+	"gotest.tools/v3/poll"
 	"gotest.tools/v3/skip"
 )
 
@@ -32,3 +40,151 @@ func TestLogsFollowTailEmpty(t *testing.T) {
 	_, err = stdcopy.StdCopy(io.Discard, io.Discard, logs)
 	assert.Check(t, err)
 }
+
+func TestLogs(t *testing.T) {
+	drivers := []string{local.Name, jsonfilelog.Name}
+
+	for _, logDriver := range drivers {
+		t.Run("driver "+logDriver, func(t *testing.T) {
+			testLogs(t, logDriver)
+		})
+	}
+}
+
+func testLogs(t *testing.T, logDriver string) {
+	defer setupTest(t)()
+	client := testEnv.APIClient()
+	ctx := context.Background()
+
+	testCases := []struct {
+		desc        string
+		logOps      types.ContainerLogsOptions
+		expectedOut string
+		expectedErr string
+		tty         bool
+	}{
+		// TTY, only one output stream
+		{
+			desc: "tty/stdout and stderr",
+			tty:  true,
+			logOps: types.ContainerLogsOptions{
+				ShowStdout: true,
+				ShowStderr: true,
+			},
+			expectedOut: "this is fineaccidents happen",
+		},
+		{
+			desc: "tty/only stdout",
+			tty:  true,
+			logOps: types.ContainerLogsOptions{
+				ShowStdout: true,
+				ShowStderr: false,
+			},
+			expectedOut: "this is fineaccidents happen",
+		},
+		{
+			desc: "tty/only stderr",
+			tty:  true,
+			logOps: types.ContainerLogsOptions{
+				ShowStdout: false,
+				ShowStderr: true,
+			},
+			expectedOut: "",
+		},
+		// Without TTY, both stdout and stderr
+		{
+			desc: "without tty/stdout and stderr",
+			tty:  false,
+			logOps: types.ContainerLogsOptions{
+				ShowStdout: true,
+				ShowStderr: true,
+			},
+			expectedOut: "this is fine",
+			expectedErr: "accidents happen",
+		},
+		{
+			desc: "without tty/only stdout",
+			tty:  false,
+			logOps: types.ContainerLogsOptions{
+				ShowStdout: true,
+				ShowStderr: false,
+			},
+			expectedOut: "this is fine",
+			expectedErr: "",
+		},
+		{
+			desc: "without tty/only stderr",
+			tty:  false,
+			logOps: types.ContainerLogsOptions{
+				ShowStdout: false,
+				ShowStderr: true,
+			},
+			expectedOut: "",
+			expectedErr: "accidents happen",
+		},
+	}
+
+	for _, tC := range testCases {
+		tC := tC
+		t.Run(tC.desc, func(t *testing.T) {
+			t.Parallel()
+			tty := tC.tty
+			id := container.Run(ctx, t, client,
+				container.WithCmd("sh", "-c", "echo -n this is fine; echo -n accidents happen >&2"),
+				container.WithTty(tty),
+				container.WithLogDriver(logDriver),
+			)
+			defer client.ContainerRemove(ctx, id, types.ContainerRemoveOptions{Force: true})
+
+			poll.WaitOn(t, container.IsStopped(ctx, client, id), poll.WithDelay(time.Millisecond*100))
+
+			logs, err := client.ContainerLogs(ctx, id, tC.logOps)
+			assert.NilError(t, err)
+			defer logs.Close()
+
+			var stdout, stderr bytes.Buffer
+			if tty {
+				// TTY, only one output stream
+				_, err = io.Copy(&stdout, logs)
+			} else {
+				_, err = stdcopy.StdCopy(&stdout, &stderr, logs)
+			}
+			assert.NilError(t, err)
+
+			stdoutStr := stdout.String()
+
+			if tty && testEnv.OSType == "windows" {
+				stdoutStr = stripEscapeCodes(t, stdoutStr)
+
+				// Special case for Windows Server 2019
+				// Check only that the raw output stream contains strings
+				// that were printed to container's stdout and stderr.
+				// This is a workaround for the backspace being outputted in an unexpected place
+				// which breaks the parsed output: https://github.com/moby/moby/issues/43710
+				if strings.Contains(testEnv.DaemonInfo.OperatingSystem, "Windows Server Version 1809") {
+					if tC.logOps.ShowStdout {
+						assert.Check(t, cmp.Contains(stdout.String(), "this is fine"))
+						assert.Check(t, cmp.Contains(stdout.String(), "accidents happen"))
+					} else {
+						assert.DeepEqual(t, stdoutStr, "")
+					}
+					return
+				}
+
+			}
+
+			assert.DeepEqual(t, stdoutStr, tC.expectedOut)
+			assert.DeepEqual(t, stderr.String(), tC.expectedErr)
+		})
+	}
+}
+
+// This hack strips the escape codes that appear in the Windows TTY output and don't have
+// any effect on the text content.
+// This doesn't handle all escape sequences, only ones that were encountered during testing.
+func stripEscapeCodes(t *testing.T, input string) string {
+	t.Logf("Stripping: %q\n", input)
+	output, err := termtest.StripANSICommands(input)
+	assert.NilError(t, err)
+	return output
+}

integration/internal/termtest/stripansi.go

@@ -0,0 +1,108 @@
+package termtest // import "github.com/docker/docker/integration/internal/termtest"
+
+import (
+	"errors"
+	"regexp"
+
+	"github.com/Azure/go-ansiterm"
+)
+
+var stripOSC = regexp.MustCompile(`\x1b\][^\x1b\a]*(\x1b\\|\a)`)
+
+// StripANSICommands attempts to strip ANSI console escape and control sequences
+// from s, returning a string containing only the final printed characters which
+// would be visible onscreen if the string was to be processed by a terminal
+// emulator. Basic cursor positioning and screen erase control sequences are
+// parsed and processed such that the output of simple CLI commands passed
+// through a Windows Pseudoterminal and then this function yields the same
+// string as if the output of those commands was redirected to a file.
+//
+// The only correct way to represent the result of processing ANSI console
+// output would be a two-dimensional array of an emulated terminal's display
+// buffer. That would be awkward to test against, so this function instead
+// attempts to render to a one-dimensional string without extra padding. This is
+// an inherently lossy process, and any attempts to render a string containing
+// many cursor positioning commands are unlikely to yield satisfactory results.
+// Handlers for several ANSI control sequences are also unimplemented; attempts
+// to parse a string containing one will panic.
+func StripANSICommands(s string, opts ...ansiterm.Option) (string, error) {
+	// Work around https://github.com/Azure/go-ansiterm/issues/34
+	s = stripOSC.ReplaceAllLiteralString(s, "")
+
+	var h stringHandler
+	p := ansiterm.CreateParser("Ground", &h, opts...)
+	_, err := p.Parse([]byte(s))
+	return h.String(), err
+}
+
+type stringHandler struct {
+	ansiterm.AnsiEventHandler
+	cursor int
+	b      []byte
+}
+
+func (h *stringHandler) Print(b byte) error {
+	if h.cursor == len(h.b) {
+		h.b = append(h.b, b)
+	} else {
+		h.b[h.cursor] = b
+	}
+	h.cursor++
+	return nil
+}
+
+func (h *stringHandler) Execute(b byte) error {
+	switch b {
+	case '\b':
+		if h.cursor > 0 {
+			if h.cursor == len(h.b) && h.b[h.cursor-1] == ' ' {
+				h.b = h.b[:len(h.b)-1]
+			}
+			h.cursor--
+		}
+	case '\r', '\n':
+		h.Print(b)
+	}
+	return nil
+}
+
+// Erase Display
+func (h *stringHandler) ED(v int) error {
+	switch v {
+	case 1: // Erase from start to cursor.
+		for i := 0; i < h.cursor; i++ {
+			h.b[i] = ' '
+		}
+	case 2, 3: // Erase whole display.
+		h.b = make([]byte, h.cursor)
+		for i := range h.b {
+			h.b[i] = ' '
+		}
+	default: // Erase from cursor to end of display.
+		h.b = h.b[:h.cursor+1]
+	}
+	return nil
+}
+
+// CUrsor Position
+func (h *stringHandler) CUP(x, y int) error {
+	if x > 1 {
+		return errors.New("termtest: cursor position not supported for X > 1")
+	}
+	if y > len(h.b) {
+		for n := len(h.b) - y; n > 0; n-- {
+			h.b = append(h.b, ' ')
+		}
+	}
+	h.cursor = y - 1
+	return nil
+}
+
+func (h stringHandler) DECTCEM(bool) error      { return nil } // Text Cursor Enable
+func (h stringHandler) SGR(v []int) error       { return nil } // Set Graphics Rendition
+func (h stringHandler) DA(attrs []string) error { return nil }
+func (h stringHandler) Flush() error            { return nil }
+
+func (h *stringHandler) String() string {
+	return string(h.b)
+}

integration/internal/termtest/stripansi_test.go

@@ -0,0 +1,26 @@
+package termtest // import "github.com/docker/docker/integration/internal/termtest"
+
+import (
+	"testing"
+
+	"gotest.tools/v3/assert"
+)
+
+func TestStripANSICommands(t *testing.T) {
+	for _, tt := range []struct{ input, want string }{
+		{
+			input: "\x1b[2J\x1b[?25l\x1b[m\x1b[Hthis is fine\b\x1b]0;C:\\bin\\sh.exe\x00\a\x1b[?25h\x1b[Ht\x1b[1;13H\x1b[?25laccidents happen \b\x1b[?25h\x1b[Ht\x1b[1;29H",
+			want:  "this is fineaccidents happen",
+		},
+		{
+			input: "\x1b[2J\x1b[m\x1b[Hthis is fine\x1b]0;C:\\bin\\sh.exe\a\x1b[?25haccidents happen",
+			want:  "this is fineaccidents happen",
+		},
+	} {
+		t.Run("", func(t *testing.T) {
+			got, err := StripANSICommands(tt.input)
+			assert.NilError(t, err)
+			assert.DeepEqual(t, tt.want, got)
+		})
+	}
+}