diff --git a/pkg/aaq-controller/aaq-evaluator/aaq-evaluator.go b/pkg/aaq-controller/aaq-evaluator/aaq-evaluator.go index 5bd019372..6b6cb9a07 100644 --- a/pkg/aaq-controller/aaq-evaluator/aaq-evaluator.go +++ b/pkg/aaq-controller/aaq-evaluator/aaq-evaluator.go @@ -2,6 +2,7 @@ package aaq_evaluator import ( "fmt" + corev1 "k8s.io/api/core/v1" "k8s.io/apimachinery/pkg/api/resource" "k8s.io/apimachinery/pkg/labels" @@ -16,6 +17,7 @@ import ( "k8s.io/kubernetes/pkg/quota/v1/evaluator/core" "k8s.io/utils/clock" "kubevirt.io/application-aware-quota/pkg/util" + "kubevirt.io/application-aware-quota/staging/src/kubevirt.io/application-aware-quota-api/pkg/apis/core/v1alpha1" ) // NewAaqEvaluator returns an evaluator that can evaluate pods with apps consideration @@ -31,8 +33,7 @@ func NewAaqEvaluator(podLister v1.PodLister, aaqEvalRegistery Registry, clock cl type AaqEvaluator struct { podEvaluator v12.Evaluator aaqEvalRegistery Registry - // knows how to list pods - podLister v1.PodLister + podLister v1.PodLister } func (aaqe *AaqEvaluator) Constraints(_ []corev1.ResourceName, _ runtime.Object) error { @@ -49,11 +50,42 @@ func (aaqe *AaqEvaluator) Handles(operation admission.Attributes) bool { } func (aaqe *AaqEvaluator) Matches(resourceQuota *corev1.ResourceQuota, item runtime.Object) (bool, error) { - return aaqe.podEvaluator.Matches(resourceQuota, item) + matchResource := len(aaqe.MatchingResources(quota.ResourceNames(resourceQuota.Status.Hard))) > 0 + matchScope := true + for _, scope := range getScopeSelectorsFromQuota(resourceQuota) { + innerMatch, err := aaqe.podMatchesScopeFunc(scope, item) + if err != nil { + return false, err + } + matchScope = matchScope && innerMatch + } + return matchResource && matchScope, nil +} + +func getScopeSelectorsFromQuota(rq *corev1.ResourceQuota) []corev1.ScopedResourceSelectorRequirement { + var selectors []corev1.ScopedResourceSelectorRequirement + for _, scope := range rq.Spec.Scopes { + selectors = append(selectors, corev1.ScopedResourceSelectorRequirement{ + ScopeName: scope, Operator: corev1.ScopeSelectorOpExists}) + } + if rq.Spec.ScopeSelector != nil { + selectors = append(selectors, rq.Spec.ScopeSelector.MatchExpressions...) + } + return selectors } func (aaqe *AaqEvaluator) MatchingScopes(item runtime.Object, scopes []corev1.ScopedResourceSelectorRequirement) ([]corev1.ScopedResourceSelectorRequirement, error) { - return aaqe.podEvaluator.MatchingScopes(item, scopes) + var matched []corev1.ScopedResourceSelectorRequirement + for _, scope := range scopes { + innerMatch, err := aaqe.podMatchesScopeFunc(scope, item) + if err != nil { + return nil, err + } + if innerMatch { + matched = append(matched, scope) + } + } + return matched, nil } func (aaqe *AaqEvaluator) UncoveredQuotaScopes(limitedScopes []corev1.ScopedResourceSelectorRequirement, matchedQuotaScopes []corev1.ScopedResourceSelectorRequirement) ([]corev1.ScopedResourceSelectorRequirement, error) { @@ -64,6 +96,17 @@ func (aaqe *AaqEvaluator) MatchingResources(input []corev1.ResourceName) []corev return input } +func (aaqe *AaqEvaluator) SourceCalculatorUsage(pod *corev1.Pod, existingPods []*corev1.Pod) (corev1.ResourceList, error) { + if len(pod.Spec.SchedulingGates) > 0 { + return corev1.ResourceList{}, nil + } + rl, err := aaqe.aaqEvalRegistery.SourceUsage(pod, existingPods) + if err != nil { + return aaqe.podEvaluator.Usage(pod) + } + return rl, err +} + func (aaqe *AaqEvaluator) Usage(item runtime.Object) (corev1.ResourceList, error) { pod, err := util.ToExternalPodOrError(item) if err != nil { @@ -106,11 +149,12 @@ func (aaqe *AaqEvaluator) UsageStats(options v12.UsageStatsOptions) (v12.UsageSt return result, fmt.Errorf("failed to list content: %v", err) } + hasVmiScope := hasVmiScopes(options.Scopes, options.ScopeSelector) + for _, pod := range existingPods { - // need to verify that the item matches the set of scopes matchesScopes := true for _, scope := range options.Scopes { - innerMatch, err := podMatchesScopeFunc(corev1.ScopedResourceSelectorRequirement{ScopeName: scope, Operator: corev1.ScopeSelectorOpExists}, pod) + innerMatch, err := aaqe.podMatchesScopeFunc(corev1.ScopedResourceSelectorRequirement{ScopeName: scope, Operator: corev1.ScopeSelectorOpExists}, pod) if err != nil { return result, nil } @@ -120,18 +164,23 @@ func (aaqe *AaqEvaluator) UsageStats(options v12.UsageStatsOptions) (v12.UsageSt } if options.ScopeSelector != nil { for _, selector := range options.ScopeSelector.MatchExpressions { - innerMatch, err := podMatchesScopeFunc(selector, pod) + innerMatch, err := aaqe.podMatchesScopeFunc(selector, pod) if err != nil { return result, nil } matchesScopes = matchesScopes && innerMatch } } - // only count usage if there was a match if matchesScopes { - usage, err := aaqe.CalculatorUsage(pod, existingPods) - if err != nil { - return result, err + var usage corev1.ResourceList + var usageErr error + if hasVmiScope { + usage, usageErr = aaqe.SourceCalculatorUsage(pod, existingPods) + } else { + usage, usageErr = aaqe.CalculatorUsage(pod, existingPods) + } + if usageErr != nil { + return result, usageErr } result.Used = quota.Add(result.Used, usage) } @@ -139,9 +188,28 @@ func (aaqe *AaqEvaluator) UsageStats(options v12.UsageStatsOptions) (v12.UsageSt return result, nil } -// todo: ask kubernetes to make this funcs global and remove all this code -// podMatchesScopeFunc is a function that knows how to evaluate if a pod matches a scope -func podMatchesScopeFunc(selector corev1.ScopedResourceSelectorRequirement, object runtime.Object) (bool, error) { +var aaqVmiScopes = map[corev1.ResourceQuotaScope]bool{ + v1alpha1.VmiStarting: true, + v1alpha1.VmiMigrating: true, +} + +func hasVmiScopes(scopes []corev1.ResourceQuotaScope, scopeSelector *corev1.ScopeSelector) bool { + for _, scope := range scopes { + if aaqVmiScopes[scope] { + return true + } + } + if scopeSelector != nil { + for _, expr := range scopeSelector.MatchExpressions { + if aaqVmiScopes[expr.ScopeName] && expr.Operator == corev1.ScopeSelectorOpExists { + return true + } + } + } + return false +} + +func (aaqe *AaqEvaluator) podMatchesScopeFunc(selector corev1.ScopedResourceSelectorRequirement, object runtime.Object) (bool, error) { pod, err := util.ToExternalPodOrError(object) if err != nil { return false, err @@ -157,13 +225,15 @@ func podMatchesScopeFunc(selector corev1.ScopedResourceSelectorRequirement, obje return !isBestEffort(pod), nil case corev1.ResourceQuotaScopePriorityClass: if selector.Operator == corev1.ScopeSelectorOpExists { - // This is just checking for existence of a priorityClass on the pod, - // no need to take the overhead of selector parsing/evaluation. return len(pod.Spec.PriorityClassName) != 0, nil } return podMatchesSelector(pod, selector) case corev1.ResourceQuotaScopeCrossNamespacePodAffinity: return usesCrossNamespacePodAffinity(pod), nil + default: + if matched, handled := aaqe.aaqEvalRegistery.MatchesScope(pod, selector.ScopeName); handled { + return matched, nil + } } return false, nil } diff --git a/pkg/aaq-controller/aaq-evaluator/aaq_socket_evaluator.go b/pkg/aaq-controller/aaq-evaluator/aaq_socket_evaluator.go index 8f360106b..da7f43d79 100644 --- a/pkg/aaq-controller/aaq-evaluator/aaq_socket_evaluator.go +++ b/pkg/aaq-controller/aaq-evaluator/aaq_socket_evaluator.go @@ -15,6 +15,14 @@ type AaqSocketCalculator struct { sidecarSocketPath string } +func (aaqsc *AaqSocketCalculator) MatchesScope(_ *corev1.Pod, _ corev1.ResourceQuotaScope) (bool, bool) { + return false, false +} + +func (aaqsc *AaqSocketCalculator) SourceUsage(pod *corev1.Pod, podsState []*corev1.Pod) (corev1.ResourceList, error, bool) { + return aaqsc.PodUsageFunc(pod, podsState) +} + func (aaqsc *AaqSocketCalculator) PodUsageFunc(pod *corev1.Pod, podsState []*corev1.Pod) (corev1.ResourceList, error, bool) { conn, err := grpc.DialSocketWithTimeout(aaqsc.sidecarSocketPath, 1) if err != nil { diff --git a/pkg/aaq-controller/aaq-evaluator/evaluator_registry.go b/pkg/aaq-controller/aaq-evaluator/evaluator_registry.go index 8d52f8ad0..1e3187efe 100644 --- a/pkg/aaq-controller/aaq-evaluator/evaluator_registry.go +++ b/pkg/aaq-controller/aaq-evaluator/evaluator_registry.go @@ -22,12 +22,20 @@ var once sync.Once type AaqCalculator interface { PodUsageFunc(pod *corev1.Pod, podsState []*corev1.Pod) (corev1.ResourceList, error, bool) + // MatchesScope returns (matched, handled). If the calculator does not recognize + // the scope it should return (false, false). + MatchesScope(pod *corev1.Pod, scope corev1.ResourceQuotaScope) (bool, bool) + // SourceUsage returns the full source-equivalent resources for a pod. + // For migration targets this returns the same as the source pod (not the delta). + SourceUsage(pod *corev1.Pod, podsState []*corev1.Pod) (corev1.ResourceList, error, bool) } type Registry interface { Add(aaqCalculator AaqCalculator) Collect(numberOfRequestedEvaluatorsSidecars uint, timeout time.Duration) error Usage(*corev1.Pod, []*corev1.Pod) (corev1.ResourceList, error) + SourceUsage(*corev1.Pod, []*corev1.Pod) (corev1.ResourceList, error) + MatchesScope(pod *corev1.Pod, scope corev1.ResourceQuotaScope) (bool, bool) } type AaqEvaluatorRegistry struct { @@ -126,6 +134,37 @@ func processSideCarSocket(socketPath string) (string, bool, error) { return socketPath, false, nil } +func (aaqe *AaqEvaluatorRegistry) MatchesScope(pod *corev1.Pod, scope corev1.ResourceQuotaScope) (bool, bool) { + for _, calculator := range aaqe.aaqCalculators { + if matched, handled := calculator.MatchesScope(pod, scope); handled { + return matched, true + } + } + return false, false +} + +func (aaqe *AaqEvaluatorRegistry) SourceUsage(pod *corev1.Pod, podsState []*corev1.Pod) (rlToRet corev1.ResourceList, acceptedErr error) { + accepted := false + for _, calculator := range aaqe.aaqCalculators { + for retries := 0; retries < aaqe.retriesOnMatchFailure; retries++ { + rl, err, match := calculator.SourceUsage(pod, podsState) + if !match && err == nil { + break + } else if err == nil { + accepted = true + rlToRet = quota.Add(rlToRet, rl) + break + } else { + log.Log.Infof("Retries: %v Error: %v ", retries, err) + } + } + } + if !accepted { + acceptedErr = fmt.Errorf("pod didn't match any usageFunc") + } + return rlToRet, acceptedErr +} + func (aaqe *AaqEvaluatorRegistry) Usage(pod *corev1.Pod, podsState []*corev1.Pod) (rlToRet corev1.ResourceList, acceptedErr error) { accepted := false for _, calculator := range aaqe.aaqCalculators { diff --git a/pkg/aaq-controller/aaq-evaluator/evaluator_test.go b/pkg/aaq-controller/aaq-evaluator/evaluator_test.go index b1c51ea8a..4f01f4eac 100644 --- a/pkg/aaq-controller/aaq-evaluator/evaluator_test.go +++ b/pkg/aaq-controller/aaq-evaluator/evaluator_test.go @@ -2,6 +2,8 @@ package aaq_evaluator import ( "fmt" + "time" + "github.com/google/go-cmp/cmp" . "github.com/onsi/ginkgo/v2" . "github.com/onsi/gomega" @@ -9,6 +11,7 @@ import ( "k8s.io/apimachinery/pkg/api/resource" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/runtime/schema" + "k8s.io/apimachinery/pkg/types" quota "k8s.io/apiserver/pkg/quota/v1" "k8s.io/apiserver/pkg/quota/v1/generic" "k8s.io/apiserver/pkg/util/feature" @@ -18,8 +21,10 @@ import ( "k8s.io/kubernetes/pkg/quota/v1/evaluator/core" "k8s.io/kubernetes/pkg/util/node" testingclock "k8s.io/utils/clock/testing" + kvv1 "kubevirt.io/api/core/v1" + built_in_usage_calculators "kubevirt.io/application-aware-quota/pkg/aaq-controller/built-in-usage-calculators" fakeinformers "kubevirt.io/application-aware-quota/pkg/tests-utils" - "time" + "kubevirt.io/application-aware-quota/staging/src/kubevirt.io/application-aware-quota-api/pkg/apis/core/v1alpha1" ) var _ = Describe("AaqEvaluator", func() { @@ -830,6 +835,118 @@ var _ = Describe("AaqEvaluator", func() { ) }) + Context("Test VMI scopes", func() { + cpu1 := corev1.ResourceList{corev1.ResourceCPU: resource.MustParse("1")} + fakeNs := "test-ns" + fakeVmiName := "test-vmi" + fakeVmiUID := "vmi-uid-123" + + makeRegistryWithVirtLauncher := func(vmiInformer, migrationInformer fakeinformers.FakeSharedIndexInformer) *AaqEvaluatorRegistry { + registry := newAaqEvaluatorsRegistry(1, "/fakeSocketSharedDirectory") + registry.Add(built_in_usage_calculators.NewVirtLauncherCalculator(nil, vmiInformer, migrationInformer, v1alpha1.VmiPodUsage)) + return registry + } + + DescribeTable("Test VmiStarting scope", func(vmiPhase kvv1.VirtualMachineInstancePhase, expectMatch bool) { + vmi := &kvv1.VirtualMachineInstance{ + ObjectMeta: metav1.ObjectMeta{Name: fakeVmiName, Namespace: fakeNs, UID: types.UID(fakeVmiUID)}, + Status: kvv1.VirtualMachineInstanceStatus{Phase: vmiPhase}, + } + pod := makeVmiPod("launcher", fakeVmiName, fakeVmiUID, fakeNs, cpu1, corev1.PodRunning) + + vmiInformer := fakeinformers.NewFakeSharedIndexInformer([]metav1.Object{vmi}) + migrationInformer := fakeinformers.NewFakeSharedIndexInformer([]metav1.Object{}) + podInformer := fakeinformers.NewFakeSharedIndexInformer([]metav1.Object{pod}) + fakeClock := testingclock.NewFakeClock(time.Now()) + eval := NewAaqEvaluator(v1.NewPodLister(podInformer.GetIndexer()), makeRegistryWithVirtLauncher(vmiInformer, migrationInformer), fakeClock) + + matched, err := eval.MatchingScopes(pod, []corev1.ScopedResourceSelectorRequirement{ + {ScopeName: v1alpha1.VmiStarting, Operator: corev1.ScopeSelectorOpExists}, + }) + Expect(err).ToNot(HaveOccurred()) + if expectMatch { + Expect(matched).To(HaveLen(1)) + } else { + Expect(matched).To(BeEmpty()) + } + }, + Entry("VMI in Pending phase should match", kvv1.Pending, true), + Entry("VMI in Scheduling phase should match", kvv1.Scheduling, true), + Entry("VMI in Scheduled phase should match", kvv1.Scheduled, true), + Entry("VMI in Running phase should not match", kvv1.Running, false), + Entry("VMI in Succeeded phase should not match", kvv1.Succeeded, false), + Entry("VMI in Failed phase should not match", kvv1.Failed, false), + ) + + It("VmiStarting should not match non-VMI pod", func() { + pod := makePod("regular-pod", "", cpu1, corev1.PodRunning) + podInformer := fakeinformers.NewFakeSharedIndexInformer([]metav1.Object{pod}) + fakeClock := testingclock.NewFakeClock(time.Now()) + eval := NewAaqEvaluator(v1.NewPodLister(podInformer.GetIndexer()), newAaqEvaluatorsRegistry(1, "/fakeSocketSharedDirectory"), fakeClock) + + matched, err := eval.MatchingScopes(pod, []corev1.ScopedResourceSelectorRequirement{ + {ScopeName: v1alpha1.VmiStarting, Operator: corev1.ScopeSelectorOpExists}, + }) + Expect(err).ToNot(HaveOccurred()) + Expect(matched).To(BeEmpty()) + }) + + DescribeTable("Test VmiMigrating scope", func(vmimPhase kvv1.VirtualMachineInstanceMigrationPhase, expectMatch bool) { + vmi := &kvv1.VirtualMachineInstance{ + ObjectMeta: metav1.ObjectMeta{Name: fakeVmiName, Namespace: fakeNs, UID: types.UID(fakeVmiUID)}, + Status: kvv1.VirtualMachineInstanceStatus{Phase: kvv1.Running}, + } + vmim := &kvv1.VirtualMachineInstanceMigration{ + ObjectMeta: metav1.ObjectMeta{Name: "test-migration", Namespace: fakeNs, UID: "vmim-uid-456"}, + Spec: kvv1.VirtualMachineInstanceMigrationSpec{VMIName: fakeVmiName}, + Status: kvv1.VirtualMachineInstanceMigrationStatus{Phase: vmimPhase}, + } + pod := makeVmiPod("launcher", fakeVmiName, fakeVmiUID, fakeNs, cpu1, corev1.PodRunning) + + vmiInformer := fakeinformers.NewFakeSharedIndexInformer([]metav1.Object{vmi}) + migrationInformer := fakeinformers.NewFakeSharedIndexInformer([]metav1.Object{vmim}) + podInformer := fakeinformers.NewFakeSharedIndexInformer([]metav1.Object{pod}) + fakeClock := testingclock.NewFakeClock(time.Now()) + eval := NewAaqEvaluator(v1.NewPodLister(podInformer.GetIndexer()), makeRegistryWithVirtLauncher(vmiInformer, migrationInformer), fakeClock) + + matched, err := eval.MatchingScopes(pod, []corev1.ScopedResourceSelectorRequirement{ + {ScopeName: v1alpha1.VmiMigrating, Operator: corev1.ScopeSelectorOpExists}, + }) + Expect(err).ToNot(HaveOccurred()) + if expectMatch { + Expect(matched).To(HaveLen(1)) + } else { + Expect(matched).To(BeEmpty()) + } + }, + Entry("active migration (Running) should match", kvv1.MigrationRunning, true), + Entry("migration Scheduling should match", kvv1.MigrationScheduling, true), + Entry("migration PreparingTarget should match", kvv1.MigrationPreparingTarget, true), + Entry("migration Succeeded should not match", kvv1.MigrationSucceeded, false), + Entry("migration Failed should not match", kvv1.MigrationFailed, false), + ) + + It("VmiMigrating should not match when no migration exists", func() { + vmi := &kvv1.VirtualMachineInstance{ + ObjectMeta: metav1.ObjectMeta{Name: fakeVmiName, Namespace: fakeNs, UID: types.UID(fakeVmiUID)}, + Status: kvv1.VirtualMachineInstanceStatus{Phase: kvv1.Running}, + } + pod := makeVmiPod("launcher", fakeVmiName, fakeVmiUID, fakeNs, cpu1, corev1.PodRunning) + + vmiInformer := fakeinformers.NewFakeSharedIndexInformer([]metav1.Object{vmi}) + migrationInformer := fakeinformers.NewFakeSharedIndexInformer([]metav1.Object{}) + podInformer := fakeinformers.NewFakeSharedIndexInformer([]metav1.Object{pod}) + fakeClock := testingclock.NewFakeClock(time.Now()) + eval := NewAaqEvaluator(v1.NewPodLister(podInformer.GetIndexer()), makeRegistryWithVirtLauncher(vmiInformer, migrationInformer), fakeClock) + + matched, err := eval.MatchingScopes(pod, []corev1.ScopedResourceSelectorRequirement{ + {ScopeName: v1alpha1.VmiMigrating, Operator: corev1.ScopeSelectorOpExists}, + }) + Expect(err).ToNot(HaveOccurred()) + Expect(matched).To(BeEmpty()) + }) + }) + Context("test calculators-registery ", func() { var registry Registry var fakeClock *testingclock.FakeClock @@ -928,6 +1045,23 @@ func (m *FakeUsageCalculator) PodUsageFunc(pod *corev1.Pod, podsState []*corev1. return m.usageFunc(pod, podsState) } +func (m *FakeUsageCalculator) MatchesScope(_ *corev1.Pod, _ corev1.ResourceQuotaScope) (bool, bool) { + return false, false +} + +func (m *FakeUsageCalculator) SourceUsage(pod *corev1.Pod, podsState []*corev1.Pod) (corev1.ResourceList, error, bool) { + return m.PodUsageFunc(pod, podsState) +} + +func makeVmiPod(name, vmiName, vmiUID, ns string, resList corev1.ResourceList, phase corev1.PodPhase) *corev1.Pod { + pod := makePod(name, "", resList, phase) + pod.Namespace = ns + pod.OwnerReferences = []metav1.OwnerReference{ + {Kind: "VirtualMachineInstance", Name: vmiName, UID: types.UID(vmiUID)}, + } + return pod +} + func makePod(name, pcName string, resList corev1.ResourceList, phase corev1.PodPhase) *corev1.Pod { return &corev1.Pod{ ObjectMeta: metav1.ObjectMeta{ diff --git a/pkg/aaq-controller/aaq-gate-controller/aaq-gate-controller.go b/pkg/aaq-controller/aaq-gate-controller/aaq-gate-controller.go index 9b646e2ce..ae66a596c 100644 --- a/pkg/aaq-controller/aaq-gate-controller/aaq-gate-controller.go +++ b/pkg/aaq-controller/aaq-gate-controller/aaq-gate-controller.go @@ -279,6 +279,8 @@ func (ctrl *AaqGateController) execute(ns string) (error, enqueueState) { if err != nil { return err, Immediate } + scopedRqs, nonScopedRqs := splitVmiScopedRqs(rqs) + podObjs, err := ctrl.podInformer.GetIndexer().ByIndex(cache.NamespaceIndex, ns) if err != nil { return err, Immediate @@ -291,6 +293,11 @@ func (ctrl *AaqGateController) execute(ns string) (error, enqueueState) { podCopy := pod.DeepCopy() podCopy.Spec.SchedulingGates = []v1.PodSchedulingGate{} + if !ctrl.passesScopedQuotaCheck(podCopy, scopedRqs) { + ctrl.recorder.Event(pod, v1.EventTypeWarning, v1.EventTypeWarning, "exceeded scoped quota") + continue + } + podToCreateAttr := k8sadmission.NewAttributesRecord(podCopy, nil, apiextensions.Kind("Pod").WithVersion("version"), podCopy.Namespace, podCopy.Name, v1alpha12.Resource("pods").WithVersion("version"), "", k8sadmission.Create, @@ -301,9 +308,9 @@ func (ctrl *AaqGateController) execute(ns string) (error, enqueueState) { return nil, Immediate } - newRq, err := resourcequota2.CheckRequest(rqs, podToCreateAttr, ctrl.aaqEvaluator, []resourcequota.LimitedResource{currPodLimitedResource}) + newRq, err := resourcequota2.CheckRequest(nonScopedRqs, podToCreateAttr, ctrl.aaqEvaluator, []resourcequota.LimitedResource{currPodLimitedResource}) if err == nil { - rqs = newRq + nonScopedRqs = newRq aaqjqc.Status.PodsInJobQueue = append(aaqjqc.Status.PodsInJobQueue, pod.Name) } else { ctrl.recorder.Event(pod, v1.EventTypeWarning, v1.EventTypeWarning, util.IgnoreRqErr(err.Error())) @@ -383,7 +390,11 @@ func (ctrl *AaqGateController) getArtificialRqsForGateController(ns string) ([]v for _, arqObj := range arqsObjs { arq := arqObj.(*v1alpha12.ApplicationAwareResourceQuota) rq := v1.ResourceQuota{ObjectMeta: metav1.ObjectMeta{Name: arq.Name, Namespace: ns}, - Spec: v1.ResourceQuotaSpec{Hard: arq.Spec.Hard}, + Spec: v1.ResourceQuotaSpec{ + Hard: arq.Spec.Hard, + Scopes: arq.Spec.Scopes, + ScopeSelector: arq.Spec.ScopeSelector, + }, Status: v1.ResourceQuotaStatus{Hard: arq.Status.Hard, Used: arq.Status.Used}, } rqs = append(rqs, rq) @@ -490,3 +501,79 @@ func (ctrl *AaqGateController) AddMapping(_, namespaceName string) { func (ctrl *AaqGateController) RemoveMapping(_, namespaceName string) { ctrl.nsQueue.Add(namespaceName) } + +var vmiScopes = map[v1.ResourceQuotaScope]bool{ + v1alpha12.VmiStarting: true, + v1alpha12.VmiMigrating: true, +} + +func splitVmiScopedRqs(rqs []v1.ResourceQuota) (scoped []v1.ResourceQuota, nonScoped []v1.ResourceQuota) { + for _, rq := range rqs { + if hasVmiScope(&rq) { + scoped = append(scoped, rq) + } else { + nonScoped = append(nonScoped, rq) + } + } + return +} + +func hasVmiScope(rq *v1.ResourceQuota) bool { + for _, scope := range rq.Spec.Scopes { + if vmiScopes[scope] { + return true + } + } + if rq.Spec.ScopeSelector != nil { + for _, expr := range rq.Spec.ScopeSelector.MatchExpressions { + if vmiScopes[expr.ScopeName] && expr.Operator == v1.ScopeSelectorOpExists { + return true + } + } + } + return false +} + +func (ctrl *AaqGateController) passesScopedQuotaCheck(pod *v1.Pod, scopedRqs []v1.ResourceQuota) bool { + if len(scopedRqs) == 0 { + return true + } + existingPods, err := ctrl.podInformer.GetIndexer().ByIndex(cache.NamespaceIndex, pod.Namespace) + if err != nil { + return true + } + var podsState []*v1.Pod + for _, obj := range existingPods { + podsState = append(podsState, obj.(*v1.Pod)) + } + usage, err := ctrl.aaqEvaluator.SourceCalculatorUsage(pod, podsState) + if err != nil { + return true + } + for i := range scopedRqs { + rq := &scopedRqs[i] + matched, matchErr := ctrl.aaqEvaluator.Matches(rq, pod) + if matchErr != nil || !matched { + continue + } + for resourceName, hardVal := range rq.Status.Hard { + usedVal := rq.Status.Used[resourceName] + podUsage, exists := usage[resourceName] + if !exists { + continue + } + newUsed := usedVal.DeepCopy() + newUsed.Add(podUsage) + if newUsed.Cmp(hardVal) > 0 { + return false + } + } + for resourceName, podUsage := range usage { + if current, exists := rq.Status.Used[resourceName]; exists { + current.Add(podUsage) + rq.Status.Used[resourceName] = current + } + } + } + return true +} diff --git a/pkg/aaq-controller/built-in-usage-calculators/virt-launcher-usage-calc.go b/pkg/aaq-controller/built-in-usage-calculators/virt-launcher-usage-calc.go index e4174c595..daa4d15f4 100644 --- a/pkg/aaq-controller/built-in-usage-calculators/virt-launcher-usage-calc.go +++ b/pkg/aaq-controller/built-in-usage-calculators/virt-launcher-usage-calc.go @@ -80,6 +80,19 @@ func (launchercalc *VirtLauncherCalculator) PodUsageFunc(pod *corev1.Pod, existi return corev1.ResourceList{}, nil, true } +func (launchercalc *VirtLauncherCalculator) SourceUsage(pod *corev1.Pod, _ []*corev1.Pod) (corev1.ResourceList, error, bool) { + if len(pod.OwnerReferences) == 0 || pod.OwnerReferences[0].Kind != v15.VirtualMachineInstanceGroupVersionKind.Kind { + return corev1.ResourceList{}, nil, false + } + vmiObj, vmiExists, err := launchercalc.vmiInformer.GetIndexer().GetByKey(fmt.Sprintf("%s/%s", pod.Namespace, pod.OwnerReferences[0].Name)) + if err != nil || !vmiExists { + return corev1.ResourceList{}, nil, false + } + vmi := vmiObj.(*v15.VirtualMachineInstance) + rl, err := launchercalc.calculateSourceUsageByConfig(pod, vmi) + return rl, err, true +} + func (launchercalc *VirtLauncherCalculator) calculateSourceUsageByConfig(pod *corev1.Pod, vmi *v15.VirtualMachineInstance) (corev1.ResourceList, error) { return launchercalc.CalculateUsageByConfig(pod, vmi, true) } @@ -343,6 +356,54 @@ func getSourcePod(pods []*corev1.Pod, vmi *v15.VirtualMachineInstance) *corev1.P return curPod } +func (launchercalc *VirtLauncherCalculator) getVmiForPod(pod *corev1.Pod) *v15.VirtualMachineInstance { + if len(pod.OwnerReferences) == 0 || pod.OwnerReferences[0].Kind != v15.VirtualMachineInstanceGroupVersionKind.Kind { + return nil + } + vmiObj, exists, err := launchercalc.vmiInformer.GetIndexer().GetByKey(fmt.Sprintf("%s/%s", pod.Namespace, pod.OwnerReferences[0].Name)) + if err != nil || !exists { + return nil + } + return vmiObj.(*v15.VirtualMachineInstance) +} + +// MatchesScope implements ScopeEvaluator for VMI-specific scopes. +// Returns (matched, handled). handled=false means this calculator doesn't know about the scope. +func (launchercalc *VirtLauncherCalculator) MatchesScope(pod *corev1.Pod, scope corev1.ResourceQuotaScope) (bool, bool) { + switch scope { + case v1alpha1.VmiStarting: + return launchercalc.isVmiStarting(pod), true + case v1alpha1.VmiMigrating: + return launchercalc.isVmiMigrationTarget(pod), true + } + return false, false +} + +func (launchercalc *VirtLauncherCalculator) isVmiStarting(pod *corev1.Pod) bool { + vmi := launchercalc.getVmiForPod(pod) + if vmi == nil { + return false + } + switch vmi.Status.Phase { + case v15.Running, v15.Succeeded, v15.Failed, v15.Unknown: + return false + default: + return true + } +} + +func (launchercalc *VirtLauncherCalculator) isVmiMigrationTarget(pod *corev1.Pod) bool { + vmi := launchercalc.getVmiForPod(pod) + if vmi == nil { + return false + } + vmim, err := getLatestVmimIfExist(vmi, pod.Namespace, launchercalc.migrationInformer) + if err != nil || vmim == nil || vmim.IsFinal() { + return false + } + return getTargetPod([]*corev1.Pod{pod}, vmim) != nil +} + const computeContainerName = "compute" func createPodWithComputeContainerOnly(pod *corev1.Pod) *corev1.Pod { diff --git a/pkg/aaq-server/handler/handler.go b/pkg/aaq-server/handler/handler.go index 99a891ba4..5bdac4748 100644 --- a/pkg/aaq-server/handler/handler.go +++ b/pkg/aaq-server/handler/handler.go @@ -116,12 +116,15 @@ func (v Handler) validateApplicationAwareResourceQuota() (*admissionv1.Admission if err := json.Unmarshal(v.request.Object.Raw, &arq); err != nil { return nil, err } + if err := validateAaqScopeSelectors(arq.Spec.ScopeSelector); err != nil { + return reviewResponse(v.request.UID, false, http.StatusForbidden, err.Error()), nil + } rq := &v1.ResourceQuota{} rq.Namespace = arq.Namespace rq.Name = createRQName() rq.Spec.Hard = arq.Spec.Hard - rq.Spec.ScopeSelector = arq.Spec.ScopeSelector - rq.Spec.Scopes = arq.Spec.Scopes + rq.Spec.ScopeSelector = filterAaqScopeSelector(arq.Spec.ScopeSelector) + rq.Spec.Scopes = filterAaqScopes(arq.Spec.Scopes) _, err := v.aaqCli.CoreV1().ResourceQuotas(arq.Namespace).Create(context.Background(), rq, metav1.CreateOptions{DryRun: []string{metav1.DryRunAll}}) if err != nil { return reviewResponse(v.request.UID, false, http.StatusForbidden, util.IgnoreRqErr(err.Error())), nil @@ -135,11 +138,14 @@ func (v Handler) validateApplicationAwareClusterResourceQuota() (*admissionv1.Ad if err := json.Unmarshal(v.request.Object.Raw, &acrq); err != nil { return nil, err } + if err := validateAaqScopeSelectors(acrq.Spec.Quota.ScopeSelector); err != nil { + return reviewResponse(v.request.UID, false, http.StatusForbidden, err.Error()), nil + } rq := &v1.ResourceQuota{} rq.Name = createRQName() rq.Spec.Hard = acrq.Spec.Quota.Hard - rq.Spec.ScopeSelector = acrq.Spec.Quota.ScopeSelector - rq.Spec.Scopes = acrq.Spec.Quota.Scopes + rq.Spec.ScopeSelector = filterAaqScopeSelector(acrq.Spec.Quota.ScopeSelector) + rq.Spec.Scopes = filterAaqScopes(acrq.Spec.Quota.Scopes) _, err := v.aaqCli.CoreV1().ResourceQuotas(v1.NamespaceDefault).Create(context.Background(), rq, metav1.CreateOptions{DryRun: []string{metav1.DryRunAll}}) if err != nil { return reviewResponse(v.request.UID, false, http.StatusForbidden, util.IgnoreRqErr(err.Error())), nil @@ -209,3 +215,46 @@ func getResourcesNames(resourceList v1.ResourceList) []v1.ResourceName { } return keys } + +func validateAaqScopeSelectors(selector *v1.ScopeSelector) error { + if selector == nil { + return nil + } + for _, expr := range selector.MatchExpressions { + if aaqCustomScopes[expr.ScopeName] && expr.Operator != v1.ScopeSelectorOpExists { + return fmt.Errorf("scope %q only supports the Exists operator", expr.ScopeName) + } + } + return nil +} + +var aaqCustomScopes = map[v1.ResourceQuotaScope]bool{ + v1alpha1.VmiStarting: true, + v1alpha1.VmiMigrating: true, +} + +func filterAaqScopes(scopes []v1.ResourceQuotaScope) []v1.ResourceQuotaScope { + var filtered []v1.ResourceQuotaScope + for _, scope := range scopes { + if !aaqCustomScopes[scope] { + filtered = append(filtered, scope) + } + } + return filtered +} + +func filterAaqScopeSelector(selector *v1.ScopeSelector) *v1.ScopeSelector { + if selector == nil { + return nil + } + var filtered []v1.ScopedResourceSelectorRequirement + for _, expr := range selector.MatchExpressions { + if !aaqCustomScopes[expr.ScopeName] { + filtered = append(filtered, expr) + } + } + if len(filtered) == 0 { + return nil + } + return &v1.ScopeSelector{MatchExpressions: filtered} +} diff --git a/staging/src/kubevirt.io/application-aware-quota-api/pkg/apis/core/v1alpha1/types.go b/staging/src/kubevirt.io/application-aware-quota-api/pkg/apis/core/v1alpha1/types.go index 619539ce7..011508a2d 100644 --- a/staging/src/kubevirt.io/application-aware-quota-api/pkg/apis/core/v1alpha1/types.go +++ b/staging/src/kubevirt.io/application-aware-quota-api/pkg/apis/core/v1alpha1/types.go @@ -202,6 +202,12 @@ const ( ResourceRequestsVmiMemory corev1.ResourceName = "requests.memory/vmi" // Short form of requested memory for the VMI, in bytes. ResourceRequestsVmiMemoryShort corev1.ResourceName = "memory/vmi" + + // VmiStarting matches pods associated with a VMI that is not yet Running + // (VMI phases: "", Pending, Scheduling, Scheduled, WaitingForSync) + VmiStarting corev1.ResourceQuotaScope = "VmiStarting" + // VmiMigrating matches target virt-launcher pods created for an active VMI migration + VmiMigrating corev1.ResourceQuotaScope = "VmiMigrating" ) // AAQPriorityClass defines the priority class of the AAQ control plane. diff --git a/test-migrating.sh b/test-migrating.sh new file mode 100755 index 000000000..62dc929aa --- /dev/null +++ b/test-migrating.sh @@ -0,0 +1,175 @@ +#!/bin/bash +set -euo pipefail + +KK="kubevirtci/cluster-up/kubectl.sh" +NS="default" +TOTAL_VMIS=5 +MAX_MIGRATING=2 + +CALC_CONFIG=$($KK get aaq aaq -o jsonpath='{.spec.configuration.vmiCalculatorConfiguration.configName}' 2>/dev/null) +echo "VMI calculator config: $CALC_CONFIG" + +case "$CALC_CONFIG" in + DedicatedVirtualResources) + CPU_RESOURCE="cpu/vmi" + MEM_RESOURCE="memory/vmi" + ;; + *) + CPU_RESOURCE="requests.cpu" + MEM_RESOURCE="requests.memory" + ;; +esac + +echo "Using resources: $CPU_RESOURCE, $MEM_RESOURCE" +echo "Will create $TOTAL_VMIS VMIs, start them, then migrate all" +echo "Quota allows $MAX_MIGRATING migrating at a time" + +NODES=$($KK get nodes --no-headers 2>/dev/null | grep -c "Ready" || true) +if [[ "$NODES" -lt 2 ]]; then + echo "ERROR: Need at least 2 nodes for migration testing, found $NODES" + exit 1 +fi +echo "Cluster has $NODES nodes" + +TEST_LABEL="migrate-test-$(date +%s)" + +cleanup() { + echo "" + echo "=== Cleanup ===" + $KK delete virtualmachineinstancemigration -l test-run="$TEST_LABEL" -n $NS --ignore-not-found 2>/dev/null || true + $KK delete vmi -l test-run="$TEST_LABEL" -n $NS --ignore-not-found 2>/dev/null || true + $KK delete arq migrating-vmi-quota -n $NS --ignore-not-found 2>/dev/null || true + $KK delete aaqjqc -n $NS --ignore-not-found 2>/dev/null || true + echo "Done." +} +trap cleanup EXIT + +echo "" +echo "=== Step 1: Create VmiMigrating scoped ARQ (max $MAX_MIGRATING migrating VMIs) ===" +$KK apply -f - </dev/null +apiVersion: kubevirt.io/v1 +kind: VirtualMachineInstance +metadata: + name: test-vmi-$i + namespace: $NS + labels: + test-run: "$TEST_LABEL" +spec: + domain: + resources: + requests: + memory: 512Mi + cpu: "1" + devices: + disks: + - name: containerdisk + disk: + bus: virtio + interfaces: + - name: default + masquerade: {} + networks: + - name: default + pod: {} + volumes: + - name: containerdisk + containerDisk: + image: quay.io/kubevirt/cirros-container-disk-demo:latest +EOF + echo " Created test-vmi-$i" +done + +echo "" +echo "Waiting for all VMIs to reach Running..." +for attempt in $(seq 1 60); do + running=$($KK get vmi -l test-run="$TEST_LABEL" -n $NS -o jsonpath='{range .items[*]}{.status.phase}{"\n"}{end}' 2>/dev/null | grep -c "Running" || true) + echo " $running/$TOTAL_VMIS Running" + if [[ "$running" -eq "$TOTAL_VMIS" ]]; then + echo "All VMIs are Running." + break + fi + sleep 5 +done + +echo "" +echo "Quota before migration:" +$KK get arq migrating-vmi-quota -n $NS -o jsonpath='{.status.used}' 2>/dev/null; echo "" + +echo "" +echo "=== Step 3: Trigger migration for all $TOTAL_VMIS VMIs ===" +for i in $(seq 1 $TOTAL_VMIS); do + $KK apply -f - </dev/null +apiVersion: kubevirt.io/v1 +kind: VirtualMachineInstanceMigration +metadata: + name: migrate-vmi-$i + namespace: $NS + labels: + test-run: "$TEST_LABEL" +spec: + vmiName: test-vmi-$i +EOF + echo " Triggered migration for test-vmi-$i" +done + +echo "" +echo "=== Step 4: Watching (expect at most $MAX_MIGRATING migrating at a time) ===" +printf "%-10s %-12s %-12s %-12s %s\n" "TIME" "MIGRATING" "SUCCEEDED" "GATED" "QUOTA USED" +printf "%-10s %-12s %-12s %-12s %s\n" "----" "---------" "---------" "-----" "----------" + +max_migrating_seen=0 +for i in $(seq 1 60); do + vmim_phases=$($KK get virtualmachineinstancemigration -l test-run="$TEST_LABEL" -n $NS -o jsonpath='{range .items[*]}{.status.phase}{"\n"}{end}' 2>/dev/null) + active=$(echo "$vmim_phases" | grep -cE "Scheduling|Scheduled|PreparingTarget|TargetReady|Running" || true) + succeeded=$(echo "$vmim_phases" | grep -c "Succeeded" || true) + + gated=$($KK get pods -n $NS -l test-run="$TEST_LABEL" -o jsonpath='{range .items[*]}{.spec.schedulingGates}{"\n"}{end}' 2>/dev/null | grep -c "ApplicationAwareQuotaGate" || true) + + if [[ $active -gt $max_migrating_seen ]]; then + max_migrating_seen=$active + fi + + used=$($KK get arq migrating-vmi-quota -n $NS -o jsonpath='{.status.used}' 2>/dev/null || echo "?") + + printf "%-10s %-12s %-12s %-12s %s\n" "$(date +%H:%M:%S)" "$active" "$succeeded" "$gated" "$used" + + if [[ "$succeeded" -eq "$TOTAL_VMIS" ]]; then + echo "" + echo "All $TOTAL_VMIS migrations completed." + break + fi + sleep 5 +done + +echo "" +echo "=== Results ===" +echo "Max actively migrating VMIs seen simultaneously: $max_migrating_seen" +if [[ $max_migrating_seen -le $MAX_MIGRATING ]]; then + echo "PASS: Never exceeded $MAX_MIGRATING migrating VMIs at a time" +else + echo "FAIL: Saw $max_migrating_seen migrating VMIs (limit was $MAX_MIGRATING)" +fi + +echo "" +echo "Final quota:" +$KK get arq migrating-vmi-quota -n $NS -o jsonpath='{.status.used}' 2>/dev/null; echo "" diff --git a/test.sh b/test.sh new file mode 100755 index 000000000..0eb95b9c2 --- /dev/null +++ b/test.sh @@ -0,0 +1,131 @@ +#!/bin/bash +set -euo pipefail + +KK="kubevirtci/cluster-up/kubectl.sh" +NS="default" +TOTAL_VMIS=10 +MAX_STARTING=2 + +CALC_CONFIG=$($KK get aaq aaq -o jsonpath='{.spec.configuration.vmiCalculatorConfiguration.configName}' 2>/dev/null) +echo "VMI calculator config: $CALC_CONFIG" + +case "$CALC_CONFIG" in + DedicatedVirtualResources) + CPU_RESOURCE="cpu/vmi" + MEM_RESOURCE="memory/vmi" + ;; + *) + CPU_RESOURCE="requests.cpu" + MEM_RESOURCE="requests.memory" + ;; +esac + +echo "Using resources: $CPU_RESOURCE, $MEM_RESOURCE" +echo "Will create $TOTAL_VMIS VMIs, quota allows $MAX_STARTING starting at a time" + +TEST_LABEL="scope-test-$(date +%s)" + +cleanup() { + echo "" + echo "=== Cleanup ===" + $KK delete vmi -l test-run="$TEST_LABEL" -n $NS --ignore-not-found 2>/dev/null || true + $KK delete arq starting-vmi-quota -n $NS --ignore-not-found 2>/dev/null || true + $KK delete aaqjqc -n $NS --ignore-not-found 2>/dev/null || true + echo "Done." +} +trap cleanup EXIT + +echo "" +echo "=== Step 1: Create VmiStarting scoped ARQ (max $MAX_STARTING starting VMIs) ===" +# Each VMI requests 1 CPU, so limit to MAX_STARTING CPUs +$KK apply -f - </dev/null +apiVersion: kubevirt.io/v1 +kind: VirtualMachineInstance +metadata: + name: test-vmi-$i + namespace: $NS + labels: + test-run: "$TEST_LABEL" +spec: + domain: + resources: + requests: + memory: 512Mi + cpu: "1" + devices: + disks: + - name: containerdisk + disk: + bus: virtio + interfaces: + - name: default + masquerade: {} + networks: + - name: default + pod: {} + volumes: + - name: containerdisk + containerDisk: + image: quay.io/kubevirt/cirros-container-disk-demo:latest +EOF + echo " Created test-vmi-$i" +done + +echo "" +echo "=== Step 3: Watching (expect at most $MAX_STARTING non-Running VMIs at a time) ===" +printf "%-10s %-6s %-8s %-8s %-10s %s\n" "TIME" "PODS" "RUNNING" "GATED" "UNGATED-ST" "QUOTA USED" +printf "%-10s %-6s %-8s %-8s %-10s %s\n" "----" "----" "-------" "-----" "----------" "----------" + +max_ungated_starting=0 +for i in $(seq 1 60); do + total_pods=$($KK get pods -n $NS -l test-run="$TEST_LABEL" --no-headers 2>/dev/null | wc -l || echo 0) + running=$($KK get vmi -l test-run="$TEST_LABEL" -n $NS -o jsonpath='{range .items[*]}{.status.phase}{"\n"}{end}' 2>/dev/null | grep -c "Running" || true) + + gated=$($KK get pods -n $NS -l test-run="$TEST_LABEL" -o jsonpath='{range .items[*]}{.spec.schedulingGates}{"\n"}{end}' 2>/dev/null | grep -c "ApplicationAwareQuotaGate" || true) + + ungated_starting=$((total_pods - gated - running)) + if [[ $ungated_starting -lt 0 ]]; then ungated_starting=0; fi + if [[ $ungated_starting -gt $max_ungated_starting ]]; then + max_ungated_starting=$ungated_starting + fi + + used=$($KK get arq starting-vmi-quota -n $NS -o jsonpath='{.status.used}' 2>/dev/null || echo "?") + + printf "%-10s %-6s %-8s %-8s %-10s %s\n" "$(date +%H:%M:%S)" "$total_pods" "$running" "$gated" "$ungated_starting" "$used" + + if [[ "$running" -eq "$TOTAL_VMIS" ]]; then + echo "" + echo "All $TOTAL_VMIS VMIs are Running." + break + fi + sleep 5 +done + +echo "" +echo "=== Results ===" +echo "Max ungated starting VMIs seen simultaneously: $max_ungated_starting" +if [[ $max_ungated_starting -le $MAX_STARTING ]]; then + echo "PASS: Never exceeded $MAX_STARTING ungated starting VMIs at a time" +else + echo "FAIL: Saw $max_ungated_starting ungated starting VMIs (limit was $MAX_STARTING)" +fi