github/kubernetes
1
0
Fork 0
mirror of https://github.com/k3s-io/kubernetes.git synced 2025-07-29 06:27:05 +00:00
Code Issues Projects Releases Wiki Activity

Merge pull request #4921 from mikedanese/too-many-gets

Browse Source 
Interactions with etcd should optionally take a value that is filled with the result
This commit is contained in:
Clayton Coleman 2015-03-04 14:18:12 -05:00
commit eea1e88cb0
12 changed files with 127 additions and 91 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
pkg/registry/controller/registry.go
View File

@ -27,7 +27,7 @@ type Registry interface {
ListControllers(ctx api.Context) (*api.ReplicationControllerList, error)
WatchControllers(ctx api.Context, label, field labels.Selector, resourceVersion string) (watch.Interface, error)
GetController(ctx api.Context, controllerID string) (*api.ReplicationController, error)
CreateController(ctx api.Context, controller *api.ReplicationController) error
UpdateController(ctx api.Context, controller *api.ReplicationController) error
CreateController(ctx api.Context, controller *api.ReplicationController) (*api.ReplicationController, error)
UpdateController(ctx api.Context, controller *api.ReplicationController) (*api.ReplicationController, error)
DeleteController(ctx api.Context, controllerID string) error
}

13
pkg/registry/controller/rest.go
View File

