Move cri server packages under pkg/cri

Organizes the cri related server packages under pkg/cri

Signed-off-by: Derek McGowan <derek@mcg.dev>

pkg/cri/io/container_io.go

@@ -0,0 +1,236 @@
+/*
+   Copyright The containerd Authors.
+
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
+   limitations under the License.
+*/
+
+package io
+
+import (
+	"errors"
+	"io"
+	"strings"
+	"sync"
+
+	"github.com/containerd/containerd/cio"
+	"github.com/sirupsen/logrus"
+
+	cioutil "github.com/containerd/containerd/pkg/ioutil"
+	"github.com/containerd/containerd/pkg/util"
+)
+
+// streamKey generates a key for the stream.
+func streamKey(id, name string, stream StreamType) string {
+	return strings.Join([]string{id, name, string(stream)}, "-")
+}
+
+// ContainerIO holds the container io.
+type ContainerIO struct {
+	id string
+
+	fifos *cio.FIFOSet
+	*stdioPipes
+
+	stdoutGroup *cioutil.WriterGroup
+	stderrGroup *cioutil.WriterGroup
+
+	closer *wgCloser
+}
+
+var _ cio.IO = &ContainerIO{}
+
+// ContainerIOOpts sets specific information to newly created ContainerIO.
+type ContainerIOOpts func(*ContainerIO) error
+
+// WithFIFOs specifies existing fifos for the container io.
+func WithFIFOs(fifos *cio.FIFOSet) ContainerIOOpts {
+	return func(c *ContainerIO) error {
+		c.fifos = fifos
+		return nil
+	}
+}
+
+// WithNewFIFOs creates new fifos for the container io.
+func WithNewFIFOs(root string, tty, stdin bool) ContainerIOOpts {
+	return func(c *ContainerIO) error {
+		fifos, err := newFifos(root, c.id, tty, stdin)
+		if err != nil {
+			return err
+		}
+		return WithFIFOs(fifos)(c)
+	}
+}
+
+// NewContainerIO creates container io.
+func NewContainerIO(id string, opts ...ContainerIOOpts) (_ *ContainerIO, err error) {
+	c := &ContainerIO{
+		id:          id,
+		stdoutGroup: cioutil.NewWriterGroup(),
+		stderrGroup: cioutil.NewWriterGroup(),
+	}
+	for _, opt := range opts {
+		if err := opt(c); err != nil {
+			return nil, err
+		}
+	}
+	if c.fifos == nil {
+		return nil, errors.New("fifos are not set")
+	}
+	// Create actual fifos.
+	stdio, closer, err := newStdioPipes(c.fifos)
+	if err != nil {
+		return nil, err
+	}
+	c.stdioPipes = stdio
+	c.closer = closer
+	return c, nil
+}
+
+// Config returns io config.
+func (c *ContainerIO) Config() cio.Config {
+	return c.fifos.Config
+}
+
+// Pipe creates container fifos and pipe container output
+// to output stream.
+func (c *ContainerIO) Pipe() {
+	wg := c.closer.wg
+	if c.stdout != nil {
+		wg.Add(1)
+		go func() {
+			if _, err := io.Copy(c.stdoutGroup, c.stdout); err != nil {
+				logrus.WithError(err).Errorf("Failed to pipe stdout of container %q", c.id)
+			}
+			c.stdout.Close()
+			c.stdoutGroup.Close()
+			wg.Done()
+			logrus.Infof("Finish piping stdout of container %q", c.id)
+		}()
+	}
+
+	if !c.fifos.Terminal && c.stderr != nil {
+		wg.Add(1)
+		go func() {
+			if _, err := io.Copy(c.stderrGroup, c.stderr); err != nil {
+				logrus.WithError(err).Errorf("Failed to pipe stderr of container %q", c.id)
+			}
+			c.stderr.Close()
+			c.stderrGroup.Close()
+			wg.Done()
+			logrus.Infof("Finish piping stderr of container %q", c.id)
+		}()
+	}
+}
+
+// Attach attaches container stdio.
+// TODO(random-liu): Use pools.Copy in docker to reduce memory usage?
+func (c *ContainerIO) Attach(opts AttachOptions) {
+	var wg sync.WaitGroup
+	key := util.GenerateID()
+	stdinKey := streamKey(c.id, "attach-"+key, Stdin)
+	stdoutKey := streamKey(c.id, "attach-"+key, Stdout)
+	stderrKey := streamKey(c.id, "attach-"+key, Stderr)
+
+	var stdinStreamRC io.ReadCloser
+	if c.stdin != nil && opts.Stdin != nil {
+		// Create a wrapper of stdin which could be closed. Note that the
+		// wrapper doesn't close the actual stdin, it only stops io.Copy.
+		// The actual stdin will be closed by stream server.
+		stdinStreamRC = cioutil.NewWrapReadCloser(opts.Stdin)
+		wg.Add(1)
+		go func() {
+			if _, err := io.Copy(c.stdin, stdinStreamRC); err != nil {
+				logrus.WithError(err).Errorf("Failed to pipe stdin for container attach %q", c.id)
+			}
+			logrus.Infof("Attach stream %q closed", stdinKey)
+			if opts.StdinOnce && !opts.Tty {
+				// Due to kubectl requirements and current docker behavior, when (opts.StdinOnce &&
+				// opts.Tty) we have to close container stdin and keep stdout and stderr open until
+				// container stops.
+				c.stdin.Close()
+				// Also closes the containerd side.
+				if err := opts.CloseStdin(); err != nil {
+					logrus.WithError(err).Errorf("Failed to close stdin for container %q", c.id)
+				}
+			} else {
+				if opts.Stdout != nil {
+					c.stdoutGroup.Remove(stdoutKey)
+				}
+				if opts.Stderr != nil {
+					c.stderrGroup.Remove(stderrKey)
+				}
+			}
+			wg.Done()
+		}()
+	}
+
+	attachStream := func(key string, close <-chan struct{}) {
+		<-close
+		logrus.Infof("Attach stream %q closed", key)
+		// Make sure stdin gets closed.
+		if stdinStreamRC != nil {
+			stdinStreamRC.Close()
+		}
+		wg.Done()
+	}
+
+	if opts.Stdout != nil {
+		wg.Add(1)
+		wc, close := cioutil.NewWriteCloseInformer(opts.Stdout)
+		c.stdoutGroup.Add(stdoutKey, wc)
+		go attachStream(stdoutKey, close)
+	}
+	if !opts.Tty && opts.Stderr != nil {
+		wg.Add(1)
+		wc, close := cioutil.NewWriteCloseInformer(opts.Stderr)
+		c.stderrGroup.Add(stderrKey, wc)
+		go attachStream(stderrKey, close)
+	}
+	wg.Wait()
+}
+
+// AddOutput adds new write closers to the container stream, and returns existing
+// write closers if there are any.
+func (c *ContainerIO) AddOutput(name string, stdout, stderr io.WriteCloser) (io.WriteCloser, io.WriteCloser) {
+	var oldStdout, oldStderr io.WriteCloser
+	if stdout != nil {
+		key := streamKey(c.id, name, Stdout)
+		oldStdout = c.stdoutGroup.Get(key)
+		c.stdoutGroup.Add(key, stdout)
+	}
+	if stderr != nil {
+		key := streamKey(c.id, name, Stderr)
+		oldStderr = c.stderrGroup.Get(key)
+		c.stderrGroup.Add(key, stderr)
+	}
+	return oldStdout, oldStderr
+}
+
+// Cancel cancels container io.
+func (c *ContainerIO) Cancel() {
+	c.closer.Cancel()
+}
+
+// Wait waits container io to finish.
+func (c *ContainerIO) Wait() {
+	c.closer.Wait()
+}
+
+// Close closes all FIFOs.
+func (c *ContainerIO) Close() error {
+	c.closer.Close()
+	if c.fifos != nil {
+		return c.fifos.Close()
+	}
+	return nil
+}

