Connect should be passed a Responder interface

For connect handlers that need to respond with a structured error or
structured object, pass an interface that hides the details of writing
an object to the response (error or runtime.Object).

Example use case:

Connect handler that accepts a body input stream, which it streams to a
pod, and then returns a structured object with info about the pod it
just created.

pkg/api/rest/rest.go

@@ -202,20 +202,24 @@ type Redirector interface {
 	ResourceLocation(ctx api.Context, id string) (remoteLocation *url.URL, transport http.RoundTripper, err error)
 }
 
-// ConnectHandler is a handler for HTTP connection requests. It extends the standard
+// Responder abstracts the normal response behavior for a REST method and is passed to callers that
-// http.Handler interface by adding a method that returns an error object if an error
+// may wish to handle the response directly in some cases, but delegate to the normal error or object
-// occurred during the handling of the request.
+// behavior in other cases.
-type ConnectHandler interface {
+type Responder interface {
-	http.Handler
+	// Object writes the provided object to the response. Invoking this method multiple times is undefined.
-
+	Object(statusCode int, obj runtime.Object)
-	// RequestError returns an error if one occurred during handling of an HTTP request
+	// Error writes the provided error to the response. This method may only be invoked once.
-	RequestError() error
+	Error(err error)
 }
 
-// Connecter is a storage object that responds to a connection request
+// Connecter is a storage object that responds to a connection request.
 type Connecter interface {
-	// Connect returns a ConnectHandler that will handle the request/response for a request
+	// Connect returns an http.Handler that will handle the request/response for a given API invocation.
-	Connect(ctx api.Context, id string, options runtime.Object) (ConnectHandler, error)
+	// The provided responder may be used for common API responses. The responder will write both status
+	// code and body, so the ServeHTTP method should exit after invoking the responder. The Handler will
+	// be used for a single API request and then discarded. The Responder is guaranteed to write to the
+	// same http.ResponseWriter passed to ServeHTTP.
+	Connect(ctx api.Context, id string, options runtime.Object, r Responder) (http.Handler, error)
 
 	// NewConnectOptions returns an empty options object that will be used to pass
 	// options to the Connect method. If nil, then a nil options object is passed to

pkg/apiserver/apiserver_test.go

@@ -338,19 +338,14 @@ func (s *SimpleStream) InputStream(version, accept string) (io.ReadCloser, bool,
 	return s, false, s.contentType, s.err
 }
 
-type SimpleConnectHandler struct {
+type OutputConnect struct {
 	response string
-	err      error
 }
 
-func (h *SimpleConnectHandler) ServeHTTP(w http.ResponseWriter, req *http.Request) {
+func (h *OutputConnect) ServeHTTP(w http.ResponseWriter, req *http.Request) {
 	w.Write([]byte(h.response))
 }
 
-func (h *SimpleConnectHandler) RequestError() error {
-	return h.err
-}
-
 func (storage *SimpleRESTStorage) Get(ctx api.Context, id string) (runtime.Object, error) {
 	storage.checkContext(ctx)
 	if id == "binary" {
@@ -448,10 +443,13 @@ func (storage *SimpleRESTStorage) ResourceLocation(ctx api.Context, id string) (
 
 // Implement Connecter
 type ConnecterRESTStorage struct {
-	connectHandler         rest.ConnectHandler
+	connectHandler http.Handler
+	handlerFunc    func() http.Handler
+
 	emptyConnectOptions    runtime.Object
 	receivedConnectOptions runtime.Object
 	receivedID             string
+	receivedResponder      rest.Responder
 	takesPath              string
 }
 
@@ -462,9 +460,13 @@ func (s *ConnecterRESTStorage) New() runtime.Object {
 	return &apiservertesting.Simple{}
 }
 
-func (s *ConnecterRESTStorage) Connect(ctx api.Context, id string, options runtime.Object) (rest.ConnectHandler, error) {
+func (s *ConnecterRESTStorage) Connect(ctx api.Context, id string, options runtime.Object, responder rest.Responder) (http.Handler, error) {
 	s.receivedConnectOptions = options
 	s.receivedID = id
+	s.receivedResponder = responder
+	if s.handlerFunc != nil {
+		return s.handlerFunc(), nil
+	}
 	return s.connectHandler, nil
 }
 
@@ -1287,7 +1289,7 @@ func TestConnect(t *testing.T) {
 	responseText := "Hello World"
 	itemID := "theID"
 	connectStorage := &ConnecterRESTStorage{
-		connectHandler: &SimpleConnectHandler{
+		connectHandler: &OutputConnect{
 			response: responseText,
 		},
 	}
@@ -1320,9 +1322,92 @@ func TestConnect(t *testing.T) {
 	}
 }
 
+func TestConnectResponderObject(t *testing.T) {
+	itemID := "theID"
+	simple := &apiservertesting.Simple{Other: "foo"}
+	connectStorage := &ConnecterRESTStorage{}
+	connectStorage.handlerFunc = func() http.Handler {
+		return http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
+			connectStorage.receivedResponder.Object(http.StatusCreated, simple)
+		})
+	}
+	storage := map[string]rest.Storage{
+		"simple":         &SimpleRESTStorage{},
+		"simple/connect": connectStorage,
+	}
+	handler := handle(storage)
+	server := httptest.NewServer(handler)
+	defer server.Close()
+
+	resp, err := http.Get(server.URL + "/api/version/namespaces/default/simple/" + itemID + "/connect")
+
+	if err != nil {
+		t.Errorf("unexpected error: %v", err)
+	}
+	if resp.StatusCode != http.StatusCreated {
+		t.Errorf("unexpected response: %#v", resp)
+	}
+	defer resp.Body.Close()
+	body, err := ioutil.ReadAll(resp.Body)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	if connectStorage.receivedID != itemID {
+		t.Errorf("Unexpected item id. Expected: %s. Actual: %s.", itemID, connectStorage.receivedID)
+	}
+	obj, err := codec.Decode(body)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !api.Semantic.DeepEqual(obj, simple) {
+		t.Errorf("Unexpected response: %#v", obj)
+	}
+}
+
+func TestConnectResponderError(t *testing.T) {
+	itemID := "theID"
+	connectStorage := &ConnecterRESTStorage{}
+	connectStorage.handlerFunc = func() http.Handler {
+		return http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
+			connectStorage.receivedResponder.Error(apierrs.NewForbidden("simple", itemID, errors.New("you are terminated")))
+		})
+	}
+	storage := map[string]rest.Storage{
+		"simple":         &SimpleRESTStorage{},
+		"simple/connect": connectStorage,
+	}
+	handler := handle(storage)
+	server := httptest.NewServer(handler)
+	defer server.Close()
+
+	resp, err := http.Get(server.URL + "/api/version/namespaces/default/simple/" + itemID + "/connect")
+
+	if err != nil {
+		t.Errorf("unexpected error: %v", err)
+	}
+	if resp.StatusCode != http.StatusForbidden {
+		t.Errorf("unexpected response: %#v", resp)
+	}
+	defer resp.Body.Close()
+	body, err := ioutil.ReadAll(resp.Body)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	if connectStorage.receivedID != itemID {
+		t.Errorf("Unexpected item id. Expected: %s. Actual: %s.", itemID, connectStorage.receivedID)
+	}
+	obj, err := codec.Decode(body)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if obj.(*unversioned.Status).Code != http.StatusForbidden {
+		t.Errorf("Unexpected response: %#v", obj)
+	}
+}
+
 func TestConnectWithOptionsRouteParams(t *testing.T) {
 	connectStorage := &ConnecterRESTStorage{
-		connectHandler:      &SimpleConnectHandler{},
+		connectHandler:      &OutputConnect{},
 		emptyConnectOptions: &apiservertesting.SimpleGetOptions{},
 	}
 	storage := map[string]rest.Storage{
@@ -1351,7 +1436,7 @@ func TestConnectWithOptions(t *testing.T) {
 	responseText := "Hello World"
 	itemID := "theID"
 	connectStorage := &ConnecterRESTStorage{
-		connectHandler: &SimpleConnectHandler{
+		connectHandler: &OutputConnect{
 			response: responseText,
 		},
 		emptyConnectOptions: &apiservertesting.SimpleGetOptions{},
@@ -1383,6 +1468,9 @@ func TestConnectWithOptions(t *testing.T) {
 	if string(body) != responseText {
 		t.Errorf("Unexpected response. Expected: %s. Actual: %s.", responseText, string(body))
 	}
+	if connectStorage.receivedResponder == nil {
+		t.Errorf("Unexpected responder")
+	}
 	opts, ok := connectStorage.receivedConnectOptions.(*apiservertesting.SimpleGetOptions)
 	if !ok {
 		t.Errorf("Unexpected options type: %#v", connectStorage.receivedConnectOptions)
@@ -1397,7 +1485,7 @@ func TestConnectWithOptionsAndPath(t *testing.T) {
 	itemID := "theID"
 	testPath := "a/b/c/def"
 	connectStorage := &ConnecterRESTStorage{
-		connectHandler: &SimpleConnectHandler{
+		connectHandler: &OutputConnect{
 			response: responseText,
 		},
 		emptyConnectOptions: &apiservertesting.SimpleGetOptions{},

pkg/apiserver/resthandler.go

@@ -179,20 +179,30 @@ func ConnectResource(connecter rest.Connecter, scope RequestScope, admit admissi
 				return
 			}
 		}
-		handler, err := connecter.Connect(ctx, name, opts)
+		handler, err := connecter.Connect(ctx, name, opts, &responder{scope: scope, req: req.Request, w: w})
 		if err != nil {
 			errorJSON(err, scope.Codec, w)
 			return
 		}
 		handler.ServeHTTP(w, req.Request)
-		err = handler.RequestError()
-		if err != nil {
-			errorJSON(err, scope.Codec, w)
-			return
-		}
 	}
 }
 
+// responder implements rest.Responder for assisting a connector in writing objects or errors.
+type responder struct {
+	scope RequestScope
+	req   *http.Request
+	w     http.ResponseWriter
+}
+
+func (r *responder) Object(statusCode int, obj runtime.Object) {
+	write(statusCode, r.scope.APIVersion, r.scope.Codec, obj, r.w, r.req)
+}
+
+func (r *responder) Error(err error) {
+	errorJSON(err, r.scope.Codec, r.w)
+}
+
 // ListResource returns a function that handles retrieving a list of resources from a rest.Storage object.
 func ListResource(r rest.Lister, rw rest.Watcher, scope RequestScope, forceWatch bool, minRequestTimeout time.Duration) restful.RouteFunction {
 	return func(req *restful.Request, res *restful.Response) {

pkg/registry/generic/rest/proxy.go

@@ -44,30 +44,32 @@ type UpgradeAwareProxyHandler struct {
 	WrapTransport  bool
 	FlushInterval  time.Duration
 	MaxBytesPerSec int64
-	err            error
+	Responder      ErrorResponder
 }
 
 const defaultFlushInterval = 200 * time.Millisecond
 
-// NewUpgradeAwareProxyHandler creates a new proxy handler with a default flush interval
+// ErrorResponder abstracts error reporting to the proxy handler to remove the need to hardcode a particular
-func NewUpgradeAwareProxyHandler(location *url.URL, transport http.RoundTripper, wrapTransport, upgradeRequired bool) *UpgradeAwareProxyHandler {
+// error format.
+type ErrorResponder interface {
+	Error(err error)
+}
+
+// NewUpgradeAwareProxyHandler 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{
 		Location:        location,
 		Transport:       transport,
 		WrapTransport:   wrapTransport,
 		UpgradeRequired: upgradeRequired,
 		FlushInterval:   defaultFlushInterval,
+		Responder:       responder,
 	}
 }
 
-// RequestError returns an error that occurred while handling request
-func (h *UpgradeAwareProxyHandler) RequestError() error {
-	return h.err
-}
-
 // ServeHTTP handles the proxy request
 func (h *UpgradeAwareProxyHandler) ServeHTTP(w http.ResponseWriter, req *http.Request) {
-	h.err = nil
 	if len(h.Location.Scheme) == 0 {
 		h.Location.Scheme = "http"
 	}
@@ -75,7 +77,7 @@ func (h *UpgradeAwareProxyHandler) ServeHTTP(w http.ResponseWriter, req *http.Re
 		return
 	}
 	if h.UpgradeRequired {
-		h.err = errors.NewBadRequest("Upgrade request required")
+		h.Responder.Error(errors.NewBadRequest("Upgrade request required"))
 		return
 	}
 
@@ -108,7 +110,7 @@ func (h *UpgradeAwareProxyHandler) ServeHTTP(w http.ResponseWriter, req *http.Re
 
 	newReq, err := http.NewRequest(req.Method, loc.String(), req.Body)
 	if err != nil {
-		h.err = err
+		h.Responder.Error(err)
 		return
 	}
 	newReq.Header = req.Header
@@ -127,27 +129,27 @@ func (h *UpgradeAwareProxyHandler) tryUpgrade(w http.ResponseWriter, req *http.R
 
 	backendConn, err := proxy.DialURL(h.Location, h.Transport)
 	if err != nil {
-		h.err = err
+		h.Responder.Error(err)
 		return true
 	}
 	defer backendConn.Close()
 
 	requestHijackedConn, _, err := w.(http.Hijacker).Hijack()
 	if err != nil {
-		h.err = err
+		h.Responder.Error(err)
 		return true
 	}
 	defer requestHijackedConn.Close()
 
 	newReq, err := http.NewRequest(req.Method, h.Location.String(), req.Body)
 	if err != nil {
-		h.err = err
+		h.Responder.Error(err)
 		return true
 	}
 	newReq.Header = req.Header
 
 	if err = newReq.Write(backendConn); err != nil {
-		h.err = err
+		h.Responder.Error(err)
 		return true
 	}
 

pkg/registry/generic/rest/proxy_test.go

@@ -26,6 +26,7 @@ import (
 	"net/http"
 	"net/http/httptest"
 	"net/url"
+	"strings"
 	"testing"
 
 	"golang.org/x/net/websocket"
@@ -33,6 +34,19 @@ import (
 	"k8s.io/kubernetes/pkg/util/proxy"
 )
 
+type fakeResponder struct {
+	called bool
+	err    error
+}
+
+func (r *fakeResponder) Error(err error) {
+	if r.called {
+		panic("called twice")
+	}
+	r.called = true
+	r.err = err
+}
+
 type SimpleBackendHandler struct {
 	requestURL     url.URL
 	requestHeader  http.Header
@@ -110,6 +124,8 @@ func TestServeHTTP(t *testing.T) {
 		responseHeader        map[string]string
 		expectedRespHeader    map[string]string
 		notExpectedRespHeader []string
+		upgradeRequired       bool
+		expectError           func(err error) bool
 	}{
 		{
 			name:         "root path, simple get",
@@ -117,6 +133,15 @@ func TestServeHTTP(t *testing.T) {
 			requestPath:  "/",
 			expectedPath: "/",
 		},
+		{
+			name:            "no upgrade header sent",
+			method:          "GET",
+			requestPath:     "/",
+			upgradeRequired: true,
+			expectError: func(err error) bool {
+				return err != nil && strings.Contains(err.Error(), "Upgrade request required")
+			},
+		},
 		{
 			name:         "simple path, get",
 			method:       "GET",
@@ -163,7 +188,7 @@ func TestServeHTTP(t *testing.T) {
 		},
 	}
 
-	for _, test := range tests {
+	for i, test := range tests {
 		func() {
 			backendResponse := "<html><head></head><body><a href=\"/test/path\">Hello</a></body></html>"
 			backendResponseHeader := test.responseHeader
@@ -179,10 +204,13 @@ func TestServeHTTP(t *testing.T) {
 			backendServer := httptest.NewServer(backendHandler)
 			defer backendServer.Close()
 
+			responder := &fakeResponder{}
 			backendURL, _ := url.Parse(backendServer.URL)
 			backendURL.Path = test.requestPath
 			proxyHandler := &UpgradeAwareProxyHandler{
-				Location: backendURL,
+				Location:        backendURL,
+				Responder:       responder,
+				UpgradeRequired: test.upgradeRequired,
 			}
 			proxyServer := httptest.NewServer(proxyHandler)
 			defer proxyServer.Close()
@@ -214,6 +242,17 @@ func TestServeHTTP(t *testing.T) {
 				t.Errorf("Error from proxy request: %v", err)
 			}
 
+			if test.expectError != nil {
+				if !responder.called {
+					t.Errorf("%d: responder was not invoked", i)
+					return
+				}
+				if !test.expectError(responder.err) {
+					t.Errorf("%d: unexpected error: %v", i, responder.err)
+				}
+				return
+			}
+
 			// Validate backend request
 			// Method
 			if backendHandler.requestMethod != test.method {
@@ -253,9 +292,8 @@ func TestServeHTTP(t *testing.T) {
 			}
 
 			// Error
-			err = proxyHandler.RequestError()
+			if responder.called {
-			if err != nil {
+				t.Errorf("Unexpected proxy handler error: %v", responder.err)
-				t.Errorf("Unexpected proxy handler error: %v", err)
 			}
 		}()
 	}

pkg/registry/pod/etcd/etcd.go

@@ -283,7 +283,7 @@ func (r *ProxyREST) NewConnectOptions() (runtime.Object, bool, string) {
 }
 
 // Connect returns a handler for the pod proxy
-func (r *ProxyREST) Connect(ctx api.Context, id string, opts runtime.Object) (rest.ConnectHandler, error) {
+func (r *ProxyREST) Connect(ctx api.Context, id string, opts runtime.Object, responder rest.Responder) (http.Handler, error) {
 	proxyOpts, ok := opts.(*api.PodProxyOptions)
 	if !ok {
 		return nil, fmt.Errorf("Invalid options object: %#v", opts)
@@ -294,7 +294,7 @@ func (r *ProxyREST) Connect(ctx api.Context, id string, opts runtime.Object) (re
 	}
 	location.Path = path.Join(location.Path, proxyOpts.Path)
 	// Return a proxy handler that uses the desired transport, wrapped with additional proxy handling (to get URL rewriting, X-Forwarded-* headers, etc)
-	return newThrottledUpgradeAwareProxyHandler(location, transport, true, false), nil
+	return newThrottledUpgradeAwareProxyHandler(location, transport, true, false, responder), nil
 }
 
 // Support both GET and POST methods. We must support GET for browsers that want to use WebSockets.
@@ -316,7 +316,7 @@ func (r *AttachREST) New() runtime.Object {
 }
 
 // Connect returns a handler for the pod exec proxy
-func (r *AttachREST) Connect(ctx api.Context, name string, opts runtime.Object) (rest.ConnectHandler, error) {
+func (r *AttachREST) Connect(ctx api.Context, name string, opts runtime.Object, responder rest.Responder) (http.Handler, error) {
 	attachOpts, ok := opts.(*api.PodAttachOptions)
 	if !ok {
 		return nil, fmt.Errorf("Invalid options object: %#v", opts)
@@ -325,7 +325,7 @@ func (r *AttachREST) Connect(ctx api.Context, name string, opts runtime.Object)
 	if err != nil {
 		return nil, err
 	}
-	return newThrottledUpgradeAwareProxyHandler(location, transport, false, true), nil
+	return newThrottledUpgradeAwareProxyHandler(location, transport, false, true, responder), nil
 }
 
 // NewConnectOptions returns the versioned object that represents exec parameters
@@ -354,16 +354,16 @@ func (r *ExecREST) New() runtime.Object {
 }
 
 // Connect returns a handler for the pod exec proxy
-func (r *ExecREST) Connect(ctx api.Context, name string, opts runtime.Object) (rest.ConnectHandler, error) {
+func (r *ExecREST) Connect(ctx api.Context, name string, opts runtime.Object, responder rest.Responder) (http.Handler, error) {
 	execOpts, ok := opts.(*api.PodExecOptions)
 	if !ok {
-		return nil, fmt.Errorf("Invalid options object: %#v", opts)
+		return nil, fmt.Errorf("invalid options object: %#v", opts)
 	}
 	location, transport, err := pod.ExecLocation(r.store, r.kubeletConn, ctx, name, execOpts)
 	if err != nil {
 		return nil, err
 	}
-	return newThrottledUpgradeAwareProxyHandler(location, transport, false, true), nil
+	return newThrottledUpgradeAwareProxyHandler(location, transport, false, true, responder), nil
 }
 
 // NewConnectOptions returns the versioned object that represents exec parameters
@@ -402,16 +402,16 @@ func (r *PortForwardREST) ConnectMethods() []string {
 }
 
 // Connect returns a handler for the pod portforward proxy
-func (r *PortForwardREST) Connect(ctx api.Context, name string, opts runtime.Object) (rest.ConnectHandler, error) {
+func (r *PortForwardREST) Connect(ctx api.Context, name string, opts runtime.Object, responder rest.Responder) (http.Handler, error) {
 	location, transport, err := pod.PortForwardLocation(r.store, r.kubeletConn, ctx, name)
 	if err != nil {
 		return nil, err
 	}
-	return newThrottledUpgradeAwareProxyHandler(location, transport, false, true), nil
+	return newThrottledUpgradeAwareProxyHandler(location, transport, false, true, responder), nil
 }
 
-func newThrottledUpgradeAwareProxyHandler(location *url.URL, transport http.RoundTripper, wrapTransport, upgradeRequired bool) *genericrest.UpgradeAwareProxyHandler {
+func newThrottledUpgradeAwareProxyHandler(location *url.URL, transport http.RoundTripper, wrapTransport, upgradeRequired bool, responder rest.Responder) *genericrest.UpgradeAwareProxyHandler {
-	handler := genericrest.NewUpgradeAwareProxyHandler(location, transport, wrapTransport, upgradeRequired)
+	handler := genericrest.NewUpgradeAwareProxyHandler(location, transport, wrapTransport, upgradeRequired, responder)
 	handler.MaxBytesPerSec = capabilities.Get().PerConnectionBandwidthLimitBytesPerSec
 	return handler
 }