github/kubernetes
1
0
Fork 0
mirror of https://github.com/k3s-io/kubernetes.git synced 2025-07-23 11:50:44 +00:00
Code Issues Projects Releases Wiki Activity

Merge pull request #807 from smarterclayton/abstract_apiserver_encoding

Browse Source 
Decouple apiserver from codec implementation
This commit is contained in:
Daniel Smith 2014-08-08 11:57:04 -07:00
commit ac6d6ec974
16 changed files with 140 additions and 128 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

42
pkg/apiserver/apiserver.go
View File

@ -33,6 +33,14 @@ import (
"github.com/golang/glog"
)
// Codec defines methods for serializing and deserializing API
// objects
type Codec interface {
Encode(obj interface{}) (data []byte, err error)
Decode(data []byte) (interface{}, error)
DecodeInto(data []byte, obj interface{}) error
}
// APIServer is an HTTPHandler that delegates to RESTStorage objects.
// It handles URLs of the form:
// ${prefix}/${storage_key}[/${object_name}]
@ -42,18 +50,24 @@ import (
type APIServer struct {
prefix string
storage map[string]RESTStorage
codec Codec
ops *Operations
mux *http.ServeMux
asyncOpWait time.Duration
}
// New creates a new APIServer object.
// 'storage' contains a map of handlers.
// 'prefix' is the hosting path prefix.
func New(storage map[string]RESTStorage, prefix string) *APIServer {
// New creates a new APIServer object. 'storage' contains a map of handlers. 'codec'
// is an interface for decoding to and from JSON. 'prefix' is the hosting path prefix.
//
// The codec will be used to decode the request body into an object pointer returned by
// RESTStorage.New(). The Create() and Update() methods should cast their argument to
// the type returned by New().
// TODO: add multitype codec serialization
func New(storage map[string]RESTStorage, codec Codec, prefix string) *APIServer {
s := &APIServer{
storage: storage,
prefix: strings.TrimRight(prefix, "/"),
storage: storage,
codec: codec,
ops: NewOperations(),
mux: http.NewServeMux(),
// Delay just long enough to handle most simple write operations
@ -153,7 +167,7 @@ func (s *APIServer) handleRESTStorage(parts []string, req *http.Request, w http.
internalError(err, w)
return
}
writeJSON(http.StatusOK, list, w)
writeJSON(http.StatusOK, s.codec, list, w)
case 2:
item, err := storage.Get(parts[1])
if IsNotFound(err) {
@ -164,7 +178,7 @@ func (s *APIServer) handleRESTStorage(parts []string, req *http.Request, w http.
internalError(err, w)
return
}
writeJSON(http.StatusOK, item, w)
writeJSON(http.StatusOK, s.codec, item, w)
default:
notFound(w, req)
}
@ -179,7 +193,8 @@ func (s *APIServer) handleRESTStorage(parts []string, req *http.Request, w http.
internalError(err, w)
return
}
obj, err := storage.Extract(body)
obj := storage.New()
err = s.codec.DecodeInto(body, obj)
if IsNotFound(err) {
notFound(w, req)
return
@ -227,7 +242,8 @@ func (s *APIServer) handleRESTStorage(parts []string, req *http.Request, w http.
internalError(err, w)
return
}
obj, err := storage.Extract(body)
obj := storage.New()
err = s.codec.DecodeInto(body, obj)
if IsNotFound(err) {
notFound(w, req)
return
@ -286,15 +302,15 @@ func (s *APIServer) finishReq(op *Operation, w http.ResponseWriter) {
status = stat.Code
}
}
writeJSON(status, obj, w)
writeJSON(status, s.codec, obj, w)
} else {
writeJSON(http.StatusAccepted, obj, w)
writeJSON(http.StatusAccepted, s.codec, obj, w)
}
}
// writeJSON renders an object as JSON to the response
func writeJSON(statusCode int, object interface{}, w http.ResponseWriter) {
output, err := api.Encode(object)
func writeJSON(statusCode int, codec Codec, object interface{}, w http.ResponseWriter) {
output, err := codec.Encode(object)
if err != nil {
internalError(err, w)
return

70
pkg/apiserver/apiserver_test.go
View File

@ -39,6 +39,8 @@ func convert(obj interface{}) (interface{}, error) {
return obj, nil
}
var codec = api.Codec
func init() {
api.AddKnownTypes("", Simple{}, SimpleList{})
api.AddKnownTypes("v1beta1", Simple{}, SimpleList{})
@ -59,8 +61,8 @@ type SimpleRESTStorage struct {
list []Simple
item Simple
deleted string
updated Simple
created Simple
updated *Simple
created *Simple
// Valid if WatchAll or WatchSingle is called
fakeWatch *watch.FakeWatcher
@ -97,14 +99,12 @@ func (storage *SimpleRESTStorage) Delete(id string) (<-chan interface{}, error)
}), nil
}
func (storage *SimpleRESTStorage) Extract(body []byte) (interface{}, error) {
var item Simple
api.DecodeInto(body, &item)
return item, storage.errors["extract"]
func (storage *SimpleRESTStorage) New() interface{} {
return &Simple{}
}
func (storage *SimpleRESTStorage) Create(obj interface{}) (<-chan interface{}, error) {
storage.created = obj.(Simple)
storage.created = obj.(*Simple)
if err := storage.errors["create"]; err != nil {
return nil, err
}
@ -117,7 +117,7 @@ func (storage *SimpleRESTStorage) Create(obj interface{}) (<-chan interface{}, e
}
func (storage *SimpleRESTStorage) Update(obj interface{}) (<-chan interface{}, error) {
storage.updated = obj.(Simple)
storage.updated = obj.(*Simple)
if err := storage.errors["update"]; err != nil {
return nil, err
}
@ -154,7 +154,7 @@ func extractBody(response *http.Response, object interface{}) (string, error) {
if err != nil {
return string(body), err
}
err = api.DecodeInto(body, object)
err = codec.DecodeInto(body, object)
return string(body), err
}
@ -178,7 +178,7 @@ func TestNotFound(t *testing.T) {
}
handler := New(map[string]RESTStorage{
"foo": &SimpleRESTStorage{},
}, "/prefix/version")
}, codec, "/prefix/version")
server := httptest.NewServer(handler)
client := http.Client{}
for k, v := range cases {
@ -199,7 +199,7 @@ func TestNotFound(t *testing.T) {
}
func TestVersion(t *testing.T) {
handler := New(map[string]RESTStorage{}, "/prefix/version")
handler := New(map[string]RESTStorage{}, codec, "/prefix/version")
server := httptest.NewServer(handler)
client := http.Client{}
@ -228,7 +228,7 @@ func TestSimpleList(t *testing.T) {
storage := map[string]RESTStorage{}
simpleStorage := SimpleRESTStorage{}
storage["simple"] = &simpleStorage
handler := New(storage, "/prefix/version")
handler := New(storage, codec, "/prefix/version")
server := httptest.NewServer(handler)
resp, err := http.Get(server.URL + "/prefix/version/simple")
@ -247,7 +247,7 @@ func TestErrorList(t *testing.T) {
errors: map[string]error{"list": fmt.Errorf("test Error")},
}
storage["simple"] = &simpleStorage
handler := New(storage, "/prefix/version")
handler := New(storage, codec, "/prefix/version")
server := httptest.NewServer(handler)
resp, err := http.Get(server.URL + "/prefix/version/simple")
@ -271,7 +271,7 @@ func TestNonEmptyList(t *testing.T) {
},
}
storage["simple"] = &simpleStorage
handler := New(storage, "/prefix/version")
handler := New(storage, codec, "/prefix/version")
server := httptest.NewServer(handler)
resp, err := http.Get(server.URL + "/prefix/version/simple")
@ -306,7 +306,7 @@ func TestGet(t *testing.T) {
},
}
storage["simple"] = &simpleStorage
handler := New(storage, "/prefix/version")
handler := New(storage, codec, "/prefix/version")
server := httptest.NewServer(handler)
resp, err := http.Get(server.URL + "/prefix/version/simple/id")
@ -327,7 +327,7 @@ func TestGetMissing(t *testing.T) {
errors: map[string]error{"get": NewNotFoundErr("simple", "id")},
}
storage["simple"] = &simpleStorage
handler := New(storage, "/prefix/version")
handler := New(storage, codec, "/prefix/version")
server := httptest.NewServer(handler)
resp, err := http.Get(server.URL + "/prefix/version/simple/id")
@ -345,7 +345,7 @@ func TestDelete(t *testing.T) {
simpleStorage := SimpleRESTStorage{}
ID := "id"
storage["simple"] = &simpleStorage
handler := New(storage, "/prefix/version")
handler := New(storage, codec, "/prefix/version")
server := httptest.NewServer(handler)
client := http.Client{}
@ -367,7 +367,7 @@ func TestDeleteMissing(t *testing.T) {
errors: map[string]error{"delete": NewNotFoundErr("simple", ID)},
}
storage["simple"] = &simpleStorage
handler := New(storage, "/prefix/version")
handler := New(storage, codec, "/prefix/version")
server := httptest.NewServer(handler)
client := http.Client{}
@ -387,13 +387,13 @@ func TestUpdate(t *testing.T) {
simpleStorage := SimpleRESTStorage{}
ID := "id"
storage["simple"] = &simpleStorage
handler := New(storage, "/prefix/version")
handler := New(storage, codec, "/prefix/version")
server := httptest.NewServer(handler)
item := Simple{
Name: "bar",
}
body, err := api.Encode(item)
body, err := codec.Encode(item)
if err != nil {
t.Errorf("unexpected error: %v", err)
}
@ -417,13 +417,13 @@ func TestUpdateMissing(t *testing.T) {
errors: map[string]error{"update": NewNotFoundErr("simple", ID)},
}
storage["simple"] = &simpleStorage
handler := New(storage, "/prefix/version")
handler := New(storage, codec, "/prefix/version")
server := httptest.NewServer(handler)
item := Simple{
Name: "bar",
}
body, err := api.Encode(item)
body, err := codec.Encode(item)
if err != nil {
t.Errorf("unexpected error: %v", err)
}
@ -441,7 +441,7 @@ func TestUpdateMissing(t *testing.T) {
}
func TestBadPath(t *testing.T) {
handler := New(map[string]RESTStorage{}, "/prefix/version")
handler := New(map[string]RESTStorage{}, codec, "/prefix/version")
server := httptest.NewServer(handler)
client := http.Client{}
@ -464,7 +464,7 @@ func TestCreate(t *testing.T) {
simpleStorage := &SimpleRESTStorage{}
handler := New(map[string]RESTStorage{
"foo": simpleStorage,
}, "/prefix/version")
}, codec, "/prefix/version")
handler.asyncOpWait = 0
server := httptest.NewServer(handler)
client := http.Client{}
@ -472,7 +472,7 @@ func TestCreate(t *testing.T) {
simple := Simple{
Name: "foo",
}
data, _ := api.Encode(simple)
data, _ := codec.Encode(simple)
request, err := http.NewRequest("POST", server.URL+"/prefix/version/foo", bytes.NewBuffer(data))
if err != nil {
t.Errorf("unexpected error: %v", err)
@ -505,12 +505,12 @@ func TestCreateNotFound(t *testing.T) {
// See https://github.com/GoogleCloudPlatform/kubernetes/pull/486#discussion_r15037092.
errors: map[string]error{"create": NewNotFoundErr("simple", "id")},
},
}, "/prefix/version")
}, codec, "/prefix/version")
server := httptest.NewServer(handler)
client := http.Client{}
simple := Simple{Name: "foo"}
data, _ := api.Encode(simple)
data, _ := codec.Encode(simple)
request, err := http.NewRequest("POST", server.URL+"/prefix/version/simple", bytes.NewBuffer(data))
if err != nil {
t.Errorf("unexpected error: %v", err)
@ -547,14 +547,14 @@ func TestSyncCreate(t *testing.T) {
}
handler := New(map[string]RESTStorage{
"foo": &storage,
}, "/prefix/version")
}, codec, "/prefix/version")
server := httptest.NewServer(handler)
client := http.Client{}
simple := Simple{
Name: "foo",
}
data, _ := api.Encode(simple)
data, _ := codec.Encode(simple)
request, err := http.NewRequest("POST", server.URL+"/prefix/version/foo?sync=true", bytes.NewBuffer(data))
if err != nil {
t.Errorf("unexpected error: %v", err)
@ -616,7 +616,7 @@ func TestAsyncDelayReturnsError(t *testing.T) {
return nil, errors.New("error")
},
}
handler := New(map[string]RESTStorage{"foo": &storage}, "/prefix/version")
handler := New(map[string]RESTStorage{"foo": &storage}, codec, "/prefix/version")
handler.asyncOpWait = time.Millisecond / 2
server := httptest.NewServer(handler)
@ -634,12 +634,12 @@ func TestAsyncCreateError(t *testing.T) {
return nil, errors.New("error")
},
}
handler := New(map[string]RESTStorage{"foo": &storage}, "/prefix/version")
handler := New(map[string]RESTStorage{"foo": &storage}, codec, "/prefix/version")
handler.asyncOpWait = 0
server := httptest.NewServer(handler)
simple := Simple{Name: "foo"}
data, _ := api.Encode(simple)
data, _ := codec.Encode(simple)
status := expectApiStatus(t, "POST", fmt.Sprintf("%s/prefix/version/foo", server.URL), data, http.StatusAccepted)
if status.Status != api.StatusWorking || status.Details == nil || status.Details.ID == "" {
@ -670,7 +670,7 @@ func TestWriteJSONDecodeError(t *testing.T) {
type T struct {
Value string
}
writeJSON(http.StatusOK, &T{"Undecodable"}, w)
writeJSON(http.StatusOK, api.Codec, &T{"Undecodable"}, w)
}))
client := http.Client{}
resp, err := client.Get(server.URL)
@ -715,11 +715,11 @@ func TestSyncCreateTimeout(t *testing.T) {
}
handler := New(map[string]RESTStorage{
"foo": &storage,
}, "/prefix/version")
}, codec, "/prefix/version")
server := httptest.NewServer(handler)
simple := Simple{Name: "foo"}
data, _ := api.Encode(simple)
data, _ := codec.Encode(simple)
itemOut := expectApiStatus(t, "POST", server.URL+"/prefix/version/foo?sync=true&timeout=4ms", data, http.StatusAccepted)
if itemOut.Status != api.StatusWorking || itemOut.Details == nil || itemOut.Details.ID == "" {
t.Errorf("Unexpected status %#v", itemOut)

5
pkg/apiserver/interfaces.go
View File

@ -24,6 +24,10 @@ import (
// RESTStorage is a generic interface for RESTful storage services
// Resources which are exported to the RESTful API of apiserver need to implement this interface.
type RESTStorage interface {
// New returns an empty object that can be used with Create and Update after request data has been put into it.
// This object must be a pointer type for use with Codec.DecodeInto([]byte, interface{})
New() interface{}
// List selects resources in the storage which match to the selector.
List(labels.Selector) (interface{}, error)
@ -35,7 +39,6 @@ type RESTStorage interface {
// Although it can return an arbitrary error value, IsNotFound(err) is true for the returned error value err when the specified resource is not found.
Delete(id string) (<-chan interface{}, error)
Extract(body []byte) (interface{}, error)
Create(interface{}) (<-chan interface{}, error)
Update(interface{}) (<-chan interface{}, error)
}

2
pkg/apiserver/minionproxy_test.go
View File

@ -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(New(nil, "/prefix"))
server := httptest.NewServer(New(nil, nil, "/prefix"))
proxy, _ := url.Parse(proxyServer.URL)
resp, err := http.Get(fmt.Sprintf("%s/proxy/minion/%s%s", server.URL, proxy.Host, "/test"))
if err != nil {

6
pkg/apiserver/operation.go
View File

@ -56,7 +56,7 @@ func (s *APIServer) handleOperation(w http.ResponseWriter, req *http.Request) {
if len(parts) == 0 {
// List outstanding operations.
list := s.ops.List()
writeJSON(http.StatusOK, list, w)
writeJSON(http.StatusOK, s.codec, list, w)
return
}
@ -68,9 +68,9 @@ func (s *APIServer) handleOperation(w http.ResponseWriter, req *http.Request) {
obj, complete := op.StatusOrResult()
if complete {
writeJSON(http.StatusOK, obj, w)
writeJSON(http.StatusOK, s.codec, obj, w)
} else {
writeJSON(http.StatusAccepted, obj, w)
writeJSON(http.StatusAccepted, s.codec, obj, w)
}
}

10
pkg/apiserver/operation_test.go
View File

@ -95,7 +95,7 @@ func TestOperationsList(t *testing.T) {
simpleStorage := &SimpleRESTStorage{}
handler := New(map[string]RESTStorage{
"foo": simpleStorage,
}, "/prefix/version")
}, codec, "/prefix/version")
handler.asyncOpWait = 0
server := httptest.NewServer(handler)
client := http.Client{}
@ -103,7 +103,7 @@ func TestOperationsList(t *testing.T) {
simple := Simple{
Name: "foo",
}
data, err := api.Encode(simple)
data, err := codec.Encode(simple)
if err != nil {
t.Errorf("unexpected error: %v", err)
}
@ -126,7 +126,7 @@ func TestOperationsList(t *testing.T) {
if err != nil {
t.Errorf("unexpected error: %v", err)
}
obj, err := api.Decode(body)
obj, err := codec.Decode(body)
if err != nil {
t.Errorf("unexpected error: %v", err)
}
@ -143,7 +143,7 @@ func TestOpGet(t *testing.T) {
simpleStorage := &SimpleRESTStorage{}
handler := New(map[string]RESTStorage{
"foo": simpleStorage,
}, "/prefix/version")
}, codec, "/prefix/version")
handler.asyncOpWait = 0
server := httptest.NewServer(handler)
client := http.Client{}
@ -151,7 +151,7 @@ func TestOpGet(t *testing.T) {
simple := Simple{
Name: "foo",
}
data, err := api.Encode(simple)
data, err := codec.Encode(simple)
t.Log(string(data))
if err != nil {
t.Errorf("unexpected error: %v", err)

4
pkg/apiserver/watch_test.go
View File

@ -42,7 +42,7 @@ func TestWatchWebsocket(t *testing.T) {
simpleStorage := &SimpleRESTStorage{}
handler := New(map[string]RESTStorage{
"foo": simpleStorage,
}, "/prefix/version")
}, codec, "/prefix/version")
server := httptest.NewServer(handler)
dest, _ := url.Parse(server.URL)
@ -92,7 +92,7 @@ func TestWatchHTTP(t *testing.T) {
simpleStorage := &SimpleRESTStorage{}
handler := New(map[string]RESTStorage{
"foo": simpleStorage,
}, "/prefix/version")
}, codec, "/prefix/version")
server := httptest.NewServer(handler)
client := http.Client{}

3
pkg/master/master.go
View File

@ -21,6 +21,7 @@ import (
"net/http"
"time"
"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
"github.com/GoogleCloudPlatform/kubernetes/pkg/apiserver"
"github.com/GoogleCloudPlatform/kubernetes/pkg/client"
"github.com/GoogleCloudPlatform/kubernetes/pkg/cloudprovider"
@ -140,5 +141,5 @@ func (m *Master) Run(myAddress, apiPrefix string) error {
// Instead of calling Run, you can call this function to get a handler for your own server.
// It is intended for testing. Only call once.
func (m *Master) ConstructHandler(apiPrefix string) http.Handler {
return apiserver.New(m.storage, apiPrefix)
return apiserver.New(m.storage, api.Codec, apiPrefix)
}

24
pkg/registry/controllerstorage.go
View File

@ -74,16 +74,14 @@ func (storage *ControllerRegistryStorage) Delete(id string) (<-chan interface{},
}), nil
}
// Extract deserializes user provided data into an api.ReplicationController.
func (storage *ControllerRegistryStorage) Extract(body []byte) (interface{}, error) {
result := api.ReplicationController{}
err := api.DecodeInto(body, &result)
return result, err
// New creates a new ReplicationController for use with Create and Update
func (storage *ControllerRegistryStorage) New() interface{} {
return &api.ReplicationController{}
}
// Create registers a given new ReplicationController instance to storage.registry.
func (storage *ControllerRegistryStorage) Create(obj interface{}) (<-chan interface{}, error) {
controller, ok := obj.(api.ReplicationController)
controller, ok := obj.(*api.ReplicationController)
if !ok {
return nil, fmt.Errorf("not a replication controller: %#v", obj)
}
@ -93,34 +91,34 @@ func (storage *ControllerRegistryStorage) Create(obj interface{}) (<-chan interf
// Pod Manifest ID should be assigned by the pod API
controller.DesiredState.PodTemplate.DesiredState.Manifest.ID = ""
if errs := api.ValidateReplicationController(&controller); len(errs) > 0 {
if errs := api.ValidateReplicationController(controller); len(errs) > 0 {
return nil, fmt.Errorf("Validation errors: %v", errs)
}
return apiserver.MakeAsync(func() (interface{}, error) {
err := storage.registry.CreateController(controller)
err := storage.registry.CreateController(*controller)
if err != nil {
return nil, err
}
return storage.waitForController(controller)
return storage.waitForController(*controller)
}), nil
}
// Update replaces a given ReplicationController instance with an existing instance in storage.registry.
func (storage *ControllerRegistryStorage) Update(obj interface{}) (<-chan interface{}, error) {
controller, ok := obj.(api.ReplicationController)
controller, ok := obj.(*api.ReplicationController)
if !ok {
return nil, fmt.Errorf("not a replication controller: %#v", obj)
}
if errs := api.ValidateReplicationController(&controller); len(errs) > 0 {
if errs := api.ValidateReplicationController(controller); len(errs) > 0 {
return nil, fmt.Errorf("Validation errors: %v", errs)
}
return apiserver.MakeAsync(func() (interface{}, error) {
err := storage.registry.UpdateController(controller)
err := storage.registry.UpdateController(*controller)
if err != nil {
return nil, err
}
return storage.waitForController(controller)
return storage.waitForController(*controller)
}), nil
}

16
pkg/registry/controllerstorage_test.go
View File

@ -123,23 +123,23 @@ func TestListControllerList(t *testing.T) {
}
}
func TestExtractControllerJson(t *testing.T) {
func TestControllerDecode(t *testing.T) {
mockRegistry := MockControllerRegistry{}
storage := ControllerRegistryStorage{
registry: &mockRegistry,
}
controller := api.ReplicationController{
controller := &api.ReplicationController{
JSONBase: api.JSONBase{
ID: "foo",
},
}
body, err := api.Encode(&controller)
body, err := api.Encode(controller)
if err != nil {
t.Errorf("unexpected error: %v", err)
}
controllerOut, err := storage.Extract(body)
if err != nil {
controllerOut := storage.New()
if err := api.DecodeInto(body, controllerOut); err != nil {
t.Errorf("unexpected error: %v", err)
}
@ -249,7 +249,7 @@ func TestCreateController(t *testing.T) {
podRegistry: &mockPodRegistry,
pollPeriod: time.Millisecond * 1,
}
controller := api.ReplicationController{
controller := &api.ReplicationController{
JSONBase: api.JSONBase{ID: "test"},
DesiredState: api.ReplicationControllerState{
Replicas: 2,
@ -314,7 +314,7 @@ func TestControllerStorageValidatesCreate(t *testing.T) {
},
}
for _, failureCase := range failureCases {
c, err := storage.Create(failureCase)
c, err := storage.Create(&failureCase)
if c != nil {
t.Errorf("Expected nil channel")
}
@ -345,7 +345,7 @@ func TestControllerStorageValidatesUpdate(t *testing.T) {
},
}
for _, failureCase := range failureCases {
c, err := storage.Update(failureCase)
c, err := storage.Update(&failureCase)
if c != nil {
t.Errorf("Expected nil channel")
}

8
pkg/registry/minionstorage.go
View File

@ -59,14 +59,12 @@ func (storage *MinionRegistryStorage) Get(id string) (interface{}, error) {
return storage.toApiMinion(id), err
}
func (storage *MinionRegistryStorage) Extract(body []byte) (interface{}, error) {
var minion api.Minion
err := api.DecodeInto(body, &minion)
return minion, err
func (storage *MinionRegistryStorage) New() interface{} {
return &api.Minion{}
}
func (storage *MinionRegistryStorage) Create(obj interface{}) (<-chan interface{}, error) {
minion, ok := obj.(api.Minion)
minion, ok := obj.(*api.Minion)
if !ok {
return nil, fmt.Errorf("not a minion: %#v", obj)
}

2
pkg/registry/minionstorage_test.go
View File

@ -38,7 +38,7 @@ func TestMinionRegistryStorage(t *testing.T) {
t.Errorf("has unexpected object")
}
c, err := ms.Create(api.Minion{JSONBase: api.JSONBase{ID: "baz"}})
c, err := ms.Create(&api.Minion{JSONBase: api.JSONBase{ID: "baz"}})
if err != nil {
t.Errorf("insert failed")
}

22
pkg/registry/podstorage.go
View File

@ -189,10 +189,8 @@ func (storage *PodRegistryStorage) Delete(id string) (<-chan interface{}, error)
}), nil
}
func (storage *PodRegistryStorage) Extract(body []byte) (interface{}, error) {
pod := api.Pod{}
err := api.DecodeInto(body, &pod)
return pod, err
func (storage *PodRegistryStorage) New() interface{} {
return &api.Pod{}
}
func (storage *PodRegistryStorage) scheduleAndCreatePod(pod api.Pod) error {
@ -207,36 +205,36 @@ func (storage *PodRegistryStorage) scheduleAndCreatePod(pod api.Pod) error {
}
func (storage *PodRegistryStorage) Create(obj interface{}) (<-chan interface{}, error) {
pod := obj.(api.Pod)
pod := obj.(*api.Pod)
if len(pod.ID) == 0 {
pod.ID = uuid.NewUUID().String()
}
pod.DesiredState.Manifest.ID = pod.ID
if errs := api.ValidatePod(&pod); len(errs) > 0 {
if errs := api.ValidatePod(pod); len(errs) > 0 {
return nil, fmt.Errorf("Validation errors: %v", errs)
}
return apiserver.MakeAsync(func() (interface{}, error) {
err := storage.scheduleAndCreatePod(pod)
err := storage.scheduleAndCreatePod(*pod)
if err != nil {
return nil, err
}
return storage.waitForPodRunning(pod)
return storage.waitForPodRunning(*pod)
}), nil
}
func (storage *PodRegistryStorage) Update(obj interface{}) (<-chan interface{}, error) {
pod := obj.(api.Pod)
if errs := api.ValidatePod(&pod); len(errs) > 0 {
pod := obj.(*api.Pod)
if errs := api.ValidatePod(pod); len(errs) > 0 {
return nil, fmt.Errorf("Validation errors: %v", errs)
}
return apiserver.MakeAsync(func() (interface{}, error) {
err := storage.registry.UpdatePod(pod)
err := storage.registry.UpdatePod(*pod)
if err != nil {
return nil, err
}
return storage.waitForPodRunning(pod)
return storage.waitForPodRunning(*pod)
}), nil
}

26
pkg/registry/podstorage_test.go
View File

@ -64,7 +64,7 @@ func TestCreatePodRegistryError(t *testing.T) {
Version: "v1beta1",
},
}
pod := api.Pod{DesiredState: desiredState}
pod := &api.Pod{DesiredState: desiredState}
ch, err := storage.Create(pod)
if err != nil {
t.Errorf("Expected %#v, Got %#v", nil, err)
@ -95,7 +95,7 @@ func TestCreatePodSchedulerError(t *testing.T) {
Version: "v1beta1",
},
}
pod := api.Pod{DesiredState: desiredState}
pod := &api.Pod{DesiredState: desiredState}
ch, err := storage.Create(pod)
if err != nil {
t.Errorf("Expected %#v, Got %#v", nil, err)
@ -127,7 +127,7 @@ func TestCreatePodSetsIds(t *testing.T) {
Version: "v1beta1",
},
}
pod := api.Pod{DesiredState: desiredState}
pod := &api.Pod{DesiredState: desiredState}
ch, err := storage.Create(pod)
if err != nil {
t.Errorf("Expected %#v, Got %#v", nil, err)
@ -208,28 +208,28 @@ func TestListPodList(t *testing.T) {
}
}
func TestExtractJson(t *testing.T) {
func TestPodDecode(t *testing.T) {
mockRegistry := MockPodRegistry{}
storage := PodRegistryStorage{
registry: &mockRegistry,
}
pod := api.Pod{
expected := &api.Pod{
JSONBase: api.JSONBase{
ID: "foo",
},
}
body, err := api.Encode(&pod)
body, err := api.Encode(expected)
if err != nil {
t.Errorf("unexpected error: %v", err)
}
podOut, err := storage.Extract(body)
if err != nil {
actual := storage.New()
if err := api.DecodeInto(body, actual); err != nil {
t.Errorf("unexpected error: %v", err)
}
if !reflect.DeepEqual(pod, podOut) {
t.Errorf("Expected %#v, found %#v", pod, podOut)
if !reflect.DeepEqual(expected, actual) {
t.Errorf("Expected %#v, Got %#v", expected, actual)
}
}
@ -373,7 +373,7 @@ func TestPodStorageValidatesCreate(t *testing.T) {
scheduler: &MockScheduler{machine: "test"},
registry: mockRegistry,
}
pod := api.Pod{}
pod := &api.Pod{}
c, err := storage.Create(pod)
if c != nil {
t.Errorf("Expected nil channel")
@ -391,7 +391,7 @@ func TestPodStorageValidatesUpdate(t *testing.T) {
scheduler: &MockScheduler{machine: "test"},
registry: mockRegistry,
}
pod := api.Pod{}
pod := &api.Pod{}
c, err := storage.Update(pod)
if c != nil {
t.Errorf("Expected nil channel")
@ -421,7 +421,7 @@ func TestCreatePod(t *testing.T) {
Version: "v1beta1",
},
}
pod := api.Pod{
pod := &api.Pod{
JSONBase: api.JSONBase{ID: "foo"},
DesiredState: desiredState,
}

18
pkg/registry/servicestorage.go
View File

@ -160,15 +160,13 @@ func (sr *ServiceRegistryStorage) Delete(id string) (<-chan interface{}, error)
}), nil
}
func (sr *ServiceRegistryStorage) Extract(body []byte) (interface{}, error) {
var svc api.Service
err := api.DecodeInto(body, &svc)
return svc, err
func (sr *ServiceRegistryStorage) New() interface{} {
return &api.Service{}
}
func (sr *ServiceRegistryStorage) Create(obj interface{}) (<-chan interface{}, error) {
srv := obj.(api.Service)
if errs := api.ValidateService(&srv); len(errs) > 0 {
srv := obj.(*api.Service)
if errs := api.ValidateService(srv); len(errs) > 0 {
return nil, fmt.Errorf("Validation errors: %v", errs)
}
return apiserver.MakeAsync(func() (interface{}, error) {
@ -200,7 +198,7 @@ func (sr *ServiceRegistryStorage) Create(obj interface{}) (<-chan interface{}, e
}
}
// TODO actually wait for the object to be fully created here.
err := sr.registry.CreateService(srv)
err := sr.registry.CreateService(*srv)
if err != nil {
return nil, err
}
@ -209,16 +207,16 @@ func (sr *ServiceRegistryStorage) Create(obj interface{}) (<-chan interface{}, e
}
func (sr *ServiceRegistryStorage) Update(obj interface{}) (<-chan interface{}, error) {
srv := obj.(api.Service)
srv := obj.(*api.Service)
if srv.ID == "" {
return nil, fmt.Errorf("ID should not be empty: %#v", srv)
}
if errs := api.ValidateService(&srv); len(errs) > 0 {
if errs := api.ValidateService(srv); len(errs) > 0 {
return nil, fmt.Errorf("Validation errors: %v", errs)
}
return apiserver.MakeAsync(func() (interface{}, error) {
// TODO: check to see if external load balancer status changed
err := sr.registry.UpdateService(srv)
err := sr.registry.UpdateService(*srv)
if err != nil {
return nil, err
}

10
pkg/registry/servicestorage_test.go
View File

@ -33,7 +33,7 @@ func TestServiceRegistry(t *testing.T) {
storage := MakeServiceRegistryStorage(memory, fakeCloud, MakeMinionRegistry(machines))
svc := api.Service{
svc := &api.Service{
JSONBase: api.JSONBase{ID: "foo"},
Selector: map[string]string{"bar": "baz"},
}
@ -68,7 +68,7 @@ func TestServiceStorageValidatesCreate(t *testing.T) {
},
}
for _, failureCase := range failureCases {
c, err := storage.Create(failureCase)
c, err := storage.Create(&failureCase)
if c != nil {
t.Errorf("Expected nil channel")
}
@ -97,7 +97,7 @@ func TestServiceStorageValidatesUpdate(t *testing.T) {
},
}
for _, failureCase := range failureCases {
c, err := storage.Update(failureCase)
c, err := storage.Update(&failureCase)
if c != nil {
t.Errorf("Expected nil channel")
}
@ -114,7 +114,7 @@ func TestServiceRegistryExternalService(t *testing.T) {
storage := MakeServiceRegistryStorage(memory, fakeCloud, MakeMinionRegistry(machines))
svc := api.Service{
svc := &api.Service{
JSONBase: api.JSONBase{ID: "foo"},
Selector: map[string]string{"bar": "baz"},
CreateExternalLoadBalancer: true,
@ -144,7 +144,7 @@ func TestServiceRegistryExternalServiceError(t *testing.T) {
storage := MakeServiceRegistryStorage(memory, fakeCloud, MakeMinionRegistry(machines))
svc := api.Service{
svc := &api.Service{
JSONBase: api.JSONBase{ID: "foo"},
Selector: map[string]string{"bar": "baz"},
CreateExternalLoadBalancer: true,