Merge pull request #1771 from vishh/native_exec

Use native exec support in docker instead of exec'ing nsinit in kubelet.

pkg/health/exec.go

@@ -18,13 +18,13 @@ package health
 
 import (
 	"fmt"
-	"os/exec"
+	"strings"
 
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
 	"github.com/golang/glog"
 )
 
-const defaultHealthyRegex = "^OK$"
+const defaultHealthyOutput = "ok"
 
 type CommandRunner interface {
 	RunInContainer(podFullName, uuid, containerName string, cmd []string) ([]byte, error)
@@ -38,11 +38,6 @@ func NewExecHealthChecker(runner CommandRunner) HealthChecker {
 	return &ExecHealthChecker{runner}
 }
 
-func IsExitError(err error) bool {
-	_, ok := err.(*exec.ExitError)
-	return ok
-}
-
 func (e *ExecHealthChecker) HealthCheck(podFullName, podUUID string, currentState api.PodState, container api.Container) (Status, error) {
 	if container.LivenessProbe.Exec == nil {
 		return Unknown, fmt.Errorf("Missing exec parameters")
@@ -50,11 +45,11 @@ func (e *ExecHealthChecker) HealthCheck(podFullName, podUUID string, currentStat
 	data, err := e.runner.RunInContainer(podFullName, podUUID, container.Name, container.LivenessProbe.Exec.Command)
 	glog.V(1).Infof("container %s failed health check: %s", podFullName, string(data))
 	if err != nil {
-		if IsExitError(err) {
-			return Unhealthy, nil
-		}
 		return Unknown, err
 	}
+	if strings.ToLower(string(data)) != defaultHealthyOutput {
+		return Unhealthy, nil
+	}
 	return Healthy, nil
 }
 

pkg/health/exec_test.go

@@ -18,7 +18,6 @@ package health
 
 import (
 	"fmt"
-	"os/exec"
 	"reflect"
 	"testing"
 
@@ -60,12 +59,10 @@ func TestExec(t *testing.T) {
 				Command: []string{"ls", "-l"},
 			},
 		}, true, []byte("OK, NOT"), fmt.Errorf("test error")},
-		// Command error
+		// Unhealthy
 		{Unhealthy, &api.LivenessProbe{
-			Exec: &api.ExecAction{
+			Exec: &api.ExecAction{Command: []string{"ls", "-l"}},
-				Command: []string{"ls", "-l"},
+		}, false, []byte("Fail"), nil},
-			},
-		}, false, []byte{}, &exec.ExitError{}},
 	}
 	for _, test := range tests {
 		fake.out = test.output

pkg/kubelet/dockertools/docker.go

@@ -17,6 +17,8 @@ limitations under the License.
 package dockertools
 
 import (
+	"bufio"
+	"bytes"
 	"errors"
 	"fmt"
 	"hash/adler32"
@@ -51,6 +53,9 @@ type DockerInterface interface {
 	InspectImage(image string) (*docker.Image, error)
 	PullImage(opts docker.PullImageOptions, auth docker.AuthConfiguration) error
 	Logs(opts docker.LogsOptions) error
+	Version() (*docker.Env, error)
+	CreateExec(docker.CreateExecOptions) (*docker.Exec, error)
+	StartExec(string, docker.StartExecOptions) error
 }
 
 // DockerID is an ID of docker container. It is a type to make it clear when we're working with docker container Ids
@@ -99,7 +104,51 @@ func NewDockerPuller(client DockerInterface, qps float32, burst int) DockerPulle
 	}
 }
 
