Make UpgradeAwareProxyHandler use transport.Dial if provided

pkg/registry/generic/rest/proxy.go

@@ -174,23 +174,56 @@ func (h *UpgradeAwareProxyHandler) tryUpgrade(w http.ResponseWriter, req *http.R
 func (h *UpgradeAwareProxyHandler) dialURL() (net.Conn, error) {
 	dialAddr := netutil.CanonicalAddr(h.Location)
 
+	var dialer func(network, addr string) (net.Conn, error)
+	if httpTransport, ok := h.Transport.(*http.Transport); ok && httpTransport.Dial != nil {
+		dialer = httpTransport.Dial
+	}
+
 	switch h.Location.Scheme {
 	case "http":
+		if dialer != nil {
+			return dialer("tcp", dialAddr)
+		}
 		return net.Dial("tcp", dialAddr)
 	case "https":
+		// TODO: this TLS logic can probably be cleaned up; it's messy in an attempt
+		// to preserve behavior that we don't know for sure is exercised.
+
 		// Get the tls config from the transport if we recognize it
 		var tlsConfig *tls.Config
+		var tlsConn *tls.Conn
+		var err error
 		if h.Transport != nil {
 			httpTransport, ok := h.Transport.(*http.Transport)
 			if ok {
 				tlsConfig = httpTransport.TLSClientConfig
 			}
 		}
+		if dialer != nil {
+			// We have a dialer; use it to open the connection, then
+			// create a tls client using the connection.
+			netConn, err := dialer("tcp", dialAddr)
+			if err != nil {
+				return nil, err
+			}
+			// tls.Client requires non-nil config
+			if tlsConfig == nil {
+				glog.Warningf("using custom dialer with no TLSClientConfig. Defaulting to InsecureSkipVerify")
+				tlsConfig = &tls.Config{
+					InsecureSkipVerify: true,
+				}
+			}
+			tlsConn = tls.Client(netConn, tlsConfig)
+			if err := tlsConn.Handshake(); err != nil {
+				return nil, err
+			}
 
-		// Dial
-		tlsConn, err := tls.Dial("tcp", dialAddr, tlsConfig)
-		if err != nil {
-			return nil, err
+		} else {
+			// Dial
+			tlsConn, err = tls.Dial("tcp", dialAddr, tlsConfig)
+			if err != nil {
+				return nil, err
+			}
 		}
 
 		// Verify
@@ -202,7 +235,7 @@ func (h *UpgradeAwareProxyHandler) dialURL() (net.Conn, error) {
 
 		return tlsConn, nil
 	default:
-		return nil, fmt.Errorf("Unknown scheme: %s", h.Location.Scheme)
+		return nil, fmt.Errorf("unknown scheme: %s", h.Location.Scheme)
 	}
 }
 

test/e2e/kubectl.go

@@ -20,7 +20,10 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"io/ioutil"
+	"net/http"
 	"path/filepath"
+	"regexp"
 	"strings"
 	"time"
 
@@ -42,6 +45,14 @@ const (
 	kubectlProxyPort         = 8011
 	guestbookStartupTimeout  = 10 * time.Minute
 	guestbookResponseTimeout = 3 * time.Minute
+	simplePodSelector        = "name=nginx"
+	simplePodName            = "nginx"
+	nginxDefaultOutput       = "Welcome to nginx!"
+	simplePodPort            = 80
+)
+
+var (
+	portForwardRegexp = regexp.MustCompile("Forwarding from 127.0.0.1:([0-9]+) -> 80")
 )
 
 var _ = Describe("Kubectl client", func() {
@@ -127,8 +138,85 @@ var _ = Describe("Kubectl client", func() {
 		})
 	})
 
+	Describe("Simple pod", func() {
+		var podPath string
+
+		BeforeEach(func() {
+			podPath = filepath.Join(testContext.RepoRoot, "examples/pod.yaml")
+			By("creating the pod")
+			runKubectl("create", "-f", podPath, fmt.Sprintf("--namespace=%v", ns))
+			checkPodsRunningReady(c, ns, []string{simplePodName}, podStartTimeout)
+
+		})
+		AfterEach(func() {
+			cleanup(podPath, ns, simplePodSelector)
+		})
+
+		It("should support exec", func() {
+			By("executing a command in the container")
+			execOutput := runKubectl("exec", fmt.Sprintf("--namespace=%v", ns), simplePodName, "echo", "running", "in", "container")
+			expectedExecOutput := "running in container"
+			if execOutput != expectedExecOutput {
+				Failf("Unexpected kubectl exec output. Wanted '%s', got '%s'", execOutput, expectedExecOutput)
+			}
+		})
+		It("should support port-forward", func() {
+			By("forwarding the container port to a local port")
+			cmd := kubectlCmd("port-forward", fmt.Sprintf("--namespace=%v", ns), "-p", simplePodName, fmt.Sprintf(":%d", simplePodPort))
+			defer func() {
+				if err := cmd.Process.Kill(); err != nil {
+					Logf("ERROR failed to kill kubectl port-forward command! The process may leak")
+				}
+			}()
+			// This is somewhat ugly but is the only way to retrieve the port that was picked
+			// by the port-forward command. We don't want to hard code the port as we have no
+			// way of guaranteeing we can pick one that isn't in use, particularly on Jenkins.
+			Logf("starting port-forward command and streaming output")
+			stdout, stderr, err := startCmdAndStreamOutput(cmd)
+			if err != nil {
+				Failf("Failed to start port-forward command: %v", err)
+			}
+			defer stdout.Close()
+			defer stderr.Close()
+
+			buf := make([]byte, 128)
+			var n int
+			Logf("reading from `kubectl port-forward` command's stderr")
+			if n, err = stderr.Read(buf); err != nil {
+				Failf("Failed to read from kubectl port-forward stderr: %v", err)
+			}
+			portForwardOutput := string(buf[:n])
+			match := portForwardRegexp.FindStringSubmatch(portForwardOutput)
+			if len(match) != 2 {
+				Failf("Failed to parse kubectl port-forward output: %s", portForwardOutput)
+			}
+			By("curling local port output")
+			localAddr := fmt.Sprintf("http://localhost:%s", match[1])
+			body, err := curl(localAddr)
+			if err != nil {
+				Failf("Failed http.Get of forwarded port (%s): %v", localAddr, err)
+			}
+			if !strings.Contains(body, nginxDefaultOutput) {
+				Failf("Container port output missing expected value. Wanted:'%s', got: %s", nginxDefaultOutput, body)
+			}
+		})
+	})
+
 })
 
+func curl(addr string) (string, error) {
+	resp, err := http.Get(addr)
+	if err != nil {
+		return "", err
+	}
+	defer resp.Body.Close()
+	body, err := ioutil.ReadAll(resp.Body)
+	if err != nil {
+		return "", err
+	}
+	return string(body[:]), nil
+}
+
 func validateGuestbookApp(c *client.Client, ns string) {
 	Logf("Waiting for frontend to serve content.")
 	if !waitForGuestbookResponse(c, "get", "", `{"data": ""}`, guestbookStartupTimeout, ns) {

test/e2e/util.go

@@ -19,6 +19,7 @@ package e2e
 import (
 	"bytes"
 	"fmt"
+	"io"
 	"io/ioutil"
 	"math"
 	"math/rand"
@@ -832,6 +833,20 @@ func runKubectl(args ...string) string {
 	return strings.TrimSpace(stdout.String())
 }
 
+func startCmdAndStreamOutput(cmd *exec.Cmd) (stdout, stderr io.ReadCloser, err error) {
+	stdout, err = cmd.StdoutPipe()
+	if err != nil {
+		return
+	}
+	stderr, err = cmd.StderrPipe()
+	if err != nil {
+		return
+	}
+	Logf("Asyncronously running '%s %s'", cmd.Path, strings.Join(cmd.Args, " "))
+	err = cmd.Start()
+	return
+}
+
 // testContainerOutput runs testContainerOutputInNamespace with the default namespace.
 func testContainerOutput(scenarioName string, c *client.Client, pod *api.Pod, containerIndex int, expectedOutput []string) {
 	testContainerOutputInNamespace(scenarioName, c, pod, containerIndex, expectedOutput, api.NamespaceDefault)