move portforward to client-go

pkg/client/tests/BUILD

@@ -13,6 +13,7 @@ go_test(
     srcs = [
         "fake_client_test.go",
         "listwatch_test.go",
+        "portfoward_test.go",
     ],
     library = ":go_default_library",
     tags = ["automanaged"],
@@ -23,13 +24,17 @@ go_test(
         "//pkg/api/v1:go_default_library",
         "//pkg/client/clientset_generated/internalclientset:go_default_library",
         "//pkg/client/clientset_generated/internalclientset/fake:go_default_library",
+        "//pkg/client/unversioned/remotecommand:go_default_library",
+        "//pkg/kubelet/server/portforward:go_default_library",
         "//vendor:k8s.io/apimachinery/pkg/apis/meta/v1",
         "//vendor:k8s.io/apimachinery/pkg/fields",
         "//vendor:k8s.io/apimachinery/pkg/runtime",
+        "//vendor:k8s.io/apimachinery/pkg/types",
         "//vendor:k8s.io/apimachinery/pkg/watch",
         "//vendor:k8s.io/client-go/pkg/api/install",
         "//vendor:k8s.io/client-go/rest",
         "//vendor:k8s.io/client-go/tools/cache",
+        "//vendor:k8s.io/client-go/tools/portforward",
         "//vendor:k8s.io/client-go/util/testing",
     ],
 )

pkg/client/tests/portfoward_test.go

