Proxy container streaming in kubelet.

pkg/kubelet/server/server.go

@@ -42,14 +42,13 @@ import (
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/types"
-	remotecommandconsts "k8s.io/apimachinery/pkg/util/remotecommand"
+	"k8s.io/apimachinery/pkg/util/proxy"
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
 	"k8s.io/apiserver/pkg/authentication/authenticator"
 	"k8s.io/apiserver/pkg/authorization/authorizer"
 	"k8s.io/apiserver/pkg/server/healthz"
 	"k8s.io/apiserver/pkg/server/httplog"
 	"k8s.io/apiserver/pkg/util/flushwriter"
-	"k8s.io/client-go/tools/remotecommand"
 	"k8s.io/kubernetes/pkg/api/legacyscheme"
 	api "k8s.io/kubernetes/pkg/apis/core"
 	"k8s.io/kubernetes/pkg/apis/core/v1/validation"
@@ -74,11 +73,11 @@ const (
 
 // Server is a http.Handler which exposes kubelet functionality over HTTP.
 type Server struct {
-	auth             AuthInterface
-	host             HostInterface
-	restfulCont      containerInterface
-	resourceAnalyzer stats.ResourceAnalyzer
-	runtime          kubecontainer.Runtime
+	auth                       AuthInterface
+	host                       HostInterface
+	restfulCont                containerInterface
+	resourceAnalyzer           stats.ResourceAnalyzer
+	redirectContainerStreaming bool
 }
 
 type TLSOptions struct {
@@ -124,11 +123,11 @@ func ListenAndServeKubeletServer(
 	tlsOptions *TLSOptions,
 	auth AuthInterface,
 	enableDebuggingHandlers,
-	enableContentionProfiling bool,
-	runtime kubecontainer.Runtime,
+	enableContentionProfiling,
+	redirectContainerStreaming bool,
 	criHandler http.Handler) {
 	glog.Infof("Starting to listen on %s:%d", address, port)
-	handler := NewServer(host, resourceAnalyzer, auth, enableDebuggingHandlers, enableContentionProfiling, runtime, criHandler)
+	handler := NewServer(host, resourceAnalyzer, auth, enableDebuggingHandlers, enableContentionProfiling, redirectContainerStreaming, criHandler)
 	s := &http.Server{
 		Addr:           net.JoinHostPort(address.String(), strconv.FormatUint(uint64(port), 10)),
 		Handler:        &handler,
@@ -146,9 +145,9 @@ func ListenAndServeKubeletServer(
 }
 
 // ListenAndServeKubeletReadOnlyServer initializes a server to respond to HTTP network requests on the Kubelet.
-func ListenAndServeKubeletReadOnlyServer(host HostInterface, resourceAnalyzer stats.ResourceAnalyzer, address net.IP, port uint, runtime kubecontainer.Runtime) {
+func ListenAndServeKubeletReadOnlyServer(host HostInterface, resourceAnalyzer stats.ResourceAnalyzer, address net.IP, port uint) {
 	glog.V(1).Infof("Starting to listen read-only on %s:%d", address, port)
-	s := NewServer(host, resourceAnalyzer, nil, false, false, runtime, nil)
+	s := NewServer(host, resourceAnalyzer, nil, false, false, false, nil)
 
 	server := &http.Server{
 		Addr:           net.JoinHostPort(address.String(), strconv.FormatUint(uint64(port), 10)),
@@ -173,12 +172,8 @@ type HostInterface interface {
 	GetCachedMachineInfo() (*cadvisorapi.MachineInfo, error)
 	GetRunningPods() ([]*v1.Pod, error)
 	RunInContainer(name string, uid types.UID, container string, cmd []string) ([]byte, error)
-	ExecInContainer(name string, uid types.UID, container string, cmd []string, in io.Reader, out, err io.WriteCloser, tty bool, resize <-chan remotecommand.TerminalSize, timeout time.Duration) error
-	AttachContainer(name string, uid types.UID, container string, in io.Reader, out, err io.WriteCloser, tty bool, resize <-chan remotecommand.TerminalSize) error
 	GetKubeletContainerLogs(podFullName, containerName string, logOptions *v1.PodLogOptions, stdout, stderr io.Writer) error
 	ServeLogs(w http.ResponseWriter, req *http.Request)
-	PortForward(name string, uid types.UID, port int32, stream io.ReadWriteCloser) error
-	StreamingConnectionIdleTimeout() time.Duration
 	ResyncInterval() time.Duration
 	GetHostname() string
 	LatestLoopEntryTime() time.Time
@@ -193,15 +188,15 @@ func NewServer(
 	resourceAnalyzer stats.ResourceAnalyzer,
 	auth AuthInterface,
 	enableDebuggingHandlers,
-	enableContentionProfiling bool,
-	runtime kubecontainer.Runtime,
+	enableContentionProfiling,
+	redirectContainerStreaming bool,
 	criHandler http.Handler) Server {
 	server := Server{
-		host:             host,
-		resourceAnalyzer: resourceAnalyzer,
-		auth:             auth,
-		restfulCont:      &filteringContainer{Container: restful.NewContainer()},
-		runtime:          runtime,
+		host:                       host,
+		resourceAnalyzer:           resourceAnalyzer,
+		auth:                       auth,
+		restfulCont:                &filteringContainer{Container: restful.NewContainer()},
+		redirectContainerStreaming: redirectContainerStreaming,
 	}
 	if auth != nil {
 		server.InstallAuthFilter()
@@ -627,6 +622,15 @@ func getPortForwardRequestParams(req *restful.Request) portForwardRequestParams
 	}
 }
 
+type responder struct {
+	errorMessage string
+}
+
+func (r *responder) Error(w http.ResponseWriter, req *http.Request, err error) {
+	glog.Errorf("Error while proxying request: %v", err)
+	http.Error(w, err.Error(), http.StatusInternalServerError)
+}
+
 // getAttach handles requests to attach to a container.
 func (s *Server) getAttach(request *restful.Request, response *restful.Response) {
 	params := getExecRequestParams(request)
@@ -643,26 +647,18 @@ func (s *Server) getAttach(request *restful.Request, response *restful.Response)
 	}
 
 	podFullName := kubecontainer.GetPodFullName(pod)
-	redirect, err := s.host.GetAttach(podFullName, params.podUID, params.containerName, *streamOpts)
+	url, err := s.host.GetAttach(podFullName, params.podUID, params.containerName, *streamOpts)
 	if err != nil {
 		streaming.WriteError(err, response.ResponseWriter)
 		return
 	}
-	if redirect != nil {
-		http.Redirect(response.ResponseWriter, request.Request, redirect.String(), http.StatusFound)
+
+	if s.redirectContainerStreaming {
+		http.Redirect(response.ResponseWriter, request.Request, url.String(), http.StatusFound)
 		return
 	}
-
-	remotecommandserver.ServeAttach(response.ResponseWriter,
-		request.Request,
-		s.host,
-		podFullName,
-		params.podUID,
-		params.containerName,
-		streamOpts,
-		s.host.StreamingConnectionIdleTimeout(),
-		remotecommandconsts.DefaultStreamCreationTimeout,
-		remotecommandconsts.SupportedStreamingProtocols)
+	handler := proxy.NewUpgradeAwareHandler(url, nil /*transport*/, false /*wrapTransport*/, false /*upgradeRequired*/, &responder{})
+	handler.ServeHTTP(response.ResponseWriter, request.Request)
 }
 
 // getExec handles requests to run a command inside a container.
@@ -681,27 +677,17 @@ func (s *Server) getExec(request *restful.Request, response *restful.Response) {
 	}
 
 	podFullName := kubecontainer.GetPodFullName(pod)
-	redirect, err := s.host.GetExec(podFullName, params.podUID, params.containerName, params.cmd, *streamOpts)
+	url, err := s.host.GetExec(podFullName, params.podUID, params.containerName, params.cmd, *streamOpts)
 	if err != nil {
 		streaming.WriteError(err, response.ResponseWriter)
 		return
 	}
-	if redirect != nil {
-		http.Redirect(response.ResponseWriter, request.Request, redirect.String(), http.StatusFound)
+	if s.redirectContainerStreaming {
+		http.Redirect(response.ResponseWriter, request.Request, url.String(), http.StatusFound)
 		return
 	}
-
-	remotecommandserver.ServeExec(response.ResponseWriter,
-		request.Request,
-		s.host,
-		podFullName,
-		params.podUID,
-		params.containerName,
-		params.cmd,
-		streamOpts,
-		s.host.StreamingConnectionIdleTimeout(),
-		remotecommandconsts.DefaultStreamCreationTimeout,
-		remotecommandconsts.SupportedStreamingProtocols)
+	handler := proxy.NewUpgradeAwareHandler(url, nil /*transport*/, false /*wrapTransport*/, false /*upgradeRequired*/, &responder{})
+	handler.ServeHTTP(response.ResponseWriter, request.Request)
 }
 
 // getRun handles requests to run a command inside a container.
@@ -758,25 +744,17 @@ func (s *Server) getPortForward(request *restful.Request, response *restful.Resp
 		return
 	}
 
-	redirect, err := s.host.GetPortForward(pod.Name, pod.Namespace, pod.UID, *portForwardOptions)
+	url, err := s.host.GetPortForward(pod.Name, pod.Namespace, pod.UID, *portForwardOptions)
 	if err != nil {
 		streaming.WriteError(err, response.ResponseWriter)
 		return
 	}
-	if redirect != nil {
-		http.Redirect(response.ResponseWriter, request.Request, redirect.String(), http.StatusFound)
+	if s.redirectContainerStreaming {
+		http.Redirect(response.ResponseWriter, request.Request, url.String(), http.StatusFound)
 		return
 	}
-
-	portforward.ServePortForward(response.ResponseWriter,
-		request.Request,
-		s.host,
-		kubecontainer.GetPodFullName(pod),
-		params.podUID,
-		portForwardOptions,
-		s.host.StreamingConnectionIdleTimeout(),
-		remotecommandconsts.DefaultStreamCreationTimeout,
-		portforward.SupportedProtocols)
+	handler := proxy.NewUpgradeAwareHandler(url, nil /*transport*/, false /*wrapTransport*/, false /*upgradeRequired*/, &responder{})
+	handler.ServeHTTP(response.ResponseWriter, request.Request)
 }
 
 // ServeHTTP responds to HTTP requests on the Kubelet.

pkg/kubelet/server/server_test.go

@@ -46,7 +46,6 @@ import (
 	"k8s.io/apiserver/pkg/authentication/user"
 	"k8s.io/apiserver/pkg/authorization/authorizer"
 	"k8s.io/client-go/tools/remotecommand"
-	utiltesting "k8s.io/client-go/util/testing"
 	api "k8s.io/kubernetes/pkg/apis/core"
 	statsapi "k8s.io/kubernetes/pkg/kubelet/apis/stats/v1alpha1"
 	// Do some initialization to decode the query parameters correctly.
@@ -203,7 +202,6 @@ type serverTestFramework struct {
 	fakeKubelet     *fakeKubelet
 	fakeAuth        *fakeAuth
 	testHTTPServer  *httptest.Server
-	criHandler      *utiltesting.FakeHandler
 }
 
 func newServerTest() *serverTestFramework {
@@ -238,17 +236,13 @@ func newServerTestWithDebug(enableDebugging bool) *serverTestFramework {
 			return authorizer.DecisionAllow, "", nil
 		},
 	}
-	fw.criHandler = &utiltesting.FakeHandler{
-		StatusCode: http.StatusOK,
-	}
 	server := NewServer(
 		fw.fakeKubelet,
 		stats.NewResourceAnalyzer(fw.fakeKubelet, time.Minute),
 		fw.fakeAuth,
 		enableDebugging,
 		false,
-		&kubecontainertesting.Mock{},
-		fw.criHandler)
+		&kubecontainertesting.Mock{})
 	fw.serverUnderTest = &server
 	fw.testHTTPServer = httptest.NewServer(fw.serverUnderTest)
 	return fw
@@ -1599,22 +1593,6 @@ func TestServePortForward(t *testing.T) {
 	}
 }
 
-func TestCRIHandler(t *testing.T) {
-	fw := newServerTest()
-	defer fw.testHTTPServer.Close()
-
-	const (
-		path  = "/cri/exec/123456abcdef"
-		query = "cmd=echo+foo"
-	)
-	resp, err := http.Get(fw.testHTTPServer.URL + path + "?" + query)
-	require.NoError(t, err)
-	assert.Equal(t, http.StatusOK, resp.StatusCode)
-	assert.Equal(t, "GET", fw.criHandler.RequestReceived.Method)
-	assert.Equal(t, path, fw.criHandler.RequestReceived.URL.Path)
-	assert.Equal(t, query, fw.criHandler.RequestReceived.URL.RawQuery)
-}
-
 func TestDebuggingDisabledHandlers(t *testing.T) {
 	fw := newServerTestWithDebug(false)
 	defer fw.testHTTPServer.Close()

pkg/kubelet/server/streaming/server.go

@@ -20,6 +20,7 @@ import (
 	"crypto/tls"
 	"errors"
 	"io"
+	"net"
 	"net/http"
 	"net/url"
 	"path"
@@ -71,6 +72,7 @@ type Config struct {
 	Addr string
 	// The optional base URL for constructing streaming URLs. If empty, the baseURL will be
 	// constructed from the serve address.
+	// Note that for port "0", the URL port will be set to actual port in use.
 	BaseURL *url.URL
 
 	// How long to leave idle connections open for.
@@ -233,10 +235,16 @@ func (s *server) Start(stayUp bool) error {
 		return errors.New("stayUp=false is not yet implemented")
 	}
 
+	listener, err := net.Listen("tcp", s.config.Addr)
+	if err != nil {
+		return err
+	}
+	// Use the actual address as baseURL host. This handles the "0" port case.
+	s.config.BaseURL.Host = listener.Addr().String()
 	if s.config.TLSConfig != nil {
-		return s.server.ListenAndServeTLS("", "") // Use certs from TLSConfig.
+		return s.server.ServeTLS(listener, "", "") // Use certs from TLSConfig.
 	} else {
-		return s.server.ListenAndServe()
+		return s.server.Serve(listener)
 	}
 }