Add max_container_log_size

Signed-off-by: Lantao Liu <[email protected]>

Author: Random-Liu

docs/config.md

@@ -27,6 +27,11 @@ The explanation and default value of each configuration item are as follows:
   # enable_tls_streaming enables the TLS streaming support.
   enable_tls_streaming = false
 
+  # max_container_log_line_size is the maximum log line size in bytes for a container.
+  # Log line longer than the limit will be split into multiple lines. -1 means no
+  # limit.
+  max_container_log_line_size = 16384
+
   # "plugins.cri.containerd" contains config related to containerd
   [plugins.cri.containerd]
 

hack/verify-lint.sh

@@ -20,7 +20,7 @@ set -o pipefail
 for d in $(find . -type d -a \( -iwholename './pkg*' -o -iwholename './cmd*' \) -not -iwholename './pkg/api*'); do
 	echo for directory ${d} ...
 	gometalinter \
-		 --exclude='error return value not checked.*(Close|Log|Print).*\(errcheck\)$' \
+		 --exclude='error return value not checked.*(Close|Log|Print|Fprint).*\(errcheck\)$' \
 		 --exclude='.*_test\.go:.*error return value not checked.*\(errcheck\)$' \
 		 --exclude='duplicate of.*_test.go.*\(dupl\)$' \
 		 --exclude='.*/mock_.*\.go:.*\(golint\)$' \

pkg/config/config.go

@@ -96,6 +96,10 @@ type PluginConfig struct {
 	SystemdCgroup bool `toml:"systemd_cgroup" json:"systemdCgroup"`
 	// EnableTLSStreaming indicates to enable the TLS streaming support.
 	EnableTLSStreaming bool `toml:"enable_tls_streaming" json:"enableTLSStreaming"`
+	// MaxContainerLogLineSize is the maximum log line size in bytes for a container.
+	// Log line longer than the limit will be split into multiple lines. Non-positive
+	// value means no limit.
+	MaxContainerLogLineSize int `toml:"max_container_log_line_size" json:"maxContainerLogSize"`
 }
 
 // Config contains all configurations for cri server.
@@ -129,13 +133,14 @@ func DefaultConfig() PluginConfig {
 				Root:   "",
 			},
 		},
