Files
Oliver BählerandGitHub 2252c530f4 feat(performance): removed duplicate client calls from all admission paths (#2054)
* chore

* perfromance improvements

Signed-off-by: Oliver Baehler <oliver@sudo-i.net>

* perfromance improvements

Signed-off-by: Oliver Baehler <oliver@sudo-i.net>

* feat(performance): removed duplicate client calls from all admission paths

Signed-off-by: Oliver Baehler <oliver@sudo-i.net>

* feat(performance): removed duplicate client calls from all admission paths

Signed-off-by: Oliver Baehler <oliver@sudo-i.net>

* feat(performance): removed duplicate client calls from all admission paths

Signed-off-by: Oliver Baehler <oliver@sudo-i.net>

* feat(performance): removed duplicate client calls from all admission paths

Signed-off-by: Oliver Baehler <oliver@sudo-i.net>

* feat(performance): removed duplicate client calls from all admission paths

Signed-off-by: Oliver Baehler <oliver@sudo-i.net>

* feat(performance): removed duplicate client calls from all admission paths

Signed-off-by: Oliver Baehler <oliver@sudo-i.net>

* feat(performance): removed duplicate client calls from all admission paths

Signed-off-by: Oliver Baehler <oliver@sudo-i.net>

---------

Signed-off-by: Oliver Baehler <oliver@sudo-i.net>
2026-07-28 14:32:42 +02:00

1352 lines
32 KiB
Go

// Copyright 2020-2026 Project Capsule Authors
// SPDX-License-Identifier: Apache-2.0
package customquota
import (
"context"
"fmt"
"reflect"
"slices"
"sort"
"time"
corev1 "k8s.io/api/core/v1"
apierrors "k8s.io/apimachinery/pkg/api/errors"
k8smeta "k8s.io/apimachinery/pkg/api/meta"
"k8s.io/apimachinery/pkg/api/resource"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
"k8s.io/apimachinery/pkg/conversion"
"k8s.io/apimachinery/pkg/labels"
"k8s.io/apimachinery/pkg/runtime"
"k8s.io/apimachinery/pkg/types"
"k8s.io/apimachinery/pkg/util/wait"
"k8s.io/client-go/util/retry"
"sigs.k8s.io/controller-runtime/pkg/client"
"sigs.k8s.io/controller-runtime/pkg/log"
"sigs.k8s.io/controller-runtime/pkg/webhook/admission"
capsulev1beta2 "github.com/projectcapsule/capsule/api/v1beta2"
"github.com/projectcapsule/capsule/internal/cache"
controller "github.com/projectcapsule/capsule/internal/controllers/customquotas"
"github.com/projectcapsule/capsule/pkg/api/meta"
ad "github.com/projectcapsule/capsule/pkg/runtime/admission"
"github.com/projectcapsule/capsule/pkg/runtime/configuration"
"github.com/projectcapsule/capsule/pkg/runtime/events"
"github.com/projectcapsule/capsule/pkg/runtime/handlers"
"github.com/projectcapsule/capsule/pkg/runtime/quota"
"github.com/projectcapsule/capsule/pkg/runtime/selectors"
)
// Might need some tuning in the.
var customAdmissionBackoff = wait.Backoff{
Steps: 6,
Duration: 20 * time.Millisecond,
Factor: 1.5,
Jitter: 0.2,
}
var ledgerMutationBackoff = wait.Backoff{
Steps: 8,
Duration: 10 * time.Millisecond,
Factor: 1.6,
Jitter: 0.2,
}
type objectCalculationHandler struct {
targetsCache *cache.CompiledTargetsCache[string]
jsonPathCache *cache.JSONPathCache
celCache *cache.CELCache
}
func ObjectCalculationHandler(
targetsCache *cache.CompiledTargetsCache[string],
jsonPathCache *cache.JSONPathCache,
celCache *cache.CELCache,
) handlers.Handler {
return &objectCalculationHandler{
targetsCache: targetsCache,
jsonPathCache: jsonPathCache,
celCache: celCache,
}
}
func (h *objectCalculationHandler) OnCreate(
c client.Client,
reader client.Reader,
decoder admission.Decoder,
recorder events.EventRecorder,
) handlers.Func {
return func(ctx context.Context, req admission.Request) *admission.Response {
dryRun := req.DryRun != nil && *req.DryRun
log := log.FromContext(ctx).WithValues(
"op", "create",
"kind", req.Kind.String(),
"namespace", req.Namespace,
"requestUID", string(req.UID),
"name", req.Name,
)
u, err := getUnstructured(req.Object)
if err != nil {
return ad.ErroredResponse(err)
}
var finalResp *admission.Response
err = retry.OnError(customAdmissionBackoff, apierrors.IsConflict, func() error {
matched, err := h.matchAllQuotas(ctx, reader, req, u)
if err != nil {
finalResp = ad.ErroredResponse(err)
return nil
}
if len(matched) == 0 {
return nil
}
evaluated, err := h.evaluateMatchedQuotas(ctx, u, matched)
if err != nil {
finalResp = ad.Denyf(
"creating resource %s/%s (%s) cannot be admitted because custom quota usage could not be calculated: %v",
req.Namespace,
req.Name,
req.Kind.String(),
err,
)
return nil
}
type appliedReservation struct {
LedgerKey types.NamespacedName
ReservationID string
}
applied := make([]appliedReservation, 0, len(evaluated))
for _, item := range evaluated {
ledgerKey := quantityLedgerKeyForMatchedQuota(item)
reservation := buildReservation(req, u, item.Usage, item.Usage, item.Key)
allowed, effectiveUsed, reserved, err := reserveCreateOnLedger(
ctx,
c,
reader,
item,
&reservation,
dryRun,
)
if err != nil {
for _, a := range applied {
_ = deleteLedgerReservation(ctx, c, reader, a.LedgerKey, a.ReservationID)
}
return err
}
if !allowed {
for _, a := range applied {
_ = deleteLedgerReservation(ctx, c, reader, a.LedgerKey, a.ReservationID)
}
available := item.Limit.DeepCopy()
available.Sub(effectiveUsed)
if available.Sign() < 0 {
available = resource.MustParse("0")
}
log.V(5).Info("denying create due to quota",
"quotaKey", item.Key,
"quotaName", item.Name,
"isGlobal", item.IsGlobal,
"requestedUsage", item.Usage.String(),
"currentUsed", effectiveUsed.String(),
"available", available.String(),
"limit", item.Limit.String(),
"inflightReserved", reserved.String(),
)
finalResp = ad.Denyf(
"creating resource exceeds limit for %s %q (requested=%s, currentUsed=%s, available=%s, limit=%s, inflightReserved=%s)",
quotaTypeName(item.IsGlobal),
item.Name,
item.Usage.String(),
effectiveUsed.String(),
available.String(),
item.Limit.String(),
reserved.String(),
)
return nil
}
if !dryRun {
applied = append(applied, appliedReservation{
LedgerKey: ledgerKey,
ReservationID: reservation.ID,
})
}
}
finalResp = nil
return nil
})
if err != nil {
if apierrors.IsConflict(err) {
return ad.Denyf(
"custom quota admission could not reserve usage due to concurrent quota updates after %d attempts; please retry the request: %v",
customAdmissionBackoff.Steps,
err,
)
}
return ad.ErroredResponse(err)
}
return finalResp
}
}
//nolint:gocognit,gocyclo,cyclop,maintidx
func (h *objectCalculationHandler) OnUpdate(
c client.Client,
reader client.Reader,
_ admission.Decoder,
recorder events.EventRecorder,
) handlers.Func {
return func(ctx context.Context, req admission.Request) *admission.Response {
dryRun := req.DryRun != nil && *req.DryRun
statusUpdate := req.SubResource == "status"
logger := log.FromContext(ctx).WithValues(
"op", "update",
"kind", req.Kind.String(),
"namespace", req.Namespace,
"name", req.Name,
"subresource", req.SubResource,
)
terminating, namespaceErr := namespaceTerminating(ctx, c, req.Namespace)
if namespaceErr != nil {
logger.Error(namespaceErr, "cannot determine whether namespace is terminating")
} else if terminating {
logger.V(5).Info("allowing update without quota processing because namespace is terminating")
return nil
}
oldObj, err := getUnstructured(req.OldObject)
if err != nil {
if statusUpdate {
logger.Error(err, "allowing status update because the previous object could not be decoded")
return nil
}
return ad.ErroredResponse(err)
}
newObj, err := getUnstructured(req.Object)
if err != nil {
if statusUpdate {
logger.Error(err, "allowing status update because the new object could not be decoded")
return nil
}
return ad.ErroredResponse(err)
}
var finalResp *admission.Response
err = retry.OnError(customAdmissionBackoff, apierrors.IsConflict, func() error {
policies, err := loadQuotaPolicySnapshot(ctx, reader, req.Namespace)
if err != nil {
if statusUpdate {
logger.Error(err, "allowing status update because quota policies could not be loaded")
finalResp = nil
return nil
}
finalResp = ad.ErroredResponse(err)
return nil
}
oldMatched, err := h.matchAllQuotasFromSnapshot(ctx, reader, req, oldObj, policies)
if err != nil {
if statusUpdate {
logger.Error(err, "allowing status update because previous quota matches could not be evaluated")
finalResp = nil
return nil
}
finalResp = ad.ErroredResponse(err)
return nil
}
newMatched, err := h.matchAllQuotasFromSnapshot(ctx, reader, req, newObj, policies)
if err != nil {
if statusUpdate {
logger.Error(err, "allowing status update because new quota matches could not be evaluated")
finalResp = nil
return nil
}
finalResp = ad.ErroredResponse(err)
return nil
}
oldEvaluated, err := h.evaluateMatchedQuotas(ctx, oldObj, oldMatched)
if err != nil {
if statusUpdate {
logger.Error(err, "allowing status update because previous quota usage could not be calculated")
finalResp = nil
return nil
}
finalResp = ad.Denyf(
"updating resource %s/%s (%s) cannot be admitted because previous custom quota usage could not be calculated: %v",
req.Namespace,
req.Name,
req.Kind.String(),
err,
)
return nil
}
newEvaluated, err := h.evaluateMatchedQuotas(ctx, newObj, newMatched)
if err != nil {
if statusUpdate {
logger.Error(err, "allowing status update because new quota usage could not be calculated")
finalResp = nil
return nil
}
finalResp = ad.Denyf(
"updating resource %s/%s (%s) cannot be admitted because new custom quota usage could not be calculated: %v",
req.Namespace,
req.Name,
req.Kind.String(),
err,
)
return nil
}
oldByKey := evaluatedByKey(oldEvaluated)
newByKey := evaluatedByKey(newEvaluated)
relevantChange := meta.LabelsChangedUnstructured(oldObj, newObj) || len(oldByKey) != len(newByKey)
if !relevantChange {
for key, oldItem := range oldByKey {
newItem, ok := newByKey[key]
if !ok || oldItem.Usage.Cmp(newItem.Usage) != 0 {
relevantChange = true
break
}
}
}
if !relevantChange {
finalResp = nil
return nil
}
type appliedUpdate struct {
LedgerKey types.NamespacedName
ReservationID string
OldUsage resource.Quantity
NewUsage resource.Quantity
PendingDelete *capsulev1beta2.QuantityLedgerPendingDelete
}
applied := make([]appliedUpdate, 0, len(oldByKey)+len(newByKey))
for _, key := range allKeys(oldByKey, newByKey) {
oldItem, hadOld := oldByKey[key]
newItem, hadNew := newByKey[key]
var base evaluatedQuota
switch {
case hadNew:
base = newItem
case hadOld:
base = oldItem
default:
continue
}
oldUsage := resource.MustParse("0")
if hadOld {
oldUsage = oldItem.Usage.DeepCopy()
}
newUsage := resource.MustParse("0")
if hadNew {
newUsage = newItem.Usage.DeepCopy()
}
ledgerKey := quantityLedgerKeyForMatchedQuota(base)
var pendingDelete *capsulev1beta2.QuantityLedgerPendingDelete
if hadOld && !hadNew {
pendingDelete = &capsulev1beta2.QuantityLedgerPendingDelete{
ID: fmt.Sprintf("%s/%s", req.UID, base.Key),
ObjectRef: capsulev1beta2.QuantityLedgerObjectRef{
APIGroup: req.Kind.Group,
APIVersion: req.Kind.Version,
Kind: req.Kind.Kind,
Namespace: oldObj.GetNamespace(),
Name: oldObj.GetName(),
UID: oldObj.GetUID(),
},
}
}
var reservation *capsulev1beta2.QuantityLedgerReservation
if hadNew {
delta := newUsage.DeepCopy()
delta.Sub(oldUsage)
quota.ClampQuantityToZero(&delta)
// Status is observed state owned by Kubernetes controllers.
// It must notify quota reconciliation, but it must never
// reserve capacity or be rejected for exceeding a quota.
if statusUpdate {
delta = resource.MustParse("0")
}
r := buildReservation(req, newObj, newUsage, delta, base.Key)
reservation = &r
}
allowed, effectiveUsed, reserved, err := replaceUsageOnLedger(
ctx,
c,
reader,
base,
oldUsage,
newUsage,
reservation,
pendingDelete,
!statusUpdate,
dryRun,
)
if err != nil {
for _, v := range slices.Backward(applied) {
_ = rollbackUsageReplacementOnLedger(
ctx,
c,
reader,
v.LedgerKey,
v.ReservationID,
v.OldUsage,
v.NewUsage,
v.PendingDelete,
)
}
return err
}
if !allowed {
for _, v := range slices.Backward(applied) {
_ = rollbackUsageReplacementOnLedger(
ctx,
c,
reader,
v.LedgerKey,
v.ReservationID,
v.OldUsage,
v.NewUsage,
v.PendingDelete,
)
}
if statusUpdate {
logger.Info("allowing status update despite quota limit")
finalResp = nil
return nil
}
available := base.Limit.DeepCopy()
available.Sub(effectiveUsed)
if available.Sign() < 0 {
available = resource.MustParse("0")
}
finalResp = ad.Denyf(
"updating resource exceeds limit for %s %q (requested=%s, currentUsed=%s, available=%s, limit=%s, inflightReserved=%s)",
quotaTypeName(base.IsGlobal),
base.Name,
newUsage.String(),
effectiveUsed.String(),
available.String(),
base.Limit.String(),
reserved.String(),
)
return nil
}
reservationID := ""
if reservation != nil {
reservationID = reservation.ID
}
if !dryRun {
applied = append(applied, appliedUpdate{
LedgerKey: ledgerKey,
ReservationID: reservationID,
OldUsage: oldUsage.DeepCopy(),
NewUsage: newUsage.DeepCopy(),
PendingDelete: pendingDelete,
})
}
}
finalResp = nil
return nil
})
if err != nil {
if statusUpdate {
logger.Error(err, "allowing status update because quota reconciliation could not be queued")
return nil
}
if apierrors.IsConflict(err) {
return ad.Denyf(
"custom quota admission could not reserve usage due to concurrent quota updates after %d attempts; please retry the request: %v",
customAdmissionBackoff.Steps,
err,
)
}
return ad.ErroredResponse(err)
}
return finalResp
}
}
func (h *objectCalculationHandler) OnDelete(
c client.Client,
reader client.Reader,
_ admission.Decoder,
recorder events.EventRecorder,
) handlers.Func {
return func(ctx context.Context, req admission.Request) *admission.Response {
if req.DryRun != nil && *req.DryRun {
return nil
}
logger := log.FromContext(ctx).WithValues(
"op", "delete",
"kind", req.Kind.String(),
"namespace", req.Namespace,
"name", req.Name,
)
// Namespace termination can fan out into many object deletions at once.
// Use the local informer for this best-effort fast path so those deletes
// do not each add an API read before quota processing can be skipped.
// A stale non-terminating result only falls back to the normal,
// conservative ledger path.
terminating, err := namespaceTerminating(ctx, c, req.Namespace)
if err != nil {
logger.Error(err, "cannot determine whether namespace is terminating")
} else if terminating {
logger.V(5).Info("allowing delete without quota processing because namespace is terminating")
return nil
}
oldObj, err := getUnstructured(req.OldObject)
if err != nil {
return ad.ErroredResponse(err)
}
uid := oldObj.GetUID()
if uid == "" {
return nil
}
objRef := capsulev1beta2.QuantityLedgerObjectRef{
APIGroup: req.Kind.Group,
APIVersion: req.Kind.Version,
Kind: req.Kind.Kind,
Namespace: oldObj.GetNamespace(),
Name: oldObj.GetName(),
UID: uid,
}
namespacedcq := &capsulev1beta2.CustomQuotaList{}
if err := reader.List(ctx, namespacedcq, client.InNamespace(req.Namespace)); err != nil {
return ad.ErroredResponse(err)
}
for _, nscq := range namespacedcq.Items {
if !nscq.Status.HasClaimUID(uid) {
continue
}
ledgerKey := types.NamespacedName{
Name: nscq.GetName(),
Namespace: nscq.GetNamespace(),
}
if err := addLedgerPendingDelete(ctx, c, reader, ledgerKey, objRef); err != nil {
return ad.ErroredResponse(err)
}
}
globalcq := &capsulev1beta2.GlobalCustomQuotaList{}
if err := reader.List(ctx, globalcq); err != nil {
return ad.ErroredResponse(err)
}
for _, gcq := range globalcq.Items {
if !gcq.Status.HasClaimUID(uid) {
continue
}
ledgerKey := types.NamespacedName{
Name: gcq.GetName(),
Namespace: configuration.ControllerNamespace(),
}
if err := addLedgerPendingDelete(ctx, c, reader, ledgerKey, objRef); err != nil {
return ad.ErroredResponse(err)
}
}
return nil
}
}
func deleteLedgerReservation(
ctx context.Context,
c client.Client,
reader client.Reader,
ledgerKey types.NamespacedName,
reservationID string,
) error {
return retry.RetryOnConflict(retry.DefaultBackoff, func() error {
ledger := &capsulev1beta2.QuantityLedger{}
if err := reader.Get(ctx, ledgerKey, ledger); err != nil {
if apierrors.IsNotFound(err) {
return nil
}
return err
}
active := make([]capsulev1beta2.QuantityLedgerReservation, 0, len(ledger.Status.Reservations))
released := resource.MustParse("0")
for _, res := range ledger.Status.Reservations {
if res.ID == reservationID {
released.Add(reservationDelta(res))
continue
}
active = append(active, res)
}
if released.Sign() == 0 {
return nil
}
allocated := ledger.Status.Allocated.DeepCopy()
allocated.Sub(released)
quota.ClampQuantityToZero(&allocated)
reserved := resource.MustParse("0")
for _, res := range active {
reserved.Add(reservationDelta(res))
}
ledger.Status.Reservations = active
ledger.Status.Reserved = reserved
ledger.Status.Allocated = allocated
return c.Status().Update(ctx, ledger)
})
}
type quotaPolicySnapshot struct {
namespaced []capsulev1beta2.CustomQuota
global []capsulev1beta2.GlobalCustomQuota
}
func loadQuotaPolicySnapshot(
ctx context.Context,
reader client.Reader,
namespace string,
) (quotaPolicySnapshot, error) {
snapshot := quotaPolicySnapshot{}
// Correctness requires an authoritative policy set. A single snapshot is
// also reused for both sides of UPDATE admission so transient readiness
// changes cannot turn an unchanged object into a false quota transition.
if namespace != "" {
list := &capsulev1beta2.CustomQuotaList{}
if err := reader.List(ctx, list, client.InNamespace(namespace)); err != nil {
return quotaPolicySnapshot{}, err
}
snapshot.namespaced = list.Items
}
global := &capsulev1beta2.GlobalCustomQuotaList{}
if err := reader.List(ctx, global); err != nil {
return quotaPolicySnapshot{}, err
}
snapshot.global = global.Items
return snapshot, nil
}
func (h *objectCalculationHandler) matchAllQuotas(
ctx context.Context,
reader client.Reader,
req admission.Request,
u unstructured.Unstructured,
) ([]quota.MatchedQuota, error) {
snapshot, err := loadQuotaPolicySnapshot(ctx, reader, req.Namespace)
if err != nil {
return nil, err
}
return h.matchAllQuotasFromSnapshot(ctx, reader, req, u, snapshot)
}
func (h *objectCalculationHandler) matchAllQuotasFromSnapshot(
ctx context.Context,
reader client.Reader,
req admission.Request,
u unstructured.Unstructured,
snapshot quotaPolicySnapshot,
) ([]quota.MatchedQuota, error) {
namespaced, err := h.matchCustomQuotasFromItems(ctx, req, u, snapshot.namespaced)
if err != nil {
return nil, err
}
global, err := h.matchGlobalCustomQuotasFromItems(ctx, reader, req, u, snapshot.global)
if err != nil {
return nil, err
}
out := make([]quota.MatchedQuota, 0, len(namespaced)+len(global))
out = append(out, namespaced...)
out = append(out, global...)
sort.SliceStable(out, func(i, j int) bool {
if out[i].Limit.Cmp(out[j].Limit) != 0 {
return out[i].Limit.Cmp(out[j].Limit) < 0
}
if out[i].IsGlobal != out[j].IsGlobal {
return out[i].IsGlobal
}
if out[i].Namespace != out[j].Namespace {
return out[i].Namespace < out[j].Namespace
}
if out[i].SourceRank != out[j].SourceRank {
return out[i].SourceRank < out[j].SourceRank
}
return out[i].Name < out[j].Name
})
return out, nil
}
func (h *objectCalculationHandler) matchCustomQuotasFromItems(
ctx context.Context,
req admission.Request,
u unstructured.Unstructured,
items []capsulev1beta2.CustomQuota,
) ([]quota.MatchedQuota, error) {
if req.Namespace == "" || len(items) == 0 {
return nil, nil
}
objLabels := labels.Set(u.GetLabels())
out := make([]quota.MatchedQuota, 0)
for _, cq := range items {
if !sourcesTargetKind(cq.Spec.Sources, req.Kind) {
continue
}
if !customQuotaReadyForAdmission(cq.Generation, cq.Status) {
// Status is observed state and cannot be blocked by quota policy.
// A NotReady quota has no reliable selector/usage model to notify,
// so skip it and allow Kubernetes to persist the status update.
if req.SubResource == "status" {
continue
}
return nil, fmt.Errorf(
"CustomQuota %s/%s is not ready for generation %d",
cq.Namespace,
cq.Name,
cq.Generation,
)
}
if !selectors.MatchesSelectors(objLabels, cq.Spec.ScopeSelectors) {
continue
}
compiledTargets, err := h.getOrCompileCustomQuotaTargets(&cq)
if err != nil {
return nil, fmt.Errorf("compile targets for CustomQuota %s/%s: %w", cq.Namespace, cq.Name, err)
}
for i, target := range compiledTargets {
if target.Group != req.Kind.Group ||
target.Version != req.Kind.Version ||
target.Kind != req.Kind.Kind {
continue
}
matches, err := controller.MatchesCompiledSelectorsWithFields(ctx, u, target.CompiledSelectors)
if err != nil {
return nil, fmt.Errorf(
"evaluate selectors for %s/%s on CustomQuota %s/%s: %w",
u.GetNamespace(),
u.GetName(),
cq.Namespace,
cq.Name,
err,
)
}
if !matches {
continue
}
out = append(out, quota.MatchedQuota{
Key: controller.MakeCustomQuotaCacheKey(cq.Namespace, cq.Name),
Name: cq.Name,
Namespace: cq.Namespace,
Path: target.Path,
CompiledPath: target.CompiledPath,
CEL: target.CEL,
CompiledCEL: target.CompiledCEL,
Operation: target.Operation,
Limit: cq.Spec.Limit.DeepCopy(),
Used: cq.Status.Usage.Used.DeepCopy(),
IsGlobal: false,
SourceRank: i,
})
}
}
return out, nil
}
func (h *objectCalculationHandler) matchGlobalCustomQuotasFromItems(
ctx context.Context,
reader client.Reader,
req admission.Request,
u unstructured.Unstructured,
items []capsulev1beta2.GlobalCustomQuota,
) ([]quota.MatchedQuota, error) {
if len(items) == 0 {
return nil, nil
}
objLabels := labels.Set(u.GetLabels())
out := make([]quota.MatchedQuota, 0)
for _, gcq := range items {
if !sourcesTargetKind(gcq.Spec.Sources, req.Kind) {
continue
}
if !customQuotaReadyForAdmission(gcq.Generation, gcq.Status.CustomQuotaStatus) {
if req.SubResource == "status" {
continue
}
applies, err := desiredGlobalQuotaAppliesToNamespace(ctx, reader, &gcq, req.Namespace)
if err != nil {
return nil, fmt.Errorf(
"evaluate namespaces for GlobalCustomQuota %s generation %d: %w",
gcq.Name,
gcq.Generation,
err,
)
}
if !applies {
continue
}
return nil, fmt.Errorf(
"GlobalCustomQuota %s is not ready for generation %d",
gcq.Name,
gcq.Generation,
)
}
if !gcq.Status.NamespacePresent("*") && !gcq.Status.NamespacePresent(req.Namespace) {
continue
}
if !selectors.MatchesSelectors(objLabels, gcq.Spec.ScopeSelectors) {
continue
}
compiledTargets, err := h.getOrCompileGlobalCustomQuotaTargets(&gcq)
if err != nil {
return nil, fmt.Errorf("compile targets for GlobalCustomQuota %s: %w", gcq.Name, err)
}
for i, target := range compiledTargets {
if target.Group != req.Kind.Group ||
target.Version != req.Kind.Version ||
target.Kind != req.Kind.Kind {
continue
}
matches, err := controller.MatchesCompiledSelectorsWithFields(ctx, u, target.CompiledSelectors)
if err != nil {
return nil, fmt.Errorf(
"evaluate selectors for %s/%s on GlobalCustomQuota %s: %w",
u.GetNamespace(),
u.GetName(),
gcq.Name,
err,
)
}
if !matches {
continue
}
out = append(out, quota.MatchedQuota{
Key: controller.MakeGlobalCustomQuotaCacheKey(gcq.Name),
Name: gcq.Name,
Namespace: "",
Path: target.Path,
CompiledPath: target.CompiledPath,
CEL: target.CEL,
CompiledCEL: target.CompiledCEL,
Operation: target.Operation,
Limit: gcq.Spec.Limit.DeepCopy(),
Used: gcq.Status.Usage.Used.DeepCopy(),
IsGlobal: true,
SourceRank: i,
})
}
}
return out, nil
}
func getUnstructured(rawExt runtime.RawExtension) (unstructured.Unstructured, error) {
var (
obj runtime.Object
scope conversion.Scope
)
err := runtime.Convert_runtime_RawExtension_To_runtime_Object(&rawExt, &obj, scope)
if err != nil {
return unstructured.Unstructured{}, err
}
innerObj, err := runtime.DefaultUnstructuredConverter.ToUnstructured(obj)
if err != nil {
return unstructured.Unstructured{}, err
}
u := unstructured.Unstructured{Object: innerObj}
return u, nil
}
func namespaceTerminating(
ctx context.Context,
reader client.Reader,
namespace string,
) (bool, error) {
if namespace == "" {
return false, nil
}
ns := &corev1.Namespace{}
if err := reader.Get(ctx, types.NamespacedName{Name: namespace}, ns); err != nil {
if apierrors.IsNotFound(err) {
return true, nil
}
return false, err
}
return ns.DeletionTimestamp != nil, nil
}
func quotaTypeName(global bool) string {
if global {
return "GlobalCustomQuota"
}
return "CustomQuota"
}
type evaluatedQuota struct {
quota.MatchedQuota
Usage resource.Quantity
}
func (h *objectCalculationHandler) evaluateMatchedQuotas(
ctx context.Context,
u unstructured.Unstructured,
matched []quota.MatchedQuota,
) ([]evaluatedQuota, error) {
log := log.FromContext(ctx)
usageByExpression := make(map[string]resource.Quantity, len(matched))
for _, mq := range matched {
// count does not use a calculation expression
if mq.Operation == quota.OpCount {
continue
}
expressionKey := matchedQuotaExpressionKey(mq)
if _, ok := usageByExpression[expressionKey]; ok {
continue
}
var (
usage resource.Quantity
err error
)
switch {
case mq.CompiledCEL != nil:
usage, err = mq.CompiledCEL.EvaluateQuantity(ctx, u)
case mq.CompiledPath != nil:
usage, err = quota.ParseQuantityFromUnstructured(u, mq.CompiledPath)
default:
err = fmt.Errorf("compiled usage expression is missing")
}
if err != nil {
return nil, fmt.Errorf(
"%s %q source path %q cel %q op %q did not resolve to a valid quantity: %w",
quotaTypeName(mq.IsGlobal),
mq.Name,
mq.Path,
mq.CEL,
mq.Operation,
err,
)
}
log.V(5).Info(
"evaluated usage",
"path", mq.Path,
"cel", mq.CEL,
"quantity", usage.String(),
)
usageByExpression[expressionKey] = usage
}
byKey := make(map[string]evaluatedQuota, len(matched))
order := make([]string, 0, len(matched))
for _, mq := range matched {
ev, ok := byKey[mq.Key]
if !ok {
ev = evaluatedQuota{
MatchedQuota: mq,
Usage: resource.MustParse("0"),
}
order = append(order, mq.Key)
}
var usage resource.Quantity
switch mq.Operation {
case quota.OpCount:
usage = *resource.NewQuantity(1, resource.DecimalSI)
case quota.OpSub:
usage = usageByExpression[matchedQuotaExpressionKey(mq)].DeepCopy()
usage.Neg()
ev.Usage.Add(usage)
quota.ClampQuantityToZero(&ev.Usage)
byKey[mq.Key] = ev
continue
case quota.OpAdd:
usage = usageByExpression[matchedQuotaExpressionKey(mq)].DeepCopy()
default:
return nil, fmt.Errorf("unsupported quota operation %q for key %q", mq.Operation, mq.Key)
}
ev.Usage.Add(usage)
byKey[mq.Key] = ev
}
out := make([]evaluatedQuota, 0, len(order))
for _, key := range order {
out = append(out, byKey[key])
}
return out, nil
}
func matchedQuotaExpressionKey(matched quota.MatchedQuota) string {
if matched.CEL != "" {
return "cel:" + matched.CEL
}
return "path:" + matched.Path
}
func addLedgerPendingDelete(
ctx context.Context,
c client.Client,
reader client.Reader,
ledgerKey types.NamespacedName,
objRef capsulev1beta2.QuantityLedgerObjectRef,
) error {
return retry.RetryOnConflict(ledgerMutationBackoff, func() error {
ledger := &capsulev1beta2.QuantityLedger{}
if err := reader.Get(ctx, ledgerKey, ledger); err != nil {
return err
}
now := metav1.Now()
for _, pd := range ledger.Status.PendingDeletes {
if pd.ObjectRef.UID != "" && pd.ObjectRef.UID == objRef.UID {
return nil
}
}
if len(ledger.Status.PendingDeletes) >= maxQuantityLedgerPendingDeletes {
return fmt.Errorf(
"quantity ledger %s has reached the maximum of %d pending deletes",
ledgerKey.String(),
maxQuantityLedgerPendingDeletes,
)
}
ledger.Status.PendingDeletes = append(ledger.Status.PendingDeletes, capsulev1beta2.QuantityLedgerPendingDelete{
ObjectRef: objRef,
CreatedAt: now,
})
return c.Status().Update(ctx, ledger)
})
}
func (h *objectCalculationHandler) getOrCompileCustomQuotaTargets(
cq *capsulev1beta2.CustomQuota,
) ([]cache.CompiledTarget, error) {
key := controller.MakeCustomQuotaCacheKey(cq.Namespace, cq.Name)
targets := customQuotaTargets(cq.Spec.Sources, cq.Status.Targets)
return h.getOrCompileCurrentTargets(key, targets)
}
func (h *objectCalculationHandler) getOrCompileGlobalCustomQuotaTargets(
gcq *capsulev1beta2.GlobalCustomQuota,
) ([]cache.CompiledTarget, error) {
key := controller.MakeGlobalCustomQuotaCacheKey(gcq.Name)
targets := customQuotaTargets(gcq.Spec.Sources, gcq.Status.Targets)
return h.getOrCompileCurrentTargets(key, targets)
}
func customQuotaTargets(
sources []capsulev1beta2.CustomQuotaSpecSource,
statusTargets []capsulev1beta2.CustomQuotaStatusTarget,
) []capsulev1beta2.CustomQuotaStatusTarget {
targets := make([]capsulev1beta2.CustomQuotaStatusTarget, 0, len(sources))
for i, source := range sources {
scope := k8smeta.RESTScopeName("")
if i < len(statusTargets) {
scope = statusTargets[i].Scope
}
targets = append(targets, capsulev1beta2.CustomQuotaStatusTarget{
GroupVersionKind: metav1.GroupVersionKind(source.GroupVersionKind()),
CustomQuotaSpecSourceConfig: source.CustomQuotaSpecSourceConfig,
Scope: scope,
})
}
return targets
}
func (h *objectCalculationHandler) getOrCompileCurrentTargets(
key string,
targets []capsulev1beta2.CustomQuotaStatusTarget,
) ([]cache.CompiledTarget, error) {
if compiled, ok := h.targetsCache.Get(key); ok && compiledTargetsCurrent(compiled, targets) {
return compiled, nil
}
compiled, err := controller.CompileTargets(h.jsonPathCache, h.celCache, targets)
if err != nil {
return nil, err
}
h.targetsCache.Set(key, compiled)
return compiled, nil
}
func compiledTargetsCurrent(
compiled []cache.CompiledTarget,
targets []capsulev1beta2.CustomQuotaStatusTarget,
) bool {
if len(compiled) != len(targets) {
return false
}
for i := range targets {
if !reflect.DeepEqual(compiled[i].CustomQuotaStatusTarget, targets[i]) {
return false
}
}
return true
}
func desiredGlobalQuotaAppliesToNamespace(
ctx context.Context,
reader client.Reader,
quota *capsulev1beta2.GlobalCustomQuota,
namespace string,
) (bool, error) {
if len(quota.Spec.NamespaceSelectors) == 0 {
return true, nil
}
if namespace == "" {
return false, nil
}
ns := &corev1.Namespace{}
if err := reader.Get(ctx, types.NamespacedName{Name: namespace}, ns); err != nil {
return false, err
}
nsLabels := labels.Set(ns.Labels)
for _, rawSelector := range quota.Spec.NamespaceSelectors {
if rawSelector.LabelSelector == nil {
continue
}
selector, err := metav1.LabelSelectorAsSelector(rawSelector.LabelSelector)
if err != nil {
return false, err
}
if selector.Matches(nsLabels) {
return true, nil
}
}
return false, nil
}
func sourcesTargetKind(
sources []capsulev1beta2.CustomQuotaSpecSource,
kind metav1.GroupVersionKind,
) bool {
for _, source := range sources {
target := source.GroupVersionKind()
if target.Group == kind.Group &&
target.Version == kind.Version &&
target.Kind == kind.Kind {
return true
}
}
return false
}
func customQuotaReadyForAdmission(
generation int64,
status capsulev1beta2.CustomQuotaStatus,
) bool {
return status.ObservedGeneration == generation &&
meta.IsStatusConditionTrue(status.Conditions, meta.ReadyCondition)
}
func evaluatedByKey(in []evaluatedQuota) map[string]evaluatedQuota {
out := make(map[string]evaluatedQuota, len(in))
for _, item := range in {
existing, ok := out[item.Key]
if !ok {
copyItem := item
copyItem.Usage = item.Usage.DeepCopy()
out[item.Key] = copyItem
continue
}
existing.Usage.Add(item.Usage)
quota.ClampQuantityToZero(&existing.Usage)
out[item.Key] = existing
}
return out
}