Exec probes should not be unbounded

In 1f270ef4e2, we added 10KB as the read
limit for http probes. we should do the same for exec probes as well.

Change-Id: If154c5c4e669829ab94839c56260a894a6714f0f

pkg/kubelet/prober/BUILD

@@ -58,6 +58,7 @@ go_test(
         "//pkg/kubelet/prober/results:go_default_library",
         "//pkg/kubelet/status:go_default_library",
         "//pkg/kubelet/status/testing:go_default_library",
+        "//pkg/kubelet/util/ioutils:go_default_library",
         "//pkg/probe:go_default_library",
         "//pkg/probe/exec:go_default_library",
         "//staging/src/k8s.io/api/core/v1:go_default_library",

pkg/kubelet/prober/prober.go

@@ -252,63 +252,68 @@ func formatURL(scheme string, host string, port int, path string) *url.URL {
 type execInContainer struct {
 	// run executes a command in a container. Combined stdout and stderr output is always returned. An
 	// error is returned if one occurred.
-	run func() ([]byte, error)
+	run    func() ([]byte, error)
+	writer io.Writer
 }
 
 func (pb *prober) newExecInContainer(container v1.Container, containerID kubecontainer.ContainerID, cmd []string, timeout time.Duration) exec.Cmd {
-	return execInContainer{func() ([]byte, error) {
+	return &execInContainer{run: func() ([]byte, error) {
 		return pb.runner.RunInContainer(containerID, cmd, timeout)
 	}}
 }
 
-func (eic execInContainer) Run() error {
+func (eic *execInContainer) Run() error {
-	return fmt.Errorf("unimplemented")
+	return nil
 }
 
-func (eic execInContainer) CombinedOutput() ([]byte, error) {
+func (eic *execInContainer) CombinedOutput() ([]byte, error) {
 	return eic.run()
 }
 
-func (eic execInContainer) Output() ([]byte, error) {
+func (eic *execInContainer) Output() ([]byte, error) {
 	return nil, fmt.Errorf("unimplemented")
 }
 
-func (eic execInContainer) SetDir(dir string) {
+func (eic *execInContainer) SetDir(dir string) {
 	//unimplemented
 }
 
-func (eic execInContainer) SetStdin(in io.Reader) {
+func (eic *execInContainer) SetStdin(in io.Reader) {
 	//unimplemented
 }
 
-func (eic execInContainer) SetStdout(out io.Writer) {
+func (eic *execInContainer) SetStdout(out io.Writer) {
+	eic.writer = out
+}
+
+func (eic *execInContainer) SetStderr(out io.Writer) {
+	eic.writer = out
+}
+
+func (eic *execInContainer) SetEnv(env []string) {
 	//unimplemented
 }
 
-func (eic execInContainer) SetStderr(out io.Writer) {
+func (eic *execInContainer) Stop() {
 	//unimplemented
 }
 
-func (eic execInContainer) SetEnv(env []string) {
+func (eic *execInContainer) Start() error {
-	//unimplemented
+	data, err := eic.run()
+	if eic.writer != nil {
+		eic.writer.Write(data)
+	}
+	return err
 }
 
-func (eic execInContainer) Stop() {
+func (eic *execInContainer) Wait() error {
-	//unimplemented
+	return nil
 }
 
-func (eic execInContainer) Start() error {
+func (eic *execInContainer) StdoutPipe() (io.ReadCloser, error) {
-	return fmt.Errorf("unimplemented")
-}
-
-func (eic execInContainer) Wait() error {
-	return fmt.Errorf("unimplemented")
-}
-
-func (eic execInContainer) StdoutPipe() (io.ReadCloser, error) {
 	return nil, fmt.Errorf("unimplemented")
 }
 
-func (eic execInContainer) StderrPipe() (io.ReadCloser, error) {
+func (eic *execInContainer) StderrPipe() (io.ReadCloser, error) {
 	return nil, fmt.Errorf("unimplemented")
 }

pkg/kubelet/prober/prober_test.go

@@ -17,10 +17,12 @@ limitations under the License.
 package prober
 
 import (
+	"bytes"
 	"errors"
 	"fmt"
 	"net/http"
 	"reflect"
+	"strings"
 	"testing"
 
 	"k8s.io/api/core/v1"
@@ -29,6 +31,7 @@ import (
 	kubecontainer "k8s.io/kubernetes/pkg/kubelet/container"
 	containertest "k8s.io/kubernetes/pkg/kubelet/container/testing"
 	"k8s.io/kubernetes/pkg/kubelet/prober/results"
+	"k8s.io/kubernetes/pkg/kubelet/util/ioutils"
 	"k8s.io/kubernetes/pkg/probe"
 	execprobe "k8s.io/kubernetes/pkg/probe/exec"
 )
@@ -329,23 +332,38 @@ func TestProbe(t *testing.T) {
 }
 
 func TestNewExecInContainer(t *testing.T) {
+	limit := 1024
+	tenKilobyte := strings.Repeat("logs-123", 128*10)
+
 	tests := []struct {
-		name string
+		name     string
-		err  error
+		stdout   string
+		expected string
+		err      error
 	}{
 		{
-			name: "no error",
+			name:     "no error",
-			err:  nil,
+			stdout:   "foo",
+			expected: "foo",
+			err:      nil,
 		},
 		{
-			name: "error - make sure we get output",
+			name:     "no error",
-			err:  errors.New("bad"),
+			stdout:   tenKilobyte,
+			expected: tenKilobyte[0:limit],
+			err:      nil,
+		},
+		{
+			name:     "error - make sure we get output",
+			stdout:   "foo",
+			expected: "foo",
+			err:      errors.New("bad"),
 		},
 	}
 
 	for _, test := range tests {
 		runner := &containertest.FakeContainerCommandRunner{
-			Stdout: "foo",
+			Stdout: test.stdout,
 			Err:    test.err,
 		}
 		prober := &prober{
@@ -357,7 +375,16 @@ func TestNewExecInContainer(t *testing.T) {
 		cmd := []string{"/foo", "bar"}
 		exec := prober.newExecInContainer(container, containerID, cmd, 0)
 
-		actualOutput, err := exec.CombinedOutput()
+		var dataBuffer bytes.Buffer
+		writer := ioutils.LimitWriter(&dataBuffer, int64(limit))
+		exec.SetStderr(writer)
+		exec.SetStdout(writer)
+		err := exec.Start()
+		if err == nil {
+			err = exec.Wait()
+		}
+		actualOutput := dataBuffer.Bytes()
+
 		if e, a := containerID, runner.ContainerID; e != a {
 			t.Errorf("%s: container id: expected %v, got %v", test.name, e, a)
 		}
@@ -365,7 +392,7 @@ func TestNewExecInContainer(t *testing.T) {
 			t.Errorf("%s: cmd: expected %v, got %v", test.name, e, a)
 		}
 		// this isn't 100% foolproof as a bug in a real ContainerCommandRunner where it fails to copy to stdout/stderr wouldn't be caught by this test
-		if e, a := "foo", string(actualOutput); e != a {
+		if e, a := test.expected, string(actualOutput); e != a {
 			t.Errorf("%s: output: expected %q, got %q", test.name, e, a)
 		}
 		if e, a := fmt.Sprintf("%v", test.err), fmt.Sprintf("%v", err); e != a {

pkg/kubelet/util/ioutils/BUILD

@@ -1,9 +1,6 @@
 package(default_visibility = ["//visibility:public"])
 
-load(
+load("@io_bazel_rules_go//go:def.bzl", "go_library", "go_test")
-    "@io_bazel_rules_go//go:def.bzl",
-    "go_library",
-)
 
 go_library(
     name = "go_default_library",
@@ -23,3 +20,10 @@ filegroup(
     srcs = [":package-srcs"],
     tags = ["automanaged"],
 )
+
+go_test(
+    name = "go_default_test",
+    srcs = ["ioutils_test.go"],
+    embed = [":go_default_library"],
+    deps = ["//vendor/github.com/stretchr/testify/assert:go_default_library"],
+)

pkg/probe/exec/BUILD

@@ -11,6 +11,7 @@ go_library(
     srcs = ["exec.go"],
     importpath = "k8s.io/kubernetes/pkg/probe/exec",
     deps = [
+        "//pkg/kubelet/util/ioutils:go_default_library",
         "//pkg/probe:go_default_library",
         "//vendor/k8s.io/klog:go_default_library",
         "//vendor/k8s.io/utils/exec:go_default_library",

pkg/probe/exec/exec.go

@@ -17,10 +17,17 @@ limitations under the License.
 package exec
 
 import (
+	"bytes"
+
+	"k8s.io/kubernetes/pkg/kubelet/util/ioutils"
 	"k8s.io/kubernetes/pkg/probe"
-	"k8s.io/utils/exec"
 
 	"k8s.io/klog"
+	"k8s.io/utils/exec"
+)
+
+const (
+	maxReadLength = 10 * 1 << 10 // 10KB
 )
 
 // New creates a Prober.
@@ -39,7 +46,17 @@ type execProber struct{}
 // from executing a command. Returns the Result status, command output, and
 // errors if any.
 func (pr execProber) Probe(e exec.Cmd) (probe.Result, string, error) {
-	data, err := e.CombinedOutput()
+	var dataBuffer bytes.Buffer
+	writer := ioutils.LimitWriter(&dataBuffer, maxReadLength)
+
+	e.SetStderr(writer)
+	e.SetStdout(writer)
+	err := e.Start()
+	if err == nil {
+		err = e.Wait()
+	}
+	data := dataBuffer.Bytes()
+
 	klog.V(4).Infof("Exec probe response: %q", string(data))
 	if err != nil {
 		exit, ok := err.(exec.ExitError)

pkg/probe/exec/exec_test.go

@@ -19,6 +19,7 @@ package exec
 import (
 	"fmt"
 	"io"
+	"strings"
 	"testing"
 
 	"k8s.io/kubernetes/pkg/probe"
@@ -28,6 +29,7 @@ type FakeCmd struct {
 	out    []byte
 	stdout []byte
 	err    error
+	writer io.Writer
 }
 
 func (f *FakeCmd) Run() error {
@@ -46,15 +48,25 @@ func (f *FakeCmd) SetDir(dir string) {}
 
 func (f *FakeCmd) SetStdin(in io.Reader) {}
 
-func (f *FakeCmd) SetStdout(out io.Writer) {}
+func (f *FakeCmd) SetStdout(out io.Writer) {
+	f.writer = out
+}
 
-func (f *FakeCmd) SetStderr(out io.Writer) {}
+func (f *FakeCmd) SetStderr(out io.Writer) {
+	f.writer = out
+}
 
 func (f *FakeCmd) SetEnv(env []string) {}
 
 func (f *FakeCmd) Stop() {}
 
-func (f *FakeCmd) Start() error { return nil }
+func (f *FakeCmd) Start() error {
+	if f.writer != nil {
+		f.writer.Write(f.out)
+		return f.err
+	}
+	return f.err
+}
 
 func (f *FakeCmd) Wait() error { return nil }
 
@@ -90,20 +102,26 @@ func (f *fakeExitError) ExitStatus() int {
 func TestExec(t *testing.T) {
 	prober := New()
 
+	tenKilobyte := strings.Repeat("logs-123", 128*10)      // 8*128*10=10240 = 10KB of text.
+	elevenKilobyte := strings.Repeat("logs-123", 8*128*11) // 8*128*11=11264 = 11KB of text.
+
 	tests := []struct {
 		expectedStatus probe.Result
 		expectError    bool
+		input          string
 		output         string
 		err            error
 	}{
 		// Ok
-		{probe.Success, false, "OK", nil},
+		{probe.Success, false, "OK", "OK", nil},
 		// Ok
-		{probe.Success, false, "OK", &fakeExitError{true, 0}},
+		{probe.Success, false, "OK", "OK", &fakeExitError{true, 0}},
+		// Ok - truncated output
+		{probe.Success, false, elevenKilobyte, tenKilobyte, nil},
 		// Run returns error
-		{probe.Unknown, true, "", fmt.Errorf("test error")},
+		{probe.Unknown, true, "", "", fmt.Errorf("test error")},
 		// Unhealthy
-		{probe.Failure, false, "Fail", &fakeExitError{true, 1}},
+		{probe.Failure, false, "Fail", "", &fakeExitError{true, 1}},
 	}
 	for i, test := range tests {
 		fake := FakeCmd{