pkg/cri/io/exec_io.go

@@ -0,0 +1,146 @@
+/*
+   Copyright The containerd Authors.
+
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
+   limitations under the License.
+*/
+
+package io
+
+import (
+	"io"
+	"sync"
+
+	"github.com/containerd/containerd/cio"
+	"github.com/sirupsen/logrus"
+
+	cioutil "github.com/containerd/containerd/pkg/ioutil"
+)
+
+// ExecIO holds the exec io.
+type ExecIO struct {
+	id    string
+	fifos *cio.FIFOSet
+	*stdioPipes
+	closer *wgCloser
+}
+
+var _ cio.IO = &ExecIO{}
+
+// NewExecIO creates exec io.
+func NewExecIO(id, root string, tty, stdin bool) (*ExecIO, error) {
+	fifos, err := newFifos(root, id, tty, stdin)
+	if err != nil {
+		return nil, err
+	}
+	stdio, closer, err := newStdioPipes(fifos)
+	if err != nil {
+		return nil, err
+	}
+	return &ExecIO{
+		id:         id,
+		fifos:      fifos,
+		stdioPipes: stdio,
+		closer:     closer,
+	}, nil
+}
+
+// Config returns io config.
+func (e *ExecIO) Config() cio.Config {
+	return e.fifos.Config
+}
+
+// Attach attaches exec stdio. The logic is similar with container io attach.
+func (e *ExecIO) Attach(opts AttachOptions) <-chan struct{} {
+	var wg sync.WaitGroup
+	var stdinStreamRC io.ReadCloser
+	if e.stdin != nil && opts.Stdin != nil {
+		stdinStreamRC = cioutil.NewWrapReadCloser(opts.Stdin)
+		wg.Add(1)
+		go func() {
+			if _, err := io.Copy(e.stdin, stdinStreamRC); err != nil {
+				logrus.WithError(err).Errorf("Failed to redirect stdin for container exec %q", e.id)
+			}
+			logrus.Infof("Container exec %q stdin closed", e.id)
+			if opts.StdinOnce && !opts.Tty {
+				e.stdin.Close()
+				if err := opts.CloseStdin(); err != nil {
+					logrus.WithError(err).Errorf("Failed to close stdin for container exec %q", e.id)
+				}
+			} else {
+				if e.stdout != nil {
+					e.stdout.Close()
+				}
+				if e.stderr != nil {
+					e.stderr.Close()
+				}
+			}
+			wg.Done()
+		}()
+	}
+
+	attachOutput := func(t StreamType, stream io.WriteCloser, out io.ReadCloser) {
+		if _, err := io.Copy(stream, out); err != nil {
+			logrus.WithError(err).Errorf("Failed to pipe %q for container exec %q", t, e.id)
+		}
+		out.Close()
+		stream.Close()
+		if stdinStreamRC != nil {
+			stdinStreamRC.Close()
+		}
+		e.closer.wg.Done()
+		wg.Done()
+		logrus.Infof("Finish piping %q of container exec %q", t, e.id)
+	}
+
+	if opts.Stdout != nil {
+		wg.Add(1)
+		// Closer should wait for this routine to be over.
+		e.closer.wg.Add(1)
+		go attachOutput(Stdout, opts.Stdout, e.stdout)
+	}
+
+	if !opts.Tty && opts.Stderr != nil {
+		wg.Add(1)
+		// Closer should wait for this routine to be over.
+		e.closer.wg.Add(1)
+		go attachOutput(Stderr, opts.Stderr, e.stderr)
+	}
+
+	done := make(chan struct{})
+	go func() {
+		wg.Wait()
+		close(done)
+	}()
+	return done
+}
+
+// Cancel cancels exec io.
+func (e *ExecIO) Cancel() {
+	e.closer.Cancel()
+}
+
+// Wait waits exec io to finish.
+func (e *ExecIO) Wait() {
+	e.closer.Wait()
+}
+
+// Close closes all FIFOs.
+func (e *ExecIO) Close() error {
+	if e.closer != nil {
+		e.closer.Close()
+	}
+	if e.fifos != nil {
+		return e.fifos.Close()
+	}
+	return nil
+}

