Refactor AttachToContainer and Logs.

pkg/kubelet/dockertools/docker.go

@@ -68,13 +68,13 @@ type DockerInterface interface {
 	ListImages(opts docker.ListImagesOptions) ([]docker.APIImages, error)
 	PullImage(opts docker.PullImageOptions, auth docker.AuthConfiguration) error
 	RemoveImage(image string) error
-	Logs(opts docker.LogsOptions) error
+	Logs(string, dockertypes.ContainerLogsOptions, StreamOptions) error
 	Version() (*docker.Env, error)
 	Info() (*docker.Env, error)
 	CreateExec(string, dockertypes.ExecConfig) (*dockertypes.ContainerExecCreateResponse, error)
 	StartExec(string, dockertypes.ExecStartCheck, StreamOptions) error
 	InspectExec(id string) (*dockertypes.ContainerExecInspect, error)
-	AttachToContainer(opts docker.AttachToContainerOptions) error
+	AttachToContainer(string, dockertypes.ContainerAttachOptions, StreamOptions) error
 }
 
 // KubeletContainerName encapsulates a pod name and a Kubernetes container name.

pkg/kubelet/dockertools/fake_docker_client.go

@@ -412,7 +412,7 @@ func (f *FakeDockerClient) RemoveContainer(id string, opts dockertypes.Container
 
 // Logs is a test-spy implementation of DockerInterface.Logs.
 // It adds an entry "logs" to the internal method call record.
-func (f *FakeDockerClient) Logs(opts docker.LogsOptions) error {
+func (f *FakeDockerClient) Logs(id string, opts dockertypes.ContainerLogsOptions, sopts StreamOptions) error {
 	f.Lock()
 	defer f.Unlock()
 	f.called = append(f.called, "logs")
@@ -462,7 +462,7 @@ func (f *FakeDockerClient) StartExec(startExec string, opts dockertypes.ExecStar
 	return nil
 }
 
-func (f *FakeDockerClient) AttachToContainer(opts docker.AttachToContainerOptions) error {
+func (f *FakeDockerClient) AttachToContainer(id string, opts dockertypes.ContainerAttachOptions, sopts StreamOptions) error {
 	f.Lock()
 	defer f.Unlock()
 	f.called = append(f.called, "attach")

pkg/kubelet/dockertools/instrumented_docker.go

@@ -139,11 +139,11 @@ func (in instrumentedDockerInterface) RemoveImage(image string) error {
 	return err
 }
 
-func (in instrumentedDockerInterface) Logs(opts docker.LogsOptions) error {
+func (in instrumentedDockerInterface) Logs(id string, opts dockertypes.ContainerLogsOptions, sopts StreamOptions) error {
 	const operation = "logs"
 	defer recordOperation(operation, time.Now())
 
-	err := in.client.Logs(opts)
+	err := in.client.Logs(id, opts, sopts)
 	recordError(operation, err)
 	return err
 }
@@ -193,11 +193,11 @@ func (in instrumentedDockerInterface) InspectExec(id string) (*dockertypes.Conta
 	return out, err
 }
 
-func (in instrumentedDockerInterface) AttachToContainer(opts docker.AttachToContainerOptions) error {
+func (in instrumentedDockerInterface) AttachToContainer(id string, opts dockertypes.ContainerAttachOptions, sopts StreamOptions) error {
 	const operation = "attach"
 	defer recordOperation(operation, time.Now())
 
-	err := in.client.AttachToContainer(opts)
+	err := in.client.AttachToContainer(id, opts, sopts)
 	recordError(operation, err)
 	return err
 }

pkg/kubelet/dockertools/kube_docker_client.go

@@ -23,7 +23,6 @@ import (
 	"fmt"
 	"io"
 	"io/ioutil"
-	"strconv"
 	"time"
 
 	"github.com/docker/docker/pkg/stdcopy"
@@ -210,21 +209,14 @@ func (d *kubeDockerClient) RemoveImage(image string) error {
 	return err
 }
 
-func (d *kubeDockerClient) Logs(opts docker.LogsOptions) error {
+func (d *kubeDockerClient) Logs(id string, opts dockertypes.ContainerLogsOptions, sopts StreamOptions) error {
-	resp, err := d.client.ContainerLogs(getDefaultContext(), dockertypes.ContainerLogsOptions{
+	opts.ContainerID = id
-		ContainerID: opts.Container,
+	resp, err := d.client.ContainerLogs(getDefaultContext(), opts)
-		ShowStdout:  opts.Stdout,
-		ShowStderr:  opts.Stderr,
-		Since:       strconv.FormatInt(opts.Since, 10),
-		Timestamps:  opts.Timestamps,
-		Follow:      opts.Follow,
-		Tail:        opts.Tail,
-	})
 	if err != nil {
 		return err
 	}
 	defer resp.Close()
-	return d.redirectResponseToOutputStream(opts.RawTerminal, opts.OutputStream, opts.ErrorStream, resp)
+	return d.redirectResponseToOutputStream(sopts.RawTerminal, sopts.OutputStream, sopts.ErrorStream, resp)
 }
 
 func (d *kubeDockerClient) Version() (*docker.Env, error) {
@@ -276,24 +268,14 @@ func (d *kubeDockerClient) InspectExec(id string) (*dockertypes.ContainerExecIns
 	return &resp, nil
 }
 
-func (d *kubeDockerClient) AttachToContainer(opts docker.AttachToContainerOptions) error {
+func (d *kubeDockerClient) AttachToContainer(id string, opts dockertypes.ContainerAttachOptions, sopts StreamOptions) error {
-	resp, err := d.client.ContainerAttach(getDefaultContext(), dockertypes.ContainerAttachOptions{
+	opts.ContainerID = id
-		ContainerID: opts.Container,
+	resp, err := d.client.ContainerAttach(getDefaultContext(), opts)
-		Stream:      opts.Stream,
-		Stdin:       opts.Stdin,
-		Stdout:      opts.Stdout,
-		Stderr:      opts.Stderr,
-		// TODO: How to deal with the *Logs* here? There is no *Logs* field in the engine-api.
-	})
 	if err != nil {
 		return err
 	}
 	defer resp.Close()
-	if opts.Success != nil {
+	return d.holdHijackedConnection(sopts.RawTerminal, sopts.InputStream, sopts.OutputStream, sopts.ErrorStream, resp)
-		opts.Success <- struct{}{}
-		<-opts.Success
-	}
-	return d.holdHijackedConnection(opts.RawTerminal, opts.InputStream, opts.OutputStream, opts.ErrorStream, resp)
 }
 
 // redirectResponseToOutputStream redirect the response stream to stdout and stderr. When tty is true, all stream will

pkg/kubelet/dockertools/manager.go

@@ -274,23 +274,22 @@ func (dm *DockerManager) GetContainerLogs(pod *api.Pod, containerID kubecontaine
 	if logOptions.SinceTime != nil {
 		since = logOptions.SinceTime.Unix()
 	}
-	opts := docker.LogsOptions{
+	opts := dockertypes.ContainerLogsOptions{
-		Container:    containerID.ID,
+		ShowStdout: true,
-		Stdout:       true,
+		ShowStderr: true,
-		Stderr:       true,
+		Since:      strconv.FormatInt(since, 10),
-		OutputStream: stdout,
-		ErrorStream:  stderr,
 		Timestamps: logOptions.Timestamps,
-		Since:        since,
 		Follow:     logOptions.Follow,
-		RawTerminal:  false,
 	}
-
 	if logOptions.TailLines != nil {
 		opts.Tail = strconv.FormatInt(*logOptions.TailLines, 10)
 	}
-
+	sopts := StreamOptions{
-	err = dm.client.Logs(opts)
+		OutputStream: stdout,
+		ErrorStream:  stderr,
+		RawTerminal:  false,
+	}
+	err = dm.client.Logs(containerID.ID, opts, sopts)
 	return
 }
 
@@ -1096,19 +1095,20 @@ func (dm *DockerManager) ExecInContainer(containerID kubecontainer.ContainerID,
 }
 
 func (dm *DockerManager) AttachContainer(containerID kubecontainer.ContainerID, stdin io.Reader, stdout, stderr io.WriteCloser, tty bool) error {
-	opts := docker.AttachToContainerOptions{
+	// TODO(random-liu): Do we really use the *Logs* field here?
-		Container:    containerID.ID,
+	opts := dockertypes.ContainerAttachOptions{
-		InputStream:  stdin,
-		OutputStream: stdout,
-		ErrorStream:  stderr,
 		Stream: true,
-		Logs:         true,
 		Stdin:  stdin != nil,
 		Stdout: stdout != nil,
 		Stderr: stderr != nil,
+	}
+	sopts := StreamOptions{
+		InputStream:  stdin,
+		OutputStream: stdout,
+		ErrorStream:  stderr,
 		RawTerminal:  tty,
 	}
-	return dm.client.AttachToContainer(opts)
+	return dm.client.AttachToContainer(containerID.ID, opts, sopts)
 }
 
 func noPodInfraContainerError(podName, podNamespace string) error {