add clusterquota reconciliation controller

Author: deads2k

pkg/client/clusteresourcequota.go

@@ -15,6 +15,7 @@ type ClusterResourceQuotaInterface interface {
 	List(opts kapi.ListOptions) (*quotaapi.ClusterResourceQuotaList, error)
 	Get(name string) (*quotaapi.ClusterResourceQuota, error)
 	Create(resourceQuota *quotaapi.ClusterResourceQuota) (*quotaapi.ClusterResourceQuota, error)
+	Update(resourceQuota *quotaapi.ClusterResourceQuota) (*quotaapi.ClusterResourceQuota, error)
 	Delete(name string) error
 	Watch(opts kapi.ListOptions) (watch.Interface, error)
 }
@@ -48,6 +49,13 @@ func (c *clusterResourceQuotas) Create(resourceQuota *quotaapi.ClusterResourceQu
 	return
 }
 
+// Update updates an existing deploymentConfig
+func (c *clusterResourceQuotas) Update(resourceQuota *quotaapi.ClusterResourceQuota) (result *quotaapi.ClusterResourceQuota, err error) {
+	result = &quotaapi.ClusterResourceQuota{}
+	err = c.r.Put().Resource("clusterresourcequotas").Name(resourceQuota.Name).Body(resourceQuota).Do().Into(result)
+	return
+}
+
 func (c *clusterResourceQuotas) Delete(name string) (err error) {
 	err = c.r.Delete().Resource("clusterresourcequotas").Name(name).Do().Error()
 	return

pkg/client/testclient/fake_clusterresourcequota.go

@@ -41,6 +41,14 @@ func (c *FakeClusterResourceQuotas) Create(inObj *quotaapi.ClusterResourceQuota)
 	return obj.(*quotaapi.ClusterResourceQuota), err
 }
 
+func (c *FakeClusterResourceQuotas) Update(inObj *quotaapi.ClusterResourceQuota) (*quotaapi.ClusterResourceQuota, error) {
+	obj, err := c.Fake.Invokes(ktestclient.NewRootUpdateAction("clusterresourcequotas", inObj), inObj)
+	if obj == nil {
+		return nil, err
+	}
+
+	return obj.(*quotaapi.ClusterResourceQuota), err
+}
 func (c *FakeClusterResourceQuotas) Delete(name string) error {
 	_, err := c.Fake.Invokes(ktestclient.NewRootDeleteAction("clusterresourcequotas", name), &quotaapi.ClusterResourceQuota{})
 	return err

pkg/cmd/server/origin/run_components.go

@@ -19,6 +19,7 @@ import (
 	"k8s.io/kubernetes/pkg/controller"
 	kresourcequota "k8s.io/kubernetes/pkg/controller/resourcequota"
 	sacontroller "k8s.io/kubernetes/pkg/controller/serviceaccount"
+	quotainstall "k8s.io/kubernetes/pkg/quota/install"
 	"k8s.io/kubernetes/pkg/registry/service/allocator"
 	etcdallocator "k8s.io/kubernetes/pkg/registry/service/allocator/etcd"
 	"k8s.io/kubernetes/pkg/serviceaccount"
@@ -53,6 +54,7 @@ import (
 	imageapi "github.com/openshift/origin/pkg/image/api"
 	quota "github.com/openshift/origin/pkg/quota"
 	quotacontroller "github.com/openshift/origin/pkg/quota/controller"
+	"github.com/openshift/origin/pkg/quota/controller/clusterquotareconciliation"
 	serviceaccountcontrollers "github.com/openshift/origin/pkg/serviceaccounts/controllers"
 )
 
@@ -507,3 +509,31 @@ func (c *MasterConfig) RunClusterQuotaMappingController() {
 		go c.ClusterQuotaMappingController.Run(5, utilwait.NeverStop)
 	})
 }
