From b0679f18bc678de463a803a2786de065d74b57cd Mon Sep 17 00:00:00 2001 From: Ananya Kumar Date: Thu, 6 Aug 2015 22:36:39 -0700 Subject: [PATCH] Add registry code --- pkg/registry/daemon/doc.go | 19 + pkg/registry/daemon/etcd/etcd.go | 76 +++ pkg/registry/daemon/etcd/etcd_test.go | 721 ++++++++++++++++++++++++++ pkg/registry/daemon/rest.go | 124 +++++ pkg/registry/generic/etcd/etcd.go | 2 +- 5 files changed, 941 insertions(+), 1 deletion(-) create mode 100644 pkg/registry/daemon/doc.go create mode 100644 pkg/registry/daemon/etcd/etcd.go create mode 100755 pkg/registry/daemon/etcd/etcd_test.go create mode 100644 pkg/registry/daemon/rest.go diff --git a/pkg/registry/daemon/doc.go b/pkg/registry/daemon/doc.go new file mode 100644 index 00000000000..abfc57ec013 --- /dev/null +++ b/pkg/registry/daemon/doc.go @@ -0,0 +1,19 @@ +/* +Copyright 2015 The Kubernetes Authors All rights reserved. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +// Package daemon provides Registry interface and it's RESTStorage +// implementation for storing Daemon api objects. +package daemon diff --git a/pkg/registry/daemon/etcd/etcd.go b/pkg/registry/daemon/etcd/etcd.go new file mode 100644 index 00000000000..1887be2e623 --- /dev/null +++ b/pkg/registry/daemon/etcd/etcd.go @@ -0,0 +1,76 @@ +/* +Copyright 2015 The Kubernetes Authors All rights reserved. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package etcd + +import ( + "k8s.io/kubernetes/pkg/api" + "k8s.io/kubernetes/pkg/fields" + "k8s.io/kubernetes/pkg/labels" + "k8s.io/kubernetes/pkg/registry/daemon" + "k8s.io/kubernetes/pkg/registry/generic" + etcdgeneric "k8s.io/kubernetes/pkg/registry/generic/etcd" + "k8s.io/kubernetes/pkg/runtime" + "k8s.io/kubernetes/pkg/storage" +) + +// rest implements a RESTStorage for daemons against etcd +type REST struct { + *etcdgeneric.Etcd +} + +// daemonPrefix is the location for daemons in etcd, only exposed +// for testing +var daemonPrefix = "/daemons" + +// NewREST returns a RESTStorage object that will work against daemons. +func NewREST(s storage.Interface) *REST { + store := &etcdgeneric.Etcd{ + NewFunc: func() runtime.Object { return &api.Daemon{} }, + + // NewListFunc returns an object capable of storing results of an etcd list. + NewListFunc: func() runtime.Object { return &api.DaemonList{} }, + // Produces a path that etcd understands, to the root of the resource + // by combining the namespace in the context with the given prefix + KeyRootFunc: func(ctx api.Context) string { + return etcdgeneric.NamespaceKeyRootFunc(ctx, daemonPrefix) + }, + // Produces a path that etcd understands, to the resource by combining + // the namespace in the context with the given prefix + KeyFunc: func(ctx api.Context, name string) (string, error) { + return etcdgeneric.NamespaceKeyFunc(ctx, daemonPrefix, name) + }, + // Retrieve the name field of a daemon + ObjectNameFunc: func(obj runtime.Object) (string, error) { + return obj.(*api.Daemon).Name, nil + }, + // Used to match objects based on labels/fields for list and watch + PredicateFunc: func(label labels.Selector, field fields.Selector) generic.Matcher { + return daemon.MatchDaemon(label, field) + }, + EndpointName: "daemons", + + // Used to validate daemon creation + CreateStrategy: daemon.Strategy, + + // Used to validate daemon updates + UpdateStrategy: daemon.Strategy, + + Storage: s, + } + + return &REST{store} +} diff --git a/pkg/registry/daemon/etcd/etcd_test.go b/pkg/registry/daemon/etcd/etcd_test.go new file mode 100755 index 00000000000..ed0b7eabe39 --- /dev/null +++ b/pkg/registry/daemon/etcd/etcd_test.go @@ -0,0 +1,721 @@ +/* +Copyright 2014 The Kubernetes Authors All rights reserved. +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + http://www.apache.org/licenses/LICENSE-2.0 +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package etcd + +import ( + "strings" + "testing" + "time" + + "github.com/coreos/go-etcd/etcd" + "k8s.io/kubernetes/pkg/api" + "k8s.io/kubernetes/pkg/api/errors" + "k8s.io/kubernetes/pkg/api/latest" + "k8s.io/kubernetes/pkg/api/rest/resttest" + "k8s.io/kubernetes/pkg/fields" + "k8s.io/kubernetes/pkg/labels" + etcdgeneric "k8s.io/kubernetes/pkg/registry/generic/etcd" + "k8s.io/kubernetes/pkg/runtime" + "k8s.io/kubernetes/pkg/storage" + etcdstorage "k8s.io/kubernetes/pkg/storage/etcd" + "k8s.io/kubernetes/pkg/tools" + "k8s.io/kubernetes/pkg/tools/etcdtest" +) + +const ( + PASS = iota + FAIL +) + +func newEtcdStorage(t *testing.T) (*tools.FakeEtcdClient, storage.Interface) { + fakeEtcdClient := tools.NewFakeEtcdClient(t) + fakeEtcdClient.TestIndex = true + helper := etcdstorage.NewEtcdStorage(fakeEtcdClient, latest.Codec, etcdtest.PathPrefix()) + return fakeEtcdClient, helper +} + +// newStorage creates a REST storage backed by etcd helpers +func newStorage(t *testing.T) (*REST, *tools.FakeEtcdClient) { + fakeEtcdClient, h := newEtcdStorage(t) + storage := NewREST(h) + return storage, fakeEtcdClient +} + +// createController is a helper function that returns a controller with the updated resource version. +func createController(storage *REST, dc api.Daemon, t *testing.T) (api.Daemon, error) { + ctx := api.WithNamespace(api.NewContext(), dc.Namespace) + obj, err := storage.Create(ctx, &dc) + if err != nil { + t.Errorf("Failed to create controller, %v", err) + } + newDc := obj.(*api.Daemon) + return *newDc, nil +} + +var validPodTemplate = api.PodTemplate{ + Template: api.PodTemplateSpec{ + ObjectMeta: api.ObjectMeta{ + Labels: map[string]string{"a": "b"}, + }, + Spec: api.PodSpec{ + Containers: []api.Container{ + { + Name: "test", + Image: "test_image", + ImagePullPolicy: api.PullIfNotPresent, + }, + }, + RestartPolicy: api.RestartPolicyAlways, + DNSPolicy: api.DNSClusterFirst, + }, + }, +} + +var validControllerSpec = api.DaemonSpec{ + Selector: validPodTemplate.Template.Labels, + Template: &validPodTemplate.Template, +} + +var validController = api.Daemon{ + ObjectMeta: api.ObjectMeta{Name: "foo", Namespace: "default"}, + Spec: validControllerSpec, +} + +// makeControllerKey constructs etcd paths to controller items enforcing namespace rules. +func makeControllerKey(ctx api.Context, id string) (string, error) { + return etcdgeneric.NamespaceKeyFunc(ctx, daemonPrefix, id) +} + +// makeControllerListKey constructs etcd paths to the root of the resource, +// not a specific controller resource +func makeControllerListKey(ctx api.Context) string { + return etcdgeneric.NamespaceKeyRootFunc(ctx, daemonPrefix) +} + +func TestEtcdCreateController(t *testing.T) { + ctx := api.NewDefaultContext() + storage, fakeClient := newStorage(t) + _, err := storage.Create(ctx, &validController) + if err != nil { + t.Errorf("unexpected error: %v", err) + } + key, _ := makeControllerKey(ctx, validController.Name) + key = etcdtest.AddPrefix(key) + resp, err := fakeClient.Get(key, false, false) + if err != nil { + t.Fatalf("Unexpected error %v", err) + } + var ctrl api.Daemon + err = latest.Codec.DecodeInto([]byte(resp.Node.Value), &ctrl) + if err != nil { + t.Errorf("unexpected error: %v", err) + } + + if ctrl.Name != "foo" { + t.Errorf("Unexpected controller: %#v %s", ctrl, resp.Node.Value) + } +} + +func TestEtcdCreateControllerAlreadyExisting(t *testing.T) { + ctx := api.NewDefaultContext() + storage, fakeClient := newStorage(t) + key, _ := makeControllerKey(ctx, validController.Name) + key = etcdtest.AddPrefix(key) + fakeClient.Set(key, runtime.EncodeOrDie(latest.Codec, &validController), 0) + + _, err := storage.Create(ctx, &validController) + if !errors.IsAlreadyExists(err) { + t.Errorf("expected already exists err, got %#v", err) + } +} + +func TestEtcdCreateControllerValidates(t *testing.T) { + ctx := api.NewDefaultContext() + storage, _ := newStorage(t) + emptyName := validController + emptyName.Name = "" + failureCases := []api.Daemon{emptyName} + for _, failureCase := range failureCases { + c, err := storage.Create(ctx, &failureCase) + if c != nil { + t.Errorf("Expected nil channel") + } + if !errors.IsInvalid(err) { + t.Errorf("Expected to get an invalid resource error, got %v", err) + } + } +} + +func TestCreateControllerWithGeneratedName(t *testing.T) { + storage, _ := newStorage(t) + controller := &api.Daemon{ + ObjectMeta: api.ObjectMeta{ + Namespace: api.NamespaceDefault, + GenerateName: "daemon-", + }, + Spec: api.DaemonSpec{ + Selector: map[string]string{"a": "b"}, + Template: &validPodTemplate.Template, + }, + } + + ctx := api.NewDefaultContext() + _, err := storage.Create(ctx, controller) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if controller.Name == "daemon-" || !strings.HasPrefix(controller.Name, "daemon-") { + t.Errorf("unexpected name: %#v", controller) + } +} + +func TestCreateControllerWithConflictingNamespace(t *testing.T) { + storage, _ := newStorage(t) + controller := &api.Daemon{ + ObjectMeta: api.ObjectMeta{Name: "test", Namespace: "not-default"}, + } + + ctx := api.NewDefaultContext() + channel, err := storage.Create(ctx, controller) + if channel != nil { + t.Error("Expected a nil channel, but we got a value") + } + errSubString := "namespace" + if err == nil { + t.Errorf("Expected an error, but we didn't get one") + } else if !errors.IsBadRequest(err) || + strings.Index(err.Error(), errSubString) == -1 { + t.Errorf("Expected a Bad Request error with the sub string '%s', got %v", errSubString, err) + } +} + +func TestEtcdGetController(t *testing.T) { + ctx := api.NewDefaultContext() + storage, fakeClient := newStorage(t) + key, _ := makeControllerKey(ctx, validController.Name) + key = etcdtest.AddPrefix(key) + fakeClient.Set(key, runtime.EncodeOrDie(latest.Codec, &validController), 0) + ctrl, err := storage.Get(ctx, validController.Name) + if err != nil { + t.Errorf("unexpected error: %v", err) + } + controller, ok := ctrl.(*api.Daemon) + if !ok { + t.Errorf("Expected a controller, got %#v", ctrl) + } + if controller.Name != validController.Name { + t.Errorf("Unexpected controller: %#v", controller) + } +} + +func TestEtcdControllerValidatesUpdate(t *testing.T) { + ctx := api.NewDefaultContext() + storage, _ := newStorage(t) + + updateController, err := createController(storage, validController, t) + if err != nil { + t.Errorf("Failed to create controller, cannot proceed with test.") + } + + updaters := []func(dc api.Daemon) (runtime.Object, bool, error){ + func(dc api.Daemon) (runtime.Object, bool, error) { + dc.UID = "newUID" + return storage.Update(ctx, &dc) + }, + func(dc api.Daemon) (runtime.Object, bool, error) { + dc.Name = "" + return storage.Update(ctx, &dc) + }, + func(dc api.Daemon) (runtime.Object, bool, error) { + dc.Spec.Template.Spec.RestartPolicy = api.RestartPolicyOnFailure + return storage.Update(ctx, &dc) + }, + func(dc api.Daemon) (runtime.Object, bool, error) { + dc.Spec.Selector = map[string]string{} + return storage.Update(ctx, &dc) + }, + } + for _, u := range updaters { + c, updated, err := u(updateController) + if c != nil || updated { + t.Errorf("Expected nil object and not created") + } + if !errors.IsInvalid(err) && !errors.IsBadRequest(err) { + t.Errorf("Expected invalid or bad request error, got %v of type %T", err, err) + } + } +} + +func TestEtcdControllerValidatesNamespaceOnUpdate(t *testing.T) { + storage, _ := newStorage(t) + ns := "newnamespace" + + // The update should fail if the namespace on the controller is set to something + // other than the namespace on the given context, even if the namespace on the + // controller is valid. + updateController, err := createController(storage, validController, t) + + newNamespaceController := validController + newNamespaceController.Namespace = ns + _, err = createController(storage, newNamespaceController, t) + + c, updated, err := storage.Update(api.WithNamespace(api.NewContext(), ns), &updateController) + if c != nil || updated { + t.Errorf("Expected nil object and not created") + } + // TODO: Be more paranoid about the type of error and make sure it has the substring + // "namespace" in it, once #5684 is fixed. Ideally this would be a NewBadRequest. + if err == nil { + t.Errorf("Expected an error, but we didn't get one") + } +} + +// TestEtcdGetControllerDifferentNamespace ensures same-name controllers in different namespaces do not clash +func TestEtcdGetControllerDifferentNamespace(t *testing.T) { + storage, fakeClient := newStorage(t) + + otherNs := "other" + ctx1 := api.NewDefaultContext() + ctx2 := api.WithNamespace(api.NewContext(), otherNs) + + key1, _ := makeControllerKey(ctx1, validController.Name) + key2, _ := makeControllerKey(ctx2, validController.Name) + + key1 = etcdtest.AddPrefix(key1) + key2 = etcdtest.AddPrefix(key2) + + fakeClient.Set(key1, runtime.EncodeOrDie(latest.Codec, &validController), 0) + otherNsController := validController + otherNsController.Namespace = otherNs + fakeClient.Set(key2, runtime.EncodeOrDie(latest.Codec, &otherNsController), 0) + + obj, err := storage.Get(ctx1, validController.Name) + ctrl1, _ := obj.(*api.Daemon) + if err != nil { + t.Errorf("unexpected error: %v", err) + } + if ctrl1.Name != "foo" { + t.Errorf("Unexpected controller: %#v", ctrl1) + } + if ctrl1.Namespace != "default" { + t.Errorf("Unexpected controller: %#v", ctrl1) + } + + obj, err = storage.Get(ctx2, validController.Name) + ctrl2, _ := obj.(*api.Daemon) + if err != nil { + t.Errorf("unexpected error: %v", err) + } + if ctrl2.Name != "foo" { + t.Errorf("Unexpected controller: %#v", ctrl2) + } + if ctrl2.Namespace != "other" { + t.Errorf("Unexpected controller: %#v", ctrl2) + } + +} + +func TestEtcdGetControllerNotFound(t *testing.T) { + ctx := api.NewDefaultContext() + storage, fakeClient := newStorage(t) + key, _ := makeControllerKey(ctx, validController.Name) + key = etcdtest.AddPrefix(key) + + fakeClient.Data[key] = tools.EtcdResponseWithError{ + R: &etcd.Response{ + Node: nil, + }, + E: tools.EtcdErrorNotFound, + } + ctrl, err := storage.Get(ctx, validController.Name) + if ctrl != nil { + t.Errorf("Unexpected non-nil controller: %#v", ctrl) + } + if !errors.IsNotFound(err) { + t.Errorf("Unexpected error returned: %#v", err) + } +} + +func TestEtcdDeleteController(t *testing.T) { + ctx := api.NewDefaultContext() + storage, fakeClient := newStorage(t) + key, _ := makeControllerKey(ctx, validController.Name) + key = etcdtest.AddPrefix(key) + + fakeClient.Set(key, runtime.EncodeOrDie(latest.Codec, &validController), 0) + obj, err := storage.Delete(ctx, validController.Name, nil) + if err != nil { + t.Errorf("unexpected error: %v", err) + } + if status, ok := obj.(*api.Status); !ok { + t.Errorf("Expected status of delete, got %#v", status) + } else if status.Status != api.StatusSuccess { + t.Errorf("Expected success, got %#v", status.Status) + } + if len(fakeClient.DeletedKeys) != 1 { + t.Errorf("Expected 1 delete, found %#v", fakeClient.DeletedKeys) + } + if fakeClient.DeletedKeys[0] != key { + t.Errorf("Unexpected key: %s, expected %s", fakeClient.DeletedKeys[0], key) + } +} + +func TestEtcdListControllers(t *testing.T) { + storage, fakeClient := newStorage(t) + ctx := api.NewDefaultContext() + key := makeControllerListKey(ctx) + key = etcdtest.AddPrefix(key) + controller := validController + controller.Name = "bar" + fakeClient.Data[key] = tools.EtcdResponseWithError{ + R: &etcd.Response{ + Node: &etcd.Node{ + Nodes: []*etcd.Node{ + { + Value: runtime.EncodeOrDie(latest.Codec, &validController), + }, + { + Value: runtime.EncodeOrDie(latest.Codec, &controller), + }, + }, + }, + }, + E: nil, + } + objList, err := storage.List(ctx, labels.Everything(), fields.Everything()) + if err != nil { + t.Errorf("unexpected error: %v", err) + } + controllers, _ := objList.(*api.DaemonList) + if len(controllers.Items) != 2 || controllers.Items[0].Name != validController.Name || controllers.Items[1].Name != controller.Name { + t.Errorf("Unexpected controller list: %#v", controllers) + } +} + +func TestEtcdListControllersNotFound(t *testing.T) { + storage, fakeClient := newStorage(t) + ctx := api.NewDefaultContext() + key := makeControllerListKey(ctx) + key = etcdtest.AddPrefix(key) + + fakeClient.Data[key] = tools.EtcdResponseWithError{ + R: &etcd.Response{}, + E: tools.EtcdErrorNotFound, + } + objList, err := storage.List(ctx, labels.Everything(), fields.Everything()) + if err != nil { + t.Errorf("unexpected error: %v", err) + } + controllers, _ := objList.(*api.DaemonList) + if len(controllers.Items) != 0 { + t.Errorf("Unexpected controller list: %#v", controllers) + } +} + +func TestEtcdListControllersLabelsMatch(t *testing.T) { + storage, fakeClient := newStorage(t) + ctx := api.NewDefaultContext() + key := makeControllerListKey(ctx) + key = etcdtest.AddPrefix(key) + + controller := validController + controller.Labels = map[string]string{"k": "v"} + controller.Name = "bar" + + fakeClient.Data[key] = tools.EtcdResponseWithError{ + R: &etcd.Response{ + Node: &etcd.Node{ + Nodes: []*etcd.Node{ + { + Value: runtime.EncodeOrDie(latest.Codec, &validController), + }, + { + Value: runtime.EncodeOrDie(latest.Codec, &controller), + }, + }, + }, + }, + E: nil, + } + testLabels := labels.SelectorFromSet(labels.Set(controller.Labels)) + objList, err := storage.List(ctx, testLabels, fields.Everything()) + if err != nil { + t.Errorf("unexpected error: %v", err) + } + controllers, _ := objList.(*api.DaemonList) + if len(controllers.Items) != 1 || controllers.Items[0].Name != controller.Name || + !testLabels.Matches(labels.Set(controllers.Items[0].Labels)) { + t.Errorf("Unexpected controller list: %#v for query with labels %#v", + controllers, testLabels) + } +} + +func TestEtcdWatchController(t *testing.T) { + ctx := api.NewDefaultContext() + storage, fakeClient := newStorage(t) + watching, err := storage.Watch(ctx, + labels.Everything(), + fields.Everything(), + "1", + ) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + fakeClient.WaitForWatchCompletion() + + select { + case _, ok := <-watching.ResultChan(): + if !ok { + t.Errorf("watching channel should be open") + } + default: + } + fakeClient.WatchInjectError <- nil + if _, ok := <-watching.ResultChan(); ok { + t.Errorf("watching channel should be closed") + } + watching.Stop() +} + +// Tests that we can watch for the creation of daemon controllers with specified labels. +func TestEtcdWatchControllersMatch(t *testing.T) { + ctx := api.WithNamespace(api.NewDefaultContext(), validController.Namespace) + storage, fakeClient := newStorage(t) + fakeClient.ExpectNotFoundGet(etcdgeneric.NamespaceKeyRootFunc(ctx, "/registry/pods")) + + watching, err := storage.Watch(ctx, + labels.SelectorFromSet(validController.Spec.Selector), + fields.Everything(), + "1", + ) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + fakeClient.WaitForWatchCompletion() + + // The watcher above is waiting for these Labels, on receiving them it should + // apply the ControllerStatus decorator, which lists pods, causing a query against + // the /registry/pods endpoint of the etcd client. + controller := &api.Daemon{ + ObjectMeta: api.ObjectMeta{ + Name: "foo", + Labels: validController.Spec.Selector, + Namespace: "default", + }, + } + controllerBytes, _ := latest.Codec.Encode(controller) + fakeClient.WatchResponse <- &etcd.Response{ + Action: "create", + Node: &etcd.Node{ + Value: string(controllerBytes), + }, + } + select { + case _, ok := <-watching.ResultChan(): + if !ok { + t.Errorf("watching channel should be open") + } + case <-time.After(time.Millisecond * 100): + t.Error("unexpected timeout from result channel") + } + watching.Stop() +} + +// Tests that we can watch for daemon controllers with specified fields. +func TestEtcdWatchControllersFields(t *testing.T) { + ctx := api.WithNamespace(api.NewDefaultContext(), validController.Namespace) + storage, fakeClient := newStorage(t) + fakeClient.ExpectNotFoundGet(etcdgeneric.NamespaceKeyRootFunc(ctx, "/registry/pods")) + + testFieldMap := map[int][]fields.Set{ + PASS: { + {"status.currentNumberScheduled": "2"}, + {"status.numberMisscheduled": "1"}, + {"status.desiredNumberScheduled": "4"}, + {"metadata.name": "foo"}, + {"status.currentNumberScheduled": "2", "status.numberMisscheduled": "1"}, + {"status.currentNumberScheduled": "2", "status.desiredNumberScheduled": "4"}, + {"status.currentNumberScheduled": "2", "metadata.name": "foo"}, + {"status.desiredNumberScheduled": "4", "metadata.name": "foo"}, + {"status.currentNumberScheduled": "2", "status.desiredNumberScheduled": "4", "metadata.name": "foo"}, + {"status.currentNumberScheduled": "2", "status.numberMisscheduled": "1", "status.desiredNumberScheduled": "4"}, + {"status.currentNumberScheduled": "2", "status.numberMisscheduled": "1", "status.desiredNumberScheduled": "4", "metadata.name": "foo"}, + }, + FAIL: { + {"status.currentNumberScheduled": "1"}, + {"status.numberMisscheduled": "0"}, + {"status.desiredNumberScheduled": "5"}, + {"metadata.name": "bar"}, + {"name": "foo"}, + {"status.replicas": "0"}, + {"status.currentNumberScheduled": "2", "status.desiredNumberScheduled": "3"}, + {"status.numberMisscheduled": "3", "status.desiredNumberScheduled": "5"}, + {"status.currentNumberScheduled": "2", "status.desiredNumberScheduled": "4", "metadata.name": "foox"}, + }, + } + testEtcdActions := []string{ + etcdstorage.EtcdCreate, + etcdstorage.EtcdSet, + etcdstorage.EtcdCAS, + etcdstorage.EtcdDelete} + + controller := &api.Daemon{ + ObjectMeta: api.ObjectMeta{ + Name: "foo", + Labels: validController.Spec.Selector, + Namespace: "default", + }, + Status: api.DaemonStatus{ + CurrentNumberScheduled: 2, + NumberMisscheduled: 1, + DesiredNumberScheduled: 4, + }, + } + controllerBytes, _ := latest.Codec.Encode(controller) + + for expectedResult, fieldSet := range testFieldMap { + for _, field := range fieldSet { + for _, action := range testEtcdActions { + watching, err := storage.Watch(ctx, + labels.Everything(), + field.AsSelector(), + "1", + ) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + var prevNode *etcd.Node = nil + node := &etcd.Node{ + Value: string(controllerBytes), + } + if action == etcdstorage.EtcdDelete { + prevNode = node + } + fakeClient.WaitForWatchCompletion() + fakeClient.WatchResponse <- &etcd.Response{ + Action: action, + Node: node, + PrevNode: prevNode, + } + + select { + case r, ok := <-watching.ResultChan(): + if expectedResult == FAIL { + t.Errorf("Unexpected result from channel %#v", r) + } + if !ok { + t.Errorf("watching channel should be open") + } + case <-time.After(time.Millisecond * 100): + if expectedResult == PASS { + t.Error("unexpected timeout from result channel") + } + } + watching.Stop() + } + } + } +} + +func TestEtcdWatchControllersNotMatch(t *testing.T) { + ctx := api.NewDefaultContext() + storage, fakeClient := newStorage(t) + fakeClient.ExpectNotFoundGet(etcdgeneric.NamespaceKeyRootFunc(ctx, "/registry/pods")) + + watching, err := storage.Watch(ctx, + labels.SelectorFromSet(labels.Set{"name": "foo"}), + fields.Everything(), + "1", + ) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + fakeClient.WaitForWatchCompletion() + + controller := &api.Daemon{ + ObjectMeta: api.ObjectMeta{ + Name: "bar", + Labels: map[string]string{ + "name": "bar", + }, + }, + } + controllerBytes, _ := latest.Codec.Encode(controller) + fakeClient.WatchResponse <- &etcd.Response{ + Action: "create", + Node: &etcd.Node{ + Value: string(controllerBytes), + }, + } + + select { + case <-watching.ResultChan(): + t.Error("unexpected result from result channel") + case <-time.After(time.Millisecond * 100): + // expected case + } +} + +func TestCreate(t *testing.T) { + storage, fakeClient := newStorage(t) + test := resttest.New(t, storage, fakeClient.SetError) + test.TestCreate( + // valid + &api.Daemon{ + Spec: api.DaemonSpec{ + Selector: map[string]string{"a": "b"}, + Template: &validPodTemplate.Template, + }, + }, + // invalid + &api.Daemon{ + Spec: api.DaemonSpec{ + Selector: map[string]string{}, + Template: &validPodTemplate.Template, + }, + }, + ) +} + +func TestDelete(t *testing.T) { + ctx := api.NewDefaultContext() + storage, fakeClient := newStorage(t) + test := resttest.New(t, storage, fakeClient.SetError) + key, _ := makeControllerKey(ctx, validController.Name) + key = etcdtest.AddPrefix(key) + + createFn := func() runtime.Object { + dc := validController + dc.ResourceVersion = "1" + fakeClient.Data[key] = tools.EtcdResponseWithError{ + R: &etcd.Response{ + Node: &etcd.Node{ + Value: runtime.EncodeOrDie(latest.Codec, &dc), + ModifiedIndex: 1, + }, + }, + } + return &dc + } + gracefulSetFn := func() bool { + // If the controller is still around after trying to delete either the delete + // failed, or we're deleting it gracefully. + if fakeClient.Data[key].R.Node != nil { + return true + } + return false + } + + test.TestDelete(createFn, gracefulSetFn) +} diff --git a/pkg/registry/daemon/rest.go b/pkg/registry/daemon/rest.go new file mode 100644 index 00000000000..31b3409c365 --- /dev/null +++ b/pkg/registry/daemon/rest.go @@ -0,0 +1,124 @@ +/* +Copyright 2015 The Kubernetes Authors All rights reserved. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package daemon + +import ( + "fmt" + "reflect" + "strconv" + + "k8s.io/kubernetes/pkg/api" + "k8s.io/kubernetes/pkg/api/validation" + "k8s.io/kubernetes/pkg/fields" + "k8s.io/kubernetes/pkg/labels" + "k8s.io/kubernetes/pkg/registry/generic" + "k8s.io/kubernetes/pkg/runtime" + "k8s.io/kubernetes/pkg/util/fielderrors" +) + +// daemonStrategy implements verification logic for daemons. +type daemonStrategy struct { + runtime.ObjectTyper + api.NameGenerator +} + +// Strategy is the default logic that applies when creating and updating Daemon objects. +var Strategy = daemonStrategy{api.Scheme, api.SimpleNameGenerator} + +// NamespaceScoped returns true because all Daemons need to be within a namespace. +func (daemonStrategy) NamespaceScoped() bool { + return true +} + +// PrepareForCreate clears the status of a daemon before creation. +func (daemonStrategy) PrepareForCreate(obj runtime.Object) { + daemon := obj.(*api.Daemon) + daemon.Status = api.DaemonStatus{} + + daemon.Generation = 1 +} + +// PrepareForUpdate clears fields that are not allowed to be set by end users on update. +func (daemonStrategy) PrepareForUpdate(obj, old runtime.Object) { + newDaemon := obj.(*api.Daemon) + oldDaemon := old.(*api.Daemon) + + // Any changes to the spec increment the generation number, any changes to the + // status should reflect the generation number of the corresponding object. We push + // the burden of managing the status onto the clients because we can't (in general) + // know here what version of spec the writer of the status has seen. It may seem like + // we can at first -- since obj contains spec -- but in the future we will probably make + // status its own object, and even if we don't, writes may be the result of a + // read-update-write loop, so the contents of spec may not actually be the spec that + // the controller has *seen*. + // + // TODO: Any changes to a part of the object that represents desired state (labels, + // annotations etc) should also increment the generation. + if !reflect.DeepEqual(oldDaemon.Spec, newDaemon.Spec) { + newDaemon.Generation = oldDaemon.Generation + 1 + } +} + +// Validate validates a new daemon. +func (daemonStrategy) Validate(ctx api.Context, obj runtime.Object) fielderrors.ValidationErrorList { + daemon := obj.(*api.Daemon) + return validation.ValidateDaemon(daemon) +} + +// AllowCreateOnUpdate is false for daemon; this means a POST is +// needed to create one +func (daemonStrategy) AllowCreateOnUpdate() bool { + return false +} + +// ValidateUpdate is the default update validation for an end user. +func (daemonStrategy) ValidateUpdate(ctx api.Context, obj, old runtime.Object) fielderrors.ValidationErrorList { + validationErrorList := validation.ValidateDaemon(obj.(*api.Daemon)) + updateErrorList := validation.ValidateDaemonUpdate(old.(*api.Daemon), obj.(*api.Daemon)) + return append(validationErrorList, updateErrorList...) +} + +func (daemonStrategy) AllowUnconditionalUpdate() bool { + return true +} + +// DaemonToSelectableFields returns a label set that represents the object. +func DaemonToSelectableFields(daemon *api.Daemon) fields.Set { + return fields.Set{ + "metadata.name": daemon.Name, + "status.currentNumberScheduled": strconv.Itoa(daemon.Status.CurrentNumberScheduled), + "status.numberMisscheduled": strconv.Itoa(daemon.Status.NumberMisscheduled), + "status.desiredNumberScheduled": strconv.Itoa(daemon.Status.DesiredNumberScheduled), + } +} + +// MatchDaemon is the filter used by the generic etcd backend to route +// watch events from etcd to clients of the apiserver only interested in specific +// labels/fields. +func MatchDaemon(label labels.Selector, field fields.Selector) generic.Matcher { + return &generic.SelectionPredicate{ + Label: label, + Field: field, + GetAttrs: func(obj runtime.Object) (labels.Set, fields.Set, error) { + daemon, ok := obj.(*api.Daemon) + if !ok { + return nil, nil, fmt.Errorf("given object is not a daemon.") + } + return labels.Set(daemon.ObjectMeta.Labels), DaemonToSelectableFields(daemon), nil + }, + } +} diff --git a/pkg/registry/generic/etcd/etcd.go b/pkg/registry/generic/etcd/etcd.go index 3242d20bc73..b1b39e57051 100644 --- a/pkg/registry/generic/etcd/etcd.go +++ b/pkg/registry/generic/etcd/etcd.go @@ -273,7 +273,7 @@ func (e *Etcd) Update(ctx api.Context, obj runtime.Object) (runtime.Object, bool return nil, nil, err } if newVersion != version { - return nil, nil, kubeerr.NewConflict(e.EndpointName, name, fmt.Errorf("the object has been modified; please apply your changes to the latest version and try again")) + return nil, nil, kubeerr.NewConflict(e.EndpointName, name, fmt.Errorf("the object has been modified; please apply your changes to the latest version and try again %+v, %+v", version, newVersion)) } } if err := rest.BeforeUpdate(e.UpdateStrategy, ctx, obj, existing); err != nil {