pkg/cri/io/helpers.go

@@ -0,0 +1,144 @@
+/*
+   Copyright The containerd Authors.
+
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
+   limitations under the License.
+*/
+
+package io
+
+import (
+	"io"
+	"os"
+	"path/filepath"
+	"sync"
+	"syscall"
+
+	"github.com/containerd/containerd/cio"
+	"golang.org/x/net/context"
+	runtime "k8s.io/cri-api/pkg/apis/runtime/v1alpha2"
+)
+
+// AttachOptions specifies how to attach to a container.
+type AttachOptions struct {
+	Stdin     io.Reader
+	Stdout    io.WriteCloser
+	Stderr    io.WriteCloser
+	Tty       bool
+	StdinOnce bool
+	// CloseStdin is the function to close container stdin.
+	CloseStdin func() error
+}
+
+// StreamType is the type of the stream, stdout/stderr.
+type StreamType string
+
+const (
+	// Stdin stream type.
+	Stdin StreamType = "stdin"
+	// Stdout stream type.
+	Stdout StreamType = StreamType(runtime.Stdout)
+	// Stderr stream type.
+	Stderr StreamType = StreamType(runtime.Stderr)
+)
+
+type wgCloser struct {
+	ctx    context.Context
+	wg     *sync.WaitGroup
+	set    []io.Closer
+	cancel context.CancelFunc
+}
+
+func (g *wgCloser) Wait() {
+	g.wg.Wait()
+}
+
+func (g *wgCloser) Close() {
+	for _, f := range g.set {
+		f.Close()
+	}
+}
+
+func (g *wgCloser) Cancel() {
+	g.cancel()
+}
+
+// newFifos creates fifos directory for a container.
+func newFifos(root, id string, tty, stdin bool) (*cio.FIFOSet, error) {
+	root = filepath.Join(root, "io")
+	if err := os.MkdirAll(root, 0700); err != nil {
+		return nil, err
+	}
+	fifos, err := cio.NewFIFOSetInDir(root, id, tty)
+	if err != nil {
+		return nil, err
+	}
+	if !stdin {
+		fifos.Stdin = ""
+	}
+	return fifos, nil
+}
+
+type stdioPipes struct {
+	stdin  io.WriteCloser
+	stdout io.ReadCloser
+	stderr io.ReadCloser
+}
+
+// newStdioPipes creates actual fifos for stdio.
+func newStdioPipes(fifos *cio.FIFOSet) (_ *stdioPipes, _ *wgCloser, err error) {
+	var (
+		f           io.ReadWriteCloser
+		set         []io.Closer
+		ctx, cancel = context.WithCancel(context.Background())
+		p           = &stdioPipes{}
+	)
+	defer func() {
+		if err != nil {
+			for _, f := range set {
+				f.Close()
+			}
+			cancel()
+		}
+	}()
+
+	if fifos.Stdin != "" {
+		if f, err = openPipe(ctx, fifos.Stdin, syscall.O_WRONLY|syscall.O_CREAT|syscall.O_NONBLOCK, 0700); err != nil {
+			return nil, nil, err
+		}
+		p.stdin = f
+		set = append(set, f)
+	}
+
+	if fifos.Stdout != "" {
+		if f, err = openPipe(ctx, fifos.Stdout, syscall.O_RDONLY|syscall.O_CREAT|syscall.O_NONBLOCK, 0700); err != nil {
+			return nil, nil, err
+		}
+		p.stdout = f
+		set = append(set, f)
+	}
+
+	if fifos.Stderr != "" {
+		if f, err = openPipe(ctx, fifos.Stderr, syscall.O_RDONLY|syscall.O_CREAT|syscall.O_NONBLOCK, 0700); err != nil {
+			return nil, nil, err
+		}
+		p.stderr = f
+		set = append(set, f)
+	}
+
+	return p, &wgCloser{
+		wg:     &sync.WaitGroup{},
+		set:    set,
+		ctx:    ctx,
+		cancel: cancel,
+	}, nil
+}