+/*
+Copyright 2015 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 tests
+
+import (
+	"bytes"
+	"fmt"
+	"io"
+	"net"
+	"net/http"
+	"net/http/httptest"
+	"net/url"
+	"os"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"k8s.io/apimachinery/pkg/types"
+	restclient "k8s.io/client-go/rest"
+	. "k8s.io/client-go/tools/portforward"
+	"k8s.io/kubernetes/pkg/client/unversioned/remotecommand"
+	"k8s.io/kubernetes/pkg/kubelet/server/portforward"
+)
+
+// fakePortForwarder simulates port forwarding for testing. It implements
+// portforward.PortForwarder.
+type fakePortForwarder struct {
+	lock sync.Mutex
+	// stores data expected from the stream per port
+	expected map[uint16]string
+	// stores data received from the stream per port
+	received map[uint16]string
+	// data to be sent to the stream per port
+	send map[uint16]string
+}
+
+var _ portforward.PortForwarder = &fakePortForwarder{}
+
+func (pf *fakePortForwarder) PortForward(name string, uid types.UID, port uint16, stream io.ReadWriteCloser) error {
+	defer stream.Close()
+
+	// read from the client
+	received := make([]byte, len(pf.expected[port]))
+	n, err := stream.Read(received)
+	if err != nil {
+		return fmt.Errorf("error reading from client for port %d: %v", port, err)
+	}
+	if n != len(pf.expected[port]) {
+		return fmt.Errorf("unexpected length read from client for port %d: got %d, expected %d. data=%q", port, n, len(pf.expected[port]), string(received))
+	}
+
+	// store the received content
+	pf.lock.Lock()
+	pf.received[port] = string(received)
+	pf.lock.Unlock()
+
+	// send the hardcoded data to the client
+	io.Copy(stream, strings.NewReader(pf.send[port]))
+
+	return nil
+}
+
+// fakePortForwardServer creates an HTTP server that can handle port forwarding
+// requests.
+func fakePortForwardServer(t *testing.T, testName string, serverSends, expectedFromClient map[uint16]string) http.HandlerFunc {
+	return http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
+		pf := &fakePortForwarder{
+			expected: expectedFromClient,
+			received: make(map[uint16]string),
+			send:     serverSends,
+		}
+		portforward.ServePortForward(w, req, pf, "pod", "uid", 0, 10*time.Second)
+
+		for port, expected := range expectedFromClient {
+			actual, ok := pf.received[port]
+			if !ok {
+				t.Errorf("%s: server didn't receive any data for port %d", testName, port)
+				continue
+			}
+
+			if expected != actual {
+				t.Errorf("%s: server expected to receive %q, got %q for port %d", testName, expected, actual, port)
+			}
+		}
+
+		for port, actual := range pf.received {
+			if _, ok := expectedFromClient[port]; !ok {
+				t.Errorf("%s: server unexpectedly received %q for port %d", testName, actual, port)
+			}
+		}
+	})
+}
+
+func TestForwardPorts(t *testing.T) {
+	tests := map[string]struct {
+		ports       []string
+		clientSends map[uint16]string
+		serverSends map[uint16]string
+	}{
+		"forward 1 port with no data either direction": {
+			ports: []string{"5000"},
+		},
+		"forward 2 ports with bidirectional data": {
+			ports: []string{"5001", "6000"},
+			clientSends: map[uint16]string{
+				5001: "abcd",
+				6000: "ghij",
+			},
+			serverSends: map[uint16]string{
+				5001: "1234",
+				6000: "5678",
+			},
+		},
+	}
+
+	for testName, test := range tests {
+		server := httptest.NewServer(fakePortForwardServer(t, testName, test.serverSends, test.clientSends))
+
+		url, _ := url.Parse(server.URL)
+		exec, err := remotecommand.NewExecutor(&restclient.Config{}, "POST", url)
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		stopChan := make(chan struct{}, 1)
+		readyChan := make(chan struct{})
+
+		pf, err := New(exec, test.ports, stopChan, readyChan, os.Stdout, os.Stderr)
+		if err != nil {
+			t.Fatalf("%s: unexpected error calling New: %v", testName, err)
+		}
+
+		doneChan := make(chan error)
+		go func() {
+			doneChan <- pf.ForwardPorts()
+		}()
+		<-pf.Ready
+
+		for port, data := range test.clientSends {
+			clientConn, err := net.Dial("tcp", fmt.Sprintf("localhost:%d", port))
+			if err != nil {
+				t.Errorf("%s: error dialing %d: %s", testName, port, err)
+				server.Close()
+				continue
+			}
+			defer clientConn.Close()
+
+			n, err := clientConn.Write([]byte(data))
+			if err != nil && err != io.EOF {
+				t.Errorf("%s: Error sending data '%s': %s", testName, data, err)
+				server.Close()
+				continue
+			}
+			if n == 0 {
+				t.Errorf("%s: unexpected write of 0 bytes", testName)
+				server.Close()
+				continue
+			}
+			b := make([]byte, 4)
+			n, err = clientConn.Read(b)
+			if err != nil && err != io.EOF {
+				t.Errorf("%s: Error reading data: %s", testName, err)
+				server.Close()
+				continue
+			}
+			if !bytes.Equal([]byte(test.serverSends[port]), b) {
+				t.Errorf("%s: expected to read '%s', got '%s'", testName, test.serverSends[port], b)
+				server.Close()
+				continue
+			}
+		}
+		// tell r.ForwardPorts to stop
+		close(stopChan)
+
+		// wait for r.ForwardPorts to actually return
+		err = <-doneChan
+		if err != nil {
+			t.Errorf("%s: unexpected error: %s", testName, err)
+		}
+		server.Close()
+	}
+
+}
+
+func TestForwardPortsReturnsErrorWhenAllBindsFailed(t *testing.T) {
+	server := httptest.NewServer(fakePortForwardServer(t, "allBindsFailed", nil, nil))
+	defer server.Close()
+
+	url, _ := url.Parse(server.URL)
+	exec, err := remotecommand.NewExecutor(&restclient.Config{}, "POST", url)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	stopChan1 := make(chan struct{}, 1)
+	defer close(stopChan1)
+	readyChan1 := make(chan struct{})
+
+	pf1, err := New(exec, []string{"5555"}, stopChan1, readyChan1, os.Stdout, os.Stderr)
+	if err != nil {
+		t.Fatalf("error creating pf1: %v", err)
+	}
+	go pf1.ForwardPorts()
+	<-pf1.Ready
+
+	stopChan2 := make(chan struct{}, 1)
+	readyChan2 := make(chan struct{})
+	pf2, err := New(exec, []string{"5555"}, stopChan2, readyChan2, os.Stdout, os.Stderr)
+	if err != nil {
+		t.Fatalf("error creating pf2: %v", err)
+	}
+	if err := pf2.ForwardPorts(); err == nil {
+		t.Fatal("expected non-nil error for pf2.ForwardPorts")
+	}
+}

pkg/client/unversioned/BUILD

@@ -60,7 +60,6 @@ filegroup(
     name = "all-srcs",
     srcs = [
         ":package-srcs",
-        "//pkg/client/unversioned/portforward:all-srcs",
         "//pkg/client/unversioned/remotecommand:all-srcs",
         "//pkg/client/unversioned/testclient/simple:all-srcs",
     ],

pkg/client/unversioned/portforward/BUILD

-package(default_visibility = ["//visibility:public"])
-
-licenses(["notice"])
-
-load(
-    "@io_bazel_rules_go//go:def.bzl",
-    "go_library",
-    "go_test",
-)
-
-go_library(
-    name = "go_default_library",
-    srcs = [
-        "doc.go",
-        "portforward.go",
-    ],
-    tags = ["automanaged"],
-    deps = [
-        "//pkg/api:go_default_library",
-        "//pkg/kubelet/server/portforward:go_default_library",
-        "//vendor:k8s.io/apimachinery/pkg/util/httpstream",
-        "//vendor:k8s.io/apimachinery/pkg/util/runtime",
-    ],
-)
-
-go_test(
-    name = "go_default_test",
-    srcs = ["portforward_test.go"],
-    library = ":go_default_library",
-    tags = ["automanaged"],
-    deps = [
-        "//pkg/client/unversioned/remotecommand:go_default_library",
-        "//pkg/kubelet/server/portforward:go_default_library",
-        "//vendor:k8s.io/apimachinery/pkg/types",
-        "//vendor:k8s.io/apimachinery/pkg/util/httpstream",
-        "//vendor:k8s.io/client-go/rest",
-    ],
-)
-
-filegroup(
-    name = "package-srcs",
-    srcs = glob(["**"]),
-    tags = ["automanaged"],
-    visibility = ["//visibility:private"],
-)
-
-filegroup(
-    name = "all-srcs",
-    srcs = [":package-srcs"],
-    tags = ["automanaged"],
-)

pkg/client/unversioned/portforward/doc.go

-/*
-Copyright 2015 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 portforward adds support for SSH-like port forwarding from the client's
-// local host to remote containers.
-package portforward // import "k8s.io/kubernetes/pkg/client/unversioned/portforward"

pkg/kubectl/cmd/BUILD

@@ -73,7 +73,6 @@ go_library(
         "//pkg/client/clientset_generated/internalclientset:go_default_library",
         "//pkg/client/clientset_generated/internalclientset/typed/core/internalversion:go_default_library",
         "//pkg/client/unversioned:go_default_library",
-        "//pkg/client/unversioned/portforward:go_default_library",
         "//pkg/client/unversioned/remotecommand:go_default_library",
         "//pkg/kubectl:go_default_library",
         "//pkg/kubectl/cmd/config:go_default_library",
@@ -125,6 +124,7 @@ go_library(
         "//vendor:k8s.io/client-go/discovery",
         "//vendor:k8s.io/client-go/rest",
         "//vendor:k8s.io/client-go/tools/clientcmd",
+        "//vendor:k8s.io/client-go/tools/portforward",
     ],
 )
 

pkg/kubectl/cmd/portforward.go

@@ -27,9 +27,9 @@ import (
 
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	restclient "k8s.io/client-go/rest"
+	"k8s.io/client-go/tools/portforward"
 	"k8s.io/kubernetes/pkg/api"
 	coreclient "k8s.io/kubernetes/pkg/client/clientset_generated/internalclientset/typed/core/internalversion"
-	"k8s.io/kubernetes/pkg/client/unversioned/portforward"
 	"k8s.io/kubernetes/pkg/client/unversioned/remotecommand"
 	"k8s.io/kubernetes/pkg/kubectl/cmd/templates"
 	cmdutil "k8s.io/kubernetes/pkg/kubectl/cmd/util"

staging/copy.sh

@@ -85,6 +85,7 @@ save "tools/auth"
 save "tools/cache"
 save "tools/clientcmd"
 save "tools/metrics"
+save "tools/portforward"
 save "transport"
 save "third_party"
 save "plugin"
@@ -107,11 +108,6 @@ find "${MAIN_REPO}/pkg/version" -maxdepth 1 -type f | xargs -I{} cp {} "${CLIENT
 mkcp "pkg/client/clientset_generated/${CLIENTSET}" "pkg/client/clientset_generated"
 mkcp "/pkg/client/record" "/pkg/client"
 
-mkcp "/pkg/client/unversioned/portforward" "/pkg/client/unversioned"
-
-# remove this test because it imports the internal clientset
-rm "${CLIENT_REPO_TEMP}"/pkg/client/unversioned/portforward/portforward_test.go
-
 pushd "${CLIENT_REPO_TEMP}" > /dev/null
 echo "generating vendor/"
 GO15VENDOREXPERIMENT=1 godep save ./...
@@ -189,7 +185,6 @@ function mvfolder {
 
 mvfolder "pkg/client/clientset_generated/${CLIENTSET}" kubernetes
 mvfolder pkg/client/record tools/record
-mvfolder pkg/client/unversioned/portforward tools/portforward
 if [ "$(find "${CLIENT_REPO_TEMP}"/pkg/client -type f -name "*.go")" ]; then
     echo "${CLIENT_REPO_TEMP}/pkg/client is expected to be empty"
     exit 1

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

@@ -21,8 +21,8 @@ import (
 	"net/http"
 	"strings"
 
-	"k8s.io/apimachinery/pkg/util/runtime"
 	"k8s.io/apimachinery/pkg/util/httpstream"
+	"k8s.io/apimachinery/pkg/util/runtime"
 )
 
 const HeaderSpdy31 = "SPDY/3.1"

staging/src/k8s.io/client-go/tools/portforward/portforward.go

@@ -30,9 +30,12 @@ import (
 	"k8s.io/apimachinery/pkg/util/httpstream"
 	"k8s.io/apimachinery/pkg/util/runtime"
 	"k8s.io/client-go/pkg/api"
-	"k8s.io/client-go/pkg/kubelet/server/portforward"
 )
 
+// TODO move to API machinery and re-unify with kubelet/server/portfoward
+// The subprotocol "portforward.k8s.io" is used for port forwarding.
+const PortForwardProtocolV1Name = "portforward.k8s.io"
+
 // PortForwarder knows how to listen for local connections and forward them to
 // a remote pod via an upgraded HTTP request.
 type PortForwarder struct {
@@ -132,7 +135,7 @@ func (pf *PortForwarder) ForwardPorts() error {
 	defer pf.Close()
 
 	var err error
-	pf.streamConn, _, err = pf.dialer.Dial(portforward.PortForwardProtocolV1Name)
+	pf.streamConn, _, err = pf.dialer.Dial(PortForwardProtocolV1Name)
 	if err != nil {
 		return fmt.Errorf("error upgrading connection: %s", err)
 	}

pkg/client/unversioned/portforward/portforward_test.go → staging/src/k8s.io/client-go/tools/portforward/portforward_test.go

@@ -17,25 +17,13 @@ limitations under the License.
 package portforward
 
 import (
-	"bytes"
-	"fmt"
-	"io"
 	"net"
-	"net/http"
-	"net/http/httptest"
-	"net/url"
 	"os"
 	"reflect"
 	"strings"
-	"sync"
 	"testing"
-	"time"
 
-	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/httpstream"
-	restclient "k8s.io/client-go/rest"
-	"k8s.io/kubernetes/pkg/client/unversioned/remotecommand"
-	"k8s.io/kubernetes/pkg/kubelet/server/portforward"
 )
 
 type fakeDialer struct {
@@ -204,195 +192,3 @@ func TestGetListener(t *testing.T) {
 
 	}
 }
-
-// fakePortForwarder simulates port forwarding for testing. It implements
-// portforward.PortForwarder.
-type fakePortForwarder struct {
-	lock sync.Mutex
-	// stores data expected from the stream per port
-	expected map[uint16]string
-	// stores data received from the stream per port
-	received map[uint16]string
-	// data to be sent to the stream per port
-	send map[uint16]string
-}
-
-var _ portforward.PortForwarder = &fakePortForwarder{}
-
-func (pf *fakePortForwarder) PortForward(name string, uid types.UID, port uint16, stream io.ReadWriteCloser) error {
-	defer stream.Close()
-
-	// read from the client
-	received := make([]byte, len(pf.expected[port]))
-	n, err := stream.Read(received)
-	if err != nil {
-		return fmt.Errorf("error reading from client for port %d: %v", port, err)
-	}
-	if n != len(pf.expected[port]) {
-		return fmt.Errorf("unexpected length read from client for port %d: got %d, expected %d. data=%q", port, n, len(pf.expected[port]), string(received))
-	}
-
-	// store the received content
-	pf.lock.Lock()
-	pf.received[port] = string(received)
-	pf.lock.Unlock()
-
-	// send the hardcoded data to the client
-	io.Copy(stream, strings.NewReader(pf.send[port]))
-
-	return nil
-}
-
-// fakePortForwardServer creates an HTTP server that can handle port forwarding
-// requests.
-func fakePortForwardServer(t *testing.T, testName string, serverSends, expectedFromClient map[uint16]string) http.HandlerFunc {
-	return http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
-		pf := &fakePortForwarder{
-			expected: expectedFromClient,
-			received: make(map[uint16]string),
-			send:     serverSends,
-		}
-		portforward.ServePortForward(w, req, pf, "pod", "uid", 0, 10*time.Second)
-
-		for port, expected := range expectedFromClient {
-			actual, ok := pf.received[port]
-			if !ok {
-				t.Errorf("%s: server didn't receive any data for port %d", testName, port)
-				continue
-			}
-
-			if expected != actual {
-				t.Errorf("%s: server expected to receive %q, got %q for port %d", testName, expected, actual, port)
-			}
-		}
-
-		for port, actual := range pf.received {
-			if _, ok := expectedFromClient[port]; !ok {
-				t.Errorf("%s: server unexpectedly received %q for port %d", testName, actual, port)
-			}
-		}
-	})
-}
-
-func TestForwardPorts(t *testing.T) {
-	tests := map[string]struct {
-		ports       []string
-		clientSends map[uint16]string
-		serverSends map[uint16]string
-	}{
-		"forward 1 port with no data either direction": {
-			ports: []string{"5000"},
-		},
-		"forward 2 ports with bidirectional data": {
-			ports: []string{"5001", "6000"},
-			clientSends: map[uint16]string{
-				5001: "abcd",
-				6000: "ghij",
-			},
-			serverSends: map[uint16]string{
-				5001: "1234",
-				6000: "5678",
-			},
-		},
-	}
-
-	for testName, test := range tests {
-		server := httptest.NewServer(fakePortForwardServer(t, testName, test.serverSends, test.clientSends))
-
-		url, _ := url.Parse(server.URL)
-		exec, err := remotecommand.NewExecutor(&restclient.Config{}, "POST", url)
-		if err != nil {
-			t.Fatal(err)
-		}
-
-		stopChan := make(chan struct{}, 1)
-		readyChan := make(chan struct{})
-
-		pf, err := New(exec, test.ports, stopChan, readyChan, os.Stdout, os.Stderr)
-		if err != nil {
-			t.Fatalf("%s: unexpected error calling New: %v", testName, err)
-		}
-
-		doneChan := make(chan error)
-		go func() {
-			doneChan <- pf.ForwardPorts()
-		}()
-		<-pf.Ready
-
-		for port, data := range test.clientSends {
-			clientConn, err := net.Dial("tcp", fmt.Sprintf("localhost:%d", port))
-			if err != nil {
-				t.Errorf("%s: error dialing %d: %s", testName, port, err)
-				server.Close()
-				continue
-			}
-			defer clientConn.Close()
-
-			n, err := clientConn.Write([]byte(data))
-			if err != nil && err != io.EOF {
-				t.Errorf("%s: Error sending data '%s': %s", testName, data, err)
-				server.Close()
-				continue
-			}
-			if n == 0 {
-				t.Errorf("%s: unexpected write of 0 bytes", testName)
-				server.Close()
-				continue
-			}
-			b := make([]byte, 4)
-			n, err = clientConn.Read(b)
-			if err != nil && err != io.EOF {
-				t.Errorf("%s: Error reading data: %s", testName, err)
-				server.Close()
-				continue
-			}
-			if !bytes.Equal([]byte(test.serverSends[port]), b) {
-				t.Errorf("%s: expected to read '%s', got '%s'", testName, test.serverSends[port], b)
-				server.Close()
-				continue
-			}
-		}
-		// tell r.ForwardPorts to stop
-		close(stopChan)
-
-		// wait for r.ForwardPorts to actually return
-		err = <-doneChan
-		if err != nil {
-			t.Errorf("%s: unexpected error: %s", testName, err)
-		}
-		server.Close()
-	}
-
-}
-
-func TestForwardPortsReturnsErrorWhenAllBindsFailed(t *testing.T) {
-	server := httptest.NewServer(fakePortForwardServer(t, "allBindsFailed", nil, nil))
-	defer server.Close()
-
-	url, _ := url.Parse(server.URL)
-	exec, err := remotecommand.NewExecutor(&restclient.Config{}, "POST", url)
-	if err != nil {
-		t.Fatal(err)
-	}
-
-	stopChan1 := make(chan struct{}, 1)
-	defer close(stopChan1)
-	readyChan1 := make(chan struct{})
-
-	pf1, err := New(exec, []string{"5555"}, stopChan1, readyChan1, os.Stdout, os.Stderr)
-	if err != nil {
-		t.Fatalf("error creating pf1: %v", err)
-	}
-	go pf1.ForwardPorts()
-	<-pf1.Ready
-
-	stopChan2 := make(chan struct{}, 1)
-	readyChan2 := make(chan struct{})
-	pf2, err := New(exec, []string{"5555"}, stopChan2, readyChan2, os.Stdout, os.Stderr)
-	if err != nil {
-		t.Fatalf("error creating pf2: %v", err)
-	}
-	if err := pf2.ForwardPorts(); err == nil {
-		t.Fatal("expected non-nil error for pf2.ForwardPorts")
-	}
-}

vendor/BUILD

@@ -12582,7 +12582,6 @@ go_library(
         "//vendor:k8s.io/apimachinery/pkg/util/httpstream",
         "//vendor:k8s.io/apimachinery/pkg/util/runtime",
         "//vendor:k8s.io/client-go/pkg/api",
-        "//vendor:k8s.io/client-go/pkg/kubelet/server/portforward",
     ],
 )
 
@@ -13999,3 +13998,11 @@ go_library(
         "//vendor:k8s.io/apimachinery/pkg/openapi",
     ],
 )
+
+go_test(
+    name = "k8s.io/client-go/tools/portforward_test",
+    srcs = ["k8s.io/client-go/tools/portforward/portforward_test.go"],
+    library = ":k8s.io/client-go/tools/portforward",
+    tags = ["automanaged"],
+    deps = ["//vendor:k8s.io/apimachinery/pkg/util/httpstream"],
+)