// Copyright 2020-2026 Project Capsule Authors // SPDX-License-Identifier: Apache-2.0 package customquotas import ( "context" "errors" "fmt" "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/client-go/util/retry" "sigs.k8s.io/controller-runtime/pkg/client" capsulev1beta2 "github.com/projectcapsule/capsule/api/v1beta2" "github.com/projectcapsule/capsule/internal/cache" "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 = 500 * time.Millisecond 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 CompiledSelectors []selectors.CompiledSelectorWithFields } func CompileTargets( jcache *cache.JSONPathCache, 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 path needed case quota.OpAdd, quota.OpSub: 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 default: return nil, fmt.Errorf("unsupported operation %q for %s", target.Operation, target.String()) } compiledSelectors, err := CompileSelectorsWithFields(jcache, 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( 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 { 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( cache *cache.JSONPathCache, 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(cache, raw) if err != nil { return nil, fmt.Errorf("compile field selector %q: %w", raw, err) } fieldMatchers = append(fieldMatchers, compiledSelector) } out = append(out, selectors.CompiledSelectorWithFields{ LabelSelector: lblSel, FieldMatchers: fieldMatchers, }) } 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, scopeSelectors []metav1.LabelSelector, namespaces ...string, ) ([]unstructured.Unstructured, error) { compiledSelectors := make([]labels.Selector, 0, len(scopeSelectors)) for _, selector := range scopeSelectors { sel, err := metav1.LabelSelectorAsSelector(&selector) if err != nil { return nil, err } compiledSelectors = append(compiledSelectors, sel) } filterByNamespace := true namespaceSet := make(map[string]struct{}, len(namespaces)) for _, ns := range namespaces { if ns == "*" { filterByNamespace = false namespaceSet = nil break } namespaceSet[ns] = struct{}{} } list := &unstructured.UnstructuredList{} list.SetGroupVersionKind(schema.GroupVersionKind{ Group: gvk.Group, Version: gvk.Version, Kind: gvk.Kind + "List", }) if err := kubeClient.List(ctx, list); err != nil { return nil, err } items := make([]unstructured.Unstructured, 0, len(list.Items)) seen := make(map[string]struct{}, len(list.Items)) for i := range list.Items { item := list.Items[i] // Skip objects that are already definitely deleting: // deletionTimestamp is set and there are no finalizers left. if item.GetDeletionTimestamp() != nil && len(item.GetFinalizers()) == 0 { continue } // Namespace filter if filterByNamespace { if _, ok := namespaceSet[item.GetNamespace()]; !ok { continue } } // Label selector filter (OR semantics) if len(compiledSelectors) > 0 { itemLabels := labels.Set(item.GetLabels()) matched := false for _, sel := range compiledSelectors { if sel.Matches(itemLabels) { matched = true break } } if !matched { continue } } key := item.GetNamespace() + "/" + item.GetName() if item.GetNamespace() == "" { key = item.GetName() } 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 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 pd.ObjectRef.UID != "" && claim.UID != "" && pd.ObjectRef.UID == claim.UID { return true } if pd.ObjectRef.APIGroup == claim.Group && pd.ObjectRef.APIVersion == claim.Version && pd.ObjectRef.Kind == claim.Kind && pd.ObjectRef.Namespace == string(claim.Namespace) && pd.ObjectRef.Name == claim.Name { return true } } return false } const unresolvedReservationRequeue = 250 * time.Millisecond func nextReservationMaterializationRequeue( now metav1.Time, res capsulev1beta2.QuantityLedgerReservation, ) time.Duration { if res.ExpiresAt == nil { return unresolvedReservationRequeue } untilExpiry := time.Until(res.ExpiresAt.Time) if untilExpiry <= 0 { return 0 } if untilExpiry < unresolvedReservationRequeue { return untilExpiry } return unresolvedReservationRequeue } func reconcileQuantityLedgerAllocation( ctx context.Context, c client.Client, 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{} if err := c.Get(ctx, key, ledger); err != nil { if apierrors.IsNotFound(err) { return nil } return err } now := metav1.Now() pendingDeleteTTL := 30 * time.Second activeReservations := make([]capsulev1beta2.QuantityLedgerReservation, 0, len(ledger.Status.Reservations)) for _, res := range ledger.Status.Reservations { materialized := reservationMaterializedLedger(res, claims) 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, "expired", expired, ) switch { case materialized: 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 _, pd := range ledger.Status.PendingDeletes { stillPresent := pendingDeleteStillPresent(pd, claims) 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, ) if stillPresent { activeDeletes = append(activeDeletes, pd) requeueAfter = minDurationPtr(requeueAfter, immediatePendingDeleteRequeue) } } reserved := resource.MustParse("0") for _, res := range activeReservations { reserved.Add(res.Usage) } allocated := observedUsed.DeepCopy() allocated.Add(reserved) quota.ClampQuantityToZero(&allocated) ledger.Status.Reservations = activeReservations ledger.Status.PendingDeletes = activeDeletes ledger.Status.Reserved = reserved ledger.Status.Allocated = allocated return c.Status().Update(ctx, ledger) }) if err != nil { return nil, err } return requeueAfter, nil } 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 { if ref.APIGroup != claim.Group || ref.APIVersion != claim.Version || ref.Kind != claim.Kind || ref.Namespace != string(claim.Namespace) || ref.Name != claim.Name { return false } // CREATE admissions often do not have a UID yet. if ref.UID != "" && claim.UID != "" { return ref.UID == claim.UID } return true }