pkg/cri/io/helpers_unix.go

@@ -0,0 +1,31 @@
+// +build !windows
+
+/*
+   Copyright The containerd Authors.
+
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
+   limitations under the License.
+*/
+
+package io
+
+import (
+	"io"
+	"os"
+
+	"github.com/containerd/fifo"
+	"golang.org/x/net/context"
+)
+
+func openPipe(ctx context.Context, fn string, flag int, perm os.FileMode) (io.ReadWriteCloser, error) {
+	return fifo.OpenFifo(ctx, fn, flag, perm)
+}

pkg/cri/io/helpers_windows.go

@@ -0,0 +1,81 @@
+// +build windows
+
+/*
+   Copyright The containerd Authors.
+
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
+   limitations under the License.
+*/
+
+package io
+
+import (
+	"io"
+	"net"
+	"os"
+	"sync"
+
+	winio "github.com/Microsoft/go-winio"
+	"github.com/pkg/errors"
+	"golang.org/x/net/context"
+)
+
+type pipe struct {
+	l      net.Listener
+	con    net.Conn
+	conErr error
+	conWg  sync.WaitGroup
+}
+
+func openPipe(ctx context.Context, fn string, flag int, perm os.FileMode) (io.ReadWriteCloser, error) {
+	l, err := winio.ListenPipe(fn, nil)
+	if err != nil {
+		return nil, err
+	}
+	p := &pipe{l: l}
+	p.conWg.Add(1)
+	go func() {
+		defer p.conWg.Done()
+		c, err := l.Accept()
+		if err != nil {
+			p.conErr = err
+			return
+		}
+		p.con = c
+	}()
+	return p, nil
+}
+
+func (p *pipe) Write(b []byte) (int, error) {
+	p.conWg.Wait()
+	if p.conErr != nil {
+		return 0, errors.Wrap(p.conErr, "connection error")
+	}
+	return p.con.Write(b)
+}
+
+func (p *pipe) Read(b []byte) (int, error) {
+	p.conWg.Wait()
+	if p.conErr != nil {
+		return 0, errors.Wrap(p.conErr, "connection error")
+	}
+	return p.con.Read(b)
+}
+
+func (p *pipe) Close() error {
+	p.l.Close()
+	p.conWg.Wait()
+	if p.con != nil {
+		return p.con.Close()
+	}
+	return p.conErr
+}

