Refactor CreateExec, StartExec and InspectExec.

pkg/kubelet/dockertools/docker.go

@@ -71,9 +71,9 @@ type DockerInterface interface {
 	Logs(opts docker.LogsOptions) error
 	Version() (*docker.Env, error)
 	Info() (*docker.Env, error)
-	CreateExec(docker.CreateExecOptions) (*docker.Exec, error)
+	CreateExec(string, dockertypes.ExecConfig) (*dockertypes.ContainerExecCreateResponse, error)
-	StartExec(string, docker.StartExecOptions) error
+	StartExec(string, dockertypes.ExecStartCheck, StreamOptions) error
-	InspectExec(id string) (*docker.ExecInspect, error)
+	InspectExec(id string) (*dockertypes.ContainerExecInspect, error)
 	AttachToContainer(opts docker.AttachToContainerOptions) error
 }
 

pkg/kubelet/dockertools/exec.go

@@ -24,7 +24,6 @@ import (
 	"time"
 
 	dockertypes "github.com/docker/engine-api/types"
-	docker "github.com/fsouza/go-dockerclient"
 	"github.com/golang/glog"
 	kubecontainer "k8s.io/kubernetes/pkg/kubelet/container"
 )
@@ -100,27 +99,25 @@ func (*NsenterExecHandler) ExecInContainer(client DockerInterface, container *do
 type NativeExecHandler struct{}
 
 func (*NativeExecHandler) ExecInContainer(client DockerInterface, container *dockertypes.ContainerJSON, cmd []string, stdin io.Reader, stdout, stderr io.WriteCloser, tty bool) error {
-	createOpts := docker.CreateExecOptions{
+	createOpts := dockertypes.ExecConfig{
-		Container:    container.ID,
 		Cmd:          cmd,
 		AttachStdin:  stdin != nil,
 		AttachStdout: stdout != nil,
 		AttachStderr: stderr != nil,
 		Tty:          tty,
 	}
-	execObj, err := client.CreateExec(createOpts)
+	execObj, err := client.CreateExec(container.ID, createOpts)
 	if err != nil {
 		return fmt.Errorf("failed to exec in container - Exec setup failed - %v", err)
 	}
-	startOpts := docker.StartExecOptions{
+	startOpts := dockertypes.ExecStartCheck{Detach: false, Tty: tty}
-		Detach:       false,
+	streamOpts := StreamOptions{
 		InputStream:  stdin,
 		OutputStream: stdout,
 		ErrorStream:  stderr,
-		Tty:          tty,
 		RawTerminal:  tty,
 	}
-	err = client.StartExec(execObj.ID, startOpts)
+	err = client.StartExec(execObj.ID, startOpts, streamOpts)
 	if err != nil {
 		return err
 	}

pkg/kubelet/dockertools/fake_docker_client.go

@@ -52,7 +52,7 @@ type FakeDockerClient struct {
 	RemovedImages sets.String
 	VersionInfo   docker.Env
 	Information   docker.Env
-	ExecInspect   *docker.ExecInspect
+	ExecInspect   *dockertypes.ContainerExecInspect
 	execCmd       []string
 	EnableSleep   bool
 }
@@ -447,15 +447,15 @@ func (f *FakeDockerClient) Info() (*docker.Env, error) {
 	return &f.Information, nil
 }
 
-func (f *FakeDockerClient) CreateExec(opts docker.CreateExecOptions) (*docker.Exec, error) {
+func (f *FakeDockerClient) CreateExec(id string, opts dockertypes.ExecConfig) (*dockertypes.ContainerExecCreateResponse, error) {
 	f.Lock()
 	defer f.Unlock()
 	f.execCmd = opts.Cmd
 	f.called = append(f.called, "create_exec")
-	return &docker.Exec{ID: "12345678"}, nil
+	return &dockertypes.ContainerExecCreateResponse{ID: "12345678"}, nil
 }
 
-func (f *FakeDockerClient) StartExec(_ string, _ docker.StartExecOptions) error {
+func (f *FakeDockerClient) StartExec(startExec string, opts dockertypes.ExecStartCheck, sopts StreamOptions) error {
 	f.Lock()
 	defer f.Unlock()
 	f.called = append(f.called, "start_exec")
@@ -469,7 +469,7 @@ func (f *FakeDockerClient) AttachToContainer(opts docker.AttachToContainerOption
 	return nil
 }
 
-func (f *FakeDockerClient) InspectExec(id string) (*docker.ExecInspect, error) {
+func (f *FakeDockerClient) InspectExec(id string) (*dockertypes.ContainerExecInspect, error) {
 	return f.ExecInspect, f.popError("inspect_exec")
 }
 

pkg/kubelet/dockertools/instrumented_docker.go

@@ -166,25 +166,25 @@ func (in instrumentedDockerInterface) Info() (*docker.Env, error) {
 	return out, err
 }
 
-func (in instrumentedDockerInterface) CreateExec(opts docker.CreateExecOptions) (*docker.Exec, error) {
+func (in instrumentedDockerInterface) CreateExec(id string, opts dockertypes.ExecConfig) (*dockertypes.ContainerExecCreateResponse, error) {
 	const operation = "create_exec"
 	defer recordOperation(operation, time.Now())
 
-	out, err := in.client.CreateExec(opts)
+	out, err := in.client.CreateExec(id, opts)
 	recordError(operation, err)
 	return out, err
 }
 
-func (in instrumentedDockerInterface) StartExec(startExec string, opts docker.StartExecOptions) error {
+func (in instrumentedDockerInterface) StartExec(startExec string, opts dockertypes.ExecStartCheck, sopts StreamOptions) error {
 	const operation = "start_exec"
 	defer recordOperation(operation, time.Now())
 
-	err := in.client.StartExec(startExec, opts)
+	err := in.client.StartExec(startExec, opts, sopts)
 	recordError(operation, err)
 	return err
 }
 
-func (in instrumentedDockerInterface) InspectExec(id string) (*docker.ExecInspect, error) {
+func (in instrumentedDockerInterface) InspectExec(id string) (*dockertypes.ContainerExecInspect, error) {
 	const operation = "inspect_exec"
 	defer recordOperation(operation, time.Now())
 

pkg/kubelet/dockertools/kube_docker_client.go

@@ -243,28 +243,19 @@ func (d *kubeDockerClient) Info() (*docker.Env, error) {
 	return convertEnv(resp)
 }
 
-func (d *kubeDockerClient) CreateExec(opts docker.CreateExecOptions) (*docker.Exec, error) {
+// TODO(random-liu): Add unit test for exec and attach functions, just like what go-dockerclient did.
-	cfg := dockertypes.ExecConfig{}
+func (d *kubeDockerClient) CreateExec(id string, opts dockertypes.ExecConfig) (*dockertypes.ContainerExecCreateResponse, error) {
-	if err := convertType(&opts, &cfg); err != nil {
+	opts.Container = id
-		return nil, err
+	resp, err := d.client.ContainerExecCreate(getDefaultContext(), opts)
-	}
-	resp, err := d.client.ContainerExecCreate(getDefaultContext(), cfg)
 	if err != nil {
 		return nil, err
 	}
-	exec := &docker.Exec{}
+	return &resp, nil
-	if err := convertType(&resp, exec); err != nil {
-		return nil, err
-	}
-	return exec, nil
 }
 
-func (d *kubeDockerClient) StartExec(startExec string, opts docker.StartExecOptions) error {
+func (d *kubeDockerClient) StartExec(startExec string, opts dockertypes.ExecStartCheck, sopts StreamOptions) error {
 	if opts.Detach {
-		return d.client.ContainerExecStart(getDefaultContext(), startExec, dockertypes.ExecStartCheck{
+		return d.client.ContainerExecStart(getDefaultContext(), startExec, opts)
-			Detach: opts.Detach,
-			Tty:    opts.Tty,
-		})
 	}
 	resp, err := d.client.ContainerExecAttach(getDefaultContext(), startExec, dockertypes.ExecConfig{
 		Detach: opts.Detach,
@@ -274,23 +265,15 @@ func (d *kubeDockerClient) StartExec(startExec string, opts docker.StartExecOpti
 		return err
 	}
 	defer resp.Close()
-	if opts.Success != nil {
+	return d.holdHijackedConnection(sopts.RawTerminal || opts.Tty, sopts.InputStream, sopts.OutputStream, sopts.ErrorStream, resp)
-		opts.Success <- struct{}{}
-		<-opts.Success
-	}
-	return d.holdHijackedConnection(opts.RawTerminal || opts.Tty, opts.InputStream, opts.OutputStream, opts.ErrorStream, resp)
 }
 
-func (d *kubeDockerClient) InspectExec(id string) (*docker.ExecInspect, error) {
+func (d *kubeDockerClient) InspectExec(id string) (*dockertypes.ContainerExecInspect, error) {
 	resp, err := d.client.ContainerExecInspect(getDefaultContext(), id)
 	if err != nil {
 		return nil, err
 	}
-	exec := &docker.ExecInspect{}
+	return &resp, nil
-	if err := convertType(&resp, exec); err != nil {
-		return nil, err
-	}
-	return exec, nil
 }
 
 func (d *kubeDockerClient) AttachToContainer(opts docker.AttachToContainerOptions) error {
@@ -367,6 +350,14 @@ func parseDockerTimestamp(s string) (time.Time, error) {
 	return time.Parse(time.RFC3339Nano, s)
 }
 
+// StreamOptions are the options used to configure the stream redirection
+type StreamOptions struct {
+	RawTerminal  bool
+	InputStream  io.Reader
+	OutputStream io.Writer
+	ErrorStream  io.Writer
+}
+
 // containerNotFoundError is the error returned by InspectContainer when container not found. We
 // add this error type for testability. We don't use the original error returned by engine-api
 // because dockertypes.containerNotFoundError is private, we can't create and inject it in our test.

pkg/kubelet/dockertools/manager.go

@@ -1014,27 +1014,25 @@ func (dm *DockerManager) defaultSecurityOpt() ([]string, error) {
 // RunInContainer run the command inside the container identified by containerID
 func (dm *DockerManager) RunInContainer(containerID kubecontainer.ContainerID, cmd []string) ([]byte, error) {
 	glog.V(2).Infof("Using docker native exec to run cmd %+v inside container %s", cmd, containerID)
-	createOpts := docker.CreateExecOptions{
+	createOpts := dockertypes.ExecConfig{
-		Container:    containerID.ID,
 		Cmd:          cmd,
 		AttachStdin:  false,
 		AttachStdout: true,
 		AttachStderr: true,
 		Tty:          false,
 	}
-	execObj, err := dm.client.CreateExec(createOpts)
+	execObj, err := dm.client.CreateExec(containerID.ID, createOpts)
 	if err != nil {
 		return nil, fmt.Errorf("failed to run in container - Exec setup failed - %v", err)
 	}
 	var buf bytes.Buffer
-	startOpts := docker.StartExecOptions{
+	startOpts := dockertypes.ExecStartCheck{Detach: false, Tty: false}
-		Detach:       false,
+	streamOpts := StreamOptions{
-		Tty:          false,
 		OutputStream: &buf,
 		ErrorStream:  &buf,
 		RawTerminal:  false,
 	}
-	err = dm.client.StartExec(execObj.ID, startOpts)
+	err = dm.client.StartExec(execObj.ID, startOpts, streamOpts)
 	if err != nil {
 		glog.V(2).Infof("StartExec With error: %v", err)
 		return nil, err
@@ -1061,7 +1059,7 @@ func (dm *DockerManager) RunInContainer(containerID kubecontainer.ContainerID, c
 }
 
 type dockerExitError struct {
-	Inspect *docker.ExecInspect
+	Inspect *dockertypes.ContainerExecInspect
 }
 
 func (d *dockerExitError) String() string {

pkg/kubelet/dockertools/manager_test.go

@@ -438,7 +438,7 @@ func TestKillContainerInPod(t *testing.T) {
 
 func TestKillContainerInPodWithPreStop(t *testing.T) {
 	manager, fakeDocker := newTestDockerManager()
-	fakeDocker.ExecInspect = &docker.ExecInspect{
+	fakeDocker.ExecInspect = &dockertypes.ContainerExecInspect{
 		Running:  false,
 		ExitCode: 0,
 	}