+
+func (c *MasterConfig) RunClusterQuotaReconciliationController() {
+	osClient, kClient := c.ResourceQuotaManagerClients()
+	resourceQuotaRegistry := quotainstall.NewRegistry(kClient)
+	groupKindsToReplenish := []unversioned.GroupKind{
+		kapi.Kind("Pod"),
+		kapi.Kind("Service"),
+		kapi.Kind("ReplicationController"),
+		kapi.Kind("PersistentVolumeClaim"),
+		kapi.Kind("Secret"),
+		kapi.Kind("ConfigMap"),
+	}
+
+	options := clusterquotareconciliation.ClusterQuotaReconcilationControllerOptions{
+		ClusterQuotaInformer: c.Informers.ClusterResourceQuotas(),
+		ClusterQuotaMapper:   c.ClusterQuotaMappingController.GetClusterQuotaMapper(),
+		ClusterQuotaClient:   osClient,
+
+		Registry:                  resourceQuotaRegistry,
+		ResyncPeriod:              defaultResourceQuotaSyncPeriod,
+		ControllerFactory:         kresourcequota.NewReplenishmentControllerFactory(c.Informers.Pods().Informer(), kClient),
+		ReplenishmentResyncPeriod: controller.StaticResyncPeriodFunc(defaultReplenishmentSyncPeriod),
+		GroupKindsToReplenish:     groupKindsToReplenish,
+	}
+	controller := clusterquotareconciliation.NewClusterQuotaReconcilationController(options)
+	c.ClusterQuotaMappingController.GetClusterQuotaMapper().AddListener(controller)
+	go controller.Run(5, utilwait.NeverStop)
+}

pkg/cmd/server/start/start_master.go

@@ -656,6 +656,7 @@ func startControllers(oc *origin.MasterConfig, kc *kubernetes.MasterConfig) erro
 	oc.RunImageImportController()
 	oc.RunOriginNamespaceController()
 	oc.RunSDNController()
+	oc.RunClusterQuotaReconciliationController()
 	oc.RunClusterQuotaMappingController()
 
 	_, _, serviceServingCertClient, err := oc.GetServiceAccountClients(bootstrappolicy.ServiceServingCertServiceAccountName)

pkg/quota/controller/clusterquotamapping/clusterquotamapping.go