pkg/cri/io/logger.go

@@ -0,0 +1,196 @@
+/*
+   Copyright The containerd Authors.
+
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
+   limitations under the License.
+*/
+
+package io
+
+import (
+	"bufio"
+	"bytes"
+	"fmt"
+	"io"
+	"io/ioutil"
+	"time"
+
+	"github.com/sirupsen/logrus"
+	runtime "k8s.io/cri-api/pkg/apis/runtime/v1alpha2"
+
+	cioutil "github.com/containerd/containerd/pkg/ioutil"
+)
+
+const (
+	// delimiter used in CRI logging format.
+	delimiter = ' '
+	// eof is end-of-line.
+	eol = '\n'
+	// timestampFormat is the timestamp format used in CRI logging format.
+	timestampFormat = time.RFC3339Nano
+	// defaultBufSize is the default size of the read buffer in bytes.
+	defaultBufSize = 4096
+)
+
+// NewDiscardLogger creates logger which discards all the input.
+func NewDiscardLogger() io.WriteCloser {
+	return cioutil.NewNopWriteCloser(ioutil.Discard)
+}
+
+// NewCRILogger returns a write closer which redirect container log into
+// log file, and decorate the log line into CRI defined format. It also
+// returns a channel which indicates whether the logger is stopped.
+// maxLen is the max length limit of a line. A line longer than the
+// limit will be cut into multiple lines.
+func NewCRILogger(path string, w io.Writer, stream StreamType, maxLen int) (io.WriteCloser, <-chan struct{}) {
+	logrus.Debugf("Start writing stream %q to log file %q", stream, path)
+	prc, pwc := io.Pipe()
+	stop := make(chan struct{})
+	go func() {
+		redirectLogs(path, prc, w, stream, maxLen)
+		close(stop)
+	}()
+	return pwc, stop
+}
+
+// bufio.ReadLine in golang eats both read errors and tailing newlines
+// (See https://golang.org/pkg/bufio/#Reader.ReadLine). When reading
+// to io.EOF, it is impossible for the caller to figure out whether
+// there is a newline at the end, for example:
+// 1) When reading "CONTENT\n", it returns "CONTENT" without error;
+// 2) When reading "CONTENT", it also returns "CONTENT" without error.
+//
+// To differentiate these 2 cases, we need to write a readLine function
+// ourselves to not ignore the error.
+//
+// The code is similar with https://golang.org/src/bufio/bufio.go?s=9537:9604#L359.
+// The only difference is that it returns all errors from `ReadSlice`.
+//
+// readLine returns err != nil if and only if line does not end with a new line.
+func readLine(b *bufio.Reader) (line []byte, isPrefix bool, err error) {
+	line, err = b.ReadSlice('\n')
+	if err == bufio.ErrBufferFull {
+		// Handle the case where "\r\n" straddles the buffer.
+		if len(line) > 0 && line[len(line)-1] == '\r' {
+			// Unread the last '\r'
+			if err := b.UnreadByte(); err != nil {
+				panic(fmt.Sprintf("invalid unread %v", err))
+			}
+			line = line[:len(line)-1]
+		}
+		return line, true, nil
+	}
+
+	if len(line) == 0 {
+		if err != nil {
+			line = nil
+		}
+		return
+	}
+
+	if line[len(line)-1] == '\n' {
+		// "ReadSlice returns err != nil if and only if line does not end in delim"
+		// (See https://golang.org/pkg/bufio/#Reader.ReadSlice).
+		if err != nil {
+			panic(fmt.Sprintf("full read with unexpected error %v", err))
+		}
+		drop := 1
+		if len(line) > 1 && line[len(line)-2] == '\r' {
+			drop = 2
+		}
+		line = line[:len(line)-drop]
+	}
+	return
+}
+
+func redirectLogs(path string, rc io.ReadCloser, w io.Writer, s StreamType, maxLen int) {
+	defer rc.Close()
+	var (
+		stream    = []byte(s)
+		delimiter = []byte{delimiter}
+		partial   = []byte(runtime.LogTagPartial)
+		full      = []byte(runtime.LogTagFull)
+		buf       [][]byte
+		length    int
+		bufSize   = defaultBufSize
+	)
+	// Make sure bufSize <= maxLen
+	if maxLen > 0 && maxLen < bufSize {
+		bufSize = maxLen
+	}
+	r := bufio.NewReaderSize(rc, bufSize)
+	writeLine := func(tag, line []byte) {
+		timestamp := time.Now().AppendFormat(nil, timestampFormat)
+		data := bytes.Join([][]byte{timestamp, stream, tag, line}, delimiter)
+		data = append(data, eol)
+		if _, err := w.Write(data); err != nil {
+			logrus.WithError(err).Errorf("Fail to write %q log to log file %q", s, path)
+			// Continue on write error to drain the container output.
+		}
+	}
+	for {
+		var stop bool
+		newLine, isPrefix, err := readLine(r)
+		// NOTE(random-liu): readLine can return actual content even if there is an error.
+		if len(newLine) > 0 {
+			// Buffer returned by ReadLine will change after
+			// next read, copy it.
+			l := make([]byte, len(newLine))
+			copy(l, newLine)
+			buf = append(buf, l)
+			length += len(l)
+		}
+		if err != nil {
+			if err == io.EOF {
+				logrus.Debugf("Getting EOF from stream %q while redirecting to log file %q", s, path)
+			} else {
+				logrus.WithError(err).Errorf("An error occurred when redirecting stream %q to log file %q", s, path)
+			}
+			if length == 0 {
+				// No content left to write, break.
+				break
+			}
+			// Stop after writing the content left in buffer.
+			stop = true
+		}
+		if maxLen > 0 && length > maxLen {
+			exceedLen := length - maxLen
+			last := buf[len(buf)-1]
+			if exceedLen > len(last) {
+				// exceedLen must <= len(last), or else the buffer
+				// should have be written in the previous iteration.
+				panic("exceed length should <= last buffer size")
+			}
+			buf[len(buf)-1] = last[:len(last)-exceedLen]
+			writeLine(partial, bytes.Join(buf, nil))
+			buf = [][]byte{last[len(last)-exceedLen:]}
+			length = exceedLen
+		}
+		if isPrefix {
+			continue
+		}
+		if stop {
+			// readLine only returns error when the message doesn't
+			// end with a newline, in that case it should be treated
+			// as a partial line.
+			writeLine(partial, bytes.Join(buf, nil))
+		} else {
+			writeLine(full, bytes.Join(buf, nil))
+		}
+		buf = nil
+		length = 0
+		if stop {
+			break
+		}
+	}
+	logrus.Debugf("Finish redirecting stream %q to log file %q", s, path)
+}

