Migrate replication controllers to generic etcd

pkg/api/rest/resttest/resttest.go

@@ -24,7 +24,9 @@ import (
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/api/errors"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/api/rest"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/runtime"
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/tools"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/util"
+	"github.com/coreos/go-etcd/etcd"
 )
 
 type Tester struct {
@@ -197,6 +199,44 @@ func (t *Tester) TestCreateRejectsNamespace(valid runtime.Object) {
 	}
 }
 
+func (t *Tester) TestDeleteInvokesValidation(invalid ...runtime.Object) {
+	for i, obj := range invalid {
+		objectMeta, err := api.ObjectMetaFor(obj)
+		if err != nil {
+			t.Fatalf("object does not have ObjectMeta: %v\n%#v", err, obj)
+		}
+		ctx := api.NewDefaultContext()
+		_, err = t.storage.(rest.GracefulDeleter).Delete(ctx, objectMeta.Name, nil)
+		if !errors.IsInvalid(err) {
+			t.Errorf("%d: Expected to get an invalid resource error, got %v", i, err)
+		}
+	}
+}
+
+func (t *Tester) TestDelete(createFn func() runtime.Object, wasGracefulFn func() bool, invalid ...runtime.Object) {
+	t.TestDeleteNonExist(createFn)
+	t.TestDeleteNoGraceful(createFn, wasGracefulFn)
+	t.TestDeleteInvokesValidation(invalid...)
+	// TODO: Test delete namespace mismatch rejection
+	// once #5684 is fixed.
+}
+
+func (t *Tester) TestDeleteNonExist(createFn func() runtime.Object) {
+	existing := createFn()
+	objectMeta, err := api.ObjectMetaFor(existing)
+	if err != nil {
+		t.Fatalf("object does not have ObjectMeta: %v\n%#v", err, existing)
+	}
+	context := api.NewDefaultContext()
+
+	t.withStorageError(&etcd.EtcdError{ErrorCode: tools.EtcdErrorCodeNotFound}, func() {
+		_, err := t.storage.(rest.GracefulDeleter).Delete(context, objectMeta.Name, nil)
+		if err == nil || !errors.IsNotFound(err) {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+	})
+}
+
 func (t *Tester) TestDeleteGraceful(createFn func() runtime.Object, expectedGrace int64, wasGracefulFn func() bool) {
 	t.TestDeleteGracefulHasDefault(createFn(), expectedGrace, wasGracefulFn)
 	t.TestDeleteGracefulUsesZeroOnNil(createFn(), 0)

pkg/controller/replication_controller.go

@@ -58,7 +58,7 @@ type RealPodControl struct {
 }
 
 // Time period of main replication controller sync loop
-const DefaultSyncPeriod = 10 * time.Second
+const DefaultSyncPeriod = 5 * time.Second
 
 func (r RealPodControl) createReplica(namespace string, controller api.ReplicationController) {
 	desiredLabels := make(labels.Set)

pkg/master/master.go

@@ -42,7 +42,7 @@ import (
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/client"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/cloudprovider"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/master/ports"
-	"github.com/GoogleCloudPlatform/kubernetes/pkg/registry/controller"
+	controlleretcd "github.com/GoogleCloudPlatform/kubernetes/pkg/registry/controller/etcd"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/registry/endpoint"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/registry/etcd"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/registry/event"
@@ -384,6 +384,8 @@ func (m *Master) init(c *Config) {
 		podStorage = podStorage.WithPodStatus(podCache)
 	}
 
+	controllerStorage := controlleretcd.NewREST(c.EtcdHelper)
+
 	// TODO: Factor out the core API registration
 	m.storage = map[string]rest.Storage{
 		"pods":         podStorage,
@@ -391,7 +393,7 @@ func (m *Master) init(c *Config) {
 		"pods/binding": bindingStorage,
 		"bindings":     bindingStorage,
 
-		"replicationControllers": controller.NewStorage(registry, podRegistry),
+		"replicationControllers": controllerStorage,
 		"services":               service.NewStorage(m.serviceRegistry, c.Cloud, m.nodeRegistry, m.portalNet, c.ClusterName),
 		"endpoints":              endpoint.NewStorage(m.endpointRegistry),
 		"minions":                nodeStorage,

pkg/registry/controller/etcd/etcd.go

+/*
+Copyright 2014 Google Inc. 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 (
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/fields"
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/labels"
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/registry/controller"
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/registry/generic"
+	etcdgeneric "github.com/GoogleCloudPlatform/kubernetes/pkg/registry/generic/etcd"
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/runtime"
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/tools"
+)
+
+// rest implements a RESTStorage for replication controllers against etcd
+type REST struct {
+	*etcdgeneric.Etcd
+}
+
+// controllerPrefix is the location for controllers in etcd, only exposed
+// for testing
+var controllerPrefix = "/registry/controllers"
+
+// NewREST returns a RESTStorage object that will work against replication controllers.
+func NewREST(h tools.EtcdHelper) *REST {
+	store := &etcdgeneric.Etcd{
+		NewFunc: func() runtime.Object { return &api.ReplicationController{} },
+
+		// NewListFunc returns an object capable of storing results of an etcd list.
+		NewListFunc: func() runtime.Object { return &api.ReplicationControllerList{} },
+		// 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, controllerPrefix)
+		},
+		// 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, controllerPrefix, name)
+		},
+		// Retrieve the name field of a replication controller
+		ObjectNameFunc: func(obj runtime.Object) (string, error) {
+			return obj.(*api.ReplicationController).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 controller.MatchController(label, field)
+		},
+		EndpointName: "replicationControllers",
+
+		// Used to validate controller creation
+		CreateStrategy: controller.Strategy,
+
+		// Used to validate controller updates
+		UpdateStrategy: controller.Strategy,
+
+		Helper: h,
+	}
+
+	return &REST{store}
+}

pkg/registry/controller/registry.go

@@ -17,7 +17,9 @@ limitations under the License.
 package controller
 
 import (
+	"fmt"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/api/rest"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/fields"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/labels"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/watch"
@@ -25,10 +27,66 @@ import (
 
 // Registry is an interface for things that know how to store ReplicationControllers.
 type Registry interface {
-	ListControllers(ctx api.Context) (*api.ReplicationControllerList, error)
+	ListControllers(ctx api.Context, label labels.Selector, field fields.Selector) (*api.ReplicationControllerList, error)
 	WatchControllers(ctx api.Context, label labels.Selector, field fields.Selector, resourceVersion string) (watch.Interface, error)
 	GetController(ctx api.Context, controllerID string) (*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
 }
+
+// storage puts strong typing around storage calls
+type storage struct {
+	rest.StandardStorage
+}
+
+// NewRegistry returns a new Registry interface for the given Storage. Any mismatched
+// types will panic.
+func NewRegistry(s rest.StandardStorage) Registry {
+	return &storage{s}
+}
+
+// List obtains a list of ReplicationControllers that match selector.
+func (s *storage) ListControllers(ctx api.Context, label labels.Selector, field fields.Selector) (*api.ReplicationControllerList, error) {
+	if !field.Empty() {
+		return nil, fmt.Errorf("field selector not supported yet")
+	}
+	obj, err := s.List(ctx, label, field)
+	if err != nil {
+		return nil, err
+	}
+	return obj.(*api.ReplicationControllerList), err
+}
+
+func (s *storage) WatchControllers(ctx api.Context, label labels.Selector, field fields.Selector, resourceVersion string) (watch.Interface, error) {
+	return s.Watch(ctx, label, field, resourceVersion)
+}
+
+func (s *storage) GetController(ctx api.Context, controllerID string) (*api.ReplicationController, error) {
+	obj, err := s.Get(ctx, controllerID)
+	if err != nil {
+		return nil, err
+	}
+	return obj.(*api.ReplicationController), nil
+}
+
+func (s *storage) CreateController(ctx api.Context, controller *api.ReplicationController) (*api.ReplicationController, error) {
+	obj, err := s.Create(ctx, controller)
+	if err != nil {
+		return nil, err
+	}
+	return obj.(*api.ReplicationController), nil
+}
+
+func (s *storage) UpdateController(ctx api.Context, controller *api.ReplicationController) (*api.ReplicationController, error) {
+	obj, _, err := s.Update(ctx, controller)
+	if err != nil {
+		return nil, err
+	}
+	return obj.(*api.ReplicationController), nil
+}
+
+func (s *storage) DeleteController(ctx api.Context, controllerID string) error {
+	_, err := s.Delete(ctx, controllerID, nil)
+	return err
+}

pkg/registry/controller/rest.go

@@ -20,13 +20,12 @@ import (
 	"fmt"
 
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
-	"github.com/GoogleCloudPlatform/kubernetes/pkg/api/rest"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/api/validation"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/fields"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/labels"
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/registry/generic"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/runtime"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/util/fielderrors"
-	"github.com/GoogleCloudPlatform/kubernetes/pkg/watch"
 )
 
 // rcStrategy implements verification logic for Replication Controllers.
@@ -66,108 +65,19 @@ func (rcStrategy) ValidateUpdate(obj, old runtime.Object) fielderrors.Validation
 	return validation.ValidateReplicationControllerUpdate(old.(*api.ReplicationController), obj.(*api.ReplicationController))
 }
 
-// PodLister is anything that knows how to list pods.
-type PodLister interface {
-	ListPods(ctx api.Context, labels labels.Selector) (*api.PodList, error)
-}
-
-// REST implements rest.Storage for the replication controller service.
-type REST struct {
-	registry  Registry
-	podLister PodLister
-	strategy  rcStrategy
-}
-
-// NewStorage returns a new rest.Storage for the given registry and PodLister.
-func NewStorage(registry Registry, podLister PodLister) *REST {
-	return &REST{
-		registry:  registry,
-		podLister: podLister,
-		strategy:  Strategy,
-	}
-}
-
-// Create registers the given ReplicationController with the system,
-// which eventually leads to the controller manager acting on its behalf.
-func (rs *REST) Create(ctx api.Context, obj runtime.Object) (runtime.Object, error) {
-	controller, ok := obj.(*api.ReplicationController)
-	if !ok {
-		return nil, fmt.Errorf("not a replication controller: %#v", obj)
-	}
-
-	if err := rest.BeforeCreate(rs.strategy, ctx, obj); err != nil {
-		return nil, err
-	}
-
-	out, err := rs.registry.CreateController(ctx, controller)
-	if err != nil {
-		err = rest.CheckGeneratedNameError(rs.strategy, err, controller)
-	}
-	return out, err
-}
-
-// Delete asynchronously deletes the ReplicationController specified by its id.
-func (rs *REST) Delete(ctx api.Context, id string) (runtime.Object, error) {
-	return &api.Status{Status: api.StatusSuccess}, rs.registry.DeleteController(ctx, id)
-}
-
-// Get obtains the ReplicationController specified by its id.
-func (rs *REST) Get(ctx api.Context, id string) (runtime.Object, error) {
-	controller, err := rs.registry.GetController(ctx, id)
-	if err != nil {
-		return nil, err
-	}
-	return controller, err
-}
-
-// List obtains a list of ReplicationControllers that match selector.
-func (rs *REST) List(ctx api.Context, label labels.Selector, field fields.Selector) (runtime.Object, error) {
-	if !field.Empty() {
-		return nil, fmt.Errorf("field selector not supported yet")
-	}
-	controllers, err := rs.registry.ListControllers(ctx)
-	if err != nil {
-		return nil, err
-	}
-	filtered := []api.ReplicationController{}
-	for _, controller := range controllers.Items {
-		if label.Matches(labels.Set(controller.Labels)) {
-			filtered = append(filtered, controller)
-		}
-	}
-	controllers.Items = filtered
-	return controllers, err
-}
-
-// New creates a new ReplicationController for use with Create and Update.
-func (*REST) New() runtime.Object {
-	return &api.ReplicationController{}
-}
-
-func (*REST) NewList() runtime.Object {
-	return &api.ReplicationControllerList{}
-}
-
-// Update replaces a given ReplicationController instance with an existing
-// instance in storage.registry.
-func (rs *REST) Update(ctx api.Context, obj runtime.Object) (runtime.Object, bool, error) {
-	controller, ok := obj.(*api.ReplicationController)
-	if !ok {
-		return nil, false, fmt.Errorf("not a replication controller: %#v", obj)
-	}
-	existingController, err := rs.registry.GetController(ctx, controller.Name)
-	if err != nil {
-		return nil, false, err
-	}
-	if err := rest.BeforeUpdate(rs.strategy, ctx, controller, existingController); err != nil {
-		return nil, false, err
-	}
-	out, err := rs.registry.UpdateController(ctx, controller)
-	return out, false, err
-}
-
-// Watch returns ReplicationController events via a watch.Interface.
-// It implements rest.Watcher.
-func (rs *REST) Watch(ctx api.Context, label labels.Selector, field fields.Selector, resourceVersion string) (watch.Interface, error) {
-	return rs.registry.WatchControllers(ctx, label, field, resourceVersion)
+// MatchController 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 MatchController(label labels.Selector, field fields.Selector) generic.Matcher {
+	return generic.MatcherFunc(
+		func(obj runtime.Object) (bool, error) {
+			if !field.Empty() {
+				return false, fmt.Errorf("field selector not supported yet")
+			}
+			controllerObj, ok := obj.(*api.ReplicationController)
+			if !ok {
+				return false, fmt.Errorf("Given object is not a replication controller.")
+			}
+			return label.Matches(labels.Set(controllerObj.Labels)), nil
+		})
 }

pkg/registry/generic/etcd/etcd.go

@@ -241,6 +241,7 @@ func (e *Etcd) UpdateWithName(ctx api.Context, name string, obj runtime.Object)
 
 // Update performs an atomic update and set of the object. Returns the result of the update
 // or an error. If the registry allows create-on-update, the create flow will be executed.
+// A bool is returned along with the object and any errors, to indicate object creation.
 func (e *Etcd) Update(ctx api.Context, obj runtime.Object) (runtime.Object, bool, error) {
 	name, err := e.ObjectNameFunc(obj)
 	if err != nil {

pkg/registry/pod/etcd/etcd_test.go

@@ -155,7 +155,7 @@ func TestDelete(t *testing.T) {
 		}
 		return fakeEtcdClient.Data["/registry/pods/default/foo"].R.Node.TTL == 30
 	}
-	test.TestDeleteNoGraceful(createFn, gracefulSetFn)
+	test.TestDelete(createFn, gracefulSetFn)
 }
 
 func expectPod(t *testing.T, out runtime.Object) (*api.Pod, bool) {