Make ResourceQuota admission and controller work generically

cmd/kube-controller-manager/app/controllermanager.go

@@ -60,6 +60,7 @@ import (
 	servicecontroller "k8s.io/kubernetes/pkg/controller/service"
 	serviceaccountcontroller "k8s.io/kubernetes/pkg/controller/serviceaccount"
 	"k8s.io/kubernetes/pkg/healthz"
+	quotainstall "k8s.io/kubernetes/pkg/quota/install"
 	"k8s.io/kubernetes/pkg/serviceaccount"
 	"k8s.io/kubernetes/pkg/util"
 	"k8s.io/kubernetes/pkg/util/wait"
@@ -220,9 +221,23 @@ func StartControllers(s *options.CMServer, kubeClient *client.Client, kubeconfig
 		glog.Infof("allocate-node-cidrs set to %v, node controller not creating routes", s.AllocateNodeCIDRs)
 	}
 
-	go resourcequotacontroller.NewResourceQuotaController(
-		clientset.NewForConfigOrDie(client.AddUserAgent(kubeconfig, "resourcequota-controller")),
-		controller.StaticResyncPeriodFunc(s.ResourceQuotaSyncPeriod.Duration)).Run(s.ConcurrentResourceQuotaSyncs, wait.NeverStop)
+	resourceQuotaControllerClient := clientset.NewForConfigOrDie(client.AddUserAgent(kubeconfig, "resourcequota-controller"))
+	resourceQuotaRegistry := quotainstall.NewRegistry(resourceQuotaControllerClient)
+	groupKindsToReplenish := []unversioned.GroupKind{
+		api.Kind("Pod"),
+		api.Kind("Service"),
+		api.Kind("ReplicationController"),
+		api.Kind("PersistentVolumeClaim"),
+		api.Kind("Secret"),
+	}
+	resourceQuotaControllerOptions := &resourcequotacontroller.ResourceQuotaControllerOptions{
+		KubeClient:            resourceQuotaControllerClient,
+		ResyncPeriod:          controller.StaticResyncPeriodFunc(s.ResourceQuotaSyncPeriod.Duration),
+		Registry:              resourceQuotaRegistry,
+		GroupKindsToReplenish: groupKindsToReplenish,
+		ControllerFactory:     resourcequotacontroller.NewReplenishmentControllerFactory(resourceQuotaControllerClient),
+	}
+	go resourcequotacontroller.NewResourceQuotaController(resourceQuotaControllerOptions).Run(s.ConcurrentResourceQuotaSyncs, wait.NeverStop)
 
 	// If apiserver is not running we should wait for some time and fail only then. This is particularly
 	// important when we start apiserver and controller manager at the same time.

contrib/mesos/pkg/controllermanager/controllermanager.go

@@ -28,6 +28,7 @@ import (
 	kubecontrollermanager "k8s.io/kubernetes/cmd/kube-controller-manager/app"
 	"k8s.io/kubernetes/cmd/kube-controller-manager/app/options"
 	"k8s.io/kubernetes/contrib/mesos/pkg/node"
+	"k8s.io/kubernetes/pkg/api"
 	"k8s.io/kubernetes/pkg/api/unversioned"
 	clientset "k8s.io/kubernetes/pkg/client/clientset_generated/internalclientset"
 	client "k8s.io/kubernetes/pkg/client/unversioned"
@@ -53,6 +54,7 @@ import (
 	servicecontroller "k8s.io/kubernetes/pkg/controller/service"
 	serviceaccountcontroller "k8s.io/kubernetes/pkg/controller/serviceaccount"
 	"k8s.io/kubernetes/pkg/healthz"
+	quotainstall "k8s.io/kubernetes/pkg/quota/install"
 	"k8s.io/kubernetes/pkg/serviceaccount"
 	"k8s.io/kubernetes/pkg/util"
 	"k8s.io/kubernetes/pkg/util/wait"
@@ -173,8 +175,23 @@ func (s *CMServer) Run(_ []string) error {
 		routeController.Run(s.NodeSyncPeriod.Duration)
 	}
 
-	go resourcequotacontroller.NewResourceQuotaController(
-		clientset.NewForConfigOrDie(client.AddUserAgent(kubeconfig, "resource-quota-controller")), controller.StaticResyncPeriodFunc(s.ResourceQuotaSyncPeriod.Duration)).Run(s.ConcurrentResourceQuotaSyncs, wait.NeverStop)
+	resourceQuotaControllerClient := clientset.NewForConfigOrDie(client.AddUserAgent(kubeconfig, "resource-quota-controller"))
+	resourceQuotaRegistry := quotainstall.NewRegistry(resourceQuotaControllerClient)
+	groupKindsToReplenish := []unversioned.GroupKind{
+		api.Kind("Pod"),
+		api.Kind("Service"),
+		api.Kind("ReplicationController"),
+		api.Kind("PersistentVolumeClaim"),
+		api.Kind("Secret"),
+	}
+	resourceQuotaControllerOptions := &resourcequotacontroller.ResourceQuotaControllerOptions{
+		KubeClient:            resourceQuotaControllerClient,
+		ResyncPeriod:          controller.StaticResyncPeriodFunc(s.ResourceQuotaSyncPeriod.Duration),
+		Registry:              resourceQuotaRegistry,
+		GroupKindsToReplenish: groupKindsToReplenish,
+		ControllerFactory:     resourcequotacontroller.NewReplenishmentControllerFactory(resourceQuotaControllerClient),
+	}
+	go resourcequotacontroller.NewResourceQuotaController(resourceQuotaControllerOptions).Run(s.ConcurrentResourceQuotaSyncs, wait.NeverStop)
 
 	// If apiserver is not running we should wait for some time and fail only then. This is particularly
 	// important when we start apiserver and controller manager at the same time.

contrib/mesos/pkg/scheduler/resources/resource_test.go

@@ -22,7 +22,6 @@ import (
 
 	"github.com/stretchr/testify/assert"
 	"k8s.io/kubernetes/pkg/api"
-	"k8s.io/kubernetes/pkg/controller/resourcequota"
 )
 
 type resources struct {
@@ -81,7 +80,6 @@ func TestResources(tst *testing.T) {
 		}
 
 		tst.Logf("Testing resource computation for %v => request=%v limit=%v", t, pod.Spec.Containers[0].Resources.Requests, pod.Spec.Containers[0].Resources.Limits)
-		tst.Logf("hasRequests: cpu => %v, mem => %v", resourcequota.PodHasRequests(pod, api.ResourceCPU), resourcequota.PodHasRequests(pod, api.ResourceMemory))
 
 		beforeCpuR, beforeCpuL, _, err := LimitedCPUForPod(pod, DefaultDefaultContainerCPULimit)
 		assert.NoError(err, "CPUForPod should not return an error")

pkg/controller/resourcequota/replenishment_controller.go

+/*
+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 resourcequota
+
+import (
+	"fmt"
+
+	"github.com/golang/glog"
+
+	"k8s.io/kubernetes/pkg/api"
+	"k8s.io/kubernetes/pkg/api/meta"
+	"k8s.io/kubernetes/pkg/api/unversioned"
+	"k8s.io/kubernetes/pkg/client/cache"
+	clientset "k8s.io/kubernetes/pkg/client/clientset_generated/internalclientset"
+	"k8s.io/kubernetes/pkg/controller"
+	"k8s.io/kubernetes/pkg/controller/framework"
+	"k8s.io/kubernetes/pkg/quota/evaluator/core"
+	"k8s.io/kubernetes/pkg/runtime"
+	utilruntime "k8s.io/kubernetes/pkg/util/runtime"
+	"k8s.io/kubernetes/pkg/watch"
+)
+
+// ReplenishmentFunc is a function that is invoked when controller sees a change
+// that may require a quota to be replenished (i.e. object deletion, or object moved to terminal state)
+type ReplenishmentFunc func(groupKind unversioned.GroupKind, namespace string, object runtime.Object)
+
+// ReplenishmentControllerOptions is an options struct that tells a factory
+// how to configure a controller that can inform the quota system it should
+// replenish quota
+type ReplenishmentControllerOptions struct {
+	// The kind monitored for replenishment
+	GroupKind unversioned.GroupKind
+	// The period that should be used to re-sync the monitored resource
+	ResyncPeriod controller.ResyncPeriodFunc
+	// The function to invoke when a change is observed that should trigger
+	// replenishment
+	ReplenishmentFunc ReplenishmentFunc
+}
+
+// PodReplenishmentUpdateFunc will replenish if the old pod was quota tracked but the new is not
+func PodReplenishmentUpdateFunc(options *ReplenishmentControllerOptions) func(oldObj, newObj interface{}) {
+	return func(oldObj, newObj interface{}) {
+		oldPod := oldObj.(*api.Pod)
+		newPod := newObj.(*api.Pod)
+		if core.QuotaPod(oldPod) && !core.QuotaPod(newPod) {
+			options.ReplenishmentFunc(options.GroupKind, newPod.Namespace, newPod)
+		}
+	}
+}
+
+// ObjectReplenenishmentDeleteFunc will replenish on every delete
+func ObjectReplenishmentDeleteFunc(options *ReplenishmentControllerOptions) func(obj interface{}) {
+	return func(obj interface{}) {
+		metaObject, err := meta.Accessor(obj)
+		if err != nil {
+			tombstone, ok := obj.(cache.DeletedFinalStateUnknown)
+			if !ok {
+				glog.Errorf("replenishment controller could not get object from tombstone %+v, could take up to %v before quota is replenished", obj, options.ResyncPeriod())
+				utilruntime.HandleError(err)
+				return
+			}
+			metaObject, err = meta.Accessor(tombstone.Obj)
+			if err != nil {
+				glog.Errorf("replenishment controller tombstone contained object that is not a meta %+v, could take up to %v before quota is replenished", tombstone.Obj, options.ResyncPeriod())
+				utilruntime.HandleError(err)
+				return
+			}
+		}
+		options.ReplenishmentFunc(options.GroupKind, metaObject.GetNamespace(), nil)
+	}
+}
+
+// ReplenishmentControllerFactory knows how to build replenishment controllers
+type ReplenishmentControllerFactory interface {
+	// NewController returns a controller configured with the specified options
+	NewController(options *ReplenishmentControllerOptions) (*framework.Controller, error)
+}
+
+// replenishmentControllerFactory implements ReplenishmentControllerFactory
+type replenishmentControllerFactory struct {
+	kubeClient clientset.Interface
+}
+
+// NewReplenishmentControllerFactory returns a factory that knows how to build controllers
+// to replenish resources when updated or deleted
+func NewReplenishmentControllerFactory(kubeClient clientset.Interface) ReplenishmentControllerFactory {
+	return &replenishmentControllerFactory{
+		kubeClient: kubeClient,
+	}
+}
+
+func (r *replenishmentControllerFactory) NewController(options *ReplenishmentControllerOptions) (*framework.Controller, error) {
+	var result *framework.Controller
+	switch options.GroupKind {
+	case api.Kind("Pod"):
+		_, result = framework.NewInformer(
+			&cache.ListWatch{
+				ListFunc: func(options api.ListOptions) (runtime.Object, error) {
+					return r.kubeClient.Core().Pods(api.NamespaceAll).List(options)
+				},
+				WatchFunc: func(options api.ListOptions) (watch.Interface, error) {
+					return r.kubeClient.Core().Pods(api.NamespaceAll).Watch(options)
+				},
+			},
+			&api.Pod{},
+			options.ResyncPeriod(),
+			framework.ResourceEventHandlerFuncs{
+				UpdateFunc: PodReplenishmentUpdateFunc(options),
+				DeleteFunc: ObjectReplenishmentDeleteFunc(options),
+			},
+		)
+	case api.Kind("Service"):
+		_, result = framework.NewInformer(
+			&cache.ListWatch{
+				ListFunc: func(options api.ListOptions) (runtime.Object, error) {
+					return r.kubeClient.Core().Services(api.NamespaceAll).List(options)
+				},
+				WatchFunc: func(options api.ListOptions) (watch.Interface, error) {
+					return r.kubeClient.Core().Services(api.NamespaceAll).Watch(options)
+				},
+			},
+			&api.Service{},
+			options.ResyncPeriod(),
+			framework.ResourceEventHandlerFuncs{
+				DeleteFunc: ObjectReplenishmentDeleteFunc(options),
+			},
+		)
+	case api.Kind("ReplicationController"):
+		_, result = framework.NewInformer(
+			&cache.ListWatch{
+				ListFunc: func(options api.ListOptions) (runtime.Object, error) {
+					return r.kubeClient.Core().ReplicationControllers(api.NamespaceAll).List(options)
+				},
+				WatchFunc: func(options api.ListOptions) (watch.Interface, error) {
+					return r.kubeClient.Core().ReplicationControllers(api.NamespaceAll).Watch(options)
+				},
+			},
+			&api.ReplicationController{},
+			options.ResyncPeriod(),
+			framework.ResourceEventHandlerFuncs{
+				DeleteFunc: ObjectReplenishmentDeleteFunc(options),
+			},
+		)
+	case api.Kind("PersistentVolumeClaim"):
+		_, result = framework.NewInformer(
+			&cache.ListWatch{
+				ListFunc: func(options api.ListOptions) (runtime.Object, error) {
+					return r.kubeClient.Core().PersistentVolumeClaims(api.NamespaceAll).List(options)
+				},
+				WatchFunc: func(options api.ListOptions) (watch.Interface, error) {
+					return r.kubeClient.Core().PersistentVolumeClaims(api.NamespaceAll).Watch(options)
+				},
+			},
+			&api.PersistentVolumeClaim{},
+			options.ResyncPeriod(),
+			framework.ResourceEventHandlerFuncs{
+				DeleteFunc: ObjectReplenishmentDeleteFunc(options),
+			},
+		)
+	case api.Kind("Secret"):
+		_, result = framework.NewInformer(
+			&cache.ListWatch{
+				ListFunc: func(options api.ListOptions) (runtime.Object, error) {
+					return r.kubeClient.Core().Secrets(api.NamespaceAll).List(options)
+				},
+				WatchFunc: func(options api.ListOptions) (watch.Interface, error) {
+					return r.kubeClient.Core().Secrets(api.NamespaceAll).Watch(options)
+				},
+			},
+			&api.PersistentVolumeClaim{},
+			options.ResyncPeriod(),
+			framework.ResourceEventHandlerFuncs{
+				DeleteFunc: ObjectReplenishmentDeleteFunc(options),
+			},
+		)
+	default:
+		return nil, fmt.Errorf("no replenishment controller available for %s", options.GroupKind)
+	}
+	return result, nil
+}

pkg/controller/resourcequota/replenishment_controller_test.go

+/*
+Copyright 2016 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 resourcequota
+
+import (
+	"testing"
+
+	"k8s.io/kubernetes/pkg/api"
+	"k8s.io/kubernetes/pkg/api/unversioned"
+	"k8s.io/kubernetes/pkg/controller"
+	"k8s.io/kubernetes/pkg/runtime"
+)
+
+// testReplenishment lets us test replenishment functions are invoked
+type testReplenishment struct {
+	groupKind unversioned.GroupKind
+	namespace string
+}
+
+// mock function that holds onto the last kind that was replenished
+func (t *testReplenishment) Replenish(groupKind unversioned.GroupKind, namespace string, object runtime.Object) {
+	t.groupKind = groupKind
+	t.namespace = namespace
+}
+
+func TestPodReplenishmentUpdateFunc(t *testing.T) {
+	mockReplenish := &testReplenishment{}
+	options := ReplenishmentControllerOptions{
+		GroupKind:         api.Kind("Pod"),
+		ReplenishmentFunc: mockReplenish.Replenish,
+		ResyncPeriod:      controller.NoResyncPeriodFunc,
+	}
+	oldPod := &api.Pod{
+		ObjectMeta: api.ObjectMeta{Namespace: "test", Name: "pod"},
+		Status:     api.PodStatus{Phase: api.PodRunning},
+	}
+	newPod := &api.Pod{
+		ObjectMeta: api.ObjectMeta{Namespace: "test", Name: "pod"},
+		Status:     api.PodStatus{Phase: api.PodFailed},
+	}
+	updateFunc := PodReplenishmentUpdateFunc(&options)
+	updateFunc(oldPod, newPod)
+	if mockReplenish.groupKind != api.Kind("Pod") {
+		t.Errorf("Unexpected group kind %v", mockReplenish.groupKind)
+	}
+	if mockReplenish.namespace != oldPod.Namespace {
+		t.Errorf("Unexpected namespace %v", mockReplenish.namespace)
+	}
+}
+
+func TestObjectReplenishmentDeleteFunc(t *testing.T) {
+	mockReplenish := &testReplenishment{}
+	options := ReplenishmentControllerOptions{
+		GroupKind:         api.Kind("Pod"),
+		ReplenishmentFunc: mockReplenish.Replenish,
+		ResyncPeriod:      controller.NoResyncPeriodFunc,
+	}
+	oldPod := &api.Pod{
+		ObjectMeta: api.ObjectMeta{Namespace: "test", Name: "pod"},
+		Status:     api.PodStatus{Phase: api.PodRunning},
+	}
+	deleteFunc := ObjectReplenishmentDeleteFunc(&options)
+	deleteFunc(oldPod)
+	if mockReplenish.groupKind != api.Kind("Pod") {
+		t.Errorf("Unexpected group kind %v", mockReplenish.groupKind)
+	}
+	if mockReplenish.namespace != oldPod.Namespace {
+		t.Errorf("Unexpected namespace %v", mockReplenish.namespace)
+	}
+}