Merge pull request #12600 from derekwaynecarr/resource_quota_request

pkg/api/resource/quantity.go

@@ -309,6 +309,14 @@ func (q *Quantity) Add(y Quantity) error {
 	return nil
 }
 
+func (q *Quantity) Sub(y Quantity) error {
+	if q.Format != y.Format {
+		return fmt.Errorf("format mismatch: %v vs. %v", q.Format, y.Format)
+	}
+	q.Amount.Sub(q.Amount, y.Amount)
+	return nil
+}
+
 // MarshalJSON implements the json.Marshaller interface.
 func (q Quantity) MarshalJSON() ([]byte, error) {
 	return []byte(`"` + q.String() + `"`), nil

pkg/controller/resourcequota/resource_quota_controller.go

@@ -17,6 +17,7 @@ limitations under the License.
 package resourcequotacontroller
 
 import (
+	"fmt"
 	"time"
 
 	"github.com/golang/glog"
@@ -163,18 +164,6 @@ func (rm *ResourceQuotaController) syncResourceQuota(quota api.ResourceQuota) (e
 		switch k {
 		case api.ResourcePods:
 			value = resource.NewQuantity(int64(len(filteredPods)), resource.DecimalSI)
-		case api.ResourceMemory:
-			val := int64(0)
-			for _, pod := range filteredPods {
-				val = val + PodMemory(pod).Value()
-			}
-			value = resource.NewQuantity(int64(val), resource.DecimalSI)
-		case api.ResourceCPU:
-			val := int64(0)
-			for _, pod := range filteredPods {
-				val = val + PodCPU(pod).MilliValue()
-			}
-			value = resource.NewMilliQuantity(int64(val), resource.DecimalSI)
 		case api.ResourceServices:
 			items, err := rm.kubeClient.Services(usage.Namespace).List(labels.Everything())
 			if err != nil {
@@ -205,6 +194,10 @@ func (rm *ResourceQuotaController) syncResourceQuota(quota api.ResourceQuota) (e
 				return err
 			}
 			value = resource.NewQuantity(int64(len(items.Items)), resource.DecimalSI)
+		case api.ResourceMemory:
+			value = PodsRequests(filteredPods, api.ResourceMemory)
+		case api.ResourceCPU:
+			value = PodsRequests(filteredPods, api.ResourceCPU)
 		}
 
 		// ignore fields we do not understand (assume another controller is tracking it)
@@ -224,38 +217,83 @@ func (rm *ResourceQuotaController) syncResourceQuota(quota api.ResourceQuota) (e
 	return nil
 }
 
-// PodCPU computes total cpu usage of a pod
-func PodCPU(pod *api.Pod) *resource.Quantity {
-	val := int64(0)
-	for j := range pod.Spec.Containers {
-		val = val + pod.Spec.Containers[j].Resources.Limits.Cpu().MilliValue()
+// PodsRequests returns sum of each resource request for each pod in list
+// If a given pod in the list does not have a request for the named resource, we log the error
+// but still attempt to get the most representative count
+func PodsRequests(pods []*api.Pod, resourceName api.ResourceName) *resource.Quantity {
+	var sum *resource.Quantity
+	for i := range pods {
+		pod := pods[i]
+		podQuantity, err := PodRequests(pod, resourceName)
+		if err != nil {
+			// log the error, but try to keep the most accurate count possible in log
+			// rationale here is that you may have had pods in a namespace that did not have
+			// explicit requests prior to adding the quota
+			glog.Infof("No explicit request for resource, pod %s/%s, %s", pod.Namespace, pod.Name, resourceName)
+		} else {
+			if sum == nil {
+				sum = podQuantity
+			} else {
+				sum.Add(*podQuantity)
+			}
+		}
 	}
-	return resource.NewMilliQuantity(int64(val), resource.DecimalSI)
+	// if list is empty
+	if sum == nil {
+		q := resource.MustParse("0")
+		sum = &q
+	}
+	return sum
 }
 
-// IsPodCPUUnbounded returns true if the cpu use is unbounded for any container in pod
-func IsPodCPUUnbounded(pod *api.Pod) bool {
+// PodRequests returns sum of each resource request across all containers in pod
+func PodRequests(pod *api.Pod, resourceName api.ResourceName) (*resource.Quantity, error) {
+	if !PodHasRequests(pod, resourceName) {
+		return nil, fmt.Errorf("Each container in pod %s/%s does not have an explicit request for resource %s.", pod.Namespace, pod.Name, resourceName)
+	}
+	var sum *resource.Quantity
 	for j := range pod.Spec.Containers {
-		container := pod.Spec.Containers[j]
-		if container.Resources.Limits.Cpu().MilliValue() == int64(0) {
-			return true
+		value, _ := pod.Spec.Containers[j].Resources.Requests[resourceName]
+		if sum == nil {
+			sum = value.Copy()
+		} else {
+			err := sum.Add(value)
+			if err != nil {
+				return sum, err
+			}
 		}
 	}
-	return false
+	// if list is empty
+	if sum == nil {
+		q := resource.MustParse("0")
+		sum = &q
+	}
+	return sum, nil
 }
 
-// IsPodMemoryUnbounded returns true if the memory use is unbounded for any container in pod
-func IsPodMemoryUnbounded(pod *api.Pod) bool {
+// PodHasRequests verifies that each container in the pod has an explicit request that is non-zero for a named resource
+func PodHasRequests(pod *api.Pod, resourceName api.ResourceName) bool {
 	for j := range pod.Spec.Containers {
-		container := pod.Spec.Containers[j]
-		if container.Resources.Limits.Memory().Value() == int64(0) {
-			return true
+		value, valueSet := pod.Spec.Containers[j].Resources.Requests[resourceName]
+		if !valueSet || value.Value() == int64(0) {
+			return false
 		}
 	}
-	return false
+	return true
+}
+
+// PodCPU computes total cpu limit across all containers in pod
+// TODO: Remove this once the mesos scheduler becomes request aware
+func PodCPU(pod *api.Pod) *resource.Quantity {
+	val := int64(0)
+	for j := range pod.Spec.Containers {
+		val = val + pod.Spec.Containers[j].Resources.Limits.Cpu().MilliValue()
+	}
+	return resource.NewMilliQuantity(int64(val), resource.DecimalSI)
 }
 
-// PodMemory computes the memory usage of a pod
+// PodMemory computes total memory limit across all containers in a pod
+// TODO: Remove this once the mesos scheduler becomes request aware
 func PodMemory(pod *api.Pod) *resource.Quantity {
 	val := int64(0)
 	for j := range pod.Spec.Containers {

plugin/pkg/admission/resourcequota/admission.go

@@ -190,52 +190,69 @@ func IncrementUsage(a admission.Attributes, status *api.ResourceQuotaStatus, cli
 			}
 		}
 	}
-	// handle memory/cpu constraints, and any diff of usage based on memory/cpu on updates
-	if a.GetResource() == "pods" && (set[api.ResourceMemory] || set[api.ResourceCPU]) {
-		pod := obj.(*api.Pod)
-		deltaCPU := resourcequotacontroller.PodCPU(pod)
-		deltaMemory := resourcequotacontroller.PodMemory(pod)
-		// if this is an update, we need to find the delta cpu/memory usage from previous state
-		if a.GetOperation() == admission.Update {
-			oldPod, err := client.Pods(a.GetNamespace()).Get(pod.Name)
-			if err != nil {
-				return false, err
+
+	if a.GetResource() == "pods" {
+		for _, resourceName := range []api.ResourceName{api.ResourceMemory, api.ResourceCPU} {
+
+			// ignore tracking the resource if its not in the quota document
+			if !set[resourceName] {
+				continue
 			}
-			oldCPU := resourcequotacontroller.PodCPU(oldPod)
-			oldMemory := resourcequotacontroller.PodMemory(oldPod)
-			deltaCPU = resource.NewMilliQuantity(deltaCPU.MilliValue()-oldCPU.MilliValue(), resource.DecimalSI)
-			deltaMemory = resource.NewQuantity(deltaMemory.Value()-oldMemory.Value(), resource.DecimalSI)
-		}
 
-		hardMem, hardMemFound := status.Hard[api.ResourceMemory]
-		if hardMemFound {
-			if set[api.ResourceMemory] && resourcequotacontroller.IsPodMemoryUnbounded(pod) {
-				return false, fmt.Errorf("Limited to %s memory, but pod has no specified memory limit", hardMem.String())
+			hard, hardFound := status.Hard[resourceName]
+			if !hardFound {
+				continue
 			}
-			used, usedFound := status.Used[api.ResourceMemory]
+
+			// if we do not yet know how much of the current resource is used, we cannot accept any request
+			used, usedFound := status.Used[resourceName]
 			if !usedFound {
-				return false, fmt.Errorf("Quota usage stats are not yet known, unable to admit resource until an accurate count is completed.")
+				return false, fmt.Errorf("Unable to admit pod until quota usage stats are calculated.")
 			}
-			if used.Value()+deltaMemory.Value() > hardMem.Value() {
-				return false, fmt.Errorf("Limited to %s memory", hardMem.String())
-			} else {
-				status.Used[api.ResourceMemory] = *resource.NewQuantity(used.Value()+deltaMemory.Value(), resource.DecimalSI)
-				dirty = true
+
+			// the amount of resource being requested, or an error if it does not make a request that is tracked
+			pod := obj.(*api.Pod)
+			delta, err := resourcequotacontroller.PodRequests(pod, resourceName)
+
+			if err != nil {
+				return false, fmt.Errorf("Must make a non-zero request for %s since it is tracked by quota.", resourceName)
 			}
-		}
-		hardCPU, hardCPUFound := status.Hard[api.ResourceCPU]
-		if hardCPUFound {
-			if set[api.ResourceCPU] && resourcequotacontroller.IsPodCPUUnbounded(pod) {
-				return false, fmt.Errorf("Limited to %s CPU, but pod has no specified cpu limit", hardCPU.String())
+
+			// if this operation is an update, we need to find the delta usage from the previous state
+			if a.GetOperation() == admission.Update {
+				oldPod, err := client.Pods(a.GetNamespace()).Get(pod.Name)
+				if err != nil {
+					return false, err
+				}
+
+				// if the previous version of the resource made a resource request, we need to subtract the old request
+				// from the current to get the actual resource request delta.  if the previous version of the pod
+				// made no request on the resource, then we get an err value.  we ignore the err value, and delta
+				// will just be equal to the total resource request on the pod since there is nothing to subtract.
+				oldRequest, err := resourcequotacontroller.PodRequests(oldPod, resourceName)
+				if err == nil {
+					err = delta.Sub(*oldRequest)
+					if err != nil {
+						return false, err
+					}
+				}
 			}
-			used, usedFound := status.Used[api.ResourceCPU]
-			if !usedFound {
-				return false, fmt.Errorf("Quota usage stats are not yet known, unable to admit resource until an accurate count is completed.")
+
+			newUsage := used.Copy()
+			newUsage.Add(*delta)
+
+			// make the most precise comparison possible
+			newUsageValue := newUsage.Value()
+			hardUsageValue := hard.Value()
+			if newUsageValue <= resource.MaxMilliValue && hardUsageValue <= resource.MaxMilliValue {
+				newUsageValue = newUsage.MilliValue()
+				hardUsageValue = hard.MilliValue()
 			}
-			if used.MilliValue()+deltaCPU.MilliValue() > hardCPU.MilliValue() {
-				return false, fmt.Errorf("Limited to %s CPU", hardCPU.String())
+
+			if newUsageValue > hardUsageValue {
+				return false, fmt.Errorf("Unable to admit pod without exceeding quota for resource %s.  Limited to %s but require %s to succeed.", resourceName, hard.String(), newUsage.String())
 			} else {
-				status.Used[api.ResourceCPU] = *resource.NewMilliQuantity(used.MilliValue()+deltaCPU.MilliValue(), resource.DecimalSI)
+				status.Used[resourceName] = *newUsage
 				dirty = true
 			}
 		}