federatedresourcequota add the support to multi-components scheduling

Signed-off-by: zhzhuang-zju <[email protected]>

Author: zhzhuang-zju

pkg/webhook/resourcebinding/validating.go

@@ -356,6 +356,9 @@ func isQuotaRelevantFieldChanged(oldRB, newRB *workv1alpha2.ResourceBinding) boo
 	if isScheduledReplicasChanged(oldRB, newRB) {
 		return true
 	}
+	if (len(oldRB.Spec.Components) != 0 || len(newRB.Spec.Components) != 0) && isScheduledClusterChanged(oldRB, newRB) {
+		return true
+	}
 	return isComponentsChanged(oldRB, newRB)
 }
 
@@ -383,6 +386,22 @@ func isScheduledReplicasChanged(oldRB, newRB *workv1alpha2.ResourceBinding) bool
 	return oldScheduledReplicas != newScheduledReplicas
 }
 
+func isScheduledClusterChanged(oldRB, newRB *workv1alpha2.ResourceBinding) bool {
+	if len(oldRB.Spec.Clusters) != len(newRB.Spec.Clusters) {
+		return true
+	}
+	oldClusterMap := make(map[string]struct{})
+	for _, c := range oldRB.Spec.Clusters {
+		oldClusterMap[c.Name] = struct{}{}
+	}
+	for _, c := range newRB.Spec.Clusters {
+		if _, exists := oldClusterMap[c.Name]; !exists {
+			return true
+		}
+	}
+	return false
+}
+
 func isComponentsChanged(oldRB, newRB *workv1alpha2.ResourceBinding) bool {
 	if len(oldRB.Spec.Components) != len(newRB.Spec.Components) {
 		return true

test/e2e/suites/base/federatedresourcequota_test.go

@@ -27,14 +27,18 @@ import (
 	"github.com/onsi/gomega"
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
+	apiextensionsv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	"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/runtime/schema"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/rand"
 	"k8s.io/cli-runtime/pkg/genericclioptions"
 	"k8s.io/utils/ptr"
+	"sigs.k8s.io/yaml"
 
 	policyv1alpha1 "github.com/karmada-io/karmada/pkg/apis/policy/v1alpha1"
 	workv1alpha2 "github.com/karmada-io/karmada/pkg/apis/work/v1alpha2"
@@ -485,3 +489,156 @@ var _ = ginkgo.Describe("FederatedResourceQuota enforcement testing", func() {
 		})
 	})
 })
