Merge pull request #22154 from sdminonne/service_nodeports_quotas

pkg/api/helpers.go

@@ -152,6 +152,7 @@ var standardQuotaResources = sets.NewString(
 	string(ResourceSecrets),
 	string(ResourcePersistentVolumeClaims),
 	string(ResourceConfigMaps),
+	string(ResourceServicesNodePorts),
 )
 
 // IsStandardQuotaResourceName returns true if the resource is known to
@@ -190,6 +191,7 @@ var integerResources = sets.NewString(
 	string(ResourceSecrets),
 	string(ResourceConfigMaps),
 	string(ResourcePersistentVolumeClaims),
+	string(ResourceServicesNodePorts),
 )
 
 // IsIntegerResourceName returns true if the resource is measured in integer values

pkg/api/types.go

@@ -2210,6 +2210,8 @@ const (
 	ResourceConfigMaps ResourceName = "configmaps"
 	// ResourcePersistentVolumeClaims, number
 	ResourcePersistentVolumeClaims ResourceName = "persistentvolumeclaims"
+	// ResourceServicesNodePorts, number
+	ResourceServicesNodePorts ResourceName = "services.nodeports"
 	// CPU request, in cores. (500m = .5 cores)
 	ResourceRequestsCPU ResourceName = "requests.cpu"
 	// Memory request, in bytes. (500Gi = 500GiB = 500 * 1024 * 1024 * 1024)

pkg/api/v1/types.go

@@ -2668,6 +2668,8 @@ const (
 	ResourceConfigMaps ResourceName = "configmaps"
 	// ResourcePersistentVolumeClaims, number
 	ResourcePersistentVolumeClaims ResourceName = "persistentvolumeclaims"
+	// ResourceServicesNodePorts, number
+	ResourceServicesNodePorts ResourceName = "services.nodeports"
 	// CPU request, in cores. (500m = .5 cores)
 	ResourceCPURequest ResourceName = "cpu.request"
 	// CPU limit, in cores. (500m = .5 cores)

pkg/controller/resourcequota/replenishment_controller.go

@@ -136,6 +136,7 @@ func (r *replenishmentControllerFactory) NewController(options *ReplenishmentCon
 			&api.Service{},
 			options.ResyncPeriod(),
 			framework.ResourceEventHandlerFuncs{
+				UpdateFunc: ServiceReplenishmentUpdateFunc(options),
 				DeleteFunc: ObjectReplenishmentDeleteFunc(options),
 			},
 		)
@@ -208,3 +209,14 @@ func (r *replenishmentControllerFactory) NewController(options *ReplenishmentCon
 	}
 	return result, nil
 }
+
+// ServiceReplenishmentUpdateFunc will replenish if the old service was quota tracked but the new is not
+func ServiceReplenishmentUpdateFunc(options *ReplenishmentControllerOptions) func(oldObj, newObj interface{}) {
+	return func(oldObj, newObj interface{}) {
+		oldService := oldObj.(*api.Service)
+		newService := newObj.(*api.Service)
+		if core.QuotaServiceType(oldService) && !core.QuotaServiceType(newService) {
+			options.ReplenishmentFunc(options.GroupKind, newService.Namespace, newService)
+		}
+	}
+}

pkg/controller/resourcequota/replenishment_controller_test.go

@@ -23,6 +23,7 @@ import (
 	"k8s.io/kubernetes/pkg/api/unversioned"
 	"k8s.io/kubernetes/pkg/controller"
 	"k8s.io/kubernetes/pkg/runtime"
+	"k8s.io/kubernetes/pkg/util/intstr"
 )
 
 // testReplenishment lets us test replenishment functions are invoked
@@ -82,3 +83,39 @@ func TestObjectReplenishmentDeleteFunc(t *testing.T) {
 		t.Errorf("Unexpected namespace %v", mockReplenish.namespace)
 	}
 }
