Implement streaming CRI methods in dockershim

pkg/kubelet/dockershim/docker_streaming.go

@@ -0,0 +1,149 @@
+/*
+Copyright 2016 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package dockershim
+
+import (
+	"bytes"
+	"fmt"
+	"io"
+	"math"
+	"time"
+
+	dockertypes "github.com/docker/engine-api/types"
+	runtimeapi "k8s.io/kubernetes/pkg/kubelet/api/v1alpha1/runtime"
+	"k8s.io/kubernetes/pkg/kubelet/dockertools"
+	"k8s.io/kubernetes/pkg/kubelet/server/streaming"
+	"k8s.io/kubernetes/pkg/kubelet/util/ioutils"
+	"k8s.io/kubernetes/pkg/util/term"
+)
+
+type streamingRuntime struct {
+	client      dockertools.DockerInterface
+	execHandler dockertools.ExecHandler
+}
+
+var _ streaming.Runtime = &streamingRuntime{}
+
+func (r *streamingRuntime) Exec(containerID string, cmd []string, in io.Reader, out, err io.WriteCloser, tty bool, resize <-chan term.Size) error {
+	return r.exec(containerID, cmd, in, out, err, tty, resize, 0)
+}
+
+// Internal version of Exec adds a timeout.
+func (r *streamingRuntime) exec(containerID string, cmd []string, in io.Reader, out, errw io.WriteCloser, tty bool, resize <-chan term.Size, timeout time.Duration) error {
+	container, err := checkContainerStatus(r.client, containerID)
+	if err != nil {
+		return err
+	}
+
+	// TODO(timstclair): Clean this up once PR#33366 merges.
+	if timeout <= 0 {
+		// Run until command exits.
+		return r.execHandler.ExecInContainer(r.client, container, cmd, in, out, errw, tty, resize)
+	}
+
+	errCh := make(chan error)
+	go func() {
+		errCh <- r.execHandler.ExecInContainer(r.client, container, cmd, in, out, errw, tty, resize)
+	}()
+
+	select {
+	case err := <-errCh:
+		return err
+	case <-time.After(timeout):
+		return streaming.ErrorTimeout("exec", timeout)
+	}
+}
+
+func (r *streamingRuntime) Attach(containerID string, in io.Reader, out, errw io.WriteCloser, resize <-chan term.Size) error {
+	container, err := checkContainerStatus(r.client, containerID)
+	if err != nil {
+		return err
+	}
+
+	tty := container.Config.Tty
+	return dockertools.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, uint16(port), stream)
+}
+
+// ExecSync executes a command in the container, and returns the stdout output.
+// If command exits with a non-zero exit code, an error is returned.
+func (ds *dockerService) ExecSync(containerID string, cmd []string, timeout time.Duration) (stdout []byte, stderr []byte, err error) {
+	var stdoutBuffer, stderrBuffer bytes.Buffer
+	err = ds.streamingRuntime.exec(containerID, cmd,
+		nil, // in
+		ioutils.WriteCloserWrapper(&stdoutBuffer),
+		ioutils.WriteCloserWrapper(&stderrBuffer),
+		false, // tty
+		nil,   // resize
+		timeout)
+	return stdoutBuffer.Bytes(), stderrBuffer.Bytes(), err
+}
+
+// Exec prepares a streaming endpoint to execute a command in the container, and returns the address.
+func (ds *dockerService) Exec(req *runtimeapi.ExecRequest) (*runtimeapi.ExecResponse, error) {
+	if ds.streamingServer == nil {
+		return nil, streaming.ErrorStreamingDisabled("exec")
+	}
+	_, err := checkContainerStatus(ds.client, req.GetContainerId())
+	if err != nil {
+		return nil, err
+	}
+	return ds.streamingServer.GetExec(req)
+}
+
+// Attach prepares a streaming endpoint to attach to a running container, and returns the address.
+func (ds *dockerService) Attach(req *runtimeapi.AttachRequest) (*runtimeapi.AttachResponse, error) {
+	if ds.streamingServer == nil {
+		return nil, streaming.ErrorStreamingDisabled("attach")
+	}
+	container, err := checkContainerStatus(ds.client, req.GetContainerId())
+	if err != nil {
+		return nil, err
+	}
+	tty := container.Config.Tty
+	return ds.streamingServer.GetAttach(req, tty)
+}
+
+// PortForward prepares a streaming endpoint to forward ports from a PodSandbox, and returns the address.
+func (ds *dockerService) PortForward(req *runtimeapi.PortForwardRequest) (*runtimeapi.PortForwardResponse, error) {
+	if ds.streamingServer == nil {
+		return nil, streaming.ErrorStreamingDisabled("port forward")
+	}
+	_, err := checkContainerStatus(ds.client, req.GetPodSandboxId())
+	if err != nil {
+		return nil, err
+	}
+	// TODO(timstclair): Verify that ports are exposed.
+	return ds.streamingServer.GetPortForward(req)
+}
+
+func checkContainerStatus(client dockertools.DockerInterface, containerID string) (*dockertypes.ContainerJSON, error) {
+	container, err := client.InspectContainer(containerID)
+	if err != nil {
+		return nil, err
+	}
+	if !container.State.Running {
+		return nil, fmt.Errorf("container not running (%s)", container.ID)
+	}
+	return container, nil
+}