@ -23,7 +23,6 @@ import (
"github.com/GoogleCloudPlatform/kubernetes/pkg/api/errors"
"github.com/GoogleCloudPlatform/kubernetes/pkg/api/rest"
"github.com/GoogleCloudPlatform/kubernetes/pkg/api/validation"
"github.com/GoogleCloudPlatform/kubernetes/pkg/apiserver"
rc "github.com/GoogleCloudPlatform/kubernetes/pkg/controller"
"github.com/GoogleCloudPlatform/kubernetes/pkg/labels"
"github.com/GoogleCloudPlatform/kubernetes/pkg/runtime"
@ -60,11 +59,11 @@ func (rs *REST) Create(ctx api.Context, obj runtime.Object) (runtime.Object, err
return nil, err
}
if err := rs.registry.CreateController(ctx, controller); err != nil {
out, err := rs.registry.CreateController(ctx, controller)
if err != nil {
err = rest.CheckGeneratedNameError(rest.ReplicationControllers, err, controller)
return apiserver.RESTResult{}, err
}
return rs.registry.GetController(ctx, controller.Name)
return out, err
}
// Delete asynchronously deletes the ReplicationController specified by its id.
@ -124,11 +123,7 @@ func (rs *REST) Update(ctx api.Context, obj runtime.Object) (runtime.Object, boo
if errs := validation.ValidateReplicationController(controller); len(errs) > 0 {
return nil, false, errors.NewInvalid("replicationController", controller.Name, errs)
}
err := rs.registry.UpdateController(ctx, controller)
if err != nil {
return nil, false, err
}
out, err := rs.registry.GetController(ctx, controller.Name)
out, err := rs.registry.UpdateController(ctx, controller)
return out, false, err
}

40
pkg/registry/etcd/etcd.go
View File

@ -152,23 +152,25 @@ func (r *Registry) GetController(ctx api.Context, controllerID string) (*api.Rep
}
// CreateController creates a new ReplicationController.
func (r *Registry) CreateController(ctx api.Context, controller *api.ReplicationController) error {
func (r *Registry) CreateController(ctx api.Context, controller *api.ReplicationController) (*api.ReplicationController, error) {
key, err := makeControllerKey(ctx, controller.Name)
if err != nil {
return err
return nil, err
}
err = r.CreateObj(key, controller, 0)
return etcderr.InterpretCreateError(err, "replicationController", controller.Name)
out := &api.ReplicationController{}
err = r.CreateObj(key, controller, out, 0)
return out, etcderr.InterpretCreateError(err, "replicationController", controller.Name)
}
// UpdateController replaces an existing ReplicationController.
func (r *Registry) UpdateController(ctx api.Context, controller *api.ReplicationController) error {
func (r *Registry) UpdateController(ctx api.Context, controller *api.ReplicationController) (*api.ReplicationController, error) {
key, err := makeControllerKey(ctx, controller.Name)
if err != nil {
return err
return nil, err
}
err = r.SetObj(key, controller, 0 /* ttl */)
return etcderr.InterpretUpdateError(err, "replicationController", controller.Name)
out := &api.ReplicationController{}
err = r.SetObj(key, controller, out, 0)
return out, etcderr.InterpretUpdateError(err, "replicationController", controller.Name)
}
// DeleteController deletes a ReplicationController specified by its ID.
@ -199,13 +201,14 @@ func (r *Registry) ListServices(ctx api.Context) (*api.ServiceList, error) {
}
// CreateService creates a new Service.
func (r *Registry) CreateService(ctx api.Context, svc *api.Service) error {
func (r *Registry) CreateService(ctx api.Context, svc *api.Service) (*api.Service, error) {
key, err := makeServiceKey(ctx, svc.Name)
if err != nil {
return err
return nil, err
}
err = r.CreateObj(key, svc, 0)
return etcderr.InterpretCreateError(err, "service", svc.Name)
out := &api.Service{}
err = r.CreateObj(key, svc, out, 0)
return out, etcderr.InterpretCreateError(err, "service", svc.Name)
}
// GetService obtains a Service specified by its name.
@ -270,13 +273,14 @@ func (r *Registry) DeleteService(ctx api.Context, name string) error {
}
// UpdateService replaces an existing Service.
func (r *Registry) UpdateService(ctx api.Context, svc *api.Service) error {
func (r *Registry) UpdateService(ctx api.Context, svc *api.Service) (*api.Service, error) {
key, err := makeServiceKey(ctx, svc.Name)
if err != nil {
return err
return nil, err
}
err = r.SetObj(key, svc, 0 /* ttl */)
return etcderr.InterpretUpdateError(err, "service", svc.Name)
out := &api.Service{}
err = r.SetObj(key, svc, out, 0)
return out, etcderr.InterpretUpdateError(err, "service", svc.Name)
}
// WatchServices begins watching for new, changed, or deleted service configurations.
@ -362,13 +366,13 @@ func (r *Registry) ListMinions(ctx api.Context) (*api.NodeList, error) {
func (r *Registry) CreateMinion(ctx api.Context, minion *api.Node) error {
// TODO: Add some validations.
err := r.CreateObj(makeNodeKey(minion.Name), minion, 0)
err := r.CreateObj(makeNodeKey(minion.Name), minion, nil, 0)
return etcderr.InterpretCreateError(err, "minion", minion.Name)
}
func (r *Registry) UpdateMinion(ctx api.Context, minion *api.Node) error {
// TODO: Add some validations.
err := r.SetObj(makeNodeKey(minion.Name), minion, 0 /* ttl */)
err := r.SetObj(makeNodeKey(minion.Name), minion, nil, 0)
return etcderr.InterpretUpdateError(err, "minion", minion.Name)
}

12
pkg/registry/etcd/etcd_test.go
View File

@ -212,7 +212,7 @@ func TestEtcdCreateController(t *testing.T) {
fakeClient := tools.NewFakeEtcdClient(t)
registry := NewTestEtcdRegistry(fakeClient)
key, _ := makeControllerKey(ctx, "foo")
err := registry.CreateController(ctx, &api.ReplicationController{
_, err := registry.CreateController(ctx, &api.ReplicationController{
ObjectMeta: api.ObjectMeta{
Name: "foo",
},
@ -242,7 +242,7 @@ func TestEtcdCreateControllerAlreadyExisting(t *testing.T) {
fakeClient.Set(key, runtime.EncodeOrDie(latest.Codec, &api.ReplicationController{ObjectMeta: api.ObjectMeta{Name: "foo"}}), 0)
registry := NewTestEtcdRegistry(fakeClient)
err := registry.CreateController(ctx, &api.ReplicationController{
_, err := registry.CreateController(ctx, &api.ReplicationController{
ObjectMeta: api.ObjectMeta{
Name: "foo",
},
@ -259,7 +259,7 @@ func TestEtcdUpdateController(t *testing.T) {
key, _ := makeControllerKey(ctx, "foo")
resp, _ := fakeClient.Set(key, runtime.EncodeOrDie(latest.Codec, &api.ReplicationController{ObjectMeta: api.ObjectMeta{Name: "foo"}}), 0)
registry := NewTestEtcdRegistry(fakeClient)
err := registry.UpdateController(ctx, &api.ReplicationController{
_, err := registry.UpdateController(ctx, &api.ReplicationController{
ObjectMeta: api.ObjectMeta{Name: "foo", ResourceVersion: strconv.FormatUint(resp.Node.ModifiedIndex, 10)},
Spec: api.ReplicationControllerSpec{
Replicas: 2,
@ -417,7 +417,7 @@ func TestEtcdCreateService(t *testing.T) {
ctx := api.NewDefaultContext()
fakeClient := tools.NewFakeEtcdClient(t)
registry := NewTestEtcdRegistry(fakeClient)
err := registry.CreateService(ctx, &api.Service{
_, err := registry.CreateService(ctx, &api.Service{
ObjectMeta: api.ObjectMeta{Name: "foo"},
})
if err != nil {
@ -447,7 +447,7 @@ func TestEtcdCreateServiceAlreadyExisting(t *testing.T) {
key, _ := makeServiceKey(ctx, "foo")
fakeClient.Set(key, runtime.EncodeOrDie(latest.Codec, &api.Service{ObjectMeta: api.ObjectMeta{Name: "foo"}}), 0)
registry := NewTestEtcdRegistry(fakeClient)
err := registry.CreateService(ctx, &api.Service{
_, err := registry.CreateService(ctx, &api.Service{
ObjectMeta: api.ObjectMeta{Name: "foo"},
})
if !errors.IsAlreadyExists(err) {
@ -572,7 +572,7 @@ func TestEtcdUpdateService(t *testing.T) {
SessionAffinity: "None",
},
}
err := registry.UpdateService(ctx, &testService)
_, err := registry.UpdateService(ctx, &testService)
if err != nil {
t.Errorf("unexpected error: %v", err)
}

6
pkg/registry/generic/etcd/etcd.go
View File

@ -148,7 +148,7 @@ func (e *Etcd) CreateWithName(ctx api.Context, name string, obj runtime.Object)
return err
}
}
err = e.Helper.CreateObj(key, obj, ttl)
err = e.Helper.CreateObj(key, obj, nil, ttl)
err = etcderr.InterpretCreateError(err, e.EndpointName, name)
if err == nil && e.Decorator != nil {
err = e.Decorator(obj)
@ -177,7 +177,7 @@ func (e *Etcd) Create(ctx api.Context, obj runtime.Object) (runtime.Object, erro
}
}
out := e.NewFunc()
if err := e.Helper.Create(key, obj, out, ttl); err != nil {
if err := e.Helper.CreateObj(key, obj, out, ttl); err != nil {
err = etcderr.InterpretCreateError(err, e.EndpointName, name)
err = rest.CheckGeneratedNameError(e.CreateStrategy, err, obj)
return nil, err
@ -209,7 +209,7 @@ func (e *Etcd) UpdateWithName(ctx api.Context, name string, obj runtime.Object)
return err
}
}
err = e.Helper.SetObj(key, obj, ttl)
err = e.Helper.SetObj(key, obj, nil, ttl)
err = etcderr.InterpretUpdateError(err, e.EndpointName, name)
if err == nil && e.Decorator != nil {
err = e.Decorator(obj)

8
pkg/registry/registrytest/controller.go
View File

@ -49,16 +49,16 @@ func (r *ControllerRegistry) GetController(ctx api.Context, ID string) (*api.Rep
return &api.ReplicationController{}, r.Err
}
func (r *ControllerRegistry) CreateController(ctx api.Context, controller *api.ReplicationController) error {
func (r *ControllerRegistry) CreateController(ctx api.Context, controller *api.ReplicationController) (*api.ReplicationController, error) {
r.Lock()
defer r.Unlock()
return r.Err
return controller, r.Err
}
func (r *ControllerRegistry) UpdateController(ctx api.Context, controller *api.ReplicationController) error {
func (r *ControllerRegistry) UpdateController(ctx api.Context, controller *api.ReplicationController) (*api.ReplicationController, error) {
r.Lock()
defer r.Unlock()
return r.Err
return controller, r.Err
}
func (r *ControllerRegistry) DeleteController(ctx api.Context, ID string) error {

8
pkg/registry/registrytest/service.go
View File

@ -71,14 +71,14 @@ func (r *ServiceRegistry) ListServices(ctx api.Context) (*api.ServiceList, error
return res, r.Err
}
func (r *ServiceRegistry) CreateService(ctx api.Context, svc *api.Service) error {
func (r *ServiceRegistry) CreateService(ctx api.Context, svc *api.Service) (*api.Service, error) {
r.mu.Lock()
defer r.mu.Unlock()
r.Service = new(api.Service)
*r.Service = *svc
r.List.Items = append(r.List.Items, *svc)
return r.Err
return svc, r.Err
}
func (r *ServiceRegistry) GetService(ctx api.Context, id string) (*api.Service, error) {
@ -98,13 +98,13 @@ func (r *ServiceRegistry) DeleteService(ctx api.Context, id string) error {
return r.Err
}
func (r *ServiceRegistry) UpdateService(ctx api.Context, svc *api.Service) error {
func (r *ServiceRegistry) UpdateService(ctx api.Context, svc *api.Service) (*api.Service, error) {
r.mu.Lock()
defer r.mu.Unlock()
r.UpdatedID = svc.Name
*r.Service = *svc
return r.Err
return svc, r.Err
}
func (r *ServiceRegistry) WatchServices(ctx api.Context, label labels.Selector, field labels.Selector, resourceVersion string) (watch.Interface, error) {

4
pkg/registry/service/registry.go
View File

@ -26,10 +26,10 @@ import (
// Registry is an interface for things that know how to store services.
type Registry interface {
ListServices(ctx api.Context) (*api.ServiceList, error)
CreateService(ctx api.Context, svc *api.Service) error
CreateService(ctx api.Context, svc *api.Service) (*api.Service, error)
GetService(ctx api.Context, name string) (*api.Service, error)
DeleteService(ctx api.Context, name string) error
UpdateService(ctx api.Context, svc *api.Service) error
UpdateService(ctx api.Context, svc *api.Service) (*api.Service, error)
WatchServices(ctx api.Context, labels, fields labels.Selector, resourceVersion string) (watch.Interface, error)
// TODO: endpoints and their implementation should be separated, setting endpoints should be

12
pkg/registry/service/rest.go
View File

@ -115,12 +115,12 @@ func (rs *REST) Create(ctx api.Context, obj runtime.Object) (runtime.Object, err
}
}
if err := rs.registry.CreateService(ctx, service); err != nil {
out, err := rs.registry.CreateService(ctx, service)
if err != nil {
rs.portalMgr.Release(net.ParseIP(service.Spec.PortalIP))
err = rest.CheckGeneratedNameError(rest.Services, err, service)
return nil, err
}
return rs.registry.GetService(ctx, service.Name)
return out, err
}
func hostsFromMinionList(list *api.NodeList) []string {
@ -213,11 +213,7 @@ func (rs *REST) Update(ctx api.Context, obj runtime.Object) (runtime.Object, boo
}
}
}
err = rs.registry.UpdateService(ctx, service)
if err != nil {
return nil, false, err
}
out, err := rs.registry.GetService(ctx, service.Name)
out, err := rs.registry.UpdateService(ctx, service)
return out, false, err
}

64
pkg/tools/etcd_tools.go
View File

@ -271,25 +271,9 @@ func (h *EtcdHelper) extractObj(response *etcd.Response, inErr error, objPtr run
}
// CreateObj adds a new object at a key unless it already exists. 'ttl' is time-to-live in seconds,
// and 0 means forever.
func (h *EtcdHelper) CreateObj(key string, obj runtime.Object, ttl uint64) error {
data, err := h.Codec.Encode(obj)
if err != nil {
return err
}
if h.ResourceVersioner != nil {
if version, err := h.ResourceVersioner.ResourceVersion(obj); err == nil && version != 0 {
return errors.New("resourceVersion may not be set on objects to be created")
}
}
_, err = h.Client.Create(key, string(data), ttl)
return err
}
// Create adds a new object at a key unless it already exists. 'ttl' is time-to-live in seconds,
// and 0 means forever. If no error is returned, out will be set to the read value from etcd.
func (h *EtcdHelper) Create(key string, obj, out runtime.Object, ttl uint64) error {
// and 0 means forever. If no error is returned and out is not nil, out will be set to the read value
// from etcd.
func (h *EtcdHelper) CreateObj(key string, obj, out runtime.Object, ttl uint64) error {
data, err := h.Codec.Encode(obj)
if err != nil {
return err
@ -303,10 +287,12 @@ func (h *EtcdHelper) Create(key string, obj, out runtime.Object, ttl uint64) err
if err != nil {
return err
}
if _, err := conversion.EnforcePtr(out); err != nil {
panic("unable to convert output object to pointer")
if out != nil {
if _, err := conversion.EnforcePtr(out); err != nil {
panic("unable to convert output object to pointer")
}
_, _, err = h.extractObj(response, err, out, false, false)
}
_, _, err = h.extractObj(response, err, out, false, false)
return err
}
@ -332,21 +318,41 @@ func (h *EtcdHelper) DeleteObj(key string, out runtime.Object) error {
}
// SetObj marshals obj via json, and stores under key. Will do an atomic update if obj's ResourceVersion
// field is set. 'ttl' is time-to-live in seconds, and 0 means forever.
func (h *EtcdHelper) SetObj(key string, obj runtime.Object, ttl uint64) error {
// field is set. 'ttl' is time-to-live in seconds, and 0 means forever. If no error is returned and out is
//not nil, out will be set to the read value from etcd.
func (h *EtcdHelper) SetObj(key string, obj, out runtime.Object, ttl uint64) error {
var response *etcd.Response
data, err := h.Codec.Encode(obj)
if err != nil {
return err
}
create := true
if h.ResourceVersioner != nil {
if version, err := h.ResourceVersioner.ResourceVersion(obj); err == nil && version != 0 {
_, err = h.Client.CompareAndSwap(key, string(data), ttl, "", version)
return err // err is shadowed!
version, err := h.ResourceVersioner.ResourceVersion(obj)
if err == nil && version != 0 {
create = false
response, err = h.Client.CompareAndSwap(key, string(data), ttl, "", version)
}
}
if err != nil {
return err
}
if create {
// Create will fail if a key already exists.
response, err = h.Client.Create(key, string(data), ttl)
}
if err != nil {
return err
}
if out != nil {
if _, err := conversion.EnforcePtr(out); err != nil {
panic("unable to convert output object to pointer")
}
_, _, err = h.extractObj(response, err, out, false, false)
}
// Create will fail if a key already exists.
_, err = h.Client.Create(key, string(data), ttl)
return err
}

45
pkg/tools/etcd_tools_test.go
View File

@ -354,7 +354,8 @@ func TestCreateObj(t *testing.T) {
obj := &api.Pod{ObjectMeta: api.ObjectMeta{Name: "foo"}}
fakeClient := NewFakeEtcdClient(t)
helper := EtcdHelper{fakeClient, testapi.Codec(), versioner}
err := helper.CreateObj("/some/key", obj, 5)
returnedObj := &api.Pod{}
err := helper.CreateObj("/some/key", obj, returnedObj, 5)
if err != nil {
t.Errorf("Unexpected error %#v", err)
}
@ -369,13 +370,27 @@ func TestCreateObj(t *testing.T) {
if e, a := uint64(5), fakeClient.LastSetTTL; e != a {
t.Errorf("Wanted %v, got %v", e, a)
}
if obj.ResourceVersion != returnedObj.ResourceVersion || obj.Name != returnedObj.Name {
t.Errorf("If set was successful but returned object did not have correct resource version")
}
}
func TestCreateObjNilOutParam(t *testing.T) {
obj := &api.Pod{ObjectMeta: api.ObjectMeta{Name: "foo"}}
fakeClient := NewFakeEtcdClient(t)
helper := EtcdHelper{fakeClient, testapi.Codec(), versioner}
err := helper.CreateObj("/some/key", obj, nil, 5)
if err != nil {
t.Errorf("Unexpected error %#v", err)
}
}
func TestSetObj(t *testing.T) {
obj := &api.Pod{ObjectMeta: api.ObjectMeta{Name: "foo"}}
fakeClient := NewFakeEtcdClient(t)
helper := EtcdHelper{fakeClient, testapi.Codec(), versioner}
err := helper.SetObj("/some/key", obj, 5)
returnedObj := &api.Pod{}
err := helper.SetObj("/some/key", obj, returnedObj, 5)
if err != nil {
t.Errorf("Unexpected error %#v", err)
}
@ -391,7 +406,9 @@ func TestSetObj(t *testing.T) {
if e, a := uint64(5), fakeClient.LastSetTTL; e != a {
t.Errorf("Wanted %v, got %v", e, a)
}
if obj.ResourceVersion != returnedObj.ResourceVersion || obj.Name != returnedObj.Name {
t.Errorf("If set was successful but returned object did not have correct resource version")
}
}
func TestSetObjWithVersion(t *testing.T) {
@ -408,7 +425,8 @@ func TestSetObjWithVersion(t *testing.T) {
}
helper := EtcdHelper{fakeClient, testapi.Codec(), versioner}
err := helper.SetObj("/some/key", obj, 7)
returnedObj := &api.Pod{}
err := helper.SetObj("/some/key", obj, returnedObj, 7)
if err != nil {
t.Fatalf("Unexpected error %#v", err)
}
@ -424,13 +442,17 @@ func TestSetObjWithVersion(t *testing.T) {
if e, a := uint64(7), fakeClient.LastSetTTL; e != a {
t.Errorf("Wanted %v, got %v", e, a)
}
if obj.ResourceVersion != returnedObj.ResourceVersion || obj.Name != returnedObj.Name {
t.Errorf("If set was successful but returned object did not have correct resource version")
}
}
func TestSetObjWithoutResourceVersioner(t *testing.T) {
obj := &api.Pod{ObjectMeta: api.ObjectMeta{Name: "foo"}}
fakeClient := NewFakeEtcdClient(t)
helper := EtcdHelper{fakeClient, testapi.Codec(), nil}
err := helper.SetObj("/some/key", obj, 3)
returnedObj := &api.Pod{}
err := helper.SetObj("/some/key", obj, returnedObj, 3)
if err != nil {
t.Errorf("Unexpected error %#v", err)
}
@ -446,6 +468,19 @@ func TestSetObjWithoutResourceVersioner(t *testing.T) {
if e, a := uint64(3), fakeClient.LastSetTTL; e != a {
t.Errorf("Wanted %v, got %v", e, a)
}
if obj.ResourceVersion != returnedObj.ResourceVersion || obj.Name != returnedObj.Name {
t.Errorf("If set was successful but returned object did not have correct resource version")
}
}
func TestSetObjNilOutParam(t *testing.T) {
obj := &api.Pod{ObjectMeta: api.ObjectMeta{Name: "foo"}}
fakeClient := NewFakeEtcdClient(t)
helper := EtcdHelper{fakeClient, testapi.Codec(), nil}
err := helper.SetObj("/some/key", obj, nil, 3)
if err != nil {
t.Errorf("Unexpected error %#v", err)
}
}
func TestAtomicUpdate(t *testing.T) {

2
test/integration/etcd_tools_test.go
View File

@ -60,7 +60,7 @@ func TestSetObj(t *testing.T) {
helper := tools.EtcdHelper{Client: client, Codec: stringCodec{}}
withEtcdKey(func(key string) {
fakeObject := fakeAPIObject("object")
if err := helper.SetObj(key, &fakeObject, 0 /* ttl */); err != nil {
if err := helper.SetObj(key, &fakeObject, nil, 0); err != nil {
t.Fatalf("unexpected error: %v", err)
}
resp, err := client.Get(key, false, false)