+
+func TestServiceReplenishmentUpdateFunc(t *testing.T) {
+	mockReplenish := &testReplenishment{}
+	options := ReplenishmentControllerOptions{
+		GroupKind:         api.Kind("Service"),
+		ReplenishmentFunc: mockReplenish.Replenish,
+		ResyncPeriod:      controller.NoResyncPeriodFunc,
+	}
+	oldService := &api.Service{
+		ObjectMeta: api.ObjectMeta{Namespace: "test", Name: "mysvc"},
+		Spec: api.ServiceSpec{
+			Type: api.ServiceTypeNodePort,
+			Ports: []api.ServicePort{{
+				Port:       80,
+				TargetPort: intstr.FromInt(80),
+			}},
+		},
+	}
+	newService := &api.Service{
+		ObjectMeta: api.ObjectMeta{Namespace: "test", Name: "mysvc"},
+		Spec: api.ServiceSpec{
+			Type: api.ServiceTypeClusterIP,
+			Ports: []api.ServicePort{{
+				Port:       80,
+				TargetPort: intstr.FromInt(80),
+			}}},
+	}
+	updateFunc := ServiceReplenishmentUpdateFunc(&options)
+	updateFunc(oldService, newService)
+	if mockReplenish.groupKind != api.Kind("Service") {
+		t.Errorf("Unexpected group kind %v", mockReplenish.groupKind)
+	}
+	if mockReplenish.namespace != oldService.Namespace {
+		t.Errorf("Unexpected namespace %v", mockReplenish.namespace)
+	}
+}

pkg/quota/evaluator/core/services.go

