// Copyright 2020-2026 Project Capsule Authors // SPDX-License-Identifier: Apache-2.0 package customquotas import ( "context" "errors" "fmt" "reflect" "sort" "strings" "time" "github.com/go-logr/logr" corev1 "k8s.io/api/core/v1" apierrors "k8s.io/apimachinery/pkg/api/errors" "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/labels" "k8s.io/apimachinery/pkg/runtime/schema" "k8s.io/apimachinery/pkg/types" "k8s.io/apiserver/pkg/cel/environment" "k8s.io/client-go/util/retry" "sigs.k8s.io/controller-runtime/pkg/client" capsulev1beta2 "github.com/projectcapsule/capsule/api/v1beta2" "github.com/projectcapsule/capsule/internal/cache" celruntime "github.com/projectcapsule/capsule/pkg/runtime/cel" "github.com/projectcapsule/capsule/pkg/runtime/jsonpath" "github.com/projectcapsule/capsule/pkg/runtime/quota" "github.com/projectcapsule/capsule/pkg/runtime/selectors" "github.com/projectcapsule/capsule/pkg/utils" ) const immediatePendingDeleteRequeue = 2 * time.Second func usagePercentage(used, limit resource.Quantity) float64 { if limit.MilliValue() <= 0 { return 0 } return (float64(used.MilliValue()) / float64(limit.MilliValue())) * 100 } type GroupedTarget struct { GVK schema.GroupVersionKind Targets []capsulev1beta2.CustomQuotaStatusTarget } type CompiledTarget struct { capsulev1beta2.CustomQuotaStatusTarget CompiledPath *jsonpath.CompiledJSONPath CompiledCEL *celruntime.CompiledExpression CompiledSelectors []selectors.CompiledSelectorWithFields } func CompileTargets( jcache *cache.JSONPathCache, ccache *cache.CELCache, targets []capsulev1beta2.CustomQuotaStatusTarget, ) ([]cache.CompiledTarget, error) { out := make([]cache.CompiledTarget, 0, len(targets)) for _, target := range targets { pt := cache.CompiledTarget{ CustomQuotaStatusTarget: target, } switch target.Operation { case quota.OpCount: // no usage expression needed case quota.OpAdd, quota.OpSub: switch { case target.Path != "" && target.CEL == "": compiledPath, err := jcache.GetOrCompile(target.Path) if err != nil { return nil, fmt.Errorf( "compile usage path %q for %s %q: %w", target.Path, target.String(), target.Operation, err, ) } pt.CompiledPath = compiledPath case target.CEL != "" && target.Path == "": compiledCEL, err := ccache.GetOrCompileQuantity(target.CEL, environment.StoredExpressions) if err != nil { return nil, fmt.Errorf( "compile usage CEL expression %q for %s %q: %w", target.CEL, target.String(), target.Operation, err, ) } pt.CompiledCEL = compiledCEL default: return nil, fmt.Errorf( "exactly one of path or cel must be set for %s %q", target.String(), target.Operation, ) } default: return nil, fmt.Errorf("unsupported operation %q for %s", target.Operation, target.String()) } compiledSelectors, err := CompileSelectorsWithFields(jcache, ccache, target.Selectors) if err != nil { return nil, fmt.Errorf( "compile selectors for %s: %w", target.String(), err, ) } pt.CompiledSelectors = compiledSelectors out = append(out, pt) } return out, nil } func MatchesCompiledSelectorsWithFields( ctx context.Context, u unstructured.Unstructured, selectors []selectors.CompiledSelectorWithFields, ) (bool, error) { if len(selectors) == 0 { return true, nil } itemLabels := labels.Set(u.GetLabels()) for _, sel := range selectors { if !sel.LabelSelector.Matches(itemLabels) { continue } allFieldsMatch := true for _, matcher := range sel.FieldMatchers { ok, err := evaluateCompiledFieldSelector(u, matcher) if err != nil { return false, err } if !ok { allFieldsMatch = false break } } if !allFieldsMatch { continue } for _, matcher := range sel.CELMatchers { ok, err := matcher.EvaluateBoolean(ctx, u) if err != nil { return false, err } if !ok { allFieldsMatch = false break } } if allFieldsMatch { return true, nil } } return false, nil } func evaluateCompiledFieldSelector( u unstructured.Unstructured, matcher selectors.CompiledFieldSelector, ) (bool, error) { switch matcher.Operator { case selectors.FieldSelectorTruthy: return jsonpath.EvaluateTruthyFromCompiled(u, matcher.Compiled) case selectors.FieldSelectorEquals: actual, err := matcher.Compiled.Execute(u) if err != nil { return false, err } return strings.TrimSpace(actual) == matcher.Value, nil case selectors.FieldSelectorNotEquals: actual, err := matcher.Compiled.Execute(u) if err != nil { return false, err } return strings.TrimSpace(actual) != matcher.Value, nil default: return false, fmt.Errorf("unsupported field selector operator %q", matcher.Operator) } } func MakeCustomQuotaCacheKey(namespace, name string) string { return namespace + "/" + name } func MakeGlobalCustomQuotaCacheKey(name string) string { return "C/" + name } func CompileSelectorsWithFields( jcache *cache.JSONPathCache, ccache *cache.CELCache, in []selectors.SelectorWithFields, ) ([]selectors.CompiledSelectorWithFields, error) { if len(in) == 0 { return nil, nil } out := make([]selectors.CompiledSelectorWithFields, 0, len(in)) for _, selector := range in { lblSel := labels.Everything() if selector.LabelSelector != nil { compiled, err := metav1.LabelSelectorAsSelector(selector.LabelSelector) if err != nil { return nil, fmt.Errorf("compile label selector with fields: %w", err) } lblSel = compiled } fieldMatchers := make([]selectors.CompiledFieldSelector, 0, len(selector.FieldSelectors)) for _, raw := range selector.FieldSelectors { compiledSelector, err := utils.CompileFieldSelector(jcache, raw) if err != nil { return nil, fmt.Errorf("compile field selector %q: %w", raw, err) } fieldMatchers = append(fieldMatchers, compiledSelector) } celMatchers := make([]*celruntime.CompiledExpression, 0, len(selector.CELExpressions)) for _, expression := range selector.CELExpressions { compiledExpression, err := ccache.GetOrCompileBoolean(expression, environment.StoredExpressions) if err != nil { return nil, fmt.Errorf("compile CEL selector %q: %w", expression, err) } celMatchers = append(celMatchers, compiledExpression) } out = append(out, selectors.CompiledSelectorWithFields{ LabelSelector: lblSel, FieldMatchers: fieldMatchers, CELMatchers: celMatchers, }) } return out, nil } func shouldIgnoreLedgerEnsureError(err error) bool { if err == nil { return false } if apierrors.IsNotFound(err) { return true } if apierrors.HasStatusCause(err, corev1.NamespaceTerminatingCause) { return true } var statusErr *apierrors.StatusError if errors.As(err, &statusErr) { if statusErr.ErrStatus.Reason == metav1.StatusReasonForbidden && strings.Contains(statusErr.ErrStatus.Message, "because it is being terminated") { return true } } return false } func getResourcesByGVK( ctx context.Context, gvk schema.GroupVersionKind, kubeClient client.Reader, namespaced bool, scopeSelectors []metav1.LabelSelector, namespaces ...string, ) ([]unstructured.Unstructured, error) { compiledSelectors, err := compileScopeSelectors(scopeSelectors) if err != nil { return nil, err } filterByNamespace, namespaceSet := namespaceFilter(namespaces) listNamespaces := resourceListNamespaces(namespaced, filterByNamespace, namespaceSet) items := make([]unstructured.Unstructured, 0) seen := make(map[string]struct{}) for _, namespace := range listNamespaces { candidates, err := listResourcesForGVK(ctx, gvk, kubeClient, namespace, compiledSelectors) if err != nil { return nil, err } for i := range candidates { item := candidates[i] if !resourceMatchesQuotaScope(item, filterByNamespace, namespaceSet, compiledSelectors) { continue } key := client.ObjectKeyFromObject(&item).String() if _, exists := seen[key]; exists { continue } seen[key] = struct{}{} items = append(items, item) } } // Sort by oldest first sort.Slice(items, func(i, j int) bool { return items[i].GetCreationTimestamp().Time.Before(items[j].GetCreationTimestamp().Time) }) return items, nil } func compileScopeSelectors(scopeSelectors []metav1.LabelSelector) ([]labels.Selector, error) { compiledSelectors := make([]labels.Selector, 0, len(scopeSelectors)) for _, selector := range scopeSelectors { compiled, err := metav1.LabelSelectorAsSelector(&selector) if err != nil { return nil, err } compiledSelectors = append(compiledSelectors, compiled) } return compiledSelectors, nil } func namespaceFilter(namespaces []string) (filter bool, namespaceSet map[string]struct{}) { namespaceSet = make(map[string]struct{}, len(namespaces)) for _, namespace := range namespaces { if namespace == "*" { return false, nil } namespaceSet[namespace] = struct{}{} } return true, namespaceSet } func resourceListNamespaces( namespaced bool, filterByNamespace bool, namespaceSet map[string]struct{}, ) []string { if !namespaced || !filterByNamespace { return []string{""} } orderedNamespaces := make([]string, 0, len(namespaceSet)) for namespace := range namespaceSet { orderedNamespaces = append(orderedNamespaces, namespace) } sort.Strings(orderedNamespaces) return orderedNamespaces } func listResourcesForGVK( ctx context.Context, gvk schema.GroupVersionKind, kubeClient client.Reader, namespace string, compiledSelectors []labels.Selector, ) ([]unstructured.Unstructured, error) { list := &unstructured.UnstructuredList{} list.SetGroupVersionKind(schema.GroupVersionKind{ Group: gvk.Group, Version: gvk.Version, Kind: gvk.Kind + "List", }) options := make([]client.ListOption, 0, 2) if namespace != "" { options = append(options, client.InNamespace(namespace)) } // A single scope selector can be pushed into the informer/API list. // Multiple selectors have OR semantics and are filtered below. if len(compiledSelectors) == 1 { options = append(options, client.MatchingLabelsSelector{Selector: compiledSelectors[0]}) } if err := kubeClient.List(ctx, list, options...); err != nil { return nil, err } return list.Items, nil } func resourceMatchesQuotaScope( item unstructured.Unstructured, filterByNamespace bool, namespaceSet map[string]struct{}, compiledSelectors []labels.Selector, ) bool { // Skip objects that are already definitely deleting: // deletionTimestamp is set and there are no finalizers left. if item.GetDeletionTimestamp() != nil && len(item.GetFinalizers()) == 0 { return false } // Namespace filtering remains necessary for cluster-wide lists and // preserves the previous behavior for cluster-scoped targets. if filterByNamespace { if _, ok := namespaceSet[item.GetNamespace()]; !ok { return false } } if len(compiledSelectors) == 0 { return true } itemLabels := labels.Set(item.GetLabels()) for _, selector := range compiledSelectors { if selector.Matches(itemLabels) { return true } } return false } func minDurationPtr(cur *time.Duration, cand time.Duration) *time.Duration { if cand < 0 { cand = 0 } if cur == nil || cand < *cur { return &cand } return cur } func pendingDeleteStillPresent( pd capsulev1beta2.QuantityLedgerPendingDelete, claims []capsulev1beta2.CustomQuotaClaimItem, ) bool { for _, claim := range claims { if sameLedgerObject(pd.ObjectRef, claim) { return true } } return false } const ( unresolvedReservationInitialRequeue = 2 * time.Second unresolvedReservationRequeue = 5 * time.Second unresolvedReservationLongRequeue = 15 * time.Second ledgerWorkDebounce = 500 * time.Millisecond ledgerWorkMaximumDelay = 2 * time.Second ) func quantityLedgerWorkDelay( now time.Time, ledger *capsulev1beta2.QuantityLedger, ) (hasWork bool, delay time.Duration) { var oldest, newest time.Time record := func(ts time.Time) { hasWork = true if ts.IsZero() { return } if oldest.IsZero() || ts.Before(oldest) { oldest = ts } if newest.IsZero() || ts.After(newest) { newest = ts } } for _, reservation := range ledger.Status.Reservations { ts := reservation.UpdatedAt.Time if ts.IsZero() { ts = reservation.CreatedAt.Time } record(ts) } for _, pendingDelete := range ledger.Status.PendingDeletes { record(pendingDelete.CreatedAt.Time) } if !hasWork || oldest.IsZero() || newest.IsZero() { return hasWork, 0 } readyAt := newest.Add(ledgerWorkDebounce) maximumAt := oldest.Add(ledgerWorkMaximumDelay) if maximumAt.Before(readyAt) { readyAt = maximumAt } if !now.Before(readyAt) { return true, 0 } return true, readyAt.Sub(now) } func nextReservationMaterializationRequeue( now metav1.Time, res capsulev1beta2.QuantityLedgerReservation, ) time.Duration { if res.ExpiresAt == nil { return unresolvedReservationRequeue } untilExpiry := res.ExpiresAt.Sub(now.Time) if untilExpiry <= 0 { return 0 } age := now.Sub(res.UpdatedAt.Time) var candidate time.Duration switch { case age < 5*time.Second: candidate = unresolvedReservationInitialRequeue case age < 30*time.Second: candidate = unresolvedReservationRequeue default: candidate = unresolvedReservationLongRequeue } if untilExpiry < candidate { return untilExpiry } return candidate } func reconcileQuantityLedgerAllocation( ctx context.Context, c client.Client, reader client.Reader, log logr.Logger, key types.NamespacedName, observedUsed resource.Quantity, claims []capsulev1beta2.CustomQuotaClaimItem, ) (*time.Duration, error) { var requeueAfter *time.Duration err := retry.RetryOnConflict(retry.DefaultBackoff, func() error { ledger := &capsulev1beta2.QuantityLedger{} // Conflicts here are normally caused by admission updating the ledger. // The informer cache can keep returning the same stale resourceVersion, // making RetryOnConflict ineffective, so retries must read directly. if err := reader.Get(ctx, key, ledger); err != nil { if apierrors.IsNotFound(err) { return nil } return err } now := metav1.Now() pendingDeleteTTL := 30 * time.Second pendingDeletePresent := make([]bool, len(ledger.Status.PendingDeletes)) confirmedTransitions := make(map[string][]capsulev1beta2.QuantityLedgerObjectRef) for i, pendingDelete := range ledger.Status.PendingDeletes { pendingDeletePresent[i] = pendingDeleteStillPresent(pendingDelete, claims) if pendingDeletePresent[i] { continue } key := ledgerReservationObjectKey(pendingDelete.ObjectRef) confirmedTransitions[key] = append(confirmedTransitions[key], pendingDelete.ObjectRef) } activeReservations := make([]capsulev1beta2.QuantityLedgerReservation, 0, len(ledger.Status.Reservations)) materializedThrough := materializedReservationPositions(ledger.Status.Reservations, claims) for i, res := range ledger.Status.Reservations { materialized := materializedThrough[ledgerReservationObjectKey(res.ObjectRef)] > i transitioned := reservationHasConfirmedTransition(res, confirmedTransitions) expired := res.ExpiresAt != nil && res.ExpiresAt.Before(&now) log.V(5).Info("evaluating ledger reservation", "ledger", key.String(), "reservationID", res.ID, "usage", res.Usage.String(), "uid", string(res.ObjectRef.UID), "group", res.ObjectRef.APIGroup, "version", res.ObjectRef.APIVersion, "kind", res.ObjectRef.Kind, "namespace", res.ObjectRef.Namespace, "name", res.ObjectRef.Name, "materialized", materialized, "transitioned", transitioned, "expired", expired, ) switch { case materialized: continue case transitioned: // A persisted update/delete moved this object out of the // matching claims before reconciliation observed its earlier // state. Its pending-delete hint confirms that the admission // operation materialized, so older reservations for the same // object can be released without waiting for their TTL. continue case expired: continue default: activeReservations = append(activeReservations, res) requeueAfter = minDurationPtr( requeueAfter, nextReservationMaterializationRequeue(now, res), ) } } activeDeletes := make([]capsulev1beta2.QuantityLedgerPendingDelete, 0, len(ledger.Status.PendingDeletes)) for i, pd := range ledger.Status.PendingDeletes { stillPresent := pendingDeletePresent[i] expired := now.Sub(pd.CreatedAt.Time) >= pendingDeleteTTL log.V(5).Info("evaluating pending delete", "ledger", key.String(), "uid", string(pd.ObjectRef.UID), "group", pd.ObjectRef.APIGroup, "version", pd.ObjectRef.APIVersion, "kind", pd.ObjectRef.Kind, "namespace", pd.ObjectRef.Namespace, "name", pd.ObjectRef.Name, "stillPresent", stillPresent, "expired", expired, ) // Pending deletes are admission hints, not durable desired state. // The admitted update/delete may fail after the webhook returns, // and a transient policy snapshot must not leave a hint that // requeues this quota forever. Once the hint expires, the current // observed claims are authoritative. if !stillPresent || expired { continue } activeDeletes = append(activeDeletes, pd) requeueAfter = minDurationPtr(requeueAfter, immediatePendingDeleteRequeue) } reserved := resource.MustParse("0") for _, res := range activeReservations { reserved.Add(quantityLedgerReservationDelta(res)) } allocated := observedUsed.DeepCopy() allocated.Add(reserved) quota.ClampQuantityToZero(&allocated) originalStatus := ledger.Status.DeepCopy() ledger.Status.Reservations = activeReservations ledger.Status.PendingDeletes = activeDeletes ledger.Status.Reserved = reserved ledger.Status.Allocated = allocated if reflect.DeepEqual(*originalStatus, ledger.Status) { return nil } return c.Status().Update(ctx, ledger) }) if err != nil { return nil, err } return requeueAfter, nil } func reservationHasConfirmedTransition( reservation capsulev1beta2.QuantityLedgerReservation, confirmed map[string][]capsulev1beta2.QuantityLedgerObjectRef, ) bool { for _, ref := range confirmed[ledgerReservationObjectKey(reservation.ObjectRef)] { if sameLedgerObjectRef(reservation.ObjectRef, ref) { return true } } return false } func quantityLedgerReservationDelta( res capsulev1beta2.QuantityLedgerReservation, ) resource.Quantity { if res.Delta == nil { return res.Usage.DeepCopy() } return res.Delta.DeepCopy() } func materializedReservationPositions( reservations []capsulev1beta2.QuantityLedgerReservation, claims []capsulev1beta2.CustomQuotaClaimItem, ) map[string]int { positions := make(map[string]int) // Reservations are kept in ledger update order. When a later update for // the same object is observed, its persisted usage also supersedes every // earlier reservation for that object. Clearing the prefix avoids holding // already-materialized deltas until their TTL after rapid updates. for i, reservation := range reservations { if reservationMaterializedLedger(reservation, claims) { positions[ledgerReservationObjectKey(reservation.ObjectRef)] = i + 1 } } return positions } func ledgerReservationObjectKey(ref capsulev1beta2.QuantityLedgerObjectRef) string { return strings.Join([]string{ ref.APIGroup, ref.APIVersion, ref.Kind, ref.Namespace, ref.Name, }, "\x00") } func reservationMaterializedLedger( res capsulev1beta2.QuantityLedgerReservation, claims []capsulev1beta2.CustomQuotaClaimItem, ) bool { for _, claim := range claims { if !sameLedgerObject(res.ObjectRef, claim) { continue } // Important for updates: // UID/name match alone is not enough. The controller must have observed // the same usage that the webhook reserved. if claim.Usage.Cmp(res.Usage) != 0 { continue } return true } return false } func sameLedgerObject( ref capsulev1beta2.QuantityLedgerObjectRef, claim capsulev1beta2.CustomQuotaClaimItem, ) bool { return sameLedgerObjectRef(ref, capsulev1beta2.QuantityLedgerObjectRef{ APIGroup: claim.Group, APIVersion: claim.Version, Kind: claim.Kind, Namespace: string(claim.Namespace), Name: claim.Name, UID: claim.UID, }) } func sameLedgerObjectRef( a capsulev1beta2.QuantityLedgerObjectRef, b capsulev1beta2.QuantityLedgerObjectRef, ) bool { if a.APIGroup != b.APIGroup || a.APIVersion != b.APIVersion || a.Kind != b.Kind || a.Namespace != b.Namespace || a.Name != b.Name { return false } // CREATE admissions often do not have a UID yet. if a.UID != "" && b.UID != "" { return a.UID == b.UID } return true }