@@ -2,7 +2,6 @@ package clusterquotamapping
 
 import (
 	"fmt"
-	"sync"
 	"time"
 
 	"github.com/golang/glog"
@@ -15,7 +14,6 @@ import (
 	"k8s.io/kubernetes/pkg/controller/framework"
 	"k8s.io/kubernetes/pkg/labels"
 	utilruntime "k8s.io/kubernetes/pkg/util/runtime"
-	"k8s.io/kubernetes/pkg/util/sets"
 	"k8s.io/kubernetes/pkg/util/wait"
 	"k8s.io/kubernetes/pkg/util/workqueue"
 
@@ -24,15 +22,6 @@ import (
 	quotaapi "github.com/openshift/origin/pkg/quota/api"
 )
 
-type ClusterQuotaMapper interface {
-	// GetClusterQuotasFor returns the list of clusterquota names that this namespace matches.  It also
-	// returns the labels associated with the namespace for the check so that callers can determine staleness
-	GetClusterQuotasFor(namespaceName string) ([]string, map[string]string)
-	// GetNamespacesFor returns the list of namespace names that this cluster quota matches.  It also
-	// returns the selector associated with the clusterquota for the check so that callers can determine staleness
-	GetNamespacesFor(quotaName string) ([]string, *unversioned.LabelSelector)
-}
-
 // Look out, here there be dragons!
 // There is a race when dealing with the DeltaFifo compression used to back a reflector for a controller that uses two
 // SharedInformers for both their watch events AND their caches.  The scenario looks like this
@@ -65,15 +54,7 @@ func NewClusterQuotaMappingController(namespaceInformer shared.NamespaceInformer
 
 		quotaQueue: workqueue.NewRateLimitingQueue(workqueue.DefaultControllerRateLimiter()),
 
-		clusterQuotaMapper: &clusterQuotaMapper{
-			requiredQuotaToSelector:    map[string]*unversioned.LabelSelector{},
-			requiredNamespaceToLabels:  map[string]map[string]string{},
-			completedQuotaToSelector:   map[string]*unversioned.LabelSelector{},
-			completedNamespaceToLabels: map[string]map[string]string{},
-
-			quotaToNamespaces: map[string]sets.String{},
-			namespaceToQuota:  map[string]sets.String{},
-		},
+		clusterQuotaMapper: NewClusterQuotaMapper(),
 	}
 
 	namespaceInformer.Informer().AddEventHandler(framework.ResourceEventHandlerFuncs{
@@ -107,204 +88,6 @@ type ClusterQuotaMappingController struct {
 	clusterQuotaMapper *clusterQuotaMapper
 }
 
-// clusterQuotaMapper gives thread safe access to the actual mappings that are being stored.
-// Many method use a shareable read lock to check status followed by a non-shareable
-// write lock which double checks the condition before proceding.  Since locks aren't escalatable
-// you have to perform the recheck because someone could have beaten you in.
-type clusterQuotaMapper struct {
-	lock sync.RWMutex
-
-	// requiredQuotaToSelector indicates the latest label selector this controller has observed for a quota
-	requiredQuotaToSelector map[string]*unversioned.LabelSelector
-	// requiredNamespaceToLabels indicates the latest labels this controller has observed for a namespace
-	requiredNamespaceToLabels map[string]map[string]string
-	// completedQuotaToSelector indicates the latest label selector this controller has scanned against namespaces
-	completedQuotaToSelector map[string]*unversioned.LabelSelector
-	// completedNamespaceToLabels indicates the latest labels this controller has scanned against cluster quotas
-	completedNamespaceToLabels map[string]map[string]string
-
-	quotaToNamespaces map[string]sets.String
-	namespaceToQuota  map[string]sets.String
-}
-
-func (m *clusterQuotaMapper) GetClusterQuotasFor(namespaceName string) ([]string, map[string]string) {
-	m.lock.RLock()
-	defer m.lock.RUnlock()
-
-	quotas, ok := m.namespaceToQuota[namespaceName]
-	if !ok {
-		return []string{}, m.completedNamespaceToLabels[namespaceName]
-	}
-	return quotas.List(), m.completedNamespaceToLabels[namespaceName]
-}
-
-func (m *clusterQuotaMapper) GetNamespacesFor(quotaName string) ([]string, *unversioned.LabelSelector) {
-	m.lock.RLock()
-	defer m.lock.RUnlock()
-
-	namespaces, ok := m.quotaToNamespaces[quotaName]
-	if !ok {
-		return []string{}, m.completedQuotaToSelector[quotaName]
-	}
-	return namespaces.List(), m.completedQuotaToSelector[quotaName]
-}
-
-// requireQuota updates the selector requirements for the given quota.  This prevents stale updates to the mapping itself.
-// returns true if a modification was made
-func (m *clusterQuotaMapper) requireQuota(quota *quotaapi.ClusterResourceQuota) bool {
-	m.lock.RLock()
-	selector, exists := m.requiredQuotaToSelector[quota.Name]
-	m.lock.RUnlock()
-
-	if selectorMatches(selector, exists, quota) {
-		return false
-	}
-
-	m.lock.Lock()
-	defer m.lock.Unlock()
-	selector, exists = m.requiredQuotaToSelector[quota.Name]
-	if selectorMatches(selector, exists, quota) {
-		return false
-	}
-
-	m.requiredQuotaToSelector[quota.Name] = quota.Spec.Selector
-	return true
-}
-
-// completeQuota updates the latest selector used to generate the mappings for this quota.  The value is returned
-// by the Get methods for the mapping so that callers can determine staleness
-func (m *clusterQuotaMapper) completeQuota(quota *quotaapi.ClusterResourceQuota) {
-	m.lock.Lock()
-	defer m.lock.Unlock()
-	m.completedQuotaToSelector[quota.Name] = quota.Spec.Selector
-}
-
-// removeQuota deletes a quota from all mappings
-func (m *clusterQuotaMapper) removeQuota(quotaName string) {
-	m.lock.Lock()
-	defer m.lock.Unlock()
-
-	delete(m.requiredQuotaToSelector, quotaName)
-	delete(m.completedQuotaToSelector, quotaName)
-	delete(m.quotaToNamespaces, quotaName)
-	for _, quotas := range m.namespaceToQuota {
-		quotas.Delete(quotaName)
-	}
-}
-
-// requireNamespace updates the label requirements for the given namespace.  This prevents stale updates to the mapping itself.
-// returns true if a modification was made
-func (m *clusterQuotaMapper) requireNamespace(namespace *kapi.Namespace) bool {
-	m.lock.RLock()
-	labels, exists := m.requiredNamespaceToLabels[namespace.Name]
-	m.lock.RUnlock()
-
-	if labelsMatch(labels, exists, namespace) {
-		return false
-	}
-
-	m.lock.Lock()
-	defer m.lock.Unlock()
-	labels, exists = m.requiredNamespaceToLabels[namespace.Name]
-	if labelsMatch(labels, exists, namespace) {
-		return false
-	}
-
-	m.requiredNamespaceToLabels[namespace.Name] = namespace.Labels
-	return true
-}
-
-// completeNamespace updates the latest labels used to generate the mappings for this namespace.  The value is returned
-// by the Get methods for the mapping so that callers can determine staleness
-func (m *clusterQuotaMapper) completeNamespace(namespace *kapi.Namespace) {
-	m.lock.Lock()
-	defer m.lock.Unlock()
-	m.completedNamespaceToLabels[namespace.Name] = namespace.Labels
-}
-
-// removeNamespace deletes a namespace from all mappings
-func (m *clusterQuotaMapper) removeNamespace(namespaceName string) {
-	m.lock.Lock()
-	defer m.lock.Unlock()
-
-	delete(m.requiredNamespaceToLabels, namespaceName)
-	delete(m.completedNamespaceToLabels, namespaceName)
-	delete(m.namespaceToQuota, namespaceName)
-	for _, namespaces := range m.quotaToNamespaces {
-		namespaces.Delete(namespaceName)
-	}
-}
-
-func selectorMatches(selector *unversioned.LabelSelector, exists bool, quota *quotaapi.ClusterResourceQuota) bool {
-	return exists && kapi.Semantic.DeepEqual(selector, quota.Spec.Selector)
-}
-func labelsMatch(labels map[string]string, exists bool, namespace *kapi.Namespace) bool {
-	return exists && kapi.Semantic.DeepEqual(labels, namespace.Labels)
-}
-
-// setMapping maps (or removes a mapping) between a clusterquota and a namespace
-// It returns whether the action worked, whether the quota is out of date, whether the namespace is out of date
-// This allows callers to decide whether to pull new information from the cache or simply skip execution
-func (m *clusterQuotaMapper) setMapping(quota *quotaapi.ClusterResourceQuota, namespace *kapi.Namespace, remove bool) (bool /*added*/, bool /*quota matches*/, bool /*namespace matches*/) {
-	m.lock.RLock()
-	selector, selectorExists := m.requiredQuotaToSelector[quota.Name]
-	labels, labelsExist := m.requiredNamespaceToLabels[namespace.Name]
-	m.lock.RUnlock()
-
-	if !selectorMatches(selector, selectorExists, quota) {
-		return false, false, true
-	}
-	if !labelsMatch(labels, labelsExist, namespace) {
-		return false, true, false
-	}
-
-	m.lock.Lock()
-	defer m.lock.Unlock()
-	selector, selectorExists = m.requiredQuotaToSelector[quota.Name]
-	labels, labelsExist = m.requiredNamespaceToLabels[namespace.Name]
-	if !selectorMatches(selector, selectorExists, quota) {
-		return false, false, true
-	}
-	if !labelsMatch(labels, labelsExist, namespace) {
-		return false, true, false
-	}
-
-	if remove {
-		namespaces, ok := m.quotaToNamespaces[quota.Name]
-		if !ok {
-			m.quotaToNamespaces[quota.Name] = sets.String{}
-		} else {
-			namespaces.Delete(namespace.Name)
-		}
-
-		quotas, ok := m.namespaceToQuota[namespace.Name]
-		if !ok {
-			m.namespaceToQuota[namespace.Name] = sets.String{}
-		} else {
-			quotas.Delete(quota.Name)
-		}
-
-		return true, true, true
-	}
-
-	namespaces, ok := m.quotaToNamespaces[quota.Name]
-	if !ok {
-		m.quotaToNamespaces[quota.Name] = sets.NewString(namespace.Name)
-	} else {
-		namespaces.Insert(namespace.Name)
-	}
-
-	quotas, ok := m.namespaceToQuota[namespace.Name]
-	if !ok {
-		m.namespaceToQuota[namespace.Name] = sets.NewString(quota.Name)
-	} else {
-		quotas.Insert(quota.Name)
-	}
-
-	return true, true, true
-
-}
-
 func (c *ClusterQuotaMappingController) GetClusterQuotaMapper() ClusterQuotaMapper {
 	return c.clusterQuotaMapper
 }