feat: support rp in clusterSchedulingPolicySnapshot controller (#160)

Author: jwtty

cmd/hubagent/workload/setup.go

@@ -215,7 +215,7 @@ func SetupControllers(ctx context.Context, wg *sync.WaitGroup, mgr ctrl.Manager,
 		if err := (&clusterschedulingpolicysnapshot.Reconciler{
 			Client:              mgr.GetClient(),
 			PlacementController: clusterResourcePlacementControllerV1Beta1,
-		}).SetupWithManager(mgr); err != nil {
+		}).SetupWithManagerForClusterSchedulingPolicySnapshot(mgr); err != nil {
 			klog.ErrorS(err, "Unable to set up the clusterSchedulingPolicySnapshot watcher")
 			return err
 		}

pkg/controllers/clusterresourceplacement/suite_test.go

@@ -136,7 +136,7 @@ var _ = BeforeSuite(func() {
 	err = (&clusterschedulingpolicysnapshot.Reconciler{
 		Client:              mgr.GetClient(),
 		PlacementController: crpController,
-	}).SetupWithManager(mgr)
+	}).SetupWithManagerForClusterSchedulingPolicySnapshot(mgr)
 	Expect(err).Should(Succeed(), "failed to create clusterSchedulingPolicySnapshot watcher")
 
 	err = (&clusterresourceplacementwatcher.Reconciler{

pkg/controllers/clusterschedulingpolicysnapshot/controller.go

@@ -22,7 +22,6 @@ import (
 	"fmt"
 	"time"
 
-	"k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/klog/v2"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
@@ -33,48 +32,58 @@ import (
 	"github.com/kubefleet-dev/kubefleet/pkg/utils/controller"
 )
 
-// Reconciler reconciles a clusterSchedulingPolicySnapshot object.
+// Reconciler reconciles a clusterSchedulingPolicySnapshot or schedulingPolicySnapshot object.
 type Reconciler struct {
 	client.Client
 
 	// PlacementController exposes the placement queue for the reconciler to push to.
 	PlacementController controller.Controller
 }
 
-// Reconcile triggers a single CRP reconcile round when scheduling policy has changed.
+// Reconcile triggers a single placement reconcile round when scheduling policy has changed.
 func (r *Reconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
-	name := req.NamespacedName
-	snapshot := fleetv1beta1.ClusterSchedulingPolicySnapshot{}
-	snapshotKRef := klog.KRef(name.Namespace, name.Name)
+	var snapshot fleetv1beta1.PolicySnapshotObj
+	var err error
+	snapshotKRef := klog.KRef(req.Namespace, req.Name)
 
 	startTime := time.Now()
-	klog.V(2).InfoS("Reconciliation starts", "clusterSchedulingPolicySnapshot", snapshotKRef)
+	klog.V(2).InfoS("Reconciliation starts", "policySnapshot", snapshotKRef)
 	defer func() {
 		latency := time.Since(startTime).Milliseconds()
-		klog.V(2).InfoS("Reconciliation ends", "clusterSchedulingPolicySnapshot", snapshotKRef, "latency", latency)
+		klog.V(2).InfoS("Reconciliation ends", "policySnapshot", snapshotKRef, "latency", latency)
 	}()
 
-	if err := r.Client.Get(ctx, name, &snapshot); err != nil {
-		if errors.IsNotFound(err) {
-			klog.V(4).InfoS("Ignoring NotFound clusterSchedulingPolicySnapshot", "clusterSchedulingPolicySnapshot", snapshotKRef)
-			return ctrl.Result{}, nil
+	if req.Namespace == "" {
+		// ClusterSchedulingPolicySnapshot (cluster-scoped)
+		var clusterSchedulingPolicySnapshot fleetv1beta1.ClusterSchedulingPolicySnapshot
+		if err = r.Client.Get(ctx, req.NamespacedName, &clusterSchedulingPolicySnapshot); err != nil {
+			klog.ErrorS(err, "Failed to get cluster scheduling policy snapshot", "policySnapshot", snapshotKRef)
+			return ctrl.Result{}, controller.NewAPIServerError(true, client.IgnoreNotFound(err))
 		}
-		klog.ErrorS(err, "Failed to get clusterSchedulingPolicySnapshot", "clusterSchedulingPolicySnapshot", snapshotKRef)
-		return ctrl.Result{}, controller.NewAPIServerError(true, err)
+		snapshot = &clusterSchedulingPolicySnapshot
+	} else {
+		// schedulingPolicySnapshot (namespaced)
+		var schedulingPolicySnapshot fleetv1beta1.SchedulingPolicySnapshot
+		if err = r.Client.Get(ctx, req.NamespacedName, &schedulingPolicySnapshot); err != nil {
+			klog.ErrorS(err, "Failed to get scheduling policy snapshot", "policySnapshot", snapshotKRef)
+			return ctrl.Result{}, controller.NewAPIServerError(true, client.IgnoreNotFound(err))
+		}
+		snapshot = &schedulingPolicySnapshot
 	}
-	crp := snapshot.Labels[fleetv1beta1.PlacementTrackingLabel]
-	if len(crp) == 0 {
-		err := fmt.Errorf("invalid label value %s", fleetv1beta1.PlacementTrackingLabel)
-		klog.ErrorS(err, "Invalid clusterSchedulingPolicySnapshot", "clusterSchedulingPolicySnapshot", snapshotKRef)
+
+	placementName, exist := snapshot.GetLabels()[fleetv1beta1.PlacementTrackingLabel]
+	if !exist || len(placementName) == 0 {
+		err := fmt.Errorf("label %s is missing or has empty value", fleetv1beta1.PlacementTrackingLabel)
+		klog.ErrorS(err, "Failed to enqueue placement due to invalid schedulingPolicySnapshot", "policySnapshot", snapshotKRef)
 		return ctrl.Result{}, controller.NewUnexpectedBehaviorError(err)
 	}
 
-	r.PlacementController.Enqueue(crp)
+	r.PlacementController.Enqueue(controller.GetObjectKeyFromNamespaceName(req.Namespace, placementName))
 	return ctrl.Result{}, nil
 }
 
-// SetupWithManager sets up the controller with the Manager.
-func (r *Reconciler) SetupWithManager(mgr ctrl.Manager) error {
+// SetupWithManagerForClusterSchedulingPolicySnapshot sets up the controller with the Manager for ClusterSchedulingPolicySnapshot.
+func (r *Reconciler) SetupWithManagerForClusterSchedulingPolicySnapshot(mgr ctrl.Manager) error {
 	return ctrl.NewControllerManagedBy(mgr).Named("clusterschedulingpolicysnapshot-watcher").
 		For(&fleetv1beta1.ClusterSchedulingPolicySnapshot{}).
 		WithEventFilter(predicate.Funcs{
@@ -87,3 +96,18 @@ func (r *Reconciler) SetupWithManager(mgr ctrl.Manager) error {
 			},
 		}).Complete(r)
 }
+
+// SetupWithManagerForSchedulingPolicySnapshot sets up the controller with the Manager for SchedulingPolicySnapshot.
+func (r *Reconciler) SetupWithManagerForSchedulingPolicySnapshot(mgr ctrl.Manager) error {
+	return ctrl.NewControllerManagedBy(mgr).Named("schedulingpolicysnapshot-watcher").
+		For(&fleetv1beta1.SchedulingPolicySnapshot{}).
+		WithEventFilter(predicate.Funcs{
+			// skipping delete and create events so that RP controller does not need to update the status.
+			DeleteFunc: func(e event.DeleteEvent) bool {
+				return false
+			},
+			CreateFunc: func(e event.CreateEvent) bool {
+				return false
+			},
+		}).Complete(r)
+}

pkg/controllers/clusterschedulingpolicysnapshot/controller_integration_test.go

@@ -27,14 +27,21 @@ import (
 	"k8s.io/apimachinery/pkg/types"
 
 	fleetv1beta1 "github.com/kubefleet-dev/kubefleet/apis/placement/v1beta1"
+	"github.com/kubefleet-dev/kubefleet/pkg/utils/controller"
 )
 
 const (
 	testCRPName      = "my-crp"
+	testRPName       = "my-rp"
 	testSnapshotName = "my-snapshot"
+	testNamespace    = "test-namespace"
+
+	eventuallyTimeout    = time.Second * 10
+	consistentlyDuration = time.Second * 10
+	interval             = time.Millisecond * 250
 )
 
-func policySnapshot() *fleetv1beta1.ClusterSchedulingPolicySnapshot {
+func clusterSchedulingPolicySnapshotForTest() *fleetv1beta1.ClusterSchedulingPolicySnapshot {
 	return &fleetv1beta1.ClusterSchedulingPolicySnapshot{
 		ObjectMeta: metav1.ObjectMeta{
 			Name: testSnapshotName,
@@ -50,28 +57,39 @@ func policySnapshot() *fleetv1beta1.ClusterSchedulingPolicySnapshot {
 	}
 }
 
-var _ = Describe("Test clusterSchedulingPolicySnapshot Controller", func() {
-	const (
-		eventuallyTimeout    = time.Second * 10
-		consistentlyDuration = time.Second * 10
-		interval             = time.Millisecond * 250
-	)
+func schedulingPolicySnapshotForTest() *fleetv1beta1.SchedulingPolicySnapshot {
+	return &fleetv1beta1.SchedulingPolicySnapshot{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      testSnapshotName,
+			Namespace: testNamespace,
+			Labels: map[string]string{
+				fleetv1beta1.PolicyIndexLabel:       "1",
+				fleetv1beta1.IsLatestSnapshotLabel:  "true",
+				fleetv1beta1.PlacementTrackingLabel: testRPName,
+			},
+		},
+		Spec: fleetv1beta1.SchedulingPolicySnapshotSpec{
+			PolicyHash: []byte("hash"),
+		},
+	}
+}
 
+var _ = Describe("Test clusterSchedulingPolicySnapshot Controller", func() {
 	var (
 		createdSnapshot = &fleetv1beta1.ClusterSchedulingPolicySnapshot{}
 	)
 
 	BeforeEach(func() {
 		fakePlacementController.ResetQueue()
 		By("By creating a new clusterSchedulingPolicySnapshot")
-		snapshot := policySnapshot()
+		snapshot := clusterSchedulingPolicySnapshotForTest()
 		Expect(k8sClient.Create(ctx, snapshot)).Should(Succeed())
 	})
 
 	Context("When creating new clusterSchedulingPolicySnapshot", func() {
 		AfterEach(func() {
 			By("By deleting snapshot")
-			createdSnapshot = policySnapshot()
+			createdSnapshot = clusterSchedulingPolicySnapshotForTest()
 			Expect(k8sClient.Delete(ctx, createdSnapshot)).Should(Succeed())
 
 			By("By checking snapshot")
@@ -97,7 +115,7 @@ var _ = Describe("Test clusterSchedulingPolicySnapshot Controller", func() {
 
 		AfterEach(func() {
 			By("By deleting snapshot")
-			createdSnapshot = policySnapshot()
+			createdSnapshot = clusterSchedulingPolicySnapshotForTest()
 			Expect(k8sClient.Delete(ctx, createdSnapshot)).Should(Succeed())
 
 			By("By checking snapshot")
@@ -147,7 +165,7 @@ var _ = Describe("Test clusterSchedulingPolicySnapshot Controller", func() {
 
 		It("Should ignore the event", func() {
 			By("By deleting snapshot")
-			createdSnapshot = policySnapshot()
+			createdSnapshot = clusterSchedulingPolicySnapshotForTest()
 			Expect(k8sClient.Delete(ctx, createdSnapshot)).Should(Succeed())
 
 			By("By checking snapshot")
@@ -162,3 +180,111 @@ var _ = Describe("Test clusterSchedulingPolicySnapshot Controller", func() {
 		})
 	})
 })
+
+var _ = Describe("Test schedulingPolicySnapshot Controller", func() {
+	var (
+		createdSnapshot = &fleetv1beta1.SchedulingPolicySnapshot{}
+		key             = controller.GetObjectKeyFromNamespaceName(testNamespace, testRPName)
+	)
+
+	BeforeEach(func() {
+		fakePlacementController.ResetQueue()
+		By("By creating a new schedulingPolicySnapshot")
+		snapshot := schedulingPolicySnapshotForTest()
+		Expect(k8sClient.Create(ctx, snapshot)).Should(Succeed())
+	})
+
+	Context("When creating new schedulingPolicySnapshot", func() {
+		AfterEach(func() {
+			By("By deleting snapshot")
+			createdSnapshot = schedulingPolicySnapshotForTest()
+			Expect(k8sClient.Delete(ctx, createdSnapshot)).Should(Succeed())
+
+			By("By checking snapshot")
+			Eventually(func() bool {
+				return errors.IsNotFound(k8sClient.Get(ctx, types.NamespacedName{Name: testSnapshotName, Namespace: testNamespace}, createdSnapshot))
+			}, eventuallyTimeout, interval).Should(BeTrue(), "snapshot should be deleted")
+		})
+
+		It("Should ignore the event", func() {
+			By("By checking placement controller queue")
+			Consistently(func() bool {
+				return fakePlacementController.Key() == ""
+			}, consistentlyDuration, interval).Should(BeTrue(), "controller should ignore the create event and not enqueue the request into the placementController queue")
+
+		})
+	})
+
+	Context("When updating schedulingPolicySnapshot", func() {
+		BeforeEach(func() {
+			By("By resetting the placement queue")
+			fakePlacementController.ResetQueue()
+		})
+
+		AfterEach(func() {
+			By("By deleting snapshot")
+			createdSnapshot = schedulingPolicySnapshotForTest()
+			Expect(k8sClient.Delete(ctx, createdSnapshot)).Should(Succeed())
+
+			By("By checking snapshot")
+			Eventually(func() bool {
+				return errors.IsNotFound(k8sClient.Get(ctx, types.NamespacedName{Name: testSnapshotName, Namespace: testNamespace}, createdSnapshot))
+			}, eventuallyTimeout, interval).Should(BeTrue(), "snapshot should be deleted")
+		})
+
+		It("Updating the spec and should enqueue the event", func() {
+			Expect(k8sClient.Get(ctx, types.NamespacedName{Name: testSnapshotName, Namespace: testNamespace}, createdSnapshot)).Should(Succeed())
+
+			By("By updating the schedulingPolicySnapshot spec")
+			createdSnapshot.Spec.PolicyHash = []byte("modified-hash")
+			Expect(k8sClient.Update(ctx, createdSnapshot)).Should(Succeed())
+
+			By("By checking placement controller queue")
+			Eventually(func() bool {
+				return fakePlacementController.Key() == key
+			}, eventuallyTimeout, interval).Should(BeTrue(), "placementController should receive the RP key")
+		})
+
+		It("Updating the status and should enqueue the event", func() {
+			Expect(k8sClient.Get(ctx, types.NamespacedName{Name: testSnapshotName, Namespace: testNamespace}, createdSnapshot)).Should(Succeed())
+
+			By("By updating the schedulingPolicySnapshot status")
+			newCondition := metav1.Condition{
+				Type:               string(fleetv1beta1.PolicySnapshotScheduled),
+				Status:             metav1.ConditionTrue,
+				Reason:             "scheduled",
+				ObservedGeneration: createdSnapshot.GetGeneration(),
+			}
+			meta.SetStatusCondition(&createdSnapshot.Status.Conditions, newCondition)
+			Expect(k8sClient.Status().Update(ctx, createdSnapshot)).Should(Succeed())
+
+			By("By checking placement controller queue")
+			Eventually(func() bool {
+				return fakePlacementController.Key() == key
+			}, eventuallyTimeout, interval).Should(BeTrue(), "placementController should receive the RP key")
+		})
+	})
+
+	Context("When deleting schedulingPolicySnapshot", func() {
+		BeforeEach(func() {
+			By("By resetting the placement queue")
+			fakePlacementController.ResetQueue()
+		})
+
+		It("Should ignore the event", func() {
+			By("By deleting snapshot")
+			createdSnapshot = schedulingPolicySnapshotForTest()
+			Expect(k8sClient.Delete(ctx, createdSnapshot)).Should(Succeed())
+
+			By("By checking snapshot")
+			Eventually(func() bool {
+				return errors.IsNotFound(k8sClient.Get(ctx, types.NamespacedName{Name: testSnapshotName, Namespace: testNamespace}, createdSnapshot))
+			}, eventuallyTimeout, interval).Should(BeTrue(), "snapshot should be deleted")
+
+			By("By checking placement controller queue")
+			Consistently(func() bool {
+				return fakePlacementController.Key() == ""
+			}, consistentlyDuration, interval).Should(BeTrue(), "controller should ignore the delete event and not enqueue the request into the placementController queue")
+		})
+	})
+})

pkg/controllers/clusterschedulingpolicysnapshot/suite_test.go

@@ -24,6 +24,8 @@ import (
 
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/client-go/kubernetes/scheme"
 	"k8s.io/client-go/rest"
 	"k8s.io/klog/v2"
@@ -80,6 +82,14 @@ var _ = BeforeSuite(func() {
 	Expect(err).Should(Succeed())
 	Expect(k8sClient).NotTo(BeNil())
 
+	By("creating a test namespace")
+	var ns = corev1.Namespace{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: testNamespace,
+		},
+	}
+	Expect(k8sClient.Create(ctx, &ns)).Should(Succeed(), "failed to create namespace")
+
 	By("starting the controller manager")
 	klog.InitFlags(flag.CommandLine)
 	flag.Parse()
@@ -98,7 +108,13 @@ var _ = BeforeSuite(func() {
 	err = (&Reconciler{
 		Client:              mgr.GetClient(),
 		PlacementController: fakePlacementController,
-	}).SetupWithManager(mgr)
+	}).SetupWithManagerForClusterSchedulingPolicySnapshot(mgr)
+	Expect(err).Should(Succeed())
+
+	err = (&Reconciler{
+		Client:              mgr.GetClient(),
+		PlacementController: fakePlacementController,
+	}).SetupWithManagerForSchedulingPolicySnapshot(mgr)
 	Expect(err).Should(Succeed())
 
 	go func() {