Merge pull request #6481 from mszacillo/quota-exceeded-condition

Add quota exceeded conditions in case scheduler fails to update resourcebinding

Author: karmada-bot

pkg/apis/work/v1alpha2/binding_types.go

@@ -432,6 +432,11 @@ const (
 	// BindingReasonUnschedulable reason in Scheduled condition means that the scheduler can't schedule
 	// the binding right now, for example due to insufficient resources in the clusters.
 	BindingReasonUnschedulable = "Unschedulable"
+
+	// BindingReasonQuotaExceeded reason in Scheduled condition means that the scheduler can't schedule
+	// the binding because the resource requirement exceeds one or more of the FederatedResourceQuotas
+	// defined in the namespace.
+	BindingReasonQuotaExceeded = "QuotaExceeded"
 )
 
 // +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object

pkg/scheduler/helper.go

@@ -19,8 +19,10 @@ package scheduler
 import (
 	"encoding/json"
 	"errors"
+	"net/http"
 	"reflect"
 
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	utilerrors "k8s.io/apimachinery/pkg/util/errors"
 	"k8s.io/klog/v2"
@@ -125,13 +127,19 @@ func getConditionByError(err error) (metav1.Condition, bool) {
 	if fitErrMatcher(err) {
 		return util.NewCondition(workv1alpha2.Scheduled, workv1alpha2.BindingReasonNoClusterFit, err.Error(), metav1.ConditionFalse), true
 	}
+
 	var aggregatedErr utilerrors.Aggregate
 	if errors.As(err, &aggregatedErr) {
 		for _, ae := range aggregatedErr.Errors() {
 			if fitErrMatcher(ae) {
 				// if aggregated NoClusterFit error got, we do not ignore error but retry scheduling.
 				return util.NewCondition(workv1alpha2.Scheduled, workv1alpha2.BindingReasonNoClusterFit, err.Error(), metav1.ConditionFalse), false
 			}
+			// ResourceBinding validation webhook will return error with "FederatedResourceQuota" if quota exceeded
+			var statusErr *apierrors.StatusError
+			if errors.As(ae, &statusErr) && statusErr.Status().Code == http.StatusForbidden && statusErr.Status().Reason == util.QuotaExceededReason {
+				return util.NewCondition(workv1alpha2.Scheduled, workv1alpha2.BindingReasonQuotaExceeded, ae.Error(), metav1.ConditionFalse), false
+			}
 		}
 	}
 	return util.NewCondition(workv1alpha2.Scheduled, workv1alpha2.BindingReasonSchedulerError, err.Error(), metav1.ConditionFalse), false

pkg/util/constants.go

@@ -20,6 +20,7 @@ import (
 	"time"
 
 	discoveryv1 "k8s.io/api/discovery/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 
 	workv1alpha2 "github.com/karmada-io/karmada/pkg/apis/work/v1alpha2"
 )
@@ -216,6 +217,11 @@ const (
 	CompletionsField = "completions"
 )
 
+const (
+	// QuotaExceededReason is a unique reason to describe QuotaExceeded events
+	QuotaExceededReason metav1.StatusReason = "QuotaExceeded"
+)
+
 // ContextKey is the key of context.
 type ContextKey string
 

pkg/webhook/resourcebinding/validating.go

@@ -26,6 +26,7 @@ import (
 	admissionv1 "k8s.io/api/admission/v1"
 	corev1 "k8s.io/api/core/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/client-go/util/retry"
 	"k8s.io/klog/v2"
 	"sigs.k8s.io/controller-runtime/pkg/client"
@@ -309,19 +310,33 @@ func (v *ValidatingAdmission) processSingleFRQ(frqItem policyv1alpha1.FederatedR
 
 	potentialNewOverallUsedForThisFRQ := addResourceLists(frqItem.Status.OverallUsed, deltaForThisFRQ)
 
-	if !isAllowed(potentialNewOverallUsedForThisFRQ, frqItem.Spec.Overall) {
-		errMsg := fmt.Sprintf("Quota exceeded for FederatedResourceQuota %s/%s. ResourceBinding %s/%s will be denied.",
+	isAllowed, errMsg := isAllowed(potentialNewOverallUsedForThisFRQ, frqItem)
+	if !isAllowed {
+		klog.Warningf("Quota exceeded for FederatedResourceQuota %s/%s. ResourceBinding %s/%s will be denied.",
 			frqItem.Namespace, frqItem.Name, rbNamespace, rbName)
-		klog.Warning(errMsg)
-		resp := admission.Denied(errMsg)
-		return nil, "", &resp
+		resp := buildDenyResponse(errMsg)
+		return nil, "", resp
 	}
 
 	msg := fmt.Sprintf("Quota check passed for FRQ %s/%s.", frqItem.Namespace, frqItem.Name)
 	klog.V(3).Infof("FRQ %s/%s will be updated. New OverallUsed: %v", frqItem.Namespace, frqItem.Name, potentialNewOverallUsedForThisFRQ)
 	return potentialNewOverallUsedForThisFRQ, msg, nil
 }
 
+func buildDenyResponse(errMsg string) *admission.Response {
+	resp := admission.Response{
+		AdmissionResponse: admissionv1.AdmissionResponse{
+			Allowed: false,
+			Result: &metav1.Status{
+				Message: errMsg,
+				Reason:  util.QuotaExceededReason,
+				Code:    int32(http.StatusForbidden),
+			},
+		},
+	}
+	return &resp
+}
+
 func calculateResourceUsage(rb *workv1alpha2.ResourceBinding) (corev1.ResourceList, error) {
 	if rb == nil || rb.Spec.ReplicaRequirements == nil || len(rb.Spec.ReplicaRequirements.ResourceRequest) == 0 || len(rb.Spec.Clusters) == 0 {
 		return corev1.ResourceList{}, nil
@@ -437,9 +452,10 @@ func addResourceLists(list1, list2 corev1.ResourceList) corev1.ResourceList {
 	return result
 }
 
-func isAllowed(requested, allowedLimits corev1.ResourceList) bool {
+func isAllowed(requested corev1.ResourceList, frqItem policyv1alpha1.FederatedResourceQuota) (bool, string) {
+	allowedLimits := frqItem.Spec.Overall
 	if allowedLimits == nil {
-		return true
+		return true, ""
 	}
 	for name, reqQty := range requested {
 		if reqQty.IsZero() {
@@ -451,11 +467,12 @@ func isAllowed(requested, allowedLimits corev1.ResourceList) bool {
 			continue
 		}
 		if reqQty.Cmp(limitQty) > 0 {
-			klog.Warningf("Quota exceeded for resource %s: requested sum %s, limit %s", name, reqQty.String(), limitQty.String())
-			return false
+			msg := fmt.Sprintf("FederatedResourceQuota(%s/%s) exceeded for resource %s: requested sum %s, limit %s.", frqItem.Namespace, frqItem.Name, name, reqQty.String(), limitQty.String())
+			klog.Warning(msg)
+			return false, msg
 		}
 	}
-	return true
+	return true, ""
 }
 
 func filterResourceListByKeys(original corev1.ResourceList, filterKeySource corev1.ResourceList) corev1.ResourceList {

pkg/webhook/resourcebinding/validating_test.go

@@ -405,10 +405,7 @@ func TestValidatingAdmission_Handle(t *testing.T) {
 			decoder:            &fakeDecoder{decodeObj: rbCreateExceeds},
 			clientObjects:      []client.Object{frqForCreateExceeds},
 			featureGateEnabled: true,
-			wantResponse: admission.Denied(
-				fmt.Sprintf("Quota exceeded for FederatedResourceQuota %s/%s. ResourceBinding %s/%s will be denied.",
-					frqForCreateExceeds.Namespace, frqForCreateExceeds.Name, rbCreateExceeds.Namespace, rbCreateExceeds.Name),
-			),
+			wantResponse:       admission.Denied("FederatedResourceQuota(quota-ns/frq-create-exceeds) exceeded for resource cpu: requested sum 200m, limit 150m."),
 		},
 		{
 			name: "update passes quota (allowed response, non-dryrun)",
@@ -449,10 +446,7 @@ func TestValidatingAdmission_Handle(t *testing.T) {
 			decoder:            &fakeDecoder{decodeObj: rbUpdateFailNew, rawDecodedObj: rbUpdateFailOld},
 			clientObjects:      []client.Object{frqForUpdateFail},
 			featureGateEnabled: true,
-			wantResponse: admission.Denied(
-				fmt.Sprintf("Quota exceeded for FederatedResourceQuota %s/%s. ResourceBinding %s/%s will be denied.",
-					frqForUpdateFail.Namespace, frqForUpdateFail.Name, rbUpdateFailNew.Namespace, rbUpdateFailNew.Name),
-			),
+			wantResponse:       admission.Denied("FederatedResourceQuota(quota-ns/frq-update-fail) exceeded for resource cpu: requested sum 110m, limit 100m."),
 		},
 	}