github
/
k3s
mirror of https://github.com/k3s-io/k3s
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

Merge pull request #15442 from smarterclayton/connect_responder 

Connect should be passed a Responder interface
pull/6/head
Dawn Chen 2015-10-13 16:51:47 -07:00
parent a529f6ccdf d4cdabf2fc
commit 61a175f0bc
6 changed files with 203 additions and 61 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

26
pkg/api/rest/rest.go
View File

@ -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
// http.Handler interface by adding a method that returns an error object if an error
// occurred during the handling of the request.
type ConnectHandler interface {
http.Handler
// RequestError returns an error if one occurred during handling of an HTTP request
RequestError() error
// Responder abstracts the normal response behavior for a REST method and is passed to callers that
// may wish to handle the response directly in some cases, but delegate to the normal error or object
// behavior in other cases.
type Responder interface {
// Object writes the provided object to the response. Invoking this method multiple times is undefined.
Object(statusCode int, obj runtime.Object)
// Error writes the provided error to the response. This method may only be invoked once.
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(ctx api.Context, id string, options runtime.Object) (ConnectHandler, error)
// Connect returns an http.Handler that will handle the request/response for a given API invocation.
// 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

114
pkg/apiserver/apiserver_test.go
View File

@ -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{},

22
pkg/apiserver/resthandler.go
View File

@ -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) {

32
pkg/registry/generic/rest/proxy.go
View File

@ -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
func NewUpgradeAwareProxyHandler(location *url.URL, transport http.RoundTripper, wrapTransport, upgradeRequired bool) *UpgradeAwareProxyHandler {
// ErrorResponder abstracts error reporting to the proxy handler to remove the need to hardcode a particular
// 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
}

48
pkg/registry/generic/rest/proxy_test.go
View File

@ -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 err != nil {
t.Errorf("Unexpected proxy handler error: %v", err)
if responder.called {
t.Errorf("Unexpected proxy handler error: %v", responder.err)
}
}()
}

22
pkg/registry/pod/etcd/etcd.go
View File

@ -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 {
handler := genericrest.NewUpgradeAwareProxyHandler(location, transport, wrapTransport, upgradeRequired)
func newThrottledUpgradeAwareProxyHandler(location *url.URL, transport http.RoundTripper, wrapTransport, upgradeRequired bool, responder rest.Responder) *genericrest.UpgradeAwareProxyHandler {
handler := genericrest.NewUpgradeAwareProxyHandler(location, transport, wrapTransport, upgradeRequired, responder)
handler.MaxBytesPerSec = capabilities.Get().PerConnectionBandwidthLimitBytesPerSec
return handler
}