+
+var _ = ginkgo.Describe("Multi-Components: FederatedResourceQuota enforcement testing", func() {
+	ginkgo.Context("The federatedResourceQuota usage should be calculated correctly", func() {
+		var (
+			frqNamespace, frqName    string
+			flinkDeploymentCRD       apiextensionsv1.CustomResourceDefinition
+			flinkDeploymentNamespace string
+			flinkDeploymentName      string
+			flinkDeploymentObj       *unstructured.Unstructured
+			flinkDeploymentGVR       schema.GroupVersionResource
+			federatedResourceQuota   *policyv1alpha1.FederatedResourceQuota
+		)
+
+		ginkgo.BeforeEach(func() {
+			ginkgo.By("create FlinkDeployment CRD on karmada control plane", func() {
+				err := yaml.Unmarshal([]byte(flinkDeploymentCRDYAML), &flinkDeploymentCRD)
+				gomega.Expect(err).ShouldNot(gomega.HaveOccurred())
+
+				framework.CreateCRD(dynamicClient, &flinkDeploymentCRD)
+				ginkgo.DeferCleanup(func() {
+					framework.RemoveCRD(dynamicClient, flinkDeploymentCRD.Name)
+				})
+			})
+
+			ginkgo.By("create FederatedResourceQuota for FlinkDeployment", func() {
+				// To avoid conflicts with other test cases, use random strings to generate unique namespaces instead of using testNamespace.
+				frqNamespace = fmt.Sprintf("karmadatest-%s", rand.String(RandomStrLength))
+				err := setupTestNamespace(frqNamespace, kubeClient)
+				gomega.Expect(err).ShouldNot(gomega.HaveOccurred())
+				ginkgo.DeferCleanup(func() {
+					framework.RemoveNamespace(kubeClient, frqNamespace)
+				})
+				frqName = federatedResourceQuotaPrefix + rand.String(RandomStrLength)
+				overall := corev1.ResourceList{
+					"cpu":    resource.MustParse("10"),
+					"memory": resource.MustParse("1Gi"),
+				}
+				federatedResourceQuota = helper.NewFederatedResourceQuotaWithOverall(frqNamespace, frqName, overall)
+				framework.CreateFederatedResourceQuota(karmadaClient, federatedResourceQuota)
+				framework.WaitFederatedResourceQuotaFitWith(karmadaClient, frqNamespace, frqName, func(frq *policyv1alpha1.FederatedResourceQuota) bool {
+					// To avoid race condition, ensure that OverallUsed is not nil first, and then flinkDeployment can be created.
+					// MoreInfo can refer to https://github.com/karmada-io/karmada/pull/6876#issuecomment-3455446833.
+					return frq.Status.Overall != nil
+				})
+				ginkgo.DeferCleanup(func() {
+					framework.RemoveFederatedResourceQuota(karmadaClient, frqNamespace, frqName)
+				})
+			})
+
+			ginkgo.By("propagate FlinkDeployment CRD to all clusters", func() {
+				cpp := helper.NewClusterPropagationPolicy("flink-deployment-cpp", []policyv1alpha1.ResourceSelector{
+					{
+						APIVersion: flinkDeploymentCRD.APIVersion,
+						Kind:       flinkDeploymentCRD.Kind,
+						Name:       flinkDeploymentCRD.Name,
+					},
+				}, policyv1alpha1.Placement{
+					ClusterAffinity: &policyv1alpha1.ClusterAffinity{
+						ClusterNames: framework.ClusterNames(),
+					},
+				})
+
+				framework.CreateClusterPropagationPolicy(karmadaClient, cpp)
+				framework.WaitCRDPresentOnClusters(karmadaClient, framework.ClusterNames(),
+					fmt.Sprintf("%s/%s", flinkDeploymentCRD.Spec.Group, "v1beta1"), flinkDeploymentCRD.Spec.Names.Kind)
+				ginkgo.DeferCleanup(func() {
+					framework.RemoveClusterPropagationPolicy(karmadaClient, cpp.Name)
+				})
+			})
+
+			ginkgo.By("create FlinkDeployment on karmada control plane", func() {
+				flinkDeploymentNamespace = frqNamespace
+				flinkDeploymentName = fmt.Sprintf("flinkdeployment-%s", rand.String(RandomStrLength))
+				flinkDeploymentObj = &unstructured.Unstructured{}
+				err := yaml.Unmarshal([]byte(flinkDeploymentCRYAML), flinkDeploymentObj)
+				gomega.Expect(err).ShouldNot(gomega.HaveOccurred())
+
+				flinkDeploymentObj.SetNamespace(flinkDeploymentNamespace)
+				flinkDeploymentObj.SetName(flinkDeploymentName)
+				err = unstructured.SetNestedField(flinkDeploymentObj.Object, int64(3), "spec", "jobManager", "replicas")
+				gomega.Expect(err).ShouldNot(gomega.HaveOccurred())
+				err = unstructured.SetNestedField(flinkDeploymentObj.Object, map[string]interface{}{
+					"cpu":    int64(2),
+					"memory": "50Mi",
+				}, "spec", "jobManager", "resource")
+				gomega.Expect(err).ShouldNot(gomega.HaveOccurred())
+				err = unstructured.SetNestedField(flinkDeploymentObj.Object, map[string]interface{}{
+					"cpu":    int64(1),
+					"memory": "100Mi",
+				}, "spec", "taskManager", "resource")
+				gomega.Expect(err).ShouldNot(gomega.HaveOccurred())
+
+				flinkDeploymentGVR = schema.GroupVersionResource{
+					Group:    flinkDeploymentObj.GroupVersionKind().Group,
+					Version:  flinkDeploymentObj.GroupVersionKind().Version,
+					Resource: "flinkdeployments",
+				}
+
+				_, err = dynamicClient.Resource(flinkDeploymentGVR).Namespace(flinkDeploymentNamespace).
+					Create(context.Background(), flinkDeploymentObj, metav1.CreateOptions{})
+				gomega.Expect(err).ShouldNot(gomega.HaveOccurred())
+				ginkgo.DeferCleanup(func() {
+					err := dynamicClient.Resource(flinkDeploymentGVR).Namespace(flinkDeploymentNamespace).
+						Delete(context.Background(), flinkDeploymentName, metav1.DeleteOptions{})
+					if err != nil && !apierrors.IsNotFound(err) {
+						// Ignore not found errors during cleanup, but other errors should be handled
+						gomega.Expect(err).ShouldNot(gomega.HaveOccurred())
+					}
+				})
+			})
+
+			ginkgo.By("propagate FlinkDeployment resource to one cluster", func() {
+				pp := helper.NewPropagationPolicy(flinkDeploymentNamespace, flinkDeploymentName, []policyv1alpha1.ResourceSelector{
+					{
+						APIVersion: flinkDeploymentObj.GetAPIVersion(),
+						Kind:       flinkDeploymentObj.GetKind(),
+						Name:       flinkDeploymentObj.GetName(),
+					},
+				}, policyv1alpha1.Placement{
+					ClusterAffinity: &policyv1alpha1.ClusterAffinity{
+						ClusterNames: framework.ClusterNames(),
+					},
+					SpreadConstraints: []policyv1alpha1.SpreadConstraint{
+						{
+							SpreadByField: policyv1alpha1.SpreadByFieldCluster,
+							MaxGroups:     1,
+							MinGroups:     1,
+						},
+					},
+					ReplicaScheduling: &policyv1alpha1.ReplicaSchedulingStrategy{
+						ReplicaSchedulingType:     policyv1alpha1.ReplicaSchedulingTypeDivided,
+						ReplicaDivisionPreference: policyv1alpha1.ReplicaDivisionPreferenceAggregated,
+					},
+				})
+				framework.CreatePropagationPolicy(karmadaClient, pp)
+				ginkgo.DeferCleanup(func() {
+					framework.RemovePropagationPolicy(karmadaClient, pp.Namespace, pp.Name)
+				})
+			})
+		})
+
+		ginkgo.It("federatedResourceQuota usage should be calculated correctly", func() {
+			framework.WaitFederatedResourceQuotaFitWith(karmadaClient, frqNamespace, frqName, func(frq *policyv1alpha1.FederatedResourceQuota) bool {
+				expectedUsed := corev1.ResourceList{
+					"cpu":    resource.MustParse("7"),
+					"memory": resource.MustParse("250Mi"),
+				}
+				return frq.Status.OverallUsed != nil && frq.Status.OverallUsed.Cpu().Equal(*expectedUsed.Cpu()) &&
+					frq.Status.OverallUsed.Memory().Equal(*expectedUsed.Memory())
+			})
+		})
+	})
+})