-type dockerContainerCommandRunner struct{}
+type dockerContainerCommandRunner struct {
+	client DockerInterface
+}
+
+// The first version of docker that supports exec natively is 1.1.3
+var dockerVersionWithExec = []uint{1, 1, 3}
+
+// Returns the major and minor version numbers of docker server.
+func (d *dockerContainerCommandRunner) getDockerServerVersion() ([]uint, error) {
+	env, err := d.client.Version()
+	if err != nil {
+		return nil, fmt.Errorf("failed to get docker server version - %s", err)
+	}
+	version := []uint{}
+	for _, entry := range *env {
+		if strings.Contains(strings.ToLower(entry), "server version") {
+			elems := strings.Split(strings.Split(entry, "=")[1], ".")
+			for _, elem := range elems {
+				val, err := strconv.ParseUint(elem, 10, 32)
+				if err != nil {
+					return nil, fmt.Errorf("failed to parse docker server version (%s) - %s", entry, err)
+				}
+				version = append(version, uint(val))
+			}
+			return version, nil
+		}
+	}
+	return nil, fmt.Errorf("docker server version missing from server version output - %+v", env)
+}
+
+func (d *dockerContainerCommandRunner) nativeExecSupportExists() (bool, error) {
+	version, err := d.getDockerServerVersion()
+	if err != nil {
+		return false, err
+	}
+	if len(dockerVersionWithExec) != len(version) {
+		return false, fmt.Errorf("unexpected docker version format. Expecting %v format, got %v", dockerVersionWithExec, version)
+	}
+	for idx, val := range dockerVersionWithExec {
+		if version[idx] < val {
+			return false, nil
+		}
+	}
+	return true, nil
+}
 
 func (d *dockerContainerCommandRunner) getRunInContainerCommand(containerID string, cmd []string) (*exec.Cmd, error) {
 	args := append([]string{"exec"}, cmd...)
@@ -108,8 +157,7 @@ func (d *dockerContainerCommandRunner) getRunInContainerCommand(containerID stri
 	return command, nil
 }
 
-// RunInContainer uses nsinit to run the command inside the container identified by containerID
+func (d *dockerContainerCommandRunner) runInContainerUsingNsinit(containerID string, cmd []string) ([]byte, error) {
-func (d *dockerContainerCommandRunner) RunInContainer(containerID string, cmd []string) ([]byte, error) {
 	c, err := d.getRunInContainerCommand(containerID, cmd)
 	if err != nil {
 		return nil, err
@@ -117,6 +165,45 @@ func (d *dockerContainerCommandRunner) RunInContainer(containerID string, cmd []
 	return c.CombinedOutput()
 }
 
+// RunInContainer uses nsinit to run the command inside the container identified by containerID
+func (d *dockerContainerCommandRunner) RunInContainer(containerID string, cmd []string) ([]byte, error) {
+	// If native exec support does not exist in the local docker daemon use nsinit.
+	useNativeExec, err := d.nativeExecSupportExists()
+	if err != nil {
+		return nil, err
+	}
+	if !useNativeExec {
+		return d.runInContainerUsingNsinit(containerID, cmd)
+	}
+	createOpts := docker.CreateExecOptions{
+		Container:    containerID,
+		Cmd:          cmd,
+		AttachStdin:  false,
+		AttachStdout: true,
+		AttachStderr: true,
+		Tty:          false,
+	}
+	execObj, err := d.client.CreateExec(createOpts)
+	if err != nil {
+		return nil, fmt.Errorf("failed to run in container - Exec setup failed - %s", err)
+	}
+	var buf bytes.Buffer
+	wrBuf := bufio.NewWriter(&buf)
+	startOpts := docker.StartExecOptions{
+		Detach:       false,
+		Tty:          false,
+		OutputStream: wrBuf,
+		ErrorStream:  wrBuf,
+		RawTerminal:  false,
+	}
+	errChan := make(chan error, 1)
+	go func() {
+		errChan <- d.client.StartExec(execObj.Id, startOpts)
+	}()
+	wrBuf.Flush()
+	return buf.Bytes(), <-errChan
+}
+
 // NewDockerContainerCommandRunner creates a ContainerCommandRunner which uses nsinit to run a command
 // inside a container.
 func NewDockerContainerCommandRunner() ContainerCommandRunner {

pkg/kubelet/dockertools/docker_test.go

@@ -122,6 +122,46 @@ func TestContainerManifestNaming(t *testing.T) {
 	}
 }
 
+func TestGetDockerServerVersion(t *testing.T) {
+	fakeDocker := &FakeDockerClient{VersionInfo: docker.Env{"Client version=1.2", "Server version=1.1.3", "Server API version=1.15"}}
+	runner := dockerContainerCommandRunner{fakeDocker}
+	version, err := runner.getDockerServerVersion()
+	if err != nil {
+		t.Errorf("got error while getting docker server version - %s", err)
+	}
+	expectedVersion := []uint{1, 1, 3}
+	if len(expectedVersion) != len(version) {
+		t.Errorf("invalid docker server version. expected: %v, got: %v", expectedVersion, version)
+	} else {
+		for idx, val := range expectedVersion {
+			if version[idx] != val {
+				t.Errorf("invalid docker server version. expected: %v, got: %v", expectedVersion, version)
+			}
+		}
+	}
+}
+
+func TestExecSupportExists(t *testing.T) {
+	fakeDocker := &FakeDockerClient{VersionInfo: docker.Env{"Client version=1.2", "Server version=1.1.3", "Server API version=1.15"}}
+	runner := dockerContainerCommandRunner{fakeDocker}
+	useNativeExec, err := runner.nativeExecSupportExists()
+	if err != nil {
+		t.Errorf("got error while checking for exec support - %s", err)
+	}
+	if !useNativeExec {
+		t.Errorf("invalid exec support check output. Expected true")
+	}
+}
+
+func TestExecSupportNotExists(t *testing.T) {
+	fakeDocker := &FakeDockerClient{VersionInfo: docker.Env{"Client version=1.2", "Server version=1.1.2", "Server API version=1.15"}}
+	runner := dockerContainerCommandRunner{fakeDocker}
+	useNativeExec, _ := runner.nativeExecSupportExists()
+	if useNativeExec {
+		t.Errorf("invalid exec support check output.")
+	}
+}
+
 func TestDockerContainerCommand(t *testing.T) {
 	runner := dockerContainerCommandRunner{}
 	containerID := "1234"

pkg/kubelet/dockertools/fake_docker_client.go

@@ -35,6 +35,7 @@ type FakeDockerClient struct {
 	Stopped       []string
 	pulled        []string
 	Created       []string
+	VersionInfo   docker.Env
 }
 
 func (f *FakeDockerClient) clearCalls() {
@@ -140,6 +141,17 @@ func (f *FakeDockerClient) PullImage(opts docker.PullImageOptions, auth docker.A
 	return f.Err
 }
 
+func (f *FakeDockerClient) Version() (*docker.Env, error) {
+	return &f.VersionInfo, nil
+}
+
+func (f *FakeDockerClient) CreateExec(_ docker.CreateExecOptions) (*docker.Exec, error) {
+	return &docker.Exec{"12345678"}, nil
+}
+func (f *FakeDockerClient) StartExec(_ string, _ docker.StartExecOptions) error {
+	return nil
+}
+
 // FakeDockerPuller is a stub implementation of DockerPuller.
 type FakeDockerPuller struct {
 	sync.Mutex