From 75b93cf7e9723561dcf958422c9faae9907a7071 Mon Sep 17 00:00:00 2001 From: Daniel Smith Date: Thu, 25 Sep 2014 14:57:41 -0700 Subject: [PATCH 1/3] Add SelfLinker --- pkg/runtime/interfaces.go | 9 ++++++ pkg/runtime/jsonbase.go | 55 ++++++++++++++++++++++++++++++++---- pkg/runtime/jsonbase_test.go | 53 ++++++++++++++++++++++++++++++++++ 3 files changed, 112 insertions(+), 5 deletions(-) diff --git a/pkg/runtime/interfaces.go b/pkg/runtime/interfaces.go index 9abf7f8a6ab..517802e7699 100644 --- a/pkg/runtime/interfaces.go +++ b/pkg/runtime/interfaces.go @@ -40,6 +40,15 @@ type ResourceVersioner interface { ResourceVersion(obj Object) (uint64, error) } +// SelfLinker provides methods for setting and retrieving the SelfLink field of an API object. +type SelfLinker interface { + SetSelfLink(obj Object, selfLink string) error + SelfLink(obj Object) (string, error) + + // Knowing ID is sometimes necssary to use a SelfLinker. + ID(obj Object) (string, error) +} + // All api types must support the Object interface. It's deliberately tiny so that this is not an onerous // burden. Implement it with a pointer reciever; this will allow us to use the go compiler to check the // one thing about our objects that it's capable of checking for us. diff --git a/pkg/runtime/jsonbase.go b/pkg/runtime/jsonbase.go index f6f7c77fe75..4bd78a82882 100644 --- a/pkg/runtime/jsonbase.go +++ b/pkg/runtime/jsonbase.go @@ -21,15 +21,16 @@ import ( "reflect" ) -// NewJSONBaseResourceVersioner returns a resourceVersioner that can set or +// NewJSONBaseResourceVersioner returns a ResourceVersioner that can set or // retrieve ResourceVersion on objects derived from JSONBase. func NewJSONBaseResourceVersioner() ResourceVersioner { - return &jsonBaseResourceVersioner{} + return jsonBaseModifier{} } -type jsonBaseResourceVersioner struct{} +// jsonBaseModifier implements ResourceVersioner and SelfLinker. +type jsonBaseModifier struct{} -func (v jsonBaseResourceVersioner) ResourceVersion(obj Object) (uint64, error) { +func (v jsonBaseModifier) ResourceVersion(obj Object) (uint64, error) { json, err := FindJSONBase(obj) if err != nil { return 0, err @@ -37,7 +38,7 @@ func (v jsonBaseResourceVersioner) ResourceVersion(obj Object) (uint64, error) { return json.ResourceVersion(), nil } -func (v jsonBaseResourceVersioner) SetResourceVersion(obj Object, version uint64) error { +func (v jsonBaseModifier) SetResourceVersion(obj Object, version uint64) error { json, err := FindJSONBase(obj) if err != nil { return err @@ -46,6 +47,36 @@ func (v jsonBaseResourceVersioner) SetResourceVersion(obj Object, version uint64 return nil } +func (v jsonBaseModifier) ID(obj Object) (string, error) { + json, err := FindJSONBase(obj) + if err != nil { + return "", err + } + return json.ID(), nil +} + +func (v jsonBaseModifier) SelfLink(obj Object) (string, error) { + json, err := FindJSONBase(obj) + if err != nil { + return "", err + } + return json.SelfLink(), nil +} + +func (v jsonBaseModifier) SetSelfLink(obj Object, selfLink string) error { + json, err := FindJSONBase(obj) + if err != nil { + return err + } + json.SetSelfLink(selfLink) + return nil +} + +// NewJSONBaseSelfLinker returns a SelfLinker that works on all JSONBase SelfLink fields. +func NewJSONBaseSelfLinker() SelfLinker { + return jsonBaseModifier{} +} + // JSONBaseInterface lets you work with a JSONBase from any of the versioned or // internal APIObjects. type JSONBaseInterface interface { @@ -57,6 +88,8 @@ type JSONBaseInterface interface { SetKind(kind string) ResourceVersion() uint64 SetResourceVersion(version uint64) + SelfLink() string + SetSelfLink(selfLink string) } type genericJSONBase struct { @@ -64,6 +97,7 @@ type genericJSONBase struct { apiVersion *string kind *string resourceVersion *uint64 + selfLink *string } func (g genericJSONBase) ID() string { @@ -98,6 +132,14 @@ func (g genericJSONBase) SetResourceVersion(version uint64) { *g.resourceVersion = version } +func (g genericJSONBase) SelfLink() string { + return *g.selfLink +} + +func (g genericJSONBase) SetSelfLink(selfLink string) { + *g.selfLink = selfLink +} + // fieldPtr puts the address of fieldName, which must be a member of v, // into dest, which must be an address of a variable to which this field's // address can be assigned. @@ -140,5 +182,8 @@ func newGenericJSONBase(v reflect.Value) (genericJSONBase, error) { if err := fieldPtr(v, "ResourceVersion", &g.resourceVersion); err != nil { return g, err } + if err := fieldPtr(v, "SelfLink", &g.selfLink); err != nil { + return g, err + } return g, nil } diff --git a/pkg/runtime/jsonbase_test.go b/pkg/runtime/jsonbase_test.go index 0e0165aacb1..0115dfee248 100644 --- a/pkg/runtime/jsonbase_test.go +++ b/pkg/runtime/jsonbase_test.go @@ -37,6 +37,7 @@ func TestGenericJSONBase(t *testing.T) { APIVersion: "a", Kind: "b", ResourceVersion: 1, + SelfLink: "some/place/only/we/know", } g, err := newGenericJSONBase(reflect.ValueOf(&j).Elem()) if err != nil { @@ -56,11 +57,15 @@ func TestGenericJSONBase(t *testing.T) { if e, a := uint64(1), jbi.ResourceVersion(); e != a { t.Errorf("expected %v, got %v", e, a) } + if e, a := "some/place/only/we/know", jbi.SelfLink(); e != a { + t.Errorf("expected %v, got %v", e, a) + } jbi.SetID("bar") jbi.SetAPIVersion("c") jbi.SetKind("d") jbi.SetResourceVersion(2) + jbi.SetSelfLink("google.com") // Prove that jbi changes the original object. if e, a := "bar", j.ID; e != a { @@ -75,6 +80,9 @@ func TestGenericJSONBase(t *testing.T) { if e, a := uint64(2), j.ResourceVersion; e != a { t.Errorf("expected %v, got %v", e, a) } + if e, a := "google.com", j.SelfLink; e != a { + t.Errorf("expected %v, got %v", e, a) + } } type MyAPIObject struct { @@ -141,3 +149,48 @@ func TestResourceVersionerOfAPI(t *testing.T) { } } } + +func TestJSONBaseSelfLinker(t *testing.T) { + table := map[string]struct { + obj Object + expect string + try string + succeed bool + }{ + "normal": { + obj: &MyAPIObject{JSONBase: JSONBase{SelfLink: "foobar"}}, + expect: "foobar", + try: "newbar", + succeed: true, + }, + "fail": { + obj: &MyIncorrectlyMarkedAsAPIObject{}, + succeed: false, + }, + } + + linker := NewJSONBaseSelfLinker() + for name, item := range table { + got, err := linker.SelfLink(item.obj) + if e, a := item.succeed, err == nil; e != a { + t.Errorf("%v: expected %v, got %v", name, e, a) + } + if e, a := item.expect, got; item.succeed && e != a { + t.Errorf("%v: expected %v, got %v", name, e, a) + } + + err = linker.SetSelfLink(item.obj, item.try) + if e, a := item.succeed, err == nil; e != a { + t.Errorf("%v: expected %v, got %v", name, e, a) + } + if item.succeed { + got, err := linker.SelfLink(item.obj) + if err != nil { + t.Errorf("%v: expected no err, got %v", name, err) + } + if e, a := item.try, got; e != a { + t.Errorf("%v: expected %v, got %v", name, e, a) + } + } + } +} From b972f722481b9138d9b77319d02a132bda2dfdf8 Mon Sep 17 00:00:00 2001 From: Daniel Smith Date: Thu, 25 Sep 2014 15:08:09 -0700 Subject: [PATCH 2/3] convert multiple return values into a struct, add SelfLinker --- pkg/api/latest/latest.go | 30 +++++++++++++++++++++++++----- pkg/api/latest/latest_test.go | 4 ++-- pkg/client/client.go | 4 ++-- pkg/master/master.go | 4 ++-- 4 files changed, 31 insertions(+), 11 deletions(-) diff --git a/pkg/api/latest/latest.go b/pkg/api/latest/latest.go index cdd0704d6a9..99c558d6a79 100644 --- a/pkg/api/latest/latest.go +++ b/pkg/api/latest/latest.go @@ -49,16 +49,36 @@ var Codec = v1beta1.Codec // TODO: when versioning changes, make this part of each API definition. var ResourceVersioner = runtime.NewJSONBaseResourceVersioner() +// SelfLinker can set or get the SelfLink field of all API types. +// TODO: when versioning changes, make this part of each API definition. +// TODO(lavalamp): Combine SelfLinker & ResourceVersioner interfaces, force all uses +// to go through the InterfacesFor method below. +var SelfLinker = runtime.NewJSONBaseSelfLinker() + +// VersionInterfaces contains the interfaces one should use for dealing with types of a particular version. +type VersionInterfaces struct { + runtime.Codec + runtime.ResourceVersioner + runtime.SelfLinker +} + // InterfacesFor returns the default Codec and ResourceVersioner for a given version // string, or an error if the version is not known. -func InterfacesFor(version string) (codec runtime.Codec, versioner runtime.ResourceVersioner, err error) { +func InterfacesFor(version string) (*VersionInterfaces, error) { switch version { case "v1beta1": - codec, versioner = v1beta1.Codec, ResourceVersioner + return &VersionInterfaces{ + Codec: v1beta1.Codec, + ResourceVersioner: ResourceVersioner, + SelfLinker: SelfLinker, + }, nil case "v1beta2": - codec, versioner = v1beta2.Codec, ResourceVersioner + return &VersionInterfaces{ + Codec: v1beta2.Codec, + ResourceVersioner: ResourceVersioner, + SelfLinker: SelfLinker, + }, nil default: - err = fmt.Errorf("unsupported storage version: %s (valid: %s)", version, strings.Join(Versions, ", ")) + return nil, fmt.Errorf("unsupported storage version: %s (valid: %s)", version, strings.Join(Versions, ", ")) } - return } diff --git a/pkg/api/latest/latest_test.go b/pkg/api/latest/latest_test.go index 4c6c6f9d5c7..5fa83191a12 100644 --- a/pkg/api/latest/latest_test.go +++ b/pkg/api/latest/latest_test.go @@ -146,11 +146,11 @@ func TestCodec(t *testing.T) { } func TestInterfacesFor(t *testing.T) { - if _, _, err := InterfacesFor(""); err == nil { + if _, err := InterfacesFor(""); err == nil { t.Fatalf("unexpected non-error: %v", err) } for i, version := range append([]string{Version, OldestVersion}, Versions...) { - if codec, versioner, err := InterfacesFor(version); err != nil || codec == nil || versioner == nil { + if vi, err := InterfacesFor(version); err != nil || vi == nil { t.Fatalf("%d: unexpected result: %v", i, err) } } diff --git a/pkg/client/client.go b/pkg/client/client.go index 0ca6da7bad1..f6a3b46ed3c 100644 --- a/pkg/client/client.go +++ b/pkg/client/client.go @@ -108,12 +108,12 @@ func New(host, version string, auth *AuthInfo) (*Client, error) { // TODO: implement version negotation (highest version supported by server) version = latest.Version } - serverCodec, _, err := latest.InterfacesFor(version) + versionInterfaces, err := latest.InterfacesFor(version) if err != nil { return nil, fmt.Errorf("API version '%s' is not recognized (valid values: %s)", version, strings.Join(latest.Versions, ", ")) } prefix := fmt.Sprintf("/api/%s/", version) - restClient, err := NewRESTClient(host, auth, prefix, serverCodec) + restClient, err := NewRESTClient(host, auth, prefix, versionInterfaces.Codec) if err != nil { return nil, fmt.Errorf("API URL '%s' is not valid: %v", host, err) } diff --git a/pkg/master/master.go b/pkg/master/master.go index 886cadbb6c4..cab1167b6b1 100644 --- a/pkg/master/master.go +++ b/pkg/master/master.go @@ -73,11 +73,11 @@ func NewEtcdHelper(etcdServers []string, version string) (helper tools.EtcdHelpe if version == "" { version = latest.Version } - codec, versioner, err := latest.InterfacesFor(version) + versionInterfaces, err := latest.InterfacesFor(version) if err != nil { return helper, err } - return tools.EtcdHelper{client, codec, versioner}, nil + return tools.EtcdHelper{client, versionInterfaces.Codec, versionInterfaces.ResourceVersioner}, nil } // New returns a new instance of Master connected to the given etcd server. From 37e505601e0b4db4ec59441f20629e1fbe87d874 Mon Sep 17 00:00:00 2001 From: Daniel Smith Date: Thu, 25 Sep 2014 17:20:28 -0700 Subject: [PATCH 3/3] add self linking to apiserver --- pkg/apiserver/apiserver.go | 16 ++--- pkg/apiserver/apiserver_test.go | 98 ++++++++++++++++++++++++------- pkg/apiserver/minionproxy_test.go | 2 +- pkg/apiserver/operation.go | 28 +++++---- pkg/apiserver/operation_test.go | 12 +++- pkg/apiserver/proxy_test.go | 2 +- pkg/apiserver/redirect_test.go | 2 +- pkg/apiserver/resthandler.go | 62 ++++++++++++++++--- pkg/apiserver/watch_test.go | 8 +-- pkg/master/master.go | 8 +-- test/integration/client_test.go | 16 ++--- 11 files changed, 186 insertions(+), 68 deletions(-) diff --git a/pkg/apiserver/apiserver.go b/pkg/apiserver/apiserver.go index e86a81bd80b..7a8c9bda5d4 100644 --- a/pkg/apiserver/apiserver.go +++ b/pkg/apiserver/apiserver.go @@ -45,11 +45,11 @@ const ( StatusUnprocessableEntity = 422 ) -// Handle returns a Handler function that expose the provided storage interfaces +// Handle returns a Handler function that exposes the provided storage interfaces // as RESTful resources at prefix, serialized by codec, and also includes the support // http resources. -func Handle(storage map[string]RESTStorage, codec runtime.Codec, prefix string) http.Handler { - group := NewAPIGroup(storage, codec) +func Handle(storage map[string]RESTStorage, codec runtime.Codec, prefix string, selfLinker runtime.SelfLinker) http.Handler { + group := NewAPIGroup(storage, codec, prefix, selfLinker) mux := http.NewServeMux() group.InstallREST(mux, prefix) @@ -72,11 +72,13 @@ type APIGroup struct { // This is a helper method for registering multiple sets of REST handlers under different // prefixes onto a server. // TODO: add multitype codec serialization -func NewAPIGroup(storage map[string]RESTStorage, codec runtime.Codec) *APIGroup { +func NewAPIGroup(storage map[string]RESTStorage, codec runtime.Codec, canonicalPrefix string, selfLinker runtime.SelfLinker) *APIGroup { return &APIGroup{RESTHandler{ - storage: storage, - codec: codec, - ops: NewOperations(), + storage: storage, + codec: codec, + canonicalPrefix: canonicalPrefix, + selfLinker: selfLinker, + ops: NewOperations(), // Delay just long enough to handle most simple write operations asyncOpWait: time.Millisecond * 25, }} diff --git a/pkg/apiserver/apiserver_test.go b/pkg/apiserver/apiserver_test.go index 44b24840f8a..c5d713a14a5 100644 --- a/pkg/apiserver/apiserver_test.go +++ b/pkg/apiserver/apiserver_test.go @@ -45,6 +45,7 @@ func convert(obj runtime.Object) (runtime.Object, error) { } var codec = latest.Codec +var selfLinker = latest.SelfLinker func init() { api.Scheme.AddKnownTypes("", &Simple{}, &SimpleList{}) @@ -193,7 +194,7 @@ func TestNotFound(t *testing.T) { } handler := Handle(map[string]RESTStorage{ "foo": &SimpleRESTStorage{}, - }, codec, "/prefix/version") + }, codec, "/prefix/version", selfLinker) server := httptest.NewServer(handler) client := http.Client{} for k, v := range cases { @@ -214,7 +215,7 @@ func TestNotFound(t *testing.T) { } func TestVersion(t *testing.T) { - handler := Handle(map[string]RESTStorage{}, codec, "/prefix/version") + handler := Handle(map[string]RESTStorage{}, codec, "/prefix/version", selfLinker) server := httptest.NewServer(handler) client := http.Client{} @@ -243,7 +244,11 @@ func TestSimpleList(t *testing.T) { storage := map[string]RESTStorage{} simpleStorage := SimpleRESTStorage{} storage["simple"] = &simpleStorage - handler := Handle(storage, codec, "/prefix/version") + selfLinker := &setTestSelfLinker{ + t: t, + expectedSet: "/prefix/version/simple", + } + handler := Handle(storage, codec, "/prefix/version", selfLinker) server := httptest.NewServer(handler) resp, err := http.Get(server.URL + "/prefix/version/simple") @@ -254,6 +259,9 @@ func TestSimpleList(t *testing.T) { if resp.StatusCode != http.StatusOK { t.Errorf("Unexpected status: %d, Expected: %d, %#v", resp.StatusCode, http.StatusOK, resp) } + if !selfLinker.called { + t.Errorf("Never set self link") + } } func TestErrorList(t *testing.T) { @@ -262,7 +270,7 @@ func TestErrorList(t *testing.T) { errors: map[string]error{"list": fmt.Errorf("test Error")}, } storage["simple"] = &simpleStorage - handler := Handle(storage, codec, "/prefix/version") + handler := Handle(storage, codec, "/prefix/version", selfLinker) server := httptest.NewServer(handler) resp, err := http.Get(server.URL + "/prefix/version/simple") @@ -286,7 +294,7 @@ func TestNonEmptyList(t *testing.T) { }, } storage["simple"] = &simpleStorage - handler := Handle(storage, codec, "/prefix/version") + handler := Handle(storage, codec, "/prefix/version", selfLinker) server := httptest.NewServer(handler) resp, err := http.Get(server.URL + "/prefix/version/simple") @@ -320,8 +328,12 @@ func TestGet(t *testing.T) { Name: "foo", }, } + selfLinker := &setTestSelfLinker{ + t: t, + expectedSet: "/prefix/version/simple/id", + } storage["simple"] = &simpleStorage - handler := Handle(storage, codec, "/prefix/version") + handler := Handle(storage, codec, "/prefix/version", selfLinker) server := httptest.NewServer(handler) resp, err := http.Get(server.URL + "/prefix/version/simple/id") @@ -334,6 +346,9 @@ func TestGet(t *testing.T) { if itemOut.Name != simpleStorage.item.Name { t.Errorf("Unexpected data: %#v, expected %#v (%s)", itemOut, simpleStorage.item, string(body)) } + if !selfLinker.called { + t.Errorf("Never set self link") + } } func TestGetMissing(t *testing.T) { @@ -342,7 +357,7 @@ func TestGetMissing(t *testing.T) { errors: map[string]error{"get": apierrs.NewNotFound("simple", "id")}, } storage["simple"] = &simpleStorage - handler := Handle(storage, codec, "/prefix/version") + handler := Handle(storage, codec, "/prefix/version", selfLinker) server := httptest.NewServer(handler) resp, err := http.Get(server.URL + "/prefix/version/simple/id") @@ -360,7 +375,7 @@ func TestDelete(t *testing.T) { simpleStorage := SimpleRESTStorage{} ID := "id" storage["simple"] = &simpleStorage - handler := Handle(storage, codec, "/prefix/version") + handler := Handle(storage, codec, "/prefix/version", selfLinker) server := httptest.NewServer(handler) client := http.Client{} @@ -382,7 +397,7 @@ func TestDeleteMissing(t *testing.T) { errors: map[string]error{"delete": apierrs.NewNotFound("simple", ID)}, } storage["simple"] = &simpleStorage - handler := Handle(storage, codec, "/prefix/version") + handler := Handle(storage, codec, "/prefix/version", selfLinker) server := httptest.NewServer(handler) client := http.Client{} @@ -402,7 +417,11 @@ func TestUpdate(t *testing.T) { simpleStorage := SimpleRESTStorage{} ID := "id" storage["simple"] = &simpleStorage - handler := Handle(storage, codec, "/prefix/version") + selfLinker := &setTestSelfLinker{ + t: t, + expectedSet: "/prefix/version/simple/" + ID, + } + handler := Handle(storage, codec, "/prefix/version", selfLinker) server := httptest.NewServer(handler) item := &Simple{ @@ -423,6 +442,9 @@ func TestUpdate(t *testing.T) { if simpleStorage.updated.Name != item.Name { t.Errorf("Unexpected update value %#v, expected %#v.", simpleStorage.updated, item) } + if !selfLinker.called { + t.Errorf("Never set self link") + } } func TestUpdateMissing(t *testing.T) { @@ -432,7 +454,7 @@ func TestUpdateMissing(t *testing.T) { errors: map[string]error{"update": apierrs.NewNotFound("simple", ID)}, } storage["simple"] = &simpleStorage - handler := Handle(storage, codec, "/prefix/version") + handler := Handle(storage, codec, "/prefix/version", selfLinker) server := httptest.NewServer(handler) item := &Simple{ @@ -459,7 +481,7 @@ func TestCreate(t *testing.T) { simpleStorage := &SimpleRESTStorage{} handler := Handle(map[string]RESTStorage{ "foo": simpleStorage, - }, codec, "/prefix/version") + }, codec, "/prefix/version", selfLinker) handler.(*defaultAPIServer).group.handler.asyncOpWait = 0 server := httptest.NewServer(handler) client := http.Client{} @@ -500,7 +522,7 @@ func TestCreateNotFound(t *testing.T) { // See https://github.com/GoogleCloudPlatform/kubernetes/pull/486#discussion_r15037092. errors: map[string]error{"create": apierrs.NewNotFound("simple", "id")}, }, - }, codec, "/prefix/version") + }, codec, "/prefix/version", selfLinker) server := httptest.NewServer(handler) client := http.Client{} @@ -533,6 +555,23 @@ func TestParseTimeout(t *testing.T) { } } +type setTestSelfLinker struct { + t *testing.T + expectedSet string + id string + called bool +} + +func (s *setTestSelfLinker) ID(runtime.Object) (string, error) { return s.id, nil } +func (*setTestSelfLinker) SelfLink(runtime.Object) (string, error) { return "", nil } +func (s *setTestSelfLinker) SetSelfLink(obj runtime.Object, selfLink string) error { + if e, a := s.expectedSet, selfLink; e != a { + s.t.Errorf("expected '%v', got '%v'", e, a) + } + s.called = true + return nil +} + func TestSyncCreate(t *testing.T) { storage := SimpleRESTStorage{ injectedFunction: func(obj runtime.Object) (runtime.Object, error) { @@ -540,14 +579,19 @@ func TestSyncCreate(t *testing.T) { return obj, nil }, } + selfLinker := &setTestSelfLinker{ + t: t, + id: "bar", + expectedSet: "/prefix/version/foo/bar", + } handler := Handle(map[string]RESTStorage{ "foo": &storage, - }, codec, "/prefix/version") + }, codec, "/prefix/version", selfLinker) server := httptest.NewServer(handler) client := http.Client{} simple := &Simple{ - Name: "foo", + Name: "bar", } data, _ := codec.Encode(simple) request, err := http.NewRequest("POST", server.URL+"/prefix/version/foo?sync=true", bytes.NewBuffer(data)) @@ -579,6 +623,9 @@ func TestSyncCreate(t *testing.T) { if response.StatusCode != http.StatusOK { t.Errorf("Unexpected status: %d, Expected: %d, %#v", response.StatusCode, http.StatusOK, response) } + if !selfLinker.called { + t.Errorf("Never set self link") + } } func expectApiStatus(t *testing.T, method, url string, data []byte, code int) *api.Status { @@ -611,7 +658,7 @@ func TestAsyncDelayReturnsError(t *testing.T) { return nil, apierrs.NewAlreadyExists("foo", "bar") }, } - handler := Handle(map[string]RESTStorage{"foo": &storage}, codec, "/prefix/version") + handler := Handle(map[string]RESTStorage{"foo": &storage}, codec, "/prefix/version", selfLinker) handler.(*defaultAPIServer).group.handler.asyncOpWait = time.Millisecond / 2 server := httptest.NewServer(handler) @@ -629,11 +676,16 @@ func TestAsyncCreateError(t *testing.T) { return nil, apierrs.NewAlreadyExists("foo", "bar") }, } - handler := Handle(map[string]RESTStorage{"foo": &storage}, codec, "/prefix/version") + selfLinker := &setTestSelfLinker{ + t: t, + id: "bar", + expectedSet: "/prefix/version/foo/bar", + } + handler := Handle(map[string]RESTStorage{"foo": &storage}, codec, "/prefix/version", selfLinker) handler.(*defaultAPIServer).group.handler.asyncOpWait = 0 server := httptest.NewServer(handler) - simple := &Simple{Name: "foo"} + simple := &Simple{Name: "bar"} data, _ := codec.Encode(simple) status := expectApiStatus(t, "POST", fmt.Sprintf("%s/prefix/version/foo", server.URL), data, http.StatusAccepted) @@ -667,6 +719,9 @@ func TestAsyncCreateError(t *testing.T) { t.Logf("Details %#v, Got %#v", *expectedStatus.Details, *finalStatus.Details) } } + if !selfLinker.called { + t.Errorf("Never set self link") + } } type UnregisteredAPIObject struct { @@ -723,7 +778,7 @@ func TestSyncCreateTimeout(t *testing.T) { } handler := Handle(map[string]RESTStorage{ "foo": &storage, - }, codec, "/prefix/version") + }, codec, "/prefix/version", selfLinker) server := httptest.NewServer(handler) simple := &Simple{Name: "foo"} @@ -753,7 +808,10 @@ func TestCORSAllowedOrigins(t *testing.T) { t.Errorf("unexpected error: %v", err) } - handler := CORS(Handle(map[string]RESTStorage{}, codec, "/prefix/version"), allowedOriginRegexps, nil, nil, "true") + handler := CORS( + Handle(map[string]RESTStorage{}, codec, "/prefix/version", selfLinker), + allowedOriginRegexps, nil, nil, "true", + ) server := httptest.NewServer(handler) client := http.Client{} diff --git a/pkg/apiserver/minionproxy_test.go b/pkg/apiserver/minionproxy_test.go index 1ceeedf3a63..d5b7eb10e73 100644 --- a/pkg/apiserver/minionproxy_test.go +++ b/pkg/apiserver/minionproxy_test.go @@ -127,7 +127,7 @@ func TestApiServerMinionProxy(t *testing.T) { proxyServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) { w.Write([]byte(req.URL.Path)) })) - server := httptest.NewServer(Handle(nil, nil, "/prefix")) + server := httptest.NewServer(Handle(nil, nil, "/prefix", selfLinker)) proxy, _ := url.Parse(proxyServer.URL) resp, err := http.Get(fmt.Sprintf("%s/proxy/minion/%s%s", server.URL, proxy.Host, "/test")) if err != nil { diff --git a/pkg/apiserver/operation.go b/pkg/apiserver/operation.go index e0a3e9a461f..8f1469a7dda 100644 --- a/pkg/apiserver/operation.go +++ b/pkg/apiserver/operation.go @@ -63,12 +63,13 @@ func (h *OperationHandler) ServeHTTP(w http.ResponseWriter, req *http.Request) { // Operation represents an ongoing action which the server is performing. type Operation struct { - ID string - result runtime.Object - awaiting <-chan runtime.Object - finished *time.Time - lock sync.Mutex - notify chan struct{} + ID string + result runtime.Object + onReceive func(runtime.Object) + awaiting <-chan runtime.Object + finished *time.Time + lock sync.Mutex + notify chan struct{} } // Operations tracks all the ongoing operations. @@ -90,13 +91,15 @@ func NewOperations() *Operations { return ops } -// NewOperation adds a new operation. It is lock-free. -func (ops *Operations) NewOperation(from <-chan runtime.Object) *Operation { +// NewOperation adds a new operation. It is lock-free. 'onReceive' will be called +// with the value read from 'from', when it is read. +func (ops *Operations) NewOperation(from <-chan runtime.Object, onReceive func(runtime.Object)) *Operation { id := atomic.AddInt64(&ops.lastID, 1) op := &Operation{ - ID: strconv.FormatInt(id, 10), - awaiting: from, - notify: make(chan struct{}), + ID: strconv.FormatInt(id, 10), + awaiting: from, + onReceive: onReceive, + notify: make(chan struct{}), } go op.wait() go ops.insert(op) @@ -159,6 +162,9 @@ func (op *Operation) wait() { op.lock.Lock() defer op.lock.Unlock() + if op.onReceive != nil { + op.onReceive(result) + } op.result = result finished := time.Now() op.finished = &finished diff --git a/pkg/apiserver/operation_test.go b/pkg/apiserver/operation_test.go index 856fe764389..679b57b39ea 100644 --- a/pkg/apiserver/operation_test.go +++ b/pkg/apiserver/operation_test.go @@ -35,7 +35,8 @@ func TestOperation(t *testing.T) { ops := NewOperations() c := make(chan runtime.Object) - op := ops.NewOperation(c) + called := make(chan struct{}) + op := ops.NewOperation(c, func(runtime.Object) { go close(called) }) // Allow context switch, so that op's ID can get added to the map and Get will work. // This is just so we can test Get. Ordinary users have no need to call Get immediately // after calling NewOperation, because it returns the operation directly. @@ -72,6 +73,11 @@ func TestOperation(t *testing.T) { t.Errorf("Unexpectedly slow completion") } + _, open := <-called + if open { + t.Errorf("expected hook to be called!") + } + time.Sleep(100 * time.Millisecond) finished := atomic.LoadInt32(&waited) if finished != waiters { @@ -107,7 +113,7 @@ func TestOperationsList(t *testing.T) { } handler := Handle(map[string]RESTStorage{ "foo": simpleStorage, - }, codec, "/prefix/version") + }, codec, "/prefix/version", selfLinker) handler.(*defaultAPIServer).group.handler.asyncOpWait = 0 server := httptest.NewServer(handler) client := http.Client{} @@ -163,7 +169,7 @@ func TestOpGet(t *testing.T) { } handler := Handle(map[string]RESTStorage{ "foo": simpleStorage, - }, codec, "/prefix/version") + }, codec, "/prefix/version", selfLinker) handler.(*defaultAPIServer).group.handler.asyncOpWait = 0 server := httptest.NewServer(handler) client := http.Client{} diff --git a/pkg/apiserver/proxy_test.go b/pkg/apiserver/proxy_test.go index b50eb67264d..e62364af84e 100644 --- a/pkg/apiserver/proxy_test.go +++ b/pkg/apiserver/proxy_test.go @@ -161,7 +161,7 @@ func TestProxy(t *testing.T) { } handler := Handle(map[string]RESTStorage{ "foo": simpleStorage, - }, codec, "/prefix/version") + }, codec, "/prefix/version", selfLinker) server := httptest.NewServer(handler) req, err := http.NewRequest( diff --git a/pkg/apiserver/redirect_test.go b/pkg/apiserver/redirect_test.go index 6425f920d8c..072c388858c 100644 --- a/pkg/apiserver/redirect_test.go +++ b/pkg/apiserver/redirect_test.go @@ -30,7 +30,7 @@ func TestRedirect(t *testing.T) { } handler := Handle(map[string]RESTStorage{ "foo": simpleStorage, - }, codec, "/prefix/version") + }, codec, "/prefix/version", selfLinker) server := httptest.NewServer(handler) dontFollow := errors.New("don't follow") diff --git a/pkg/apiserver/resthandler.go b/pkg/apiserver/resthandler.go index 7f4dc1d6742..50ec2cd97bc 100644 --- a/pkg/apiserver/resthandler.go +++ b/pkg/apiserver/resthandler.go @@ -18,19 +18,24 @@ package apiserver import ( "net/http" + "path" "time" "github.com/GoogleCloudPlatform/kubernetes/pkg/api" "github.com/GoogleCloudPlatform/kubernetes/pkg/httplog" "github.com/GoogleCloudPlatform/kubernetes/pkg/labels" "github.com/GoogleCloudPlatform/kubernetes/pkg/runtime" + + "github.com/golang/glog" ) type RESTHandler struct { - storage map[string]RESTStorage - codec runtime.Codec - ops *Operations - asyncOpWait time.Duration + storage map[string]RESTStorage + codec runtime.Codec + canonicalPrefix string + selfLinker runtime.SelfLinker + ops *Operations + asyncOpWait time.Duration } // ServeHTTP handles requests to all RESTStorage objects. @@ -50,6 +55,37 @@ func (h *RESTHandler) ServeHTTP(w http.ResponseWriter, req *http.Request) { h.handleRESTStorage(parts, req, w, storage) } +// Sets the SelfLink field of the object. +func (h *RESTHandler) setSelfLink(obj runtime.Object, req *http.Request) error { + newURL := *req.URL + newURL.Path = path.Join(h.canonicalPrefix, req.URL.Path) + newURL.RawQuery = "" + newURL.Fragment = "" + return h.selfLinker.SetSelfLink(obj, newURL.String()) +} + +// Like setSelfLink, but appends the object's id. +func (h *RESTHandler) setSelfLinkAddID(obj runtime.Object, req *http.Request) error { + id, err := h.selfLinker.ID(obj) + if err != nil { + return err + } + newURL := *req.URL + newURL.Path = path.Join(h.canonicalPrefix, req.URL.Path, id) + newURL.RawQuery = "" + newURL.Fragment = "" + return h.selfLinker.SetSelfLink(obj, newURL.String()) +} + +// curry adapts either of the self link setting functions into a function appropriate for operation's hook. +func curry(f func(runtime.Object, *http.Request) error, req *http.Request) func(runtime.Object) { + return func(obj runtime.Object) { + if err := f(obj, req); err != nil { + glog.Errorf("unable to set self link for %#v: %v", obj, err) + } + } +} + // handleRESTStorage is the main dispatcher for a storage object. It switches on the HTTP method, and then // on path length, according to the following table: // Method Path Action @@ -86,6 +122,10 @@ func (h *RESTHandler) handleRESTStorage(parts []string, req *http.Request, w htt errorJSON(err, h.codec, w) return } + if err := h.setSelfLink(list, req); err != nil { + errorJSON(err, h.codec, w) + return + } writeJSON(http.StatusOK, h.codec, list, w) case 2: item, err := storage.Get(ctx, parts[1]) @@ -93,6 +133,10 @@ func (h *RESTHandler) handleRESTStorage(parts []string, req *http.Request, w htt errorJSON(err, h.codec, w) return } + if err := h.setSelfLink(item, req); err != nil { + errorJSON(err, h.codec, w) + return + } writeJSON(http.StatusOK, h.codec, item, w) default: notFound(w, req) @@ -119,7 +163,7 @@ func (h *RESTHandler) handleRESTStorage(parts []string, req *http.Request, w htt errorJSON(err, h.codec, w) return } - op := h.createOperation(out, sync, timeout) + op := h.createOperation(out, sync, timeout, curry(h.setSelfLinkAddID, req)) h.finishReq(op, req, w) case "DELETE": @@ -132,7 +176,7 @@ func (h *RESTHandler) handleRESTStorage(parts []string, req *http.Request, w htt errorJSON(err, h.codec, w) return } - op := h.createOperation(out, sync, timeout) + op := h.createOperation(out, sync, timeout, nil) h.finishReq(op, req, w) case "PUT": @@ -156,7 +200,7 @@ func (h *RESTHandler) handleRESTStorage(parts []string, req *http.Request, w htt errorJSON(err, h.codec, w) return } - op := h.createOperation(out, sync, timeout) + op := h.createOperation(out, sync, timeout, curry(h.setSelfLink, req)) h.finishReq(op, req, w) default: @@ -165,8 +209,8 @@ func (h *RESTHandler) handleRESTStorage(parts []string, req *http.Request, w htt } // createOperation creates an operation to process a channel response. -func (h *RESTHandler) createOperation(out <-chan runtime.Object, sync bool, timeout time.Duration) *Operation { - op := h.ops.NewOperation(out) +func (h *RESTHandler) createOperation(out <-chan runtime.Object, sync bool, timeout time.Duration, onReceive func(runtime.Object)) *Operation { + op := h.ops.NewOperation(out, onReceive) if sync { op.WaitFor(timeout) } else if h.asyncOpWait != 0 { diff --git a/pkg/apiserver/watch_test.go b/pkg/apiserver/watch_test.go index 575e843c84e..3b734a888e1 100644 --- a/pkg/apiserver/watch_test.go +++ b/pkg/apiserver/watch_test.go @@ -49,7 +49,7 @@ func TestWatchWebsocket(t *testing.T) { _ = ResourceWatcher(simpleStorage) // Give compile error if this doesn't work. handler := Handle(map[string]RESTStorage{ "foo": simpleStorage, - }, codec, "/prefix/version") + }, codec, "/prefix/version", selfLinker) server := httptest.NewServer(handler) dest, _ := url.Parse(server.URL) @@ -95,7 +95,7 @@ func TestWatchHTTP(t *testing.T) { simpleStorage := &SimpleRESTStorage{} handler := Handle(map[string]RESTStorage{ "foo": simpleStorage, - }, codec, "/prefix/version") + }, codec, "/prefix/version", selfLinker) server := httptest.NewServer(handler) client := http.Client{} @@ -148,7 +148,7 @@ func TestWatchParamParsing(t *testing.T) { simpleStorage := &SimpleRESTStorage{} handler := Handle(map[string]RESTStorage{ "foo": simpleStorage, - }, codec, "/prefix/version") + }, codec, "/prefix/version", selfLinker) server := httptest.NewServer(handler) dest, _ := url.Parse(server.URL) @@ -210,7 +210,7 @@ func TestWatchProtocolSelection(t *testing.T) { simpleStorage := &SimpleRESTStorage{} handler := Handle(map[string]RESTStorage{ "foo": simpleStorage, - }, codec, "/prefix/version") + }, codec, "/prefix/version", selfLinker) server := httptest.NewServer(handler) client := http.Client{} diff --git a/pkg/master/master.go b/pkg/master/master.go index cab1167b6b1..08bc3f34bb4 100644 --- a/pkg/master/master.go +++ b/pkg/master/master.go @@ -152,19 +152,19 @@ func (m *Master) init(cloud cloudprovider.Interface, podInfoGetter client.PodInf } // API_v1beta1 returns the resources and codec for API version v1beta1. -func (m *Master) API_v1beta1() (map[string]apiserver.RESTStorage, runtime.Codec) { +func (m *Master) API_v1beta1() (map[string]apiserver.RESTStorage, runtime.Codec, string, runtime.SelfLinker) { storage := make(map[string]apiserver.RESTStorage) for k, v := range m.storage { storage[k] = v } - return storage, v1beta1.Codec + return storage, v1beta1.Codec, "/api/v1beta1", latest.SelfLinker } // API_v1beta2 returns the resources and codec for API version v1beta2. -func (m *Master) API_v1beta2() (map[string]apiserver.RESTStorage, runtime.Codec) { +func (m *Master) API_v1beta2() (map[string]apiserver.RESTStorage, runtime.Codec, string, runtime.SelfLinker) { storage := make(map[string]apiserver.RESTStorage) for k, v := range m.storage { storage[k] = v } - return storage, v1beta2.Codec + return storage, v1beta2.Codec, "/api/v1beta1", latest.SelfLinker } diff --git a/test/integration/client_test.go b/test/integration/client_test.go index e3627118dd2..b5cd44fec67 100644 --- a/test/integration/client_test.go +++ b/test/integration/client_test.go @@ -45,20 +45,22 @@ func TestClient(t *testing.T) { m := master.New(&master.Config{ EtcdHelper: helper, }) - s1, c1 := m.API_v1beta1() - s2, c2 := m.API_v1beta2() + s1, c1, loc1, sl1 := m.API_v1beta1() + s2, c2, loc2, sl2 := m.API_v1beta2() testCases := map[string]struct { - Storage map[string]apiserver.RESTStorage - Codec runtime.Codec + Storage map[string]apiserver.RESTStorage + Codec runtime.Codec + location string + selfLinker runtime.SelfLinker }{ - "v1beta1": {s1, c1}, - "v1beta2": {s2, c2}, + "v1beta1": {s1, c1, loc1, sl1}, + "v1beta2": {s2, c2, loc2, sl2}, } for apiVersion, values := range testCases { deleteAllEtcdKeys() - s := httptest.NewServer(apiserver.Handle(values.Storage, values.Codec, fmt.Sprintf("/api/%s/", apiVersion))) + s := httptest.NewServer(apiserver.Handle(values.Storage, values.Codec, fmt.Sprintf("/api/%s/", apiVersion), values.selfLinker)) client := client.NewOrDie(s.URL, apiVersion, nil) info, err := client.ServerVersion()