451 lines
14 KiB
Diff
451 lines
14 KiB
Diff
diff --git a/internal/config/conmonmgr/conmonmgr.go b/internal/config/conmonmgr/conmonmgr.go
|
|
index 857437c3f..e95e27484 100644
|
|
--- a/internal/config/conmonmgr/conmonmgr.go
|
|
+++ b/internal/config/conmonmgr/conmonmgr.go
|
|
@@ -1,6 +1,7 @@
|
|
package conmonmgr
|
|
|
|
import (
|
|
+ "bytes"
|
|
"path"
|
|
"strings"
|
|
|
|
@@ -10,11 +11,15 @@ import (
|
|
"github.com/sirupsen/logrus"
|
|
)
|
|
|
|
-var versionSupportsSync = semver.MustParse("2.0.19")
|
|
+var (
|
|
+ versionSupportsSync = semver.MustParse("2.0.19")
|
|
+ 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 +42,7 @@ func New(conmonPath string) (*ConmonManager, error) {
|
|
}
|
|
|
|
c.initializeSupportsSync()
|
|
+ c.initializeSupportsLogGlobalSizeMax(conmonPath)
|
|
return c, nil
|
|
}
|
|
|
|
@@ -49,6 +55,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"
|
|
diff --git a/internal/config/conmonmgr/conmonmgr_test.go b/internal/config/conmonmgr/conmonmgr_test.go
|
|
index fae450cb3..dbcca9232 100644
|
|
--- a/internal/config/conmonmgr/conmonmgr_test.go
|
|
+++ b/internal/config/conmonmgr/conmonmgr_test.go
|
|
@@ -70,7 +70,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),
|
|
+ runner.EXPECT().CombinedOutput(gomock.Any(), gomock.Any()).Return([]byte("conmon version 2.2.2"), nil),
|
|
)
|
|
|
|
// When
|
|
@@ -170,4 +170,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))
|
|
+ })
|
|
+ })
|
|
})
|
|
diff --git a/internal/oci/oci.go b/internal/oci/oci.go
|
|
index 1ea398e13..b377c9a9e 100644
|
|
--- a/internal/oci/oci.go
|
|
+++ b/internal/oci/oci.go
|
|
@@ -35,6 +35,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
|
|
diff --git a/internal/oci/runtime_oci.go b/internal/oci/runtime_oci.go
|
|
index 208ea884f..a5b698f8f 100644
|
|
--- a/internal/oci/runtime_oci.go
|
|
+++ b/internal/oci/runtime_oci.go
|
|
@@ -486,6 +486,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")
|
|
}
|
|
@@ -661,7 +664,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,
|
|
@@ -680,6 +683,20 @@ func (r *runtimeOCI) ExecSyncContainer(ctx context.Context, c *Container, comman
|
|
}, nil
|
|
}
|
|
|
|
+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 os.ReadFile(path)
|
|
+}
|
|
+
|
|
// UpdateContainer updates container resources
|
|
func (r *runtimeOCI) UpdateContainer(ctx context.Context, c *Container, res *rspec.LinuxResources) error {
|
|
if c.Spoofed() {
|
|
diff --git a/internal/oci/runtime_oci_test.go b/internal/oci/runtime_oci_test.go
|
|
index 37ee8a1e9..7120f0872 100644
|
|
--- a/internal/oci/runtime_oci_test.go
|
|
+++ b/internal/oci/runtime_oci_test.go
|
|
@@ -3,6 +3,7 @@ package oci_test
|
|
import (
|
|
"context"
|
|
"math/rand"
|
|
+ "os"
|
|
"os/exec"
|
|
"time"
|
|
|
|
@@ -142,6 +143,44 @@ 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(os.WriteFile(fileName, test.contents, 0o644)).To(BeNil())
|
|
+ found, err := oci.TruncateAndReadFile(context.Background(), fileName, test.size)
|
|
+ Expect(err).To(BeNil())
|
|
+ Expect(found).To(Equal(test.expected))
|
|
+ })
|
|
+ }
|
|
+ })
|
|
})
|
|
|
|
func waitContainerStopAndFailAfterTimeout(ctx context.Context,
|
|
diff --git a/internal/oci/runtime_vm.go b/internal/oci/runtime_vm.go
|
|
index 9dcc54892..c49686ee3 100644
|
|
--- a/internal/oci/runtime_vm.go
|
|
+++ b/internal/oci/runtime_vm.go
|
|
@@ -39,6 +39,7 @@ import (
|
|
"k8s.io/client-go/tools/remotecommand"
|
|
types "k8s.io/cri-api/pkg/apis/runtime/v1"
|
|
kubecontainer "k8s.io/kubernetes/pkg/kubelet/container"
|
|
+ kioutil "k8s.io/kubernetes/pkg/kubelet/util/ioutils"
|
|
utilexec "k8s.io/utils/exec"
|
|
)
|
|
|
|
@@ -307,8 +308,8 @@ func (r *runtimeVM) ExecSyncContainer(ctx context.Context, c *Container, command
|
|
defer log.Debugf(ctx, "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(ctx, c, command, timeout, nil, stdout, stderr, c.terminal, nil)
|
|
if err != nil {
|
|
diff --git a/pkg/config/config.go b/pkg/config/config.go
|
|
index 3128a81f4..402fcf3c2 100644
|
|
--- a/pkg/config/config.go
|
|
+++ b/pkg/config/config.go
|
|
@@ -1183,6 +1183,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)
|
|
diff --git a/test/ctr.bats b/test/ctr.bats
|
|
index be4122af8..b822468a2 100644
|
|
--- a/test/ctr.bats
|
|
+++ b/test/ctr.bats
|
|
@@ -514,6 +514,14 @@ function check_oci_annotation() {
|
|
crictl exec --sync "$ctr_id" /bin/sh -c "[[ -t 1 ]]"
|
|
}
|
|
|
|
+@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
|
|
diff --git a/vendor/k8s.io/kubernetes/pkg/kubelet/util/ioutils/ioutils.go b/vendor/k8s.io/kubernetes/pkg/kubelet/util/ioutils/ioutils.go
|
|
new file mode 100644
|
|
index 000000000..1b2b5a6d5
|
|
--- /dev/null
|
|
+++ b/vendor/k8s.io/kubernetes/pkg/kubelet/util/ioutils/ioutils.go
|
|
@@ -0,0 +1,70 @@
|
|
+/*
|
|
+Copyright 2016 The Kubernetes 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 ioutils
|
|
+
|
|
+import "io"
|
|
+
|
|
+// writeCloserWrapper represents a WriteCloser whose closer operation is noop.
|
|
+type writeCloserWrapper struct {
|
|
+ Writer io.Writer
|
|
+}
|
|
+
|
|
+func (w *writeCloserWrapper) Write(buf []byte) (int, error) {
|
|
+ return w.Writer.Write(buf)
|
|
+}
|
|
+
|
|
+func (w *writeCloserWrapper) Close() error {
|
|
+ return nil
|
|
+}
|
|
+
|
|
+// WriteCloserWrapper returns a writeCloserWrapper.
|
|
+func WriteCloserWrapper(w io.Writer) io.WriteCloser {
|
|
+ return &writeCloserWrapper{w}
|
|
+}
|
|
+
|
|
+// LimitWriter is a copy of the standard library ioutils.LimitReader,
|
|
+// applied to the writer interface.
|
|
+// LimitWriter returns a Writer that writes to w
|
|
+// but stops with EOF after n bytes.
|
|
+// The underlying implementation is a *LimitedWriter.
|
|
+func LimitWriter(w io.Writer, n int64) io.Writer { return &LimitedWriter{w, n} }
|
|
+
|
|
+// A LimitedWriter writes to W but limits the amount of
|
|
+// data returned to just N bytes. Each call to Write
|
|
+// updates N to reflect the new amount remaining.
|
|
+// Write returns EOF when N <= 0 or when the underlying W returns EOF.
|
|
+type LimitedWriter struct {
|
|
+ W io.Writer // underlying writer
|
|
+ N int64 // max bytes remaining
|
|
+}
|
|
+
|
|
+func (l *LimitedWriter) Write(p []byte) (n int, err error) {
|
|
+ if l.N <= 0 {
|
|
+ return 0, io.ErrShortWrite
|
|
+ }
|
|
+ truncated := false
|
|
+ if int64(len(p)) > l.N {
|
|
+ p = p[0:l.N]
|
|
+ truncated = true
|
|
+ }
|
|
+ n, err = l.W.Write(p)
|
|
+ l.N -= int64(n)
|
|
+ if err == nil && truncated {
|
|
+ err = io.ErrShortWrite
|
|
+ }
|
|
+ return
|
|
+}
|
|
diff --git a/vendor/modules.txt b/vendor/modules.txt
|
|
index 6bc34e160..1f3f15b79 100644
|
|
--- a/vendor/modules.txt
|
|
+++ b/vendor/modules.txt
|
|
@@ -1741,6 +1741,7 @@ k8s.io/kubernetes/pkg/kubelet/cri/streaming
|
|
k8s.io/kubernetes/pkg/kubelet/cri/streaming/portforward
|
|
k8s.io/kubernetes/pkg/kubelet/cri/streaming/remotecommand
|
|
k8s.io/kubernetes/pkg/kubelet/types
|
|
+k8s.io/kubernetes/pkg/kubelet/util/ioutils
|
|
k8s.io/kubernetes/pkg/proxy
|
|
k8s.io/kubernetes/pkg/proxy/config
|
|
k8s.io/kubernetes/pkg/proxy/healthcheck
|