[1.20] oci: add support for capping memory and disk usage from exec sync output

internal/config/conmonmgr/conmonmgr.go

@@ -1,6 +1,7 @@
 package conmonmgr
 
 import (
+	"bytes"
 	"path"
 	"strings"
 
@@ -11,10 +12,12 @@ import (
 )
 
 var versionSupportsSync = semver.MustParse("2.0.19")
+var versionSupportsLogGlobalSizeMax = semver.MustParse("2.1.2")
 
 type ConmonManager struct {
-	conmonVersion *semver.Version
-	supportsSync  bool
+	conmonVersion            *semver.Version
+	supportsSync             bool
+	supportsLogGlobalSizeMax bool
 }
 
 // this function is heavily based on github.com/containers/common#probeConmon
@@ -37,6 +40,7 @@ func New(conmonPath string) (*ConmonManager, error) {
 	}
 
 	c.initializeSupportsSync()
+	c.initializeSupportsLogGlobalSizeMax(conmonPath)
 	return c, nil
 }
 
@@ -49,6 +53,26 @@ func (c *ConmonManager) parseConmonVersion(versionString string) error {
 	return nil
 }
 
+func (c *ConmonManager) initializeSupportsLogGlobalSizeMax(conmonPath string) {
+	c.supportsLogGlobalSizeMax = c.conmonVersion.GTE(versionSupportsLogGlobalSizeMax)
+	if !c.supportsLogGlobalSizeMax {
+		// Read help output as a fallback in case the feature was backported to conmon,
+		// but the version wasn't bumped.
+		helpOutput, err := cmdrunner.CombinedOutput(conmonPath, "--help")
+		c.supportsLogGlobalSizeMax = err == nil && bytes.Contains(helpOutput, []byte("--log-global-size-max"))
+	}
+	verb := "does not"
+	if c.supportsLogGlobalSizeMax {
+		verb = "does"
+	}
+
+	logrus.Infof("Conmon %s support the --log-global-size-max option", verb)
+}
+
+func (c *ConmonManager) SupportsLogGlobalSizeMax() bool {
+	return c.supportsLogGlobalSizeMax
+}
+
 func (c *ConmonManager) initializeSupportsSync() {
 	c.supportsSync = c.conmonVersion.GTE(versionSupportsSync)
 	verb := "does not"

internal/config/conmonmgr/conmonmgr_test.go

@@ -61,20 +61,7 @@ var _ = t.Describe("ConmonManager", func() {
 		It("should succeed when output expected", func() {
 			// Given
 			gomock.InOrder(
-				runner.EXPECT().CombinedOutput(gomock.Any(), gomock.Any()).Return([]byte("conmon version 2.0.0"), nil),
-			)
-
-			// When
-			mgr, err := New(validPath)
-
-			// Then
-			Expect(err).To(BeNil())
-			Expect(mgr).ToNot(BeNil())
-		})
-		It("should succeed when output expected", func() {
-			// Given
-			gomock.InOrder(
-				runner.EXPECT().CombinedOutput(gomock.Any(), gomock.Any()).Return([]byte("conmon version 2.0.0"), nil),
+				runner.EXPECT().CombinedOutput(gomock.Any(), gomock.Any()).Return([]byte("conmon version 2.2.2"), nil),
 			)
 
 			// When
@@ -174,4 +161,108 @@ var _ = t.Describe("ConmonManager", func() {
 			Expect(mgr.SupportsSync()).To(Equal(true))
 		})
 	})
+	t.Describe("initializeSupportsLogGlobalSizeMax", func() {
+		var mgr *ConmonManager
+		BeforeEach(func() {
+			runner = runnerMock.NewMockCommandRunner(mockCtrl)
+			cmdrunner.SetMocked(runner)
+			mgr = new(ConmonManager)
+		})
+		It("should be false when major version less", func() {
+			// Given
+			gomock.InOrder(
+				runner.EXPECT().CombinedOutput(gomock.Any(), gomock.Any()).Return([]byte{}, errors.New("cmd failed")),
+			)
+			err := mgr.parseConmonVersion("1.1.2")
+			Expect(err).To(BeNil())
+			// When
+			mgr.initializeSupportsLogGlobalSizeMax("")
+
+			// Then
+			Expect(mgr.SupportsLogGlobalSizeMax()).To(Equal(false))
+		})
+		It("should be true when major version greater", func() {
+			// Given
+			err := mgr.parseConmonVersion("3.1.1")
+			Expect(err).To(BeNil())
+
+			// When
+			mgr.initializeSupportsLogGlobalSizeMax("")
+
+			// Then
+			Expect(mgr.SupportsLogGlobalSizeMax()).To(Equal(true))
+		})
+		It("should be false when minor version less", func() {
+			// Given
+			gomock.InOrder(
+				runner.EXPECT().CombinedOutput(gomock.Any(), gomock.Any()).Return([]byte{}, errors.New("cmd failed")),
+			)
+			err := mgr.parseConmonVersion("2.0.2")
+			Expect(err).To(BeNil())
+			// When
+			mgr.initializeSupportsLogGlobalSizeMax("")
+
+			// Then
+			Expect(mgr.SupportsLogGlobalSizeMax()).To(Equal(false))
+		})
+		It("should be true when minor version greater", func() {
+			// Given
+			err := mgr.parseConmonVersion("2.2.2")
+			Expect(err).To(BeNil())
+
+			// When
+			mgr.initializeSupportsLogGlobalSizeMax("")
+
+			// Then
+			Expect(mgr.SupportsLogGlobalSizeMax()).To(Equal(true))
+		})
+		It("should be false when patch version less", func() {
+			// Given
+			gomock.InOrder(
+				runner.EXPECT().CombinedOutput(gomock.Any(), gomock.Any()).Return([]byte{}, errors.New("cmd failed")),
+			)
+			err := mgr.parseConmonVersion("2.1.1")
+			Expect(err).To(BeNil())
+			// When
+			mgr.initializeSupportsLogGlobalSizeMax("")
+
+			// Then
+			Expect(mgr.SupportsLogGlobalSizeMax()).To(Equal(false))
+		})
+		It("should be true when patch version greater", func() {
+			// Given
+			err := mgr.parseConmonVersion("2.1.3")
+			Expect(err).To(BeNil())
+
+			// When
+			mgr.initializeSupportsLogGlobalSizeMax("")
+
+			// Then
+			Expect(mgr.SupportsLogGlobalSizeMax()).To(Equal(true))
+		})
+		It("should be true when version equal", func() {
+			// Given
+			err := mgr.parseConmonVersion("2.1.2")
+			Expect(err).To(BeNil())
+
+			// When
+			mgr.initializeSupportsLogGlobalSizeMax("")
+			// Then
+			Expect(mgr.SupportsLogGlobalSizeMax()).To(Equal(true))
+		})
+		It("should be true if feature backported", func() {
+			// Given
+			gomock.InOrder(
+				runner.EXPECT().CombinedOutput(gomock.Any(), gomock.Any()).Return([]byte("--log-global-size-max"), nil),
+			)
+			err := mgr.parseConmonVersion("0.0.0")
+			Expect(err).To(BeNil())
+
+			// When
+			mgr.initializeSupportsLogGlobalSizeMax("")
+
+			// Then
+			Expect(mgr.SupportsLogGlobalSizeMax()).To(Equal(true))
+		})
+	})
 })

internal/oci/oci.go

@@ -32,6 +32,11 @@ const (
 	// killContainerTimeout is the timeout that we wait for the container to
 	// be SIGKILLed.
 	killContainerTimeout = 2 * time.Minute
+
+	// maxExecSyncSize is the maximum size of exec sync output CRI-O will process.
+	// It is set to the amount of logs allowed in the dockershim implementation:
+	// https://github.com/kubernetes/kubernetes/pull/82514
+	maxExecSyncSize = 16 * 1024 * 1024
 )
 
 // Runtime is the generic structure holding both global and specific

internal/oci/runtime_oci.go

@@ -385,6 +385,9 @@ func (r *runtimeOCI) ExecSyncContainer(ctx context.Context, c *Container, comman
 	if r.config.ConmonSupportsSync() {
 		args = append(args, "--sync")
 	}
+	if r.config.ConmonSupportsLogGlobalSizeMax() {
+		args = append(args, "--log-global-size-max", strconv.Itoa(maxExecSyncSize))
+	}
 	if c.terminal {
 		args = append(args, "-t")
 	}
@@ -491,7 +494,7 @@ func (r *runtimeOCI) ExecSyncContainer(ctx context.Context, c *Container, comman
 	// ExecSyncResponse we have to read the logfile.
 	// XXX: Currently runC dups the same console over both stdout and stderr,
 	//      so we can't differentiate between the two.
-	logBytes, err := ioutil.ReadFile(logPath)
+	logBytes, err := TruncateAndReadFile(ctx, logPath, maxExecSyncSize)
 	if err != nil {
 		return nil, &ExecSyncError{
 			Stdout:   stdoutBuf,
@@ -630,6 +633,20 @@ func pidAndpgidFromFile(pidFile string) (pid, pgid int, _ error) {
 	return pid, pgid, err
 }
 
+func TruncateAndReadFile(ctx context.Context, path string, size int64) ([]byte, error) {
+	info, err := os.Stat(path)
+	if err != nil {
+		return nil, err
+	}
+	if info.Size() > size {
+		log.Errorf(ctx, "exec sync output in file %s has size %d which is longer than expected size of %d", path, info.Size(), size)
+		if err := os.Truncate(path, size); err != nil {
+			return nil, err
+		}
+	}
+	return ioutil.ReadFile(path)
+}
+
 // UpdateContainer updates container resources
 func (r *runtimeOCI) UpdateContainer(c *Container, res *rspec.LinuxResources) error {
 	if c.Spoofed() {

internal/oci/runtime_oci_test.go

@@ -0,0 +1,58 @@
+package oci_test
+
+import (
+	"context"
+	"io/ioutil"
+
+	"github.com/cri-o/cri-o/internal/oci"
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+)
+
+const (
+	shortTimeout  int64 = 1
+	mediumTimeout int64 = 3
+	longTimeout   int64 = 15
+)
+
+// The actual test suite
+var _ = t.Describe("Oci", func() {
+	Context("TruncateAndReadFile", func() {
+		tests := []struct {
+			title    string
+			contents []byte
+			expected []byte
+			fail     bool
+			size     int64
+		}{
+			{
+				title:    "should read file if size is smaller than limit",
+				contents: []byte("abcd"),
+				expected: []byte("abcd"),
+				size:     5,
+			},
+			{
+				title:    "should read only size if size is same as limit",
+				contents: []byte("abcd"),
+				expected: []byte("abcd"),
+				size:     4,
+			},
+			{
+				title:    "should read only size if size is larger than limit",
+				contents: []byte("abcd"),
+				expected: []byte("abc"),
+				size:     3,
+			},
+		}
+		for _, test := range tests {
+			test := test
+			It(test.title, func() {
+				fileName := t.MustTempFile("to-read")
+				Expect(ioutil.WriteFile(fileName, test.contents, 0644)).To(BeNil())
+				found, err := oci.TruncateAndReadFile(context.Background(), fileName, test.size)
+				Expect(err).To(BeNil())
+				Expect(found).To(Equal(test.expected))
+			})
+		}
+	})
+})

internal/oci/runtime_vm.go

@@ -31,6 +31,7 @@ import (
 	"golang.org/x/sys/unix"
 	"k8s.io/client-go/tools/remotecommand"
 	kubecontainer "k8s.io/kubernetes/pkg/kubelet/container"
+	kioutil "k8s.io/kubernetes/pkg/kubelet/util/ioutils"
 	utilexec "k8s.io/utils/exec"
 )
 
@@ -309,8 +310,8 @@ func (r *runtimeVM) ExecSyncContainer(ctx context.Context, c *Container, command
 	defer logrus.Debug("runtimeVM.ExecSyncContainer() end")
 
 	var stdoutBuf, stderrBuf bytes.Buffer
-	stdout := cioutil.NewNopWriteCloser(&stdoutBuf)
-	stderr := cioutil.NewNopWriteCloser(&stderrBuf)
+	stdout := kioutil.WriteCloserWrapper(kioutil.LimitWriter(&stdoutBuf, maxExecSyncSize))
+	stderr := kioutil.WriteCloserWrapper(kioutil.LimitWriter(&stderrBuf, maxExecSyncSize))
 
 	exitCode, err := r.execContainerCommon(c, command, timeout, nil, stdout, stderr, c.terminal, nil)
 	if err != nil {

internal/version/version.go

@@ -21,7 +21,7 @@ import (
 )
 
 // Version is the version of the build.
-const Version = "1.20.7"
+const Version = "1.20.8"
 
 // Variables injected during build-time
 var (

pkg/config/config.go

@@ -993,6 +993,10 @@ func (c *RuntimeConfig) ConmonSupportsSync() bool {
 	return c.conmonManager.SupportsSync()
 }
 
+func (c *RuntimeConfig) ConmonSupportsLogGlobalSizeMax() bool {
+	return c.conmonManager.SupportsLogGlobalSizeMax()
+}
+
 func (c *RuntimeConfig) ValidatePinnsPath(executable string) error {
 	var err error
 	c.PinnsPath, err = validateExecutablePath(executable, c.PinnsPath)

test/ctr.bats

@@ -457,6 +457,14 @@ function wait_until_exit() {
 	crictl rm -f "$ctr_id"
 }
 
+@test "ctr execsync should cap output" {
+	start_crio
+
+	ctr_id=$(crictl run "$TESTDATA"/container_sleep.json "$TESTDATA"/sandbox_config.json)
+
+	[[ $(crictl exec --sync "$ctr_id" /bin/sh -c "for i in $(seq 1 50000000); do echo -n 'a'; done" | wc -c) -le 16777216 ]]
+}
+
 @test "ctr device add" {
 	# In an user namespace we can only bind mount devices from the host, not mknod
 	# https://github.com/opencontainers/runc/blob/master/libcontainer/rootfs_linux.go#L480-L481