agent: add configurable container pipe size cmdline option

Adds a cmdline option to configure the stdout/stderr pipe sizes.
Uses `F_SETPIPE_SZ` to resize the write side of the pipe after
creation.

Example Cmdline option: `agent.container_pipe_size=2097152`

fixes kata-containers#755

Signed-off-by: Alex Price <aprice@atlassian.com>

agent.go

@@ -177,6 +177,9 @@ var debugConsoleVSockPort = uint32(0)
 // Timeout waiting for a device to be hotplugged
 var hotplugTimeout = 3 * time.Second
 
+// Size of the stdout/stderr pipes created for each container.
+var containerPipeSize = uint32(0)
+
 // commType is used to denote the communication channel type used.
 type commType int
 

config.go

@@ -27,6 +27,7 @@ const (
 	debugConsoleFlag      = optionPrefix + "debug_console"
 	debugConsoleVPortFlag = optionPrefix + "debug_console_vport"
 	hotplugTimeoutFlag    = optionPrefix + "hotplug_timeout"
+	containerPipeSizeFlag = optionPrefix + "container_pipe_size"
 	kernelCmdlineFile     = "/proc/cmdline"
 	traceModeStatic       = "static"
 	traceModeDynamic      = "dynamic"
@@ -132,6 +133,12 @@ func (c *agentConfig) parseCmdlineOption(option string) error {
 		if timeout > 0 {
 			hotplugTimeout = timeout
 		}
+	case containerPipeSizeFlag:
+		size, err := strconv.ParseUint(split[valuePosition], 10, 32)
+		if err != nil {
+			return err
+		}
+		containerPipeSize = uint32(size)
 	case traceModeFlag:
 		switch split[valuePosition] {
 		case traceTypeIsolated:

config_test.go

@@ -451,3 +451,43 @@ func TestParseCmdlineOptionHotplugTimeout(t *testing.T) {
 		assert.Equal(d.expectedHotplugTimeout, hotplugTimeout, "test %d (%+v)", i, d)
 	}
 }
+
+func TestParseCmdlineOptionContainerPipeSize(t *testing.T) {
+	assert := assert.New(t)
+
+	a := &agentConfig{}
+
+	type testData struct {
+		option                    string
+		shouldErr                 bool
+		expectedContainerPipeSize uint32
+	}
+
+	data := []testData{
+		{"", false, 0},
+		{"container_pip_siz", false, 0},
+		{"container_pipe_size", false, 0},
+		{"container_pipe_size=3", false, 0},
+		{"agnt.container_pipe_size=3", false, 0},
+		{"agent.container_pipe_size=3", false, 3},
+		{"agent.container_pipe_size=2097152", false, 2097152},
+		{"agent.container_pipe_size=-1", true, 0},
+		{"agent.container_pipe_size=foobar", true, 0},
+		{"agent.container_pipe_size=5.0", true, 0},
+		{"agent.container_pipe_size=0", false, 0},
+	}
+
+	for i, d := range data {
+		// reset the container pipe size
+		containerPipeSize = uint32(0)
+
+		err := a.parseCmdlineOption(d.option)
+		if d.shouldErr {
+			assert.Error(err)
+		} else {
+			assert.NoError(err)
+		}
+
+		assert.Equal(d.expectedContainerPipeSize, containerPipeSize, "test %d (%+v)", i, d)
+	}
+}

grpc.go

@@ -344,12 +344,12 @@ func buildProcess(agentProcess *pb.Process, procID string, init bool) (*process,
 		return nil, err
 	}
 
-	rStdout, wStdout, err := os.Pipe()
+	rStdout, wStdout, err := createExtendedPipe()
 	if err != nil {
 		return nil, err
 	}
 
-	rStderr, wStderr, err := os.Pipe()
+	rStderr, wStderr, err := createExtendedPipe()
 	if err != nil {
 		return nil, err
 	}

util.go

@@ -0,0 +1,24 @@
+package main
+
+import (
+	"os"
+	"syscall"
+)
+
+// createExtendedPipe creates a pipe.
+// Optionally extends the pipe if containerPipeSize is positive.
+func createExtendedPipe() (*os.File, *os.File, error) {
+	r, w, err := os.Pipe()
+	if err != nil {
+		return nil, nil, err
+	}
+	if containerPipeSize > 0 {
+		extendPipe(r, w)
+	}
+	return r, w, nil
+}
+
+// extendPipe extends the write side of the pipe to value of containerPipeSize
+func extendPipe(r, w *os.File) {
+	syscall.Syscall(syscall.SYS_FCNTL, w.Fd(), syscall.F_SETPIPE_SZ, uintptr(containerPipeSize))
+}

util_test.go

@@ -0,0 +1,63 @@
+package main
+
+import (
+	"io/ioutil"
+	"os"
+	"strconv"
+	"strings"
+	"syscall"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestCreateExtendedPipe(t *testing.T) {
+	assert := assert.New(t)
+
+	// Test the default
+	containerPipeSize = 0
+	_, w, err := createExtendedPipe()
+	assert.NoError(err)
+	size, err := getPipeSize(w)
+	assert.Equal(syscall.Errno(0), err)
+	assert.Equal(uint32(65536), size) // Default = 16 pages
+
+	maxSize, err := getPipeMaxSize()
+	assert.NoError(err)
+
+	// Test setting to the max size
+	containerPipeSize = maxSize
+	_, w, err = createExtendedPipe()
+	assert.NoError(err)
+	size, err = getPipeSize(w)
+	assert.Equal(syscall.Errno(0), err)
+	assert.Equal(containerPipeSize, size)
+
+	// Test setting to a single page
+	containerPipeSize = 4096
+	_, w, err = createExtendedPipe()
+	assert.NoError(err)
+	size, err = getPipeSize(w)
+	assert.Equal(syscall.Errno(0), err)
+	assert.Equal(containerPipeSize, size)
+}
+
+func getPipeSize(f *os.File) (uint32, error){
+	r1, _, err := syscall.Syscall(syscall.SYS_FCNTL, f.Fd(), syscall.F_GETPIPE_SZ, 0)
+	return uint32(r1), err
+}
+
+func getPipeMaxSize() (uint32, error) {
+	f, err := os.Open("/proc/sys/fs/pipe-max-size")
+	if err != nil {
+		return 0, err
+	}
+	defer f.Close()
+	b, err := ioutil.ReadAll(f)
+	if err != nil {
+		return 0, err
+	}
+	s := strings.Trim(string(b), "\n")
+	u, err := strconv.ParseUint(s, 10, 32)
+	return uint32(u), err
+}