Clean up code in dockertools

cmd/kubelet/app/server.go

@@ -939,18 +939,6 @@ func RunDockershim(c *componentconfig.KubeletConfiguration, dockershimRootDir st
 	dockerClient := dockertools.ConnectToDockerOrDie(c.DockerEndpoint, c.RuntimeRequestTimeout.Duration,
 		c.ImagePullProgressDeadline.Duration)
 
-	// Initialize docker exec handler.
-	var dockerExecHandler dockertools.ExecHandler
-	switch c.DockerExecHandlerName {
-	case "native":
-		dockerExecHandler = &dockertools.NativeExecHandler{}
-	case "nsenter":
-		dockerExecHandler = &dockertools.NsenterExecHandler{}
-	default:
-		glog.Warningf("Unknown Docker exec handler %q; defaulting to native", c.DockerExecHandlerName)
-		dockerExecHandler = &dockertools.NativeExecHandler{}
-	}
-
 	// Initialize network plugin settings.
 	binDir := c.CNIBinDir
 	if binDir == "" {
@@ -976,7 +964,7 @@ func RunDockershim(c *componentconfig.KubeletConfiguration, dockershimRootDir st
 	}
 
 	ds, err := dockershim.NewDockerService(dockerClient, c.SeccompProfileRoot, c.PodInfraContainerImage,
-		streamingConfig, &pluginSettings, c.RuntimeCgroups, c.CgroupDriver, dockerExecHandler, dockershimRootDir,
+		streamingConfig, &pluginSettings, c.RuntimeCgroups, c.CgroupDriver, c.DockerExecHandlerName, dockershimRootDir,
 		!c.DockerEnableSharedPID)
 	if err != nil {
 		return err

pkg/kubelet/dockershim/BUILD

@@ -21,6 +21,7 @@ go_library(
         "docker_sandbox.go",
         "docker_service.go",
         "docker_streaming.go",
+        "exec.go",
         "helpers.go",
         "naming.go",
         "security_context.go",
@@ -37,8 +38,8 @@ go_library(
         "//pkg/kubelet/container:go_default_library",
         "//pkg/kubelet/dockershim/cm:go_default_library",
         "//pkg/kubelet/dockershim/errors:go_default_library",
+        "//pkg/kubelet/dockershim/securitycontext:go_default_library",
         "//pkg/kubelet/dockertools:go_default_library",
-        "//pkg/kubelet/dockertools/securitycontext:go_default_library",
         "//pkg/kubelet/leaky:go_default_library",
         "//pkg/kubelet/network:go_default_library",
         "//pkg/kubelet/network/cni:go_default_library",
@@ -49,14 +50,18 @@ go_library(
         "//pkg/kubelet/types:go_default_library",
         "//pkg/kubelet/util/cache:go_default_library",
         "//pkg/kubelet/util/ioutils:go_default_library",
+        "//pkg/security/apparmor:go_default_library",
+        "//pkg/util/exec:go_default_library",
         "//pkg/util/hash:go_default_library",
+        "//pkg/util/term:go_default_library",
         "//vendor/github.com/blang/semver:go_default_library",
         "//vendor/github.com/docker/engine-api/types:go_default_library",
         "//vendor/github.com/docker/engine-api/types/container:go_default_library",
         "//vendor/github.com/docker/engine-api/types/filters:go_default_library",
         "//vendor/github.com/docker/engine-api/types/strslice:go_default_library",
         "//vendor/github.com/docker/go-connections/nat:go_default_library",
         "//vendor/github.com/golang/glog:go_default_library",
+        "//vendor/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/util/errors:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/util/sets:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/util/wait:go_default_library",
@@ -86,9 +91,9 @@ go_test(
         "//pkg/kubelet/container:go_default_library",
         "//pkg/kubelet/container/testing:go_default_library",
         "//pkg/kubelet/dockershim/errors:go_default_library",
+        "//pkg/kubelet/dockershim/securitycontext:go_default_library",
         "//pkg/kubelet/dockershim/testing:go_default_library",
         "//pkg/kubelet/dockertools:go_default_library",
-        "//pkg/kubelet/dockertools/securitycontext:go_default_library",
         "//pkg/kubelet/network:go_default_library",
         "//pkg/kubelet/network/testing:go_default_library",
         "//pkg/kubelet/types:go_default_library",
@@ -119,6 +124,7 @@ filegroup(
         "//pkg/kubelet/dockershim/cm:all-srcs",
         "//pkg/kubelet/dockershim/errors:all-srcs",
         "//pkg/kubelet/dockershim/remote:all-srcs",
+        "//pkg/kubelet/dockershim/securitycontext:all-srcs",
         "//pkg/kubelet/dockershim/testing:all-srcs",
     ],
     tags = ["automanaged"],

pkg/kubelet/dockershim/docker_image.go

@@ -17,6 +17,8 @@ limitations under the License.
 package dockershim
 
 import (
+	"fmt"
+
 	dockertypes "github.com/docker/engine-api/types"
 	runtimeapi "k8s.io/kubernetes/pkg/kubelet/api/v1alpha1/runtime"
 	"k8s.io/kubernetes/pkg/kubelet/dockertools"
@@ -80,7 +82,7 @@ func (ds *dockerService) PullImage(image *runtimeapi.ImageSpec, auth *runtimeapi
 		return "", err
 	}
 
-	return dockertools.GetImageRef(ds.client, image.Image)
+	return getImageRef(ds.client, image.Image)
 }
 
 // RemoveImage removes the image.
@@ -101,3 +103,21 @@ func (ds *dockerService) RemoveImage(image *runtimeapi.ImageSpec) error {
 	_, err = ds.client.RemoveImage(image.Image, dockertypes.ImageRemoveOptions{PruneChildren: true})
 	return err
 }
+
+// getImageRef returns the image digest if exists, or else returns the image ID.
+func getImageRef(client dockertools.DockerInterface, image string) (string, error) {
+	img, err := client.InspectImageByRef(image)
+	if err != nil {
+		return "", err
+	}
+	if img == nil {
+		return "", fmt.Errorf("unable to inspect image %s", image)
+	}
+
+	// Returns the digest if it exist.
+	if len(img.RepoDigests) > 0 {
+		return img.RepoDigests[0], nil
+	}
+
+	return img.ID, nil
+}

pkg/kubelet/dockershim/docker_service.go

@@ -20,12 +20,14 @@ import (
 	"fmt"
 	"io"
 	"net/http"
+	"strconv"
 	"time"
 
 	"github.com/blang/semver"
 	dockertypes "github.com/docker/engine-api/types"
 	"github.com/golang/glog"
 
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/kubernetes/pkg/api/v1"
 	"k8s.io/kubernetes/pkg/apis/componentconfig"
 	internalapi "k8s.io/kubernetes/pkg/kubelet/api"
@@ -147,12 +149,23 @@ var internalLabelKeys []string = []string{containerTypeLabelKey, containerLogPat
 
 // NOTE: Anything passed to DockerService should be eventually handled in another way when we switch to running the shim as a different process.
 func NewDockerService(client dockertools.DockerInterface, seccompProfileRoot string, podSandboxImage string, streamingConfig *streaming.Config,
-	pluginSettings *NetworkPluginSettings, cgroupsName string, kubeCgroupDriver string, execHandler dockertools.ExecHandler, dockershimRootDir string, disableSharedPID bool) (DockerService, error) {
+	pluginSettings *NetworkPluginSettings, cgroupsName string, kubeCgroupDriver string, execHandlerName, dockershimRootDir string, disableSharedPID bool) (DockerService, error) {
 	c := dockertools.NewInstrumentedDockerInterface(client)
 	checkpointHandler, err := NewPersistentCheckpointHandler(dockershimRootDir)
 	if err != nil {
 		return nil, err
 	}
+	var execHandler ExecHandler
+	switch execHandlerName {
+	case "native":
+		execHandler = &NativeExecHandler{}
+	case "nsenter":
+		execHandler = &NsenterExecHandler{}
+	default:
+		glog.Warningf("Unknown Docker exec handler %q; defaulting to native", execHandlerName)
+		execHandler = &NativeExecHandler{}
+	}
+
 	ds := &dockerService{
 		seccompProfileRoot: seccompProfileRoot,
 		client:             c,
@@ -478,7 +491,32 @@ func (d *dockerLegacyService) GetContainerLogs(pod *v1.Pod, containerID kubecont
 	if err != nil {
 		return err
 	}
-	return dockertools.GetContainerLogs(d.client, pod, containerID, logOptions, stdout, stderr, container.Config.Tty)
+
+	var since int64
+	if logOptions.SinceSeconds != nil {
+		t := metav1.Now().Add(-time.Duration(*logOptions.SinceSeconds) * time.Second)
+		since = t.Unix()
+	}
+	if logOptions.SinceTime != nil {
+		since = logOptions.SinceTime.Unix()
+	}
+	opts := dockertypes.ContainerLogsOptions{
+		ShowStdout: true,
+		ShowStderr: true,
+		Since:      strconv.FormatInt(since, 10),
+		Timestamps: logOptions.Timestamps,
+		Follow:     logOptions.Follow,
+	}
+	if logOptions.TailLines != nil {
+		opts.Tail = strconv.FormatInt(*logOptions.TailLines, 10)
+	}
+
+	sopts := dockertools.StreamOptions{
+		OutputStream: stdout,
+		ErrorStream:  stderr,
+		RawTerminal:  container.Config.Tty,
+	}
+	return d.client.Logs(containerID.ID, opts, sopts)
 }
 
 // criSupportedLogDrivers are log drivers supported by native CRI integration.

pkg/kubelet/dockershim/docker_streaming.go

@@ -21,19 +21,23 @@ import (
 	"fmt"
 	"io"
 	"math"
+	"os/exec"
+	"strings"
 	"time"
 
 	dockertypes "github.com/docker/engine-api/types"
+	"github.com/golang/glog"
 	"k8s.io/kubernetes/pkg/client/unversioned/remotecommand"
 	runtimeapi "k8s.io/kubernetes/pkg/kubelet/api/v1alpha1/runtime"
+	kubecontainer "k8s.io/kubernetes/pkg/kubelet/container"
 	"k8s.io/kubernetes/pkg/kubelet/dockertools"
 	"k8s.io/kubernetes/pkg/kubelet/server/streaming"
 	"k8s.io/kubernetes/pkg/kubelet/util/ioutils"
 )
 
 type streamingRuntime struct {
 	client      dockertools.DockerInterface
-	execHandler dockertools.ExecHandler
+	execHandler ExecHandler
 }
 
 var _ streaming.Runtime = &streamingRuntime{}
@@ -57,14 +61,14 @@ func (r *streamingRuntime) Attach(containerID string, in io.Reader, out, errw io
 		return err
 	}
 
-	return dockertools.AttachContainer(r.client, containerID, in, out, errw, tty, resize)
+	return attachContainer(r.client, containerID, in, out, errw, tty, resize)
 }
 
 func (r *streamingRuntime) PortForward(podSandboxID string, port int32, stream io.ReadWriteCloser) error {
 	if port < 0 || port > math.MaxUint16 {
 		return fmt.Errorf("invalid port %d", port)
 	}
-	return dockertools.PortForward(r.client, podSandboxID, port, stream)
+	return portForward(r.client, podSandboxID, port, stream)
 }
 
 // ExecSync executes a command in the container, and returns the stdout output.
@@ -128,3 +132,83 @@ func checkContainerStatus(client dockertools.DockerInterface, containerID string
 	}
 	return container, nil
 }
+
+func attachContainer(client dockertools.DockerInterface, containerID string, stdin io.Reader, stdout, stderr io.WriteCloser, tty bool, resize <-chan remotecommand.TerminalSize) error {
+	// Have to start this before the call to client.AttachToContainer because client.AttachToContainer is a blocking
+	// call :-( Otherwise, resize events don't get processed and the terminal never resizes.
+	kubecontainer.HandleResizing(resize, func(size remotecommand.TerminalSize) {
+		client.ResizeContainerTTY(containerID, int(size.Height), int(size.Width))
+	})
+
+	// TODO(random-liu): Do we really use the *Logs* field here?
+	opts := dockertypes.ContainerAttachOptions{
+		Stream: true,
+		Stdin:  stdin != nil,
+		Stdout: stdout != nil,
+		Stderr: stderr != nil,
+	}
+	sopts := dockertools.StreamOptions{
+		InputStream:  stdin,
+		OutputStream: stdout,
+		ErrorStream:  stderr,
+		RawTerminal:  tty,
+	}
+	return client.AttachToContainer(containerID, opts, sopts)
+}
+
+func portForward(client dockertools.DockerInterface, podInfraContainerID string, port int32, stream io.ReadWriteCloser) error {
+	container, err := client.InspectContainer(podInfraContainerID)
+	if err != nil {
+		return err
+	}
+
+	if !container.State.Running {
+		return fmt.Errorf("container not running (%s)", container.ID)
+	}
+
+	containerPid := container.State.Pid
+	socatPath, lookupErr := exec.LookPath("socat")
+	if lookupErr != nil {
+		return fmt.Errorf("unable to do port forwarding: socat not found.")
+	}
+
+	args := []string{"-t", fmt.Sprintf("%d", containerPid), "-n", socatPath, "-", fmt.Sprintf("TCP4:localhost:%d", port)}
+
+	nsenterPath, lookupErr := exec.LookPath("nsenter")
+	if lookupErr != nil {
+		return fmt.Errorf("unable to do port forwarding: nsenter not found.")
+	}
+
+	commandString := fmt.Sprintf("%s %s", nsenterPath, strings.Join(args, " "))
+	glog.V(4).Infof("executing port forwarding command: %s", commandString)
+
+	command := exec.Command(nsenterPath, args...)
+	command.Stdout = stream
+
+	stderr := new(bytes.Buffer)
+	command.Stderr = stderr
+
+	// If we use Stdin, command.Run() won't return until the goroutine that's copying
+	// from stream finishes. Unfortunately, if you have a client like telnet connected
+	// via port forwarding, as long as the user's telnet client is connected to the user's
+	// local listener that port forwarding sets up, the telnet session never exits. This
+	// means that even if socat has finished running, command.Run() won't ever return
+	// (because the client still has the connection and stream open).
+	//
+	// The work around is to use StdinPipe(), as Wait() (called by Run()) closes the pipe
+	// when the command (socat) exits.
+	inPipe, err := command.StdinPipe()
+	if err != nil {
+		return fmt.Errorf("unable to do port forwarding: error creating stdin pipe: %v", err)
+	}
+	go func() {
+		io.Copy(inPipe, stream)
+		inPipe.Close()
+	}()
+
+	if err := command.Run(); err != nil {
+		return fmt.Errorf("%v: %s", err, stderr.String())
+	}
+
+	return nil
+}

pkg/kubelet/dockertools/exec.go → pkg/kubelet/dockershim/exec.go

@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package dockertools
+package dockershim
 
 import (
 	"fmt"
@@ -25,22 +25,44 @@ import (
 
 	dockertypes "github.com/docker/engine-api/types"
 	"github.com/golang/glog"
+
 	"k8s.io/kubernetes/pkg/client/unversioned/remotecommand"
 	kubecontainer "k8s.io/kubernetes/pkg/kubelet/container"
+	"k8s.io/kubernetes/pkg/kubelet/dockertools"
 	utilexec "k8s.io/kubernetes/pkg/util/exec"
 	"k8s.io/kubernetes/pkg/util/term"
 )
 
 // ExecHandler knows how to execute a command in a running Docker container.
 type ExecHandler interface {
-	ExecInContainer(client DockerInterface, container *dockertypes.ContainerJSON, cmd []string, stdin io.Reader, stdout, stderr io.WriteCloser, tty bool, resize <-chan remotecommand.TerminalSize, timeout time.Duration) error
+	ExecInContainer(client dockertools.DockerInterface, container *dockertypes.ContainerJSON, cmd []string, stdin io.Reader, stdout, stderr io.WriteCloser, tty bool, resize <-chan remotecommand.TerminalSize, timeout time.Duration) error
 }
 
 // NsenterExecHandler executes commands in Docker containers using nsenter.
 type NsenterExecHandler struct{}
 
+type dockerExitError struct {
+	Inspect *dockertypes.ContainerExecInspect
+}
+
+func (d *dockerExitError) String() string {
+	return d.Error()
+}
+
+func (d *dockerExitError) Error() string {
+	return fmt.Sprintf("Error executing in Docker Container: %d", d.Inspect.ExitCode)
+}
+
+func (d *dockerExitError) Exited() bool {
+	return !d.Inspect.Running
+}
+
+func (d *dockerExitError) ExitStatus() int {
+	return d.Inspect.ExitCode
+}
+
 // TODO should we support nsenter in a container, running with elevated privs and --pid=host?
-func (*NsenterExecHandler) ExecInContainer(client DockerInterface, container *dockertypes.ContainerJSON, cmd []string, stdin io.Reader, stdout, stderr io.WriteCloser, tty bool, resize <-chan remotecommand.TerminalSize, timeout time.Duration) error {
+func (*NsenterExecHandler) ExecInContainer(client dockertools.DockerInterface, container *dockertypes.ContainerJSON, cmd []string, stdin io.Reader, stdout, stderr io.WriteCloser, tty bool, resize <-chan remotecommand.TerminalSize, timeout time.Duration) error {
 	nsenter, err := exec.LookPath("nsenter")
 	if err != nil {
 		return fmt.Errorf("exec unavailable - unable to locate nsenter")
@@ -111,7 +133,7 @@ func (*NsenterExecHandler) ExecInContainer(client DockerInterface, container *do
 // NativeExecHandler executes commands in Docker containers using Docker's exec API.
 type NativeExecHandler struct{}
 
-func (*NativeExecHandler) ExecInContainer(client DockerInterface, container *dockertypes.ContainerJSON, cmd []string, stdin io.Reader, stdout, stderr io.WriteCloser, tty bool, resize <-chan remotecommand.TerminalSize, timeout time.Duration) error {
+func (*NativeExecHandler) ExecInContainer(client dockertools.DockerInterface, container *dockertypes.ContainerJSON, cmd []string, stdin io.Reader, stdout, stderr io.WriteCloser, tty bool, resize <-chan remotecommand.TerminalSize, timeout time.Duration) error {
 	createOpts := dockertypes.ExecConfig{
 		Cmd:          cmd,
 		AttachStdin:  stdin != nil,
@@ -131,7 +153,7 @@ func (*NativeExecHandler) ExecInContainer(client DockerInterface, container *doc
 	})
 
 	startOpts := dockertypes.ExecStartCheck{Detach: false, Tty: tty}
-	streamOpts := StreamOptions{
+	streamOpts := dockertools.StreamOptions{
 		InputStream:  stdin,
 		OutputStream: stdout,
 		ErrorStream:  stderr,