Unify generic proxy code in apimachinery

pkg/registry/core/node/rest/proxy.go

@@ -23,9 +23,9 @@ import (
 	"path"
 
 	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/util/proxy"
 	genericapirequest "k8s.io/apiserver/pkg/endpoints/request"
 	genericregistry "k8s.io/apiserver/pkg/registry/generic/registry"
-	genericrest "k8s.io/apiserver/pkg/registry/generic/rest"
 	"k8s.io/apiserver/pkg/registry/rest"
 	"k8s.io/kubernetes/pkg/api"
 	"k8s.io/kubernetes/pkg/capabilities"
@@ -75,8 +75,8 @@ func (r *ProxyREST) Connect(ctx genericapirequest.Context, id string, opts runti
 	return newThrottledUpgradeAwareProxyHandler(location, transport, true, false, responder), nil
 }
 
-func newThrottledUpgradeAwareProxyHandler(location *url.URL, transport http.RoundTripper, wrapTransport, upgradeRequired bool, responder rest.Responder) *genericrest.UpgradeAwareProxyHandler {
-	handler := genericrest.NewUpgradeAwareProxyHandler(location, transport, wrapTransport, upgradeRequired, responder)
+func newThrottledUpgradeAwareProxyHandler(location *url.URL, transport http.RoundTripper, wrapTransport, upgradeRequired bool, responder rest.Responder) *proxy.UpgradeAwareHandler {
+	handler := proxy.NewUpgradeAwareHandler(location, transport, wrapTransport, upgradeRequired, responder)
 	handler.MaxBytesPerSec = capabilities.Get().PerConnectionBandwidthLimitBytesPerSec
 	return handler
 }

pkg/registry/core/pod/rest/subresources.go

@@ -23,10 +23,12 @@ import (
 	"path"
 
 	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/util/proxy"
 	genericapirequest "k8s.io/apiserver/pkg/endpoints/request"
+	genericfeatures "k8s.io/apiserver/pkg/features"
 	genericregistry "k8s.io/apiserver/pkg/registry/generic/registry"
-	genericrest "k8s.io/apiserver/pkg/registry/generic/rest"
 	"k8s.io/apiserver/pkg/registry/rest"
+	utilfeature "k8s.io/apiserver/pkg/util/feature"
 	"k8s.io/kubernetes/pkg/api"
 	"k8s.io/kubernetes/pkg/capabilities"
 	"k8s.io/kubernetes/pkg/kubelet/client"
@@ -189,9 +191,9 @@ func (r *PortForwardREST) Connect(ctx genericapirequest.Context, name string, op
 	return newThrottledUpgradeAwareProxyHandler(location, transport, false, true, true, responder), nil
 }
 
-func newThrottledUpgradeAwareProxyHandler(location *url.URL, transport http.RoundTripper, wrapTransport, upgradeRequired, interceptRedirects bool, responder rest.Responder) *genericrest.UpgradeAwareProxyHandler {
-	handler := genericrest.NewUpgradeAwareProxyHandler(location, transport, wrapTransport, upgradeRequired, responder)
-	handler.InterceptRedirects = interceptRedirects
+func newThrottledUpgradeAwareProxyHandler(location *url.URL, transport http.RoundTripper, wrapTransport, upgradeRequired, interceptRedirects bool, responder rest.Responder) *proxy.UpgradeAwareHandler {
+	handler := proxy.NewUpgradeAwareHandler(location, transport, wrapTransport, upgradeRequired, responder)
+	handler.InterceptRedirects = interceptRedirects && utilfeature.DefaultFeatureGate.Enabled(genericfeatures.StreamingProxyRedirects)
 	handler.MaxBytesPerSec = capabilities.Get().PerConnectionBandwidthLimitBytesPerSec
 	return handler
 }

pkg/registry/core/service/proxy.go

@@ -23,8 +23,8 @@ import (
 	"path"
 
 	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/util/proxy"
 	genericapirequest "k8s.io/apiserver/pkg/endpoints/request"
-	genericrest "k8s.io/apiserver/pkg/registry/generic/rest"
 	"k8s.io/apiserver/pkg/registry/rest"
 	"k8s.io/kubernetes/pkg/api"
 	"k8s.io/kubernetes/pkg/capabilities"
@@ -71,8 +71,8 @@ func (r *ProxyREST) Connect(ctx genericapirequest.Context, id string, opts runti
 	return newThrottledUpgradeAwareProxyHandler(location, transport, true, false, responder), nil
 }
 
-func newThrottledUpgradeAwareProxyHandler(location *url.URL, transport http.RoundTripper, wrapTransport, upgradeRequired bool, responder rest.Responder) *genericrest.UpgradeAwareProxyHandler {
-	handler := genericrest.NewUpgradeAwareProxyHandler(location, transport, wrapTransport, upgradeRequired, responder)
+func newThrottledUpgradeAwareProxyHandler(location *url.URL, transport http.RoundTripper, wrapTransport, upgradeRequired bool, responder rest.Responder) *proxy.UpgradeAwareHandler {
+	handler := proxy.NewUpgradeAwareHandler(location, transport, wrapTransport, upgradeRequired, responder)
 	handler.MaxBytesPerSec = capabilities.Get().PerConnectionBandwidthLimitBytesPerSec
 	return handler
 }

staging/src/k8s.io/apiserver/pkg/util/proxy/doc.go → staging/src/k8s.io/apimachinery/pkg/util/proxy/doc.go

@@ -14,5 +14,5 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-// Package proxy provides transport and upgrade support for proxies
-package proxy // import "k8s.io/apiserver/pkg/util/proxy"
+// Package proxy provides transport and upgrade support for proxies.
+package proxy // import "k8s.io/apimachinery/pkg/util/proxy"

staging/src/k8s.io/apiserver/pkg/registry/generic/rest/proxy.go → staging/src/k8s.io/apimachinery/pkg/util/proxy/upgradeaware.go

 /*
-Copyright 2014 The Kubernetes Authors.
+Copyright 2017 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.
@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package rest
+package proxy
 
 import (
 	"context"
@@ -32,16 +32,13 @@ import (
 	"k8s.io/apimachinery/pkg/util/httpstream"
 	utilnet "k8s.io/apimachinery/pkg/util/net"
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
-	genericfeatures "k8s.io/apiserver/pkg/features"
-	utilfeature "k8s.io/apiserver/pkg/util/feature"
-	"k8s.io/apiserver/pkg/util/proxy"
 
 	"github.com/golang/glog"
 	"github.com/mxk/go-flowrate/flowrate"
 )
 
-// UpgradeAwareProxyHandler is a handler for proxy requests that may require an upgrade
-type UpgradeAwareProxyHandler struct {
+// UpgradeAwareHandler is a handler for proxy requests that may require an upgrade
+type UpgradeAwareHandler struct {
 	UpgradeRequired bool
 	Location        *url.URL
 	// Transport provides an optional round tripper to use to proxy. If nil, the default proxy transport is used
@@ -64,10 +61,10 @@ type ErrorResponder interface {
 	Error(err error)
 }
 
-// NewUpgradeAwareProxyHandler creates a new proxy handler with a default flush interval. Responder is required for returning
+// NewUpgradeAwareHandler creates a new proxy handler with a default flush interval. Responder is required for returning
 // errors to the caller.
-func NewUpgradeAwareProxyHandler(location *url.URL, transport http.RoundTripper, wrapTransport, upgradeRequired bool, responder ErrorResponder) *UpgradeAwareProxyHandler {
-	return &UpgradeAwareProxyHandler{
+func NewUpgradeAwareHandler(location *url.URL, transport http.RoundTripper, wrapTransport, upgradeRequired bool, responder ErrorResponder) *UpgradeAwareHandler {
+	return &UpgradeAwareHandler{
 		Location:        location,
 		Transport:       transport,
 		WrapTransport:   wrapTransport,
@@ -78,7 +75,7 @@ func NewUpgradeAwareProxyHandler(location *url.URL, transport http.RoundTripper,
 }
 
 // ServeHTTP handles the proxy request
-func (h *UpgradeAwareProxyHandler) ServeHTTP(w http.ResponseWriter, req *http.Request) {
+func (h *UpgradeAwareHandler) ServeHTTP(w http.ResponseWriter, req *http.Request) {
 	if len(h.Location.Scheme) == 0 {
 		h.Location.Scheme = "http"
 	}
@@ -129,7 +126,7 @@ func (h *UpgradeAwareProxyHandler) ServeHTTP(w http.ResponseWriter, req *http.Re
 }
 
 // tryUpgrade returns true if the request was handled.
-func (h *UpgradeAwareProxyHandler) tryUpgrade(w http.ResponseWriter, req *http.Request) bool {
+func (h *UpgradeAwareHandler) tryUpgrade(w http.ResponseWriter, req *http.Request) bool {
 	if !httpstream.IsUpgradeRequest(req) {
 		return false
 	}
@@ -144,7 +141,7 @@ func (h *UpgradeAwareProxyHandler) tryUpgrade(w http.ResponseWriter, req *http.R
 	// Only append X-Forwarded-For in the upgrade path, since httputil.NewSingleHostReverseProxy
 	// handles this in the non-upgrade path.
 	utilnet.AppendForwardedForHeader(clone)
-	if h.InterceptRedirects && utilfeature.DefaultFeatureGate.Enabled(genericfeatures.StreamingProxyRedirects) {
+	if h.InterceptRedirects {
 		backendConn, rawResponse, err = utilnet.ConnectWithRedirects(req.Method, h.Location, clone.Header, req.Body, h)
 	} else {
 		clone.URL = h.Location
@@ -214,8 +211,8 @@ func (h *UpgradeAwareProxyHandler) tryUpgrade(w http.ResponseWriter, req *http.R
 }
 
 // Dial dials the backend at req.URL and writes req to it.
-func (h *UpgradeAwareProxyHandler) Dial(req *http.Request) (net.Conn, error) {
-	conn, err := proxy.DialURL(req.URL, h.Transport)
+func (h *UpgradeAwareHandler) Dial(req *http.Request) (net.Conn, error) {
+	conn, err := DialURL(req.URL, h.Transport)
 	if err != nil {
 		return nil, fmt.Errorf("error dialing backend: %v", err)
 	}
@@ -228,9 +225,9 @@ func (h *UpgradeAwareProxyHandler) Dial(req *http.Request) (net.Conn, error) {
 	return conn, err
 }
 
-var _ utilnet.Dialer = &UpgradeAwareProxyHandler{}
+var _ utilnet.Dialer = &UpgradeAwareHandler{}
 
-func (h *UpgradeAwareProxyHandler) defaultProxyTransport(url *url.URL, internalTransport http.RoundTripper) http.RoundTripper {
+func (h *UpgradeAwareHandler) defaultProxyTransport(url *url.URL, internalTransport http.RoundTripper) http.RoundTripper {
 	scheme := url.Scheme
 	host := url.Host
 	suffix := h.Location.Path
@@ -238,7 +235,7 @@ func (h *UpgradeAwareProxyHandler) defaultProxyTransport(url *url.URL, internalT
 		suffix += "/"
 	}
 	pathPrepend := strings.TrimSuffix(url.Path, suffix)
-	rewritingTransport := &proxy.Transport{
+	rewritingTransport := &Transport{
 		Scheme:       scheme,
 		Host:         host,
 		PathPrepend:  pathPrepend,

staging/src/k8s.io/apiserver/pkg/registry/generic/rest/proxy_test.go → staging/src/k8s.io/apimachinery/pkg/util/proxy/upgradeaware_test.go

 /*
-Copyright 2014 The Kubernetes Authors.
+Copyright 2017 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.
@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package rest
+package proxy
 
 import (
 	"bytes"
@@ -43,9 +43,6 @@ import (
 
 	"k8s.io/apimachinery/pkg/util/httpstream"
 	utilnet "k8s.io/apimachinery/pkg/util/net"
-	"k8s.io/apiserver/pkg/features"
-	utilfeature "k8s.io/apiserver/pkg/util/feature"
-	"k8s.io/apiserver/pkg/util/proxy"
 )
 
 const fakeStatusCode = 567
@@ -248,7 +245,7 @@ func TestServeHTTP(t *testing.T) {
 			responder := &fakeResponder{t: t}
 			backendURL, _ := url.Parse(backendServer.URL)
 			backendURL.Path = test.requestPath
-			proxyHandler := &UpgradeAwareProxyHandler{
+			proxyHandler := &UpgradeAwareHandler{
 				Location:        backendURL,
 				Responder:       responder,
 				UpgradeRequired: test.upgradeRequired,
@@ -402,8 +399,6 @@ func TestProxyUpgrade(t *testing.T) {
 		},
 	}
 
-	// Enable StreamingProxyRedirects for test.
-	utilfeature.DefaultFeatureGate.Set(string(features.StreamingProxyRedirects) + "=true")
 	for k, tc := range testcases {
 		for _, redirect := range []bool{false, true} {
 			tcName := k
@@ -428,7 +423,7 @@ func TestProxyUpgrade(t *testing.T) {
 
 				serverURL, _ := url.Parse(backendServer.URL)
 				serverURL.Path = backendPath
-				proxyHandler := &UpgradeAwareProxyHandler{
+				proxyHandler := &UpgradeAwareHandler{
 					Location:           serverURL,
 					Transport:          tc.ProxyTransport,
 					InterceptRedirects: redirect,
@@ -479,7 +474,7 @@ func TestProxyUpgradeErrorResponse(t *testing.T) {
 			return &fakeConn{err: expectedErr}, nil
 		}
 		responder = &fakeResponder{t: t, w: w}
-		proxyHandler := &UpgradeAwareProxyHandler{
+		proxyHandler := &UpgradeAwareHandler{
 			Location: &url.URL{
 				Host: "fake-backend",
 			},
@@ -545,11 +540,11 @@ func TestDefaultProxyTransport(t *testing.T) {
 	for _, test := range tests {
 		locURL, _ := url.Parse(test.location)
 		URL, _ := url.Parse(test.url)
-		h := UpgradeAwareProxyHandler{
+		h := UpgradeAwareHandler{
 			Location: locURL,
 		}
 		result := h.defaultProxyTransport(URL, nil)
-		transport := result.(*corsRemovingTransport).RoundTripper.(*proxy.Transport)
+		transport := result.(*corsRemovingTransport).RoundTripper.(*Transport)
 		if transport.Scheme != test.expectedScheme {
 			t.Errorf("%s: unexpected scheme. Actual: %s, Expected: %s", test.name, transport.Scheme, test.expectedScheme)
 		}
@@ -721,7 +716,7 @@ func TestProxyRequestContentLengthAndTransferEncoding(t *testing.T) {
 
 		responder := &fakeResponder{t: t}
 		backendURL, _ := url.Parse(downstreamServer.URL)
-		proxyHandler := &UpgradeAwareProxyHandler{
+		proxyHandler := &UpgradeAwareHandler{
 			Location:        backendURL,
 			Responder:       responder,
 			UpgradeRequired: false,
@@ -799,28 +794,3 @@ P7y5NeJnE7X6XkyC35zrsJRkz7orE8MCIHdDjsI8pjyNDeGqwUCDWE/a6DrmIDwe
 emHSqMN2YvChAiEAnxLCM9NWaenOsaIoP+J1rDuvw+4499nJKVqGuVrSCRkCIEqK
 4KSchPMc3x8M/uhw9oWTtKFmjA/PPh0FsWCdKrEy
 -----END RSA PRIVATE KEY-----`)
-
-// localhostCert was generated from crypto/tls/generate_cert.go with the following command:
-//     go run generate_cert.go  --rsa-bits 512 --host 127.0.0.1,::1,example.com --ca --start-date "Jan 1 00:00:00 1970" --duration=1000000h
-var localhostCert = []byte(`-----BEGIN CERTIFICATE-----
-MIIBjzCCATmgAwIBAgIRAKpi2WmTcFrVjxrl5n5YDUEwDQYJKoZIhvcNAQELBQAw
-EjEQMA4GA1UEChMHQWNtZSBDbzAgFw03MDAxMDEwMDAwMDBaGA8yMDg0MDEyOTE2
-MDAwMFowEjEQMA4GA1UEChMHQWNtZSBDbzBcMA0GCSqGSIb3DQEBAQUAA0sAMEgC
-QQC9fEbRszP3t14Gr4oahV7zFObBI4TfA5i7YnlMXeLinb7MnvT4bkfOJzE6zktn
-59zP7UiHs3l4YOuqrjiwM413AgMBAAGjaDBmMA4GA1UdDwEB/wQEAwICpDATBgNV
-HSUEDDAKBggrBgEFBQcDATAPBgNVHRMBAf8EBTADAQH/MC4GA1UdEQQnMCWCC2V4
-YW1wbGUuY29thwR/AAABhxAAAAAAAAAAAAAAAAAAAAABMA0GCSqGSIb3DQEBCwUA
-A0EAUsVE6KMnza/ZbodLlyeMzdo7EM/5nb5ywyOxgIOCf0OOLHsPS9ueGLQX9HEG
-//yjTXuhNcUugExIjM/AIwAZPQ==
------END CERTIFICATE-----`)
-
-// localhostKey is the private key for localhostCert.
-var localhostKey = []byte(`-----BEGIN RSA PRIVATE KEY-----
-MIIBOwIBAAJBAL18RtGzM/e3XgavihqFXvMU5sEjhN8DmLtieUxd4uKdvsye9Phu
-R84nMTrOS2fn3M/tSIezeXhg66quOLAzjXcCAwEAAQJBAKcRxH9wuglYLBdI/0OT
-BLzfWPZCEw1vZmMR2FF1Fm8nkNOVDPleeVGTWoOEcYYlQbpTmkGSxJ6ya+hqRi6x
-goECIQDx3+X49fwpL6B5qpJIJMyZBSCuMhH4B7JevhGGFENi3wIhAMiNJN5Q3UkL
-IuSvv03kaPR5XVQ99/UeEetUgGvBcABpAiBJSBzVITIVCGkGc7d+RCf49KTCIklv
-bGWObufAR8Ni4QIgWpILjW8dkGg8GOUZ0zaNA6Nvt6TIv2UWGJ4v5PoV98kCIQDx
-rIiZs5QbKdycsv9gQJzwQAogC8o04X3Zz3dsoX+h4A==
------END RSA PRIVATE KEY-----`)

staging/src/k8s.io/apiserver/pkg/endpoints/handlers/proxy.go

@@ -33,12 +33,12 @@ import (
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/util/httpstream"
 	"k8s.io/apimachinery/pkg/util/net"
+	proxyutil "k8s.io/apimachinery/pkg/util/proxy"
 	"k8s.io/apiserver/pkg/endpoints/handlers/responsewriters"
 	"k8s.io/apiserver/pkg/endpoints/metrics"
 	"k8s.io/apiserver/pkg/endpoints/request"
 	"k8s.io/apiserver/pkg/registry/rest"
 	"k8s.io/apiserver/pkg/server/httplog"
-	proxyutil "k8s.io/apiserver/pkg/util/proxy"
 
 	"github.com/golang/glog"
 )

staging/src/k8s.io/kube-aggregator/pkg/apiserver/handler_proxy.go

@@ -29,10 +29,10 @@ import (
 	"k8s.io/apimachinery/pkg/util/httpstream"
 	"k8s.io/apimachinery/pkg/util/httpstream/spdy"
 	utilnet "k8s.io/apimachinery/pkg/util/net"
+	"k8s.io/apimachinery/pkg/util/proxy"
 	"k8s.io/apiserver/pkg/endpoints/handlers/responsewriters"
 	genericapirequest "k8s.io/apiserver/pkg/endpoints/request"
 	genericfeatures "k8s.io/apiserver/pkg/features"
-	genericrest "k8s.io/apiserver/pkg/registry/generic/rest"
 	utilfeature "k8s.io/apiserver/pkg/util/feature"
 	restclient "k8s.io/client-go/rest"
 	"k8s.io/client-go/transport"
@@ -146,7 +146,7 @@ func (r *proxyHandler) ServeHTTP(w http.ResponseWriter, req *http.Request) {
 		transport.SetAuthProxyHeaders(newReq, user.GetName(), user.GetGroups(), user.GetExtra())
 	}
 
-	handler := genericrest.NewUpgradeAwareProxyHandler(location, proxyRoundTripper, true, upgrade, &responder{w: w})
+	handler := proxy.NewUpgradeAwareHandler(location, proxyRoundTripper, true, upgrade, &responder{w: w})
 	handler.ServeHTTP(w, newReq)
 }