pkg/cri/io/logger_test.go

@@ -0,0 +1,258 @@
+/*
+   Copyright The containerd Authors.
+
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
+   limitations under the License.
+*/
+
+package io
+
+import (
+	"bytes"
+	"io/ioutil"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	runtime "k8s.io/cri-api/pkg/apis/runtime/v1alpha2"
+
+	cioutil "github.com/containerd/containerd/pkg/ioutil"
+)
+
+func TestRedirectLogs(t *testing.T) {
+	// defaultBufSize is even number
+	const maxLen = defaultBufSize * 4
+	for desc, test := range map[string]struct {
+		input   string
+		stream  StreamType
+		maxLen  int
+		tag     []runtime.LogTag
+		content []string
+	}{
+		"stdout log": {
+			input:  "test stdout log 1\ntest stdout log 2\n",
+			stream: Stdout,
+			maxLen: maxLen,
+			tag: []runtime.LogTag{
+				runtime.LogTagFull,
+				runtime.LogTagFull,
+			},
+			content: []string{
+				"test stdout log 1",
+				"test stdout log 2",
+			},
+		},
+		"stderr log": {
+			input:  "test stderr log 1\ntest stderr log 2\n",
+			stream: Stderr,
+			maxLen: maxLen,
+			tag: []runtime.LogTag{
+				runtime.LogTagFull,
+				runtime.LogTagFull,
+			},
+			content: []string{
+				"test stderr log 1",
+				"test stderr log 2",
+			},
+		},
+		"log ends without newline": {
+			input:  "test stderr log 1\ntest stderr log 2",
+			stream: Stderr,
+			maxLen: maxLen,
+			tag: []runtime.LogTag{
+				runtime.LogTagFull,
+				runtime.LogTagPartial,
+			},
+			content: []string{
+				"test stderr log 1",
+				"test stderr log 2",
+			},
+		},
+		"log length equal to buffer size": {
+			input:  strings.Repeat("a", defaultBufSize) + "\n" + strings.Repeat("a", defaultBufSize) + "\n",
+			stream: Stdout,
+			maxLen: maxLen,
+			tag: []runtime.LogTag{
+				runtime.LogTagFull,
+				runtime.LogTagFull,
+			},
+			content: []string{
+				strings.Repeat("a", defaultBufSize),
+				strings.Repeat("a", defaultBufSize),
+			},
+		},
+		"log length longer than buffer size": {
+			input:  strings.Repeat("a", defaultBufSize*2+10) + "\n" + strings.Repeat("a", defaultBufSize*2+20) + "\n",
+			stream: Stdout,
+			maxLen: maxLen,
+			tag: []runtime.LogTag{
+				runtime.LogTagFull,
+				runtime.LogTagFull,
+			},
+			content: []string{
+				strings.Repeat("a", defaultBufSize*2+10),
+				strings.Repeat("a", defaultBufSize*2+20),
+			},
+		},
+		"log length equal to max length": {
+			input:  strings.Repeat("a", maxLen) + "\n" + strings.Repeat("a", maxLen) + "\n",
+			stream: Stdout,
+			maxLen: maxLen,
+			tag: []runtime.LogTag{
+				runtime.LogTagFull,
+				runtime.LogTagFull,
+			},
+			content: []string{
+				strings.Repeat("a", maxLen),
+				strings.Repeat("a", maxLen),
+			},
+		},
+		"log length exceed max length by 1": {
+			input:  strings.Repeat("a", maxLen+1) + "\n" + strings.Repeat("a", maxLen+1) + "\n",
+			stream: Stdout,
+			maxLen: maxLen,
+			tag: []runtime.LogTag{
+				runtime.LogTagPartial,
+				runtime.LogTagFull,
+				runtime.LogTagPartial,
+				runtime.LogTagFull,
+			},
+			content: []string{
+				strings.Repeat("a", maxLen),
+				"a",
+				strings.Repeat("a", maxLen),
+				"a",
+			},
+		},
+		"log length longer than max length": {
+			input:  strings.Repeat("a", maxLen*2) + "\n" + strings.Repeat("a", maxLen*2+1) + "\n",
+			stream: Stdout,
+			maxLen: maxLen,
+			tag: []runtime.LogTag{
+				runtime.LogTagPartial,
+				runtime.LogTagFull,
+				runtime.LogTagPartial,
+				runtime.LogTagPartial,
+				runtime.LogTagFull,
+			},
+			content: []string{
+				strings.Repeat("a", maxLen),
+				strings.Repeat("a", maxLen),
+				strings.Repeat("a", maxLen),
+				strings.Repeat("a", maxLen),
+				"a",
+			},
+		},
+		"max length shorter than buffer size": {
+			input:  strings.Repeat("a", defaultBufSize*3/2+10) + "\n" + strings.Repeat("a", defaultBufSize*3/2+20) + "\n",
+			stream: Stdout,
+			maxLen: defaultBufSize / 2,
+			tag: []runtime.LogTag{
+				runtime.LogTagPartial,
+				runtime.LogTagPartial,
+				runtime.LogTagPartial,
+				runtime.LogTagFull,
+				runtime.LogTagPartial,
+				runtime.LogTagPartial,
+				runtime.LogTagPartial,
+				runtime.LogTagFull,
+			},
+			content: []string{
+				strings.Repeat("a", defaultBufSize*1/2),
+				strings.Repeat("a", defaultBufSize*1/2),
+				strings.Repeat("a", defaultBufSize*1/2),
+				strings.Repeat("a", 10),
+				strings.Repeat("a", defaultBufSize*1/2),
+				strings.Repeat("a", defaultBufSize*1/2),
+				strings.Repeat("a", defaultBufSize*1/2),
+				strings.Repeat("a", 20),
+			},
+		},
+		"log length longer than max length, and (maxLen % defaultBufSize != 0)": {
+			input:  strings.Repeat("a", defaultBufSize*2+10) + "\n" + strings.Repeat("a", defaultBufSize*2+20) + "\n",
+			stream: Stdout,
+			maxLen: defaultBufSize * 3 / 2,
+			tag: []runtime.LogTag{
+				runtime.LogTagPartial,
+				runtime.LogTagFull,
+				runtime.LogTagPartial,
+				runtime.LogTagFull,
+			},
+			content: []string{
+				strings.Repeat("a", defaultBufSize*3/2),
+				strings.Repeat("a", defaultBufSize*1/2+10),
+				strings.Repeat("a", defaultBufSize*3/2),
+				strings.Repeat("a", defaultBufSize*1/2+20),
+			},
+		},
+		"no limit if max length is 0": {
+			input:  strings.Repeat("a", defaultBufSize*10+10) + "\n" + strings.Repeat("a", defaultBufSize*10+20) + "\n",
+			stream: Stdout,
+			maxLen: 0,
+			tag: []runtime.LogTag{
+				runtime.LogTagFull,
+				runtime.LogTagFull,
+			},
+			content: []string{
+				strings.Repeat("a", defaultBufSize*10+10),
+				strings.Repeat("a", defaultBufSize*10+20),
+			},
+		},
+		"no limit if max length is negative": {
+			input:  strings.Repeat("a", defaultBufSize*10+10) + "\n" + strings.Repeat("a", defaultBufSize*10+20) + "\n",
+			stream: Stdout,
+			maxLen: -1,
+			tag: []runtime.LogTag{
+				runtime.LogTagFull,
+				runtime.LogTagFull,
+			},
+			content: []string{
+				strings.Repeat("a", defaultBufSize*10+10),
+				strings.Repeat("a", defaultBufSize*10+20),
+			},
+		},
+		"log length longer than buffer size with tailing \\r\\n": {
+			input:  strings.Repeat("a", defaultBufSize-1) + "\r\n" + strings.Repeat("a", defaultBufSize-1) + "\r\n",
+			stream: Stdout,
+			maxLen: -1,
+			tag: []runtime.LogTag{
+				runtime.LogTagFull,
+				runtime.LogTagFull,
+			},
+			content: []string{
+				strings.Repeat("a", defaultBufSize-1),
+				strings.Repeat("a", defaultBufSize-1),
+			},
+		},
+	} {
+		t.Logf("TestCase %q", desc)
+		rc := ioutil.NopCloser(strings.NewReader(test.input))
+		buf := bytes.NewBuffer(nil)
+		wc := cioutil.NewNopWriteCloser(buf)
+		redirectLogs("test-path", rc, wc, test.stream, test.maxLen)
+		output := buf.String()
+		lines := strings.Split(output, "\n")
+		lines = lines[:len(lines)-1] // Discard empty string after last \n
+		assert.Len(t, lines, len(test.content))
+		for i := range lines {
+			fields := strings.SplitN(lines[i], string([]byte{delimiter}), 4)
+			require.Len(t, fields, 4)
+			_, err := time.Parse(timestampFormat, fields[0])
+			assert.NoError(t, err)
+			assert.EqualValues(t, test.stream, fields[1])
+			assert.Equal(t, string(test.tag[i]), fields[2])
+			assert.Equal(t, test.content[i], fields[3])
+		}
+	}
+}