-		StreamServerAddress: "",
-		StreamServerPort:    "10010",
-		EnableSelinux:       false,
-		EnableTLSStreaming:  false,
-		SandboxImage:        "k8s.gcr.io/pause:3.1",
-		StatsCollectPeriod:  10,
-		SystemdCgroup:       false,
+		StreamServerAddress:     "",
+		StreamServerPort:        "10010",
+		EnableSelinux:           false,
+		EnableTLSStreaming:      false,
+		SandboxImage:            "k8s.gcr.io/pause:3.1",
+		StatsCollectPeriod:      10,
+		SystemdCgroup:           false,
+		MaxContainerLogLineSize: 16 * 1024,
 		Registry: Registry{
 			Mirrors: map[string]Mirror{
 				"docker.io": {

pkg/ioutil/write_closer.go

@@ -16,7 +16,10 @@ limitations under the License.
 
 package ioutil
 
-import "io"
+import (
+	"io"
+	"sync"
+)
 
 // writeCloseInformer wraps passed in write closer with a close channel.
 // Caller could wait on the close channel for the write closer to be
@@ -66,3 +69,34 @@ func (n *nopWriteCloser) Write(p []byte) (int, error) {
 func (n *nopWriteCloser) Close() error {
 	return nil
 }
+
+// serialWriteCloser wraps a write closer and makes sure all writes
+// are done in serial.
+// Parallel write won't intersect with each other. Use case:
+// 1) Pipe: Write content longer than PIPE_BUF.
+//    See http://man7.org/linux/man-pages/man7/pipe.7.html
+// 2) <3.14 Linux Kernel: write is not atomic
+//    See http://man7.org/linux/man-pages/man2/write.2.html
+type serialWriteCloser struct {
+	mu sync.Mutex
+	wc io.WriteCloser
+}
+
+// NewSerialWriteCloser creates a SerialWriteCloser from a write closer.
+func NewSerialWriteCloser(wc io.WriteCloser) io.WriteCloser {
+	return &serialWriteCloser{wc: wc}
+}
+
+// Write writes a group of byte arrays in order atomically.
+func (s *serialWriteCloser) Write(data []byte) (int, error) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	return s.wc.Write(data)
+}
+
+// Close closes the write closer.
+func (s *serialWriteCloser) Close() error {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	return s.wc.Close()
+}

pkg/ioutil/write_closer_test.go

@@ -17,9 +17,16 @@ limitations under the License.
 package ioutil
 
 import (
+	"io/ioutil"
+	"os"
+	"sort"
+	"strconv"
+	"strings"
+	"sync"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 )
 
 func TestWriteCloseInformer(t *testing.T) {
@@ -47,3 +54,55 @@ func TestWriteCloseInformer(t *testing.T) {
 		assert.Fail(t, "write closer not closed")
 	}
 }
+
+func TestSerialWriteCloser(t *testing.T) {
+	const (
+		// Test 10 times to make sure it always pass.
+		testCount = 10
+
+		goroutine = 10
+		dataLen   = 100000
+	)
+	for n := 0; n < testCount; n++ {
+		testData := make([][]byte, goroutine)
+		for i := 0; i < goroutine; i++ {
+			testData[i] = []byte(repeatNumber(i, dataLen) + "\n")
+		}
+
+		f, err := ioutil.TempFile("/tmp", "serial-write-closer")
+		require.NoError(t, err)
+		defer os.RemoveAll(f.Name())
+		defer f.Close()
+		wc := NewSerialWriteCloser(f)
+		defer wc.Close()
+
+		// Write data in parallel
+		var wg sync.WaitGroup
+		wg.Add(goroutine)
+		for i := 0; i < goroutine; i++ {
+			go func(id int) {
+				n, err := wc.Write(testData[id])
+				assert.NoError(t, err)
+				assert.Equal(t, dataLen+1, n)
+				wg.Done()
+			}(i)
+		}
+		wg.Wait()
+		wc.Close()
+
+		// Check test result
+		content, err := ioutil.ReadFile(f.Name())
+		require.NoError(t, err)
+		resultData := strings.Split(strings.TrimSpace(string(content)), "\n")
+		require.Len(t, resultData, goroutine)
+		sort.Strings(resultData)
+		for i := 0; i < goroutine; i++ {
+			expected := repeatNumber(i, dataLen)
+			assert.Equal(t, expected, resultData[i])
+		}
+	}
+}
+
+func repeatNumber(num, count int) string {
+	return strings.Repeat(strconv.Itoa(num), count)
+}

pkg/server/container_log_reopen.go

@@ -36,7 +36,7 @@ func (c *criService) ReopenContainerLog(ctx context.Context, r *runtime.ReopenCo
 	}
 
 	// Create new container logger and replace the existing ones.
-	stdoutWC, stderrWC, err := createContainerLoggers(container.LogPath, container.Config.GetTty())
+	stdoutWC, stderrWC, err := c.createContainerLoggers(container.LogPath, container.Config.GetTty())
 	if err != nil {
 		return nil, err
 	}

pkg/server/container_start.go

@@ -18,6 +18,7 @@ package server
 
 import (
 	"io"
+	"os"
 	"time"
 
 	"github.com/containerd/containerd"
@@ -29,6 +30,7 @@ import (
 	runtime "k8s.io/kubernetes/pkg/kubelet/apis/cri/runtime/v1alpha2"
 
 	ctrdutil "github.com/containerd/cri/pkg/containerd/util"
+	cioutil "github.com/containerd/cri/pkg/ioutil"
 	cio "github.com/containerd/cri/pkg/server/io"
 	containerstore "github.com/containerd/cri/pkg/store/container"
 	sandboxstore "github.com/containerd/cri/pkg/store/sandbox"
@@ -97,20 +99,10 @@ func (c *criService) startContainer(ctx context.Context,
 	}
 
 	ioCreation := func(id string) (_ containerdio.IO, err error) {
-		stdoutWC, stderrWC, err := createContainerLoggers(meta.LogPath, config.GetTty())
+		stdoutWC, stderrWC, err := c.createContainerLoggers(meta.LogPath, config.GetTty())
 		if err != nil {
 			return nil, errors.Wrap(err, "failed to create container loggers")
 		}
-		defer func() {
-			if err != nil {
-				if stdoutWC != nil {
-					stdoutWC.Close()
-				}
-				if stderrWC != nil {
-					stderrWC.Close()
-				}
-			}
-		}()
 		cntr.IO.AddOutput("log", stdoutWC, stderrWC)
 		cntr.IO.Pipe()
 		return cntr.IO, nil
@@ -142,24 +134,36 @@ func (c *criService) startContainer(ctx context.Context,
 	return nil
 }
 
-// Create container loggers and return write closer for stdout and stderr.
-func createContainerLoggers(logPath string, tty bool) (stdout io.WriteCloser, stderr io.WriteCloser, err error) {
+// createContainerLoggers creates container loggers and return write closer for stdout and stderr.
+func (c *criService) createContainerLoggers(logPath string, tty bool) (stdout io.WriteCloser, stderr io.WriteCloser, err error) {
 	if logPath != "" {
 		// Only generate container log when log path is specified.
-		if stdout, err = cio.NewCRILogger(logPath, cio.Stdout); err != nil {
-			return nil, nil, errors.Wrap(err, "failed to start container stdout logger")
+		f, err := os.OpenFile(logPath, os.O_CREATE|os.O_APPEND|os.O_WRONLY, 0640)
+		if err != nil {
+			return nil, nil, errors.Wrap(err, "failed to create and open log file")
 		}
 		defer func() {
 			if err != nil {
-				stdout.Close()
+				f.Close()
 			}
 		}()
+		var stdoutCh, stderrCh <-chan struct{}
+		wc := cioutil.NewSerialWriteCloser(f)
+		stdout, stdoutCh = cio.NewCRILogger(logPath, wc, cio.Stdout, c.config.MaxContainerLogLineSize)
 		// Only redirect stderr when there is no tty.
 		if !tty {
-			if stderr, err = cio.NewCRILogger(logPath, cio.Stderr); err != nil {
-				return nil, nil, errors.Wrap(err, "failed to start container stderr logger")
-			}
+			stderr, stderrCh = cio.NewCRILogger(logPath, wc, cio.Stderr, c.config.MaxContainerLogLineSize)
 		}
+		go func() {
+			if stdoutCh != nil {
+				<-stdoutCh
+			}
+			if stderrCh != nil {
+				<-stderrCh
+			}
+			logrus.Debugf("Finish redirecting log file %q, closing it", logPath)
+			f.Close()
+		}()
 	} else {
 		stdout = cio.NewDiscardLogger()
 		stderr = cio.NewDiscardLogger()