Expose exec and logs via WebSockets

Not all clients and systems can support SPDY protocols. This commit adds
support for two new websocket protocols, one to handle streaming of pod
logs from a pod, and the other to allow exec to be tunneled over
websocket.

Browser support for chunked encoding is still poor, and web consoles
that wish to show pod logs may need to make compromises to display the
output. The /pods/<name>/log endpoint now supports websocket upgrade to
the 'binary.k8s.io' subprotocol, which sends chunks of logs as binary to
the client. Messages are written as logs are streamed from the container
daemon, so flushing should be unaffected.

Browser support for raw communication over SDPY is not possible, and
some languages lack libraries for it and HTTP/2. The Kubelet supports
upgrade to WebSocket instead of SPDY, and will multiplex STDOUT/IN/ERR
over websockets by prepending each binary message with a single byte
representing the channel (0 for IN, 1 for OUT, and 2 for ERR). Because
framing on WebSockets suffers from head-of-line blocking, clients and
other server code should ensure that no particular stream blocks. An
alternative subprotocol 'base64.channel.k8s.io' base64 encodes the body
and uses '0'-'9' to represent the channel for ease of use in browsers.

test/e2e/pods.go

@@ -17,10 +17,15 @@ limitations under the License.
 package e2e
 
 import (
+	"bytes"
 	"fmt"
+	"io"
 	"strconv"
+	"strings"
 	"time"
 
+	"golang.org/x/net/websocket"
+
 	"k8s.io/kubernetes/pkg/api"
 	"k8s.io/kubernetes/pkg/api/errors"
 	"k8s.io/kubernetes/pkg/api/resource"
@@ -602,6 +607,154 @@ var _ = Describe("Pods", func() {
 		}, 0, defaultObservationTimeout)
 	})
 
+	It("should support remote command execution over websockets", func() {
+		config, err := loadConfig()
+		if err != nil {
+			Failf("Unable to get base config: %v", err)
+		}
+		podClient := framework.Client.Pods(framework.Namespace.Name)
+
+		By("creating the pod")
+		name := "pod-exec-websocket-" + string(util.NewUUID())
+		pod := &api.Pod{
+			ObjectMeta: api.ObjectMeta{
+				Name: name,
+			},
+			Spec: api.PodSpec{
+				Containers: []api.Container{
+					{
+						Name:    "main",
+						Image:   "gcr.io/google_containers/busybox",
+						Command: []string{"/bin/sh", "-c", "echo container is alive; sleep 600"},
+					},
+				},
+			},
+		}
+
+		By("submitting the pod to kubernetes")
+		defer func() {
+			By("deleting the pod")
+			podClient.Delete(pod.Name, api.NewDeleteOptions(0))
+		}()
+		pod, err = podClient.Create(pod)
+		if err != nil {
+			Failf("Failed to create pod: %v", err)
+		}
+
+		expectNoError(framework.WaitForPodRunning(pod.Name))
+
+		req := framework.Client.Get().
+			Namespace(framework.Namespace.Name).
+			Resource("pods").
+			Name(pod.Name).
+			Suffix("exec").
+			Param("stderr", "1").
+			Param("stdout", "1").
+			Param("container", pod.Spec.Containers[0].Name).
+			Param("command", "cat").
+			Param("command", "/etc/resolv.conf")
+
+		url := req.URL()
+		ws, err := OpenWebSocketForURL(url, config, []string{"channel.k8s.io"})
+		if err != nil {
+			Failf("Failed to open websocket to %s: %v", url.String(), err)
+		}
+		defer ws.Close()
+
+		buf := &bytes.Buffer{}
+		for {
+			var msg []byte
+			if err := websocket.Message.Receive(ws, &msg); err != nil {
+				if err == io.EOF {
+					break
+				}
+				Failf("Failed to read completely from websocket %s: %v", url.String(), err)
+			}
+			if len(msg) == 0 {
+				continue
+			}
+			if msg[0] != 1 {
+				Failf("Got message from server that didn't start with channel 1 (STDOUT): %v", msg)
+			}
+			buf.Write(msg[1:])
+		}
+		if buf.Len() == 0 {
+			Failf("Unexpected output from server")
+		}
+		if !strings.Contains(buf.String(), "nameserver") {
+			Failf("Expected to find 'nameserver' in %q", buf.String())
+		}
+	})
+
+	It("should support retrieving logs from the container over websockets", func() {
+		config, err := loadConfig()
+		if err != nil {
+			Failf("Unable to get base config: %v", err)
+		}
+		podClient := framework.Client.Pods(framework.Namespace.Name)
+
+		By("creating the pod")
+		name := "pod-logs-websocket-" + string(util.NewUUID())
+		pod := &api.Pod{
+			ObjectMeta: api.ObjectMeta{
+				Name: name,
+			},
+			Spec: api.PodSpec{
+				Containers: []api.Container{
+					{
+						Name:    "main",
+						Image:   "gcr.io/google_containers/busybox",
+						Command: []string{"/bin/sh", "-c", "echo container is alive; sleep 600"},
+					},
+				},
+			},
+		}
+
+		By("submitting the pod to kubernetes")
+		defer func() {
+			By("deleting the pod")
+			podClient.Delete(pod.Name, api.NewDeleteOptions(0))
+		}()
+		pod, err = podClient.Create(pod)
+		if err != nil {
+			Failf("Failed to create pod: %v", err)
+		}
+
+		expectNoError(framework.WaitForPodRunning(pod.Name))
+
+		req := framework.Client.Get().
+			Namespace(framework.Namespace.Name).
+			Resource("pods").
+			Name(pod.Name).
+			Suffix("log").
+			Param("container", pod.Spec.Containers[0].Name)
+
+		url := req.URL()
+
+		ws, err := OpenWebSocketForURL(url, config, []string{"binary.k8s.io"})
+		if err != nil {
+			Failf("Failed to open websocket to %s: %v", url.String(), err)
+		}
+		defer ws.Close()
+		buf := &bytes.Buffer{}
+		for {
+			var msg []byte
+			if err := websocket.Message.Receive(ws, &msg); err != nil {
+				if err == io.EOF {
+					break
+				}
+				Failf("Failed to read completely from websocket %s: %v", url.String(), err)
+			}
+			if len(msg) == 0 {
+				continue
+			}
+			buf.Write(msg)
+		}
+		if buf.String() != "container is alive\n" {
+			Failf("Unexpected websocket logs:\n%s", buf.String())
+		}
+	})
+
 	// The following tests for remote command execution and port forwarding are
 	// commented out because the GCE environment does not currently have nsenter
 	// in the kubelet's PATH, nor does it have socat installed. Once we figure