@@ -19,6 +19,7 @@ package core
 import (
 	"k8s.io/kubernetes/pkg/admission"
 	"k8s.io/kubernetes/pkg/api"
+	"k8s.io/kubernetes/pkg/api/resource"
 	clientset "k8s.io/kubernetes/pkg/client/clientset_generated/internalclientset"
 	"k8s.io/kubernetes/pkg/quota"
 	"k8s.io/kubernetes/pkg/quota/generic"
@@ -27,7 +28,10 @@ import (
 
 // NewServiceEvaluator returns an evaluator that can evaluate service quotas
 func NewServiceEvaluator(kubeClient clientset.Interface) quota.Evaluator {
-	allResources := []api.ResourceName{api.ResourceServices}
+	allResources := []api.ResourceName{
+		api.ResourceServices,
+		api.ResourceServicesNodePorts,
+	}
 	return &generic.GenericEvaluator{
 		Name:              "Evaluator.Service",
 		InternalGroupKind: api.Kind("Service"),
@@ -37,9 +41,31 @@ func NewServiceEvaluator(kubeClient clientset.Interface) quota.Evaluator {
 		MatchedResourceNames: allResources,
 		MatchesScopeFunc:     generic.MatchesNoScopeFunc,
 		ConstraintsFunc:      generic.ObjectCountConstraintsFunc(api.ResourceServices),
-		UsageFunc:            generic.ObjectCountUsageFunc(api.ResourceServices),
+		UsageFunc:            ServiceUsageFunc,
 		ListFuncByNamespace: func(namespace string, options api.ListOptions) (runtime.Object, error) {
 			return kubeClient.Core().Services(namespace).List(options)
 		},
 	}
 }
+
+// ServiceUsageFunc knows how to measure usage associated with services
+func ServiceUsageFunc(object runtime.Object) api.ResourceList {
+	result := api.ResourceList{}
+	if service, ok := object.(*api.Service); ok {
+		result[api.ResourceServices] = resource.MustParse("1")
+		switch service.Spec.Type {
+		case api.ServiceTypeNodePort:
+			result[api.ResourceServicesNodePorts] = resource.MustParse("1")
+		}
+	}
+	return result
+}
+
+// QuotaServiceType returns true if the service type is eligible to track against a quota
+func QuotaServiceType(service *api.Service) bool {
+	switch service.Spec.Type {
+	case api.ServiceTypeNodePort:
+		return true
+	}
+	return false
+}

plugin/pkg/admission/resourcequota/admission.go

@@ -175,7 +175,7 @@ func (q *quotaAdmission) Admit(a admission.Attributes) (err error) {
 			return admission.NewForbidden(a, fmt.Errorf("Failed quota: %s: %v", resourceQuota.Name, err))
 		}
 		if !hasUsageStats(resourceQuota) {
-			return admission.NewForbidden(a, fmt.Errorf("Status unknown for quota: %s", resourceQuota.Name))
+			return admission.NewForbidden(a, fmt.Errorf("status unknown for quota: %s", resourceQuota.Name))
 		}
 		resourceQuotas = append(resourceQuotas, resourceQuota)
 	}
@@ -235,7 +235,7 @@ func (q *quotaAdmission) Admit(a admission.Attributes) (err error) {
 				failedUsed := quota.Mask(resourceQuota.Status.Used, exceeded)
 				failedHard := quota.Mask(resourceQuota.Status.Hard, exceeded)
 				return admission.NewForbidden(a,
-					fmt.Errorf("Exceeded quota: %s, requested: %s, used: %s, limited: %s",
+					fmt.Errorf("exceeded quota: %s, requested: %s, used: %s, limited: %s",
 						resourceQuota.Name,
 						prettyPrint(failedRequestedUsage),
 						prettyPrint(failedUsed),

test/e2e/resource_quota.go

@@ -66,7 +66,7 @@ var _ = KubeDescribe("ResourceQuota", func() {
 		Expect(err).NotTo(HaveOccurred())
 
 		By("Creating a Service")
-		service := newTestServiceForQuota("test-service")
+		service := newTestServiceForQuota("test-service", api.ServiceTypeClusterIP)
 		service, err = f.Client.Services(f.Namespace.Name).Create(service)
 		Expect(err).NotTo(HaveOccurred())
 
@@ -131,6 +131,94 @@ var _ = KubeDescribe("ResourceQuota", func() {
 		Expect(err).NotTo(HaveOccurred())
 	})
 
+	It("should create a ResourceQuota and capture the life of a nodePort service.", func() {
+		By("Creating a ResourceQuota")
+		quotaName := "test-quota"
+		resourceQuota := newTestResourceQuota(quotaName)
+		resourceQuota, err := createResourceQuota(f.Client, f.Namespace.Name, resourceQuota)
+		Expect(err).NotTo(HaveOccurred())
+
+		By("Ensuring resource quota status is calculated")
+		usedResources := api.ResourceList{}
+		usedResources[api.ResourceQuotas] = resource.MustParse("1")
+		err = waitForResourceQuota(f.Client, f.Namespace.Name, quotaName, usedResources)
+		Expect(err).NotTo(HaveOccurred())
+
+		By("Creating a NodePort type Service")
+		service := newTestServiceForQuota("test-service", api.ServiceTypeNodePort)
+		service, err = f.Client.Services(f.Namespace.Name).Create(service)
+		Expect(err).NotTo(HaveOccurred())
+
+		By("Ensuring resource quota status captures service creation")
+		usedResources = api.ResourceList{}
+		usedResources[api.ResourceQuotas] = resource.MustParse("1")
+		usedResources[api.ResourceServices] = resource.MustParse("1")
+		usedResources[api.ResourceServicesNodePorts] = resource.MustParse("1")
+		err = waitForResourceQuota(f.Client, f.Namespace.Name, quotaName, usedResources)
+		Expect(err).NotTo(HaveOccurred())
+
+		By("Deleting a Service")
+		err = f.Client.Services(f.Namespace.Name).Delete(service.Name)
+		Expect(err).NotTo(HaveOccurred())
+
+		By("Ensuring resource quota status released usage")
+		usedResources[api.ResourceServices] = resource.MustParse("0")
+		usedResources[api.ResourceServicesNodePorts] = resource.MustParse("0")
+		err = waitForResourceQuota(f.Client, f.Namespace.Name, quotaName, usedResources)
+		Expect(err).NotTo(HaveOccurred())
+	})
+
+	It("should create a ResourceQuota and capture the life of a nodePort service updated to clusterIP.", func() {
+		By("Creating a ResourceQuota")
+		quotaName := "test-quota"
+		resourceQuota := newTestResourceQuota(quotaName)
+		resourceQuota, err := createResourceQuota(f.Client, f.Namespace.Name, resourceQuota)
+		Expect(err).NotTo(HaveOccurred())
+
+		By("Ensuring resource quota status is calculated")
+		usedResources := api.ResourceList{}
+		usedResources[api.ResourceQuotas] = resource.MustParse("1")
+		err = waitForResourceQuota(f.Client, f.Namespace.Name, quotaName, usedResources)
+		Expect(err).NotTo(HaveOccurred())
+
+		By("Creating a NodePort type Service")
+		service := newTestServiceForQuota("test-service", api.ServiceTypeNodePort)
+		service, err = f.Client.Services(f.Namespace.Name).Create(service)
+		Expect(err).NotTo(HaveOccurred())
+
+		By("Ensuring resource quota status captures service creation")
+		usedResources = api.ResourceList{}
+		usedResources[api.ResourceQuotas] = resource.MustParse("1")
+		usedResources[api.ResourceServices] = resource.MustParse("1")
+		usedResources[api.ResourceServicesNodePorts] = resource.MustParse("1")
+		err = waitForResourceQuota(f.Client, f.Namespace.Name, quotaName, usedResources)
+		Expect(err).NotTo(HaveOccurred())
+
+		By("Updating the service type to clusterIP")
+		service.Spec.Type = api.ServiceTypeClusterIP
+		service.Spec.Ports[0].NodePort = 0
+		_, err = f.Client.Services(f.Namespace.Name).Update(service)
+		Expect(err).NotTo(HaveOccurred())
+
+		By("Checking resource quota status capture service update")
+		usedResources = api.ResourceList{}
+		usedResources[api.ResourceQuotas] = resource.MustParse("1")
+		usedResources[api.ResourceServices] = resource.MustParse("1")
+		usedResources[api.ResourceServicesNodePorts] = resource.MustParse("0")
+		err = waitForResourceQuota(f.Client, f.Namespace.Name, quotaName, usedResources)
+		Expect(err).NotTo(HaveOccurred())
+
+		By("Deleting a Service")
+		err = f.Client.Services(f.Namespace.Name).Delete(service.Name)
+		Expect(err).NotTo(HaveOccurred())
+
+		By("Ensuring resource quota status released usage")
+		usedResources[api.ResourceServices] = resource.MustParse("0")
+		usedResources[api.ResourceServicesNodePorts] = resource.MustParse("0")
+		err = waitForResourceQuota(f.Client, f.Namespace.Name, quotaName, usedResources)
+		Expect(err).NotTo(HaveOccurred())
+	})
+
 	It("should create a ResourceQuota and capture the life of a pod.", func() {
 		By("Creating a ResourceQuota")
 		quotaName := "test-quota"
@@ -488,6 +576,7 @@ func newTestResourceQuota(name string) *api.ResourceQuota {
 	hard := api.ResourceList{}
 	hard[api.ResourcePods] = resource.MustParse("5")
 	hard[api.ResourceServices] = resource.MustParse("10")
+	hard[api.ResourceServicesNodePorts] = resource.MustParse("1")
 	hard[api.ResourceReplicationControllers] = resource.MustParse("10")
 	hard[api.ResourceQuotas] = resource.MustParse("1")
 	hard[api.ResourceCPU] = resource.MustParse("1")
@@ -572,12 +661,13 @@ func newTestReplicationControllerForQuota(name, image string, replicas int) *api
 }
 
 // newTestServiceForQuota returns a simple service
-func newTestServiceForQuota(name string) *api.Service {
+func newTestServiceForQuota(name string, serviceType api.ServiceType) *api.Service {
 	return &api.Service{
 		ObjectMeta: api.ObjectMeta{
 			Name: name,
 		},
 		Spec: api.ServiceSpec{
+			Type: serviceType,
 			Ports: []api.ServicePort{{
 				Port:       80,
 				TargetPort: intstr.FromInt(80),