Add redirect support to SpdyRoundTripper

Add support for following redirects to the SpdyRoundTripper. This is
necessary for clients using it directly (e.g. the apiserver talking
directly to the kubelet) because the CRI streaming server issues a
redirect for streaming requests.

Also extract common logic for following redirects.

pkg/client/unversioned/remotecommand/remotecommand.go

@@ -78,7 +78,7 @@ func NewExecutor(config *restclient.Config, method string, url *url.URL) (Stream
 		return nil, err
 	}
 
-	upgradeRoundTripper := spdy.NewRoundTripper(tlsConfig)
+	upgradeRoundTripper := spdy.NewRoundTripper(tlsConfig, true)
 	wrapper, err := restclient.HTTPWrappersForConfig(config, upgradeRoundTripper)
 	if err != nil {
 		return nil, err

pkg/kubelet/server/server_test.go

@@ -1119,7 +1119,7 @@ func TestServeExecInContainerIdleTimeout(t *testing.T) {
 
 	url := fw.testHTTPServer.URL + "/exec/" + podNamespace + "/" + podName + "/" + expectedContainerName + "?c=ls&c=-a&" + api.ExecStdinParam + "=1"
 
-	upgradeRoundTripper := spdy.NewSpdyRoundTripper(nil)
+	upgradeRoundTripper := spdy.NewSpdyRoundTripper(nil, true)
 	c := &http.Client{Transport: upgradeRoundTripper}
 
 	resp, err := c.Post(url, "", nil)
@@ -1304,7 +1304,7 @@ func testExecAttach(t *testing.T, verb string) {
 				return http.ErrUseLastResponse
 			}
 		} else {
-			upgradeRoundTripper = spdy.NewRoundTripper(nil)
+			upgradeRoundTripper = spdy.NewRoundTripper(nil, true)
 			c = &http.Client{Transport: upgradeRoundTripper}
 		}
 
@@ -1442,7 +1442,7 @@ func TestServePortForwardIdleTimeout(t *testing.T) {
 
 	url := fw.testHTTPServer.URL + "/portForward/" + podNamespace + "/" + podName
 
-	upgradeRoundTripper := spdy.NewRoundTripper(nil)
+	upgradeRoundTripper := spdy.NewRoundTripper(nil, true)
 	c := &http.Client{Transport: upgradeRoundTripper}
 
 	resp, err := c.Post(url, "", nil)
@@ -1552,11 +1552,20 @@ func TestServePortForward(t *testing.T) {
 			url = fmt.Sprintf("%s/portForward/%s/%s", fw.testHTTPServer.URL, podNamespace, podName)
 		}
 
-		upgradeRoundTripper := spdy.NewRoundTripper(nil)
-		c := &http.Client{Transport: upgradeRoundTripper}
-		// Don't follow redirects, since we want to inspect the redirect response.
-		c.CheckRedirect = func(*http.Request, []*http.Request) error {
-			return http.ErrUseLastResponse
+		var (
+			upgradeRoundTripper httpstream.UpgradeRoundTripper
+			c                   *http.Client
+		)
+
+		if len(test.responseLocation) > 0 {
+			c = &http.Client{}
+			// Don't follow redirects, since we want to inspect the redirect response.
+			c.CheckRedirect = func(*http.Request, []*http.Request) error {
+				return http.ErrUseLastResponse
+			}
+		} else {
+			upgradeRoundTripper = spdy.NewRoundTripper(nil, true)
+			c = &http.Client{Transport: upgradeRoundTripper}
 		}
 
 		resp, err := c.Post(url, "", nil)

staging/src/k8s.io/apimachinery/pkg/util/httpstream/spdy/roundtripper.go

@@ -18,9 +18,11 @@ package spdy
 
 import (
 	"bufio"
+	"bytes"
 	"crypto/tls"
 	"encoding/base64"
 	"fmt"
+	"io"
 	"io/ioutil"
 	"net"
 	"net/http"
@@ -33,6 +35,7 @@ import (
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/serializer"
 	"k8s.io/apimachinery/pkg/util/httpstream"
+	utilnet "k8s.io/apimachinery/pkg/util/net"
 	"k8s.io/apimachinery/third_party/forked/golang/netutil"
 )
 
@@ -59,25 +62,49 @@ type SpdyRoundTripper struct {
 	// proxier knows which proxy to use given a request, defaults to http.ProxyFromEnvironment
 	// Used primarily for mocking the proxy discovery in tests.
 	proxier func(req *http.Request) (*url.URL, error)
+
+	// followRedirects indicates if the round tripper should examine responses for redirects and
+	// follow them.
+	followRedirects bool
 }
 
+var _ utilnet.TLSClientConfigHolder = &SpdyRoundTripper{}
+var _ httpstream.UpgradeRoundTripper = &SpdyRoundTripper{}
+var _ utilnet.Dialer = &SpdyRoundTripper{}
+
 // NewRoundTripper creates a new SpdyRoundTripper that will use
 // the specified tlsConfig.
-func NewRoundTripper(tlsConfig *tls.Config) httpstream.UpgradeRoundTripper {
-	return NewSpdyRoundTripper(tlsConfig)
+func NewRoundTripper(tlsConfig *tls.Config, followRedirects bool) httpstream.UpgradeRoundTripper {
+	return NewSpdyRoundTripper(tlsConfig, followRedirects)
 }
 
 // NewSpdyRoundTripper creates a new SpdyRoundTripper that will use
 // the specified tlsConfig. This function is mostly meant for unit tests.
-func NewSpdyRoundTripper(tlsConfig *tls.Config) *SpdyRoundTripper {
-	return &SpdyRoundTripper{tlsConfig: tlsConfig}
+func NewSpdyRoundTripper(tlsConfig *tls.Config, followRedirects bool) *SpdyRoundTripper {
+	return &SpdyRoundTripper{tlsConfig: tlsConfig, followRedirects: followRedirects}
 }
 
-// implements pkg/util/net.TLSClientConfigHolder for proper TLS checking during proxying with a spdy roundtripper
+// TLSClientConfig implements pkg/util/net.TLSClientConfigHolder for proper TLS checking during
+// proxying with a spdy roundtripper.
 func (s *SpdyRoundTripper) TLSClientConfig() *tls.Config {
 	return s.tlsConfig
 }
 
+// Dial implements k8s.io/apimachinery/pkg/util/net.Dialer.
+func (s *SpdyRoundTripper) Dial(req *http.Request) (net.Conn, error) {
+	conn, err := s.dial(req)
+	if err != nil {
+		return nil, err
+	}
+
+	if err := req.Write(conn); err != nil {
+		conn.Close()
+		return nil, err
+	}
+
+	return conn, nil
+}
+
 // dial dials the host specified by req, using TLS if appropriate, optionally
 // using a proxy server if one is configured via environment variables.
 func (s *SpdyRoundTripper) dial(req *http.Request) (net.Conn, error) {
@@ -213,24 +240,39 @@ func (s *SpdyRoundTripper) proxyAuth(proxyURL *url.URL) string {
 // clients may call SpdyRoundTripper.Connection() to retrieve the upgraded
 // connection.
 func (s *SpdyRoundTripper) RoundTrip(req *http.Request) (*http.Response, error) {
-	// TODO what's the best way to clone the request?
-	r := *req
-	req = &r
-	req.Header.Add(httpstream.HeaderConnection, httpstream.HeaderUpgrade)
-	req.Header.Add(httpstream.HeaderUpgrade, HeaderSpdy31)
+	header := utilnet.CloneHeader(req.Header)
+	header.Add(httpstream.HeaderConnection, httpstream.HeaderUpgrade)
+	header.Add(httpstream.HeaderUpgrade, HeaderSpdy31)
+
+	var (
+		conn        net.Conn
+		rawResponse []byte
+		err         error
+	)
 
-	conn, err := s.dial(req)
-	if err != nil {
-		return nil, err
+	if s.followRedirects {
+		conn, rawResponse, err = utilnet.ConnectWithRedirects(req.Method, req.URL, header, req.Body, s)
+	} else {
+		clone := utilnet.CloneRequest(req)
+		clone.Header = header
+		conn, err = s.Dial(clone)
 	}
-
-	err = req.Write(conn)
 	if err != nil {
 		return nil, err
 	}
 
-	resp, err := http.ReadResponse(bufio.NewReader(conn), req)
+	responseReader := bufio.NewReader(
+		io.MultiReader(
+			bytes.NewBuffer(rawResponse),
+			conn,
+		),
+	)
+
+	resp, err := http.ReadResponse(responseReader, nil)
 	if err != nil {
+		if conn != nil {
+			conn.Close()
+		}
 		return nil, err
 	}