Merge pull request #1893 from crhym3/fix-1149-req-body

Replace custom ProxyServer (kubecfg/kubectl -proxy) with httputil.ReverseProxy

cmd/kubecfg/kubecfg.go

@@ -263,7 +263,10 @@ func main() {
 				open.Start("http://localhost:8001/static/")
 			}()
 		}
-		server := kubecfg.NewProxyServer(*www, kubeClient)
+		server, err := kubecfg.NewProxyServer(*www, clientConfig)
+		if err != nil {
+			glog.Fatalf("Error creating proxy server: %v", err)
+		}
 		glog.Fatal(server.Serve())
 	}
 

pkg/client/request_test.go

@@ -271,6 +271,38 @@ func TestUnacceptableParamNames(t *testing.T) {
 	}
 }
 
+func TestBody(t *testing.T) {
+	const data = "test payload"
+
+	f, err := ioutil.TempFile("", "test_body")
+	if err != nil {
+		t.Fatalf("TempFile error: %v", err)
+	}
+	if _, err := f.WriteString(data); err != nil {
+		t.Fatalf("TempFile.WriteString error: %v", err)
+	}
+	f.Close()
+
+	c := NewOrDie(&Config{})
+	tests := []interface{}{[]byte(data), f.Name(), strings.NewReader(data)}
+	for i, tt := range tests {
+		r := c.Post().Body(tt)
+		if r.err != nil {
+			t.Errorf("%d: r.Body(%#v) error: %v", i, tt, r.err)
+			continue
+		}
+		buf := make([]byte, len(data))
+		if _, err := r.body.Read(buf); err != nil {
+			t.Errorf("%d: r.body.Read error: %v", i, err)
+			continue
+		}
+		body := string(buf)
+		if body != data {
+			t.Errorf("%d: r.body = %q; want %q", i, body, data)
+		}
+	}
+}
+
 func TestSetPollPeriod(t *testing.T) {
 	c := NewOrDie(&Config{})
 	r := c.Get()

pkg/kubecfg/proxy_server.go

@@ -17,32 +17,37 @@ limitations under the License.
 package kubecfg
 
 import (
-	"fmt"
 	"net/http"
+	"net/http/httputil"
+	"net/url"
+	"strings"
 
-	"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
-	"github.com/GoogleCloudPlatform/kubernetes/pkg/api/latest"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/client"
 )
 
 // ProxyServer is a http.Handler which proxies Kubernetes APIs to remote API server.
 type ProxyServer struct {
-	Client *client.Client
-}
-
-func newFileHandler(prefix, base string) http.Handler {
-	return http.StripPrefix(prefix, http.FileServer(http.Dir(base)))
+	httputil.ReverseProxy
 }
 
 // NewProxyServer creates and installs a new ProxyServer.
 // It automatically registers the created ProxyServer to http.DefaultServeMux.
-func NewProxyServer(filebase string, kubeClient *client.Client) *ProxyServer {
-	server := &ProxyServer{
-		Client: kubeClient,
+func NewProxyServer(filebase string, cfg *client.Config) (*ProxyServer, error) {
+	prefix := cfg.Prefix
+	if prefix == "" {
+		prefix = "/api"
 	}
-	http.Handle("/api/", server)
+	target, err := url.Parse(singleJoiningSlash(cfg.Host, prefix))
+	if err != nil {
+		return nil, err
+	}
+	proxy := newProxyServer(target)
+	if proxy.Transport, err = client.TransportFor(cfg); err != nil {
+		return nil, err
+	}
+	http.Handle("/api/", http.StripPrefix("/api/", proxy))
 	http.Handle("/static/", newFileHandler("/static/", filebase))
-	return server
+	return proxy, nil
 }
 
 // Serve starts the server (http.DefaultServeMux) on TCP port 8001, loops forever.
@@ -50,37 +55,27 @@ func (s *ProxyServer) Serve() error {
 	return http.ListenAndServe(":8001", nil)
 }
 
-func (s *ProxyServer) doError(w http.ResponseWriter, err error) {
-	w.WriteHeader(http.StatusInternalServerError)
-	w.Header().Add("Content-type", "application/json")
-	data, _ := latest.Codec.Encode(&api.Status{
-		Status:  api.StatusFailure,
-		Message: fmt.Sprintf("internal error: %#v", err),
-	})
-	w.Write(data)
+func newProxyServer(target *url.URL) *ProxyServer {
+	director := func(req *http.Request) {
+		req.URL.Scheme = target.Scheme
+		req.URL.Host = target.Host
+		req.URL.Path = singleJoiningSlash(target.Path, req.URL.Path)
+	}
+	return &ProxyServer{ReverseProxy: httputil.ReverseProxy{Director: director}}
 }
 
-func (s *ProxyServer) ServeHTTP(w http.ResponseWriter, r *http.Request) {
-	url := r.URL
-	selector := url.Query().Get("labels")
-	fieldSelector := url.Query().Get("fields")
-	result := s.Client.
-		Verb(r.Method).
-		AbsPath(r.URL.Path).
-		ParseSelectorParam("labels", selector).
-		ParseSelectorParam("fields", fieldSelector).
-		Body(r.Body).
-		Do()
-	if result.Error() != nil {
-		s.doError(w, result.Error())
-		return
+func newFileHandler(prefix, base string) http.Handler {
+	return http.StripPrefix(prefix, http.FileServer(http.Dir(base)))
 }
-	data, err := result.Raw()
-	if err != nil {
-		s.doError(w, err)
-		return
+
+func singleJoiningSlash(a, b string) string {
+	aslash := strings.HasSuffix(a, "/")
+	bslash := strings.HasPrefix(b, "/")
+	switch {
+	case aslash && bslash:
+		return a + b[1:]
+	case !aslash && !bslash:
+		return a + "/" + b
 	}
-	w.Header().Add("Content-type", "application/json")
-	w.WriteHeader(http.StatusOK)
-	w.Write(data)
+	return a + b
 }

pkg/kubecfg/proxy_server_test.go

@@ -17,43 +17,90 @@ limitations under the License.
 package kubecfg
 
 import (
+	"fmt"
 	"io/ioutil"
 	"net/http"
 	"net/http/httptest"
+	"net/url"
+	"path/filepath"
+	"strings"
 	"testing"
 )
 
 func TestFileServing(t *testing.T) {
-	data := "This is test data"
+	const (
+		fname = "test.txt"
+		data  = "This is test data"
+	)
 	dir, err := ioutil.TempDir("", "data")
 	if err != nil {
-		t.Errorf("Unexpected error: %v", err)
+		t.Fatalf("error creating tmp dir: %v", err)
 	}
-	err = ioutil.WriteFile(dir+"/test.txt", []byte(data), 0755)
-	if err != nil {
-		t.Errorf("Unexpected error: %v", err)
+	if err := ioutil.WriteFile(filepath.Join(dir, fname), []byte(data), 0755); err != nil {
+		t.Fatalf("error writing tmp file: %v", err)
 	}
-	prefix := "/foo/"
+
+	const prefix = "/foo/"
 	handler := newFileHandler(prefix, dir)
 	server := httptest.NewServer(handler)
-	client := http.Client{}
-	req, err := http.NewRequest("GET", server.URL+prefix+"test.txt", nil)
+	defer server.Close()
+
+	url := server.URL + prefix + fname
+	res, err := http.Get(url)
 	if err != nil {
-		t.Errorf("Unexpected error: %v", err)
-	}
-	res, err := client.Do(req)
-	if err != nil {
-		t.Errorf("Unexpected error: %v", err)
+		t.Fatalf("http.Get(%q) error: %v", url, err)
 	}
 	defer res.Body.Close()
+
+	if res.StatusCode != http.StatusOK {
+		t.Errorf("res.StatusCode = %d; want %d", res.StatusCode, http.StatusOK)
+	}
 	b, err := ioutil.ReadAll(res.Body)
 	if err != nil {
-		t.Errorf("Unexpected error: %v", err)
-	}
-	if res.StatusCode != http.StatusOK {
-		t.Errorf("Unexpected status: %d", res.StatusCode)
+		t.Fatalf("error reading resp body: %v", err)
 	}
 	if string(b) != data {
-		t.Errorf("Data doesn't match: %s vs %s", string(b), data)
+		t.Errorf("have %q; want %q", string(b), data)
+	}
+}
+
+func TestAPIRequests(t *testing.T) {
+	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		b, err := ioutil.ReadAll(r.Body)
+		if err != nil {
+			http.Error(w, err.Error(), http.StatusInternalServerError)
+			return
+		}
+		fmt.Fprintf(w, "%s %s %s", r.Method, r.RequestURI, string(b))
+	}))
+	defer ts.Close()
+
+	// httptest.NewServer should always generate a valid URL.
+	target, _ := url.Parse(ts.URL)
+	proxy := newProxyServer(target)
+
+	tests := []struct{ method, body string }{
+		{"GET", ""},
+		{"DELETE", ""},
+		{"POST", "test payload"},
+		{"PUT", "test payload"},
+	}
+
+	const path = "/api/test?fields=ID%3Dfoo&labels=key%3Dvalue"
+	for i, tt := range tests {
+		r, err := http.NewRequest(tt.method, path, strings.NewReader(tt.body))
+		if err != nil {
+			t.Errorf("error creating request: %v", err)
+			continue
+		}
+		w := httptest.NewRecorder()
+		proxy.ServeHTTP(w, r)
+		if w.Code != http.StatusOK {
+			t.Errorf("%d: proxy.ServeHTTP w.Code = %d; want %d", i, w.Code, http.StatusOK)
+		}
+		want := strings.Join([]string{tt.method, path, tt.body}, " ")
+		if w.Body.String() != want {
+			t.Errorf("%d: response body = %q; want %q", i, w.Body.String(), want)
+		}
 	}
 }

pkg/kubectl/cmd/cmd.go

@@ -138,7 +138,7 @@ func getFlagInt(cmd *cobra.Command, flag string) int {
 	return v
 }
 
-func getKubeClient(cmd *cobra.Command) *client.Client {
+func getKubeConfig(cmd *cobra.Command) *client.Config {
 	config := &client.Config{}
 
 	var host string
@@ -184,6 +184,12 @@ func getKubeClient(cmd *cobra.Command) *client.Client {
 	// The API version (e.g. v1beta1), not the binary version.
 	config.Version = getFlagString(cmd, "api-version")
 
+	return config
+}
+
+func getKubeClient(cmd *cobra.Command) *client.Client {
+	config := getKubeConfig(cmd)
+
 	// The binary version.
 	matchVersion := getFlagBool(cmd, "match-server-version")
 

pkg/kubectl/cmd/proxy.go

@@ -32,7 +32,8 @@ func NewCmdProxy(out io.Writer) *cobra.Command {
 		Run: func(cmd *cobra.Command, args []string) {
 			port := getFlagInt(cmd, "port")
 			glog.Infof("Starting to serve on localhost:%d", port)
-			server := kubectl.NewProxyServer(getFlagString(cmd, "www"), getKubeClient(cmd), port)
+			server, err := kubectl.NewProxyServer(getFlagString(cmd, "www"), getKubeConfig(cmd), port)
+			checkErr(err)
 			glog.Fatal(server.Serve())
 		},
 	}

pkg/kubectl/proxy_server.go

@@ -19,32 +19,37 @@ package kubectl
 import (
 	"fmt"
 	"net/http"
+	"net/http/httputil"
+	"net/url"
+	"strings"
 
-	"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
-	"github.com/GoogleCloudPlatform/kubernetes/pkg/api/latest"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/client"
 )
 
 // ProxyServer is a http.Handler which proxies Kubernetes APIs to remote API server.
 type ProxyServer struct {
-	Client *client.Client
+	httputil.ReverseProxy
 	Port int
 }
 
-func newFileHandler(prefix, base string) http.Handler {
-	return http.StripPrefix(prefix, http.FileServer(http.Dir(base)))
-}
-
 // NewProxyServer creates and installs a new ProxyServer.
 // It automatically registers the created ProxyServer to http.DefaultServeMux.
-func NewProxyServer(filebase string, kubeClient *client.Client, port int) *ProxyServer {
-	server := &ProxyServer{
-		Client: kubeClient,
-		Port:   port,
+func NewProxyServer(filebase string, cfg *client.Config, port int) (*ProxyServer, error) {
+	prefix := cfg.Prefix
+	if prefix == "" {
+		prefix = "/api"
 	}
-	http.Handle("/api/", server)
+	target, err := url.Parse(singleJoiningSlash(cfg.Host, prefix))
+	if err != nil {
+		return nil, err
+	}
+	proxy := newProxyServer(target)
+	if proxy.Transport, err = client.TransportFor(cfg); err != nil {
+		return nil, err
+	}
+	http.Handle("/api/", http.StripPrefix("/api/", proxy))
 	http.Handle("/static/", newFileHandler("/static/", filebase))
-	return server
+	return proxy, nil
 }
 
 // Serve starts the server (http.DefaultServeMux) on TCP port 8001, loops forever.
@@ -53,37 +58,27 @@ func (s *ProxyServer) Serve() error {
 	return http.ListenAndServe(addr, nil)
 }
 
-func (s *ProxyServer) doError(w http.ResponseWriter, err error) {
-	w.WriteHeader(http.StatusInternalServerError)
-	w.Header().Add("Content-type", "application/json")
-	data, _ := latest.Codec.Encode(&api.Status{
-		Status:  api.StatusFailure,
-		Message: fmt.Sprintf("internal error: %#v", err),
-	})
-	w.Write(data)
+func newProxyServer(target *url.URL) *ProxyServer {
+	director := func(req *http.Request) {
+		req.URL.Scheme = target.Scheme
+		req.URL.Host = target.Host
+		req.URL.Path = singleJoiningSlash(target.Path, req.URL.Path)
+	}
+	return &ProxyServer{ReverseProxy: httputil.ReverseProxy{Director: director}}
 }
 
-func (s *ProxyServer) ServeHTTP(w http.ResponseWriter, r *http.Request) {
-	url := r.URL
-	selector := url.Query().Get("labels")
-	fieldSelector := url.Query().Get("fields")
-	result := s.Client.
-		Verb(r.Method).
-		AbsPath(r.URL.Path).
-		ParseSelectorParam("labels", selector).
-		ParseSelectorParam("fields", fieldSelector).
-		Body(r.Body).
-		Do()
-	if result.Error() != nil {
-		s.doError(w, result.Error())
-		return
+func newFileHandler(prefix, base string) http.Handler {
+	return http.StripPrefix(prefix, http.FileServer(http.Dir(base)))
 }
-	data, err := result.Raw()
-	if err != nil {
-		s.doError(w, err)
-		return
+
+func singleJoiningSlash(a, b string) string {
+	aslash := strings.HasSuffix(a, "/")
+	bslash := strings.HasPrefix(b, "/")
+	switch {
+	case aslash && bslash:
+		return a + b[1:]
+	case !aslash && !bslash:
+		return a + "/" + b
 	}
-	w.Header().Add("Content-type", "application/json")
-	w.WriteHeader(http.StatusOK)
-	w.Write(data)
+	return a + b
 }

pkg/kubectl/proxy_server_test.go

@@ -17,43 +17,90 @@ limitations under the License.
 package kubectl
 
 import (
+	"fmt"
 	"io/ioutil"
 	"net/http"
 	"net/http/httptest"
+	"net/url"
+	"path/filepath"
+	"strings"
 	"testing"
 )
 
 func TestFileServing(t *testing.T) {
-	data := "This is test data"
+	const (
+		fname = "test.txt"
+		data  = "This is test data"
+	)
 	dir, err := ioutil.TempDir("", "data")
 	if err != nil {
-		t.Errorf("Unexpected error: %v", err)
+		t.Fatalf("error creating tmp dir: %v", err)
 	}
-	err = ioutil.WriteFile(dir+"/test.txt", []byte(data), 0755)
-	if err != nil {
-		t.Errorf("Unexpected error: %v", err)
+	if err := ioutil.WriteFile(filepath.Join(dir, fname), []byte(data), 0755); err != nil {
+		t.Fatalf("error writing tmp file: %v", err)
 	}
-	prefix := "/foo/"
+
+	const prefix = "/foo/"
 	handler := newFileHandler(prefix, dir)
 	server := httptest.NewServer(handler)
-	client := http.Client{}
-	req, err := http.NewRequest("GET", server.URL+prefix+"test.txt", nil)
+	defer server.Close()
+
+	url := server.URL + prefix + fname
+	res, err := http.Get(url)
 	if err != nil {
-		t.Errorf("Unexpected error: %v", err)
-	}
-	res, err := client.Do(req)
-	if err != nil {
-		t.Errorf("Unexpected error: %v", err)
+		t.Fatalf("http.Get(%q) error: %v", url, err)
 	}
 	defer res.Body.Close()
+
+	if res.StatusCode != http.StatusOK {
+		t.Errorf("res.StatusCode = %d; want %d", res.StatusCode, http.StatusOK)
+	}
 	b, err := ioutil.ReadAll(res.Body)
 	if err != nil {
-		t.Errorf("Unexpected error: %v", err)
-	}
-	if res.StatusCode != http.StatusOK {
-		t.Errorf("Unexpected status: %d", res.StatusCode)
+		t.Fatalf("error reading resp body: %v", err)
 	}
 	if string(b) != data {
-		t.Errorf("Data doesn't match: %s vs %s", string(b), data)
+		t.Errorf("have %q; want %q", string(b), data)
+	}
+}
+
+func TestAPIRequests(t *testing.T) {
+	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		b, err := ioutil.ReadAll(r.Body)
+		if err != nil {
+			http.Error(w, err.Error(), http.StatusInternalServerError)
+			return
+		}
+		fmt.Fprintf(w, "%s %s %s", r.Method, r.RequestURI, string(b))
+	}))
+	defer ts.Close()
+
+	// httptest.NewServer should always generate a valid URL.
+	target, _ := url.Parse(ts.URL)
+	proxy := newProxyServer(target)
+
+	tests := []struct{ method, body string }{
+		{"GET", ""},
+		{"DELETE", ""},
+		{"POST", "test payload"},
+		{"PUT", "test payload"},
+	}
+
+	const path = "/api/test?fields=ID%3Dfoo&labels=key%3Dvalue"
+	for i, tt := range tests {
+		r, err := http.NewRequest(tt.method, path, strings.NewReader(tt.body))
+		if err != nil {
+			t.Errorf("error creating request: %v", err)
+			continue
+		}
+		w := httptest.NewRecorder()
+		proxy.ServeHTTP(w, r)
+		if w.Code != http.StatusOK {
+			t.Errorf("%d: proxy.ServeHTTP w.Code = %d; want %d", i, w.Code, http.StatusOK)
+		}
+		want := strings.Join([]string{tt.method, path, tt.body}, " ")
+		if w.Body.String() != want {
+			t.Errorf("%d: response body = %q; want %q", i, w.Body.String(), want)
+		}
 	}
 }