feat: refactor clusterresourceplacementwatcher to support rp (#149)

---------

Signed-off-by: Zhiying Lin <[email protected]>

Author: zhiying-lin

CLAUDE.md

@@ -16,6 +16,10 @@ KubeFleet is a CNCF sandbox project that provides multi-cluster application mana
 # Build all binaries
 make build
 
+# Run specific agent binaries directly
+make run-hubagent
+make run-memberagent
+
 # Run all tests (unit + integration)
 make test
 
@@ -27,19 +31,28 @@ make integration-test
 
 # Run E2E tests
 make e2e-tests
+
+# Run custom E2E tests with labels
+make e2e-tests-custom
 ```
 
 ### Code Quality
 ```bash
 # Run linting (required before commits)
 make lint
 
+# Run full linting (slower but more thorough)
+make lint-full
+
 # Run static analysis
 make staticcheck
 
 # Format code
 make fmt
 
+# Run go vet
+make vet
+
 # Run all quality checks
 make reviewable
 ```
@@ -64,23 +77,42 @@ make setup-clusters MEMBER_CLUSTER_COUNT=5
 # Run parallel E2E tests (default - excludes custom tests)
 make e2e-tests
 
+# Collect logs after E2E tests
+make collect-e2e-logs
+
 # Clean up test clusters
 make clean-e2e-tests
 ```
 
+### Docker and Images
+```bash
+# Build and push all images
+make push
+
+# Build individual images
+make docker-build-hub-agent
+make docker-build-member-agent
+make docker-build-refresh-token
+```
+
 ## Architecture Overview
 
 ### Core API Types
 - **ClusterResourcePlacement (CRP)**: Main API for placing resources across clusters with scheduling policies
 - **MemberCluster**: Represents a member cluster with identity and heartbeat settings  
 - **ClusterResourceBinding**: Represents scheduling decisions binding resources to clusters
 - **Work**: Contains manifests to be applied on member clusters
+- **ClusterResourceSnapshot**: Immutable snapshots of resources to be placed
+- **ClusterSchedulingPolicySnapshot**: Immutable snapshots of scheduling policies
 
 ### Key Controllers
 - **ClusterResourcePlacement Controller** (`pkg/controllers/clusterresourceplacement/`): Manages CRP lifecycle
 - **Scheduler** (`pkg/scheduler/`): Makes placement decisions using pluggable framework
 - **WorkGenerator** (`pkg/controllers/workgenerator/`): Generates Work objects from bindings
 - **WorkApplier** (`pkg/controllers/workapplier/`): Applies Work manifests on member clusters
+- **Resource Placement Watchers**: Monitor and react to changes in placement decisions
+- **ClusterResourceBinding Watcher** (`pkg/controllers/clusterresourcebindingwatcher/`): Watches binding changes
+- **ClusterResourcePlacement Watcher** (`pkg/controllers/clusterresourceplacementwatcher/`): Watches placement changes
 
 ### Scheduler Framework
 The scheduler uses a pluggable architecture similar to Kubernetes scheduler:
@@ -107,23 +139,39 @@ cmd/hubagent/           # Hub agent main and setup
 cmd/memberagent/        # Member agent main and setup
 ```
 
-## Testing Guidelines
+## Development Notes
+
+- Always run `make reviewable` before submitting PRs
+- Follow [Uber Go Style Guide](https://github.com/uber-go/guide/blob/master/style.md) when possible
+- Favor standard library over third-party libraries
+- Controllers should be thoroughly tested with integration tests
+- New scheduler plugins should implement both Filter and Score interfaces
+- Use existing patterns from similar controllers when adding new functionality
+- Property providers should implement the `PropertyProvider` interface
+- PR titles must use prefixes: `feat:`, `fix:`, `docs:`, `test:`, `chore:`, `ci:`, `perf:`, `refactor:`, `revert:`
+
+## Testing Patterns
 
 ### Unit Tests
 - Use `testify` for assertions
 - Controllers use `envtest` for integration testing with real etcd
 - Mock external dependencies with `gomock`
+- Unit test files: `<go_file>_test.go` in same directory
+- Table-driven test style preferred
 
-### E2E Tests  
-- Located in `test/e2e/`
+### Integration Tests  
+- Located in `test/integration/` and `test/scheduler/`
 - Use Ginkgo/Gomega framework
 - Tests run against real Kind clusters
+- Files named: `<go_file>_integration_test.go`
 - Separate test suites for different placement strategies
 
-### Integration Tests
-- Located in `test/integration/`
+### E2E Tests
+- Located in `test/e2e/`
+- Use Ginkgo/Gomega framework  
 - Test cross-controller interactions
 - Use shared test manifests in `test/integration/manifests/`
+- Run with `make e2e-tests` against 3 Kind clusters
 
 ## Key Patterns
 
@@ -148,10 +196,7 @@ All controllers follow standard Kubernetes controller patterns:
 - v1beta1 APIs are current stable version
 - Feature flags control API version enablement
 
-## Development Notes
-
-- Always run `make reviewable` before submitting PRs
-- Controllers should be thoroughly tested with integration tests
-- New scheduler plugins should implement both Filter and Score interfaces
-- Use existing patterns from similar controllers when adding new functionality
-- Property providers should implement the `PropertyProvider` interface
+### Watcher Pattern
+- Resource placement watchers monitor CRP and binding changes
+- Event-driven architecture for responsive placement decisions
+- Separate watchers for different resource types to enable focused reconciliation

cmd/hubagent/workload/setup.go

@@ -197,7 +197,7 @@ func SetupControllers(ctx context.Context, wg *sync.WaitGroup, mgr ctrl.Manager,
 		klog.Info("Setting up clusterResourcePlacement watcher")
 		if err := (&clusterresourceplacementwatcher.Reconciler{
 			PlacementController: clusterResourcePlacementControllerV1Beta1,
-		}).SetupWithManager(mgr); err != nil {
+		}).SetupWithManagerForClusterResourcePlacement(mgr); err != nil {
 			klog.ErrorS(err, "Unable to set up the clusterResourcePlacement watcher")
 			return err
 		}

pkg/controllers/clusterresourceplacement/suite_test.go

@@ -141,7 +141,7 @@ var _ = BeforeSuite(func() {
 
 	err = (&clusterresourceplacementwatcher.Reconciler{
 		PlacementController: crpController,
-	}).SetupWithManager(mgr)
+	}).SetupWithManagerForClusterResourcePlacement(mgr)
 	Expect(err).Should(Succeed(), "failed to create clusterResourcePlacement watcher")
 
 	err = (&clusterresourcebindingwatcher.Reconciler{

pkg/controllers/clusterresourceplacementwatcher/suite_test.go

@@ -23,6 +23,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"
@@ -79,6 +81,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()
@@ -96,7 +106,12 @@ var _ = BeforeSuite(func() {
 
 	err = (&Reconciler{
 		PlacementController: fakePlacementController,
-	}).SetupWithManager(mgr)
+	}).SetupWithManagerForClusterResourcePlacement(mgr)
+	Expect(err).Should(Succeed())
+
+	err = (&Reconciler{
+		PlacementController: fakePlacementController,
+	}).SetupWithManagerForResourcePlacement(mgr)
 	Expect(err).Should(Succeed())
 
 	go func() {

pkg/controllers/clusterresourceplacementwatcher/watcher.go

@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-// Package clusterresourceplacementwatcher features a controller to watch the clusterResourcePlacement changes.
+// Package clusterresourceplacementwatcher features a controller to watch the clusterResourcePlacement and resourcePlacement changes.
 package clusterresourceplacementwatcher
 
 import (
@@ -29,30 +29,38 @@ import (
 	"github.com/kubefleet-dev/kubefleet/pkg/utils/controller"
 )
 
-// Reconciler reconciles a clusterResourcePlacement object.
+// Reconciler reconciles both ClusterResourcePlacement and ResourcePlacement objects.
 type Reconciler struct {
 	// PlacementController maintains a rate limited queue which used to store
-	// the name of the clusterResourcePlacement and a reconcile function to consume the items in queue.
+	// the name of the placement objects and a reconcile function to consume the items in queue.
 	PlacementController controller.Controller
 }
 
-// Reconcile triggers a single CRP reconcile round if CRP has changed.
+// Reconcile triggers a single placement reconcile round if placement has changed.
 func (r *Reconciler) Reconcile(_ context.Context, req ctrl.Request) (ctrl.Result, error) {
 	startTime := time.Now()
-	klog.V(2).InfoS("ClusterResourcePlacementWatcher reconciliation starts", "clusterResourcePlacement", req.Name)
+	klog.V(2).InfoS("PlacementWatcher reconciliation starts", "placement", req.NamespacedName)
 	defer func() {
 		latency := time.Since(startTime).Milliseconds()
-		klog.V(2).InfoS("ClusterResourcePlacementWatcher reconciliation ends", "clusterResourcePlacement", req.Name, "latency", latency)
+		klog.V(2).InfoS("PlacementWatcher reconciliation ends", "placement", req.NamespacedName, "latency", latency)
 	}()
 
-	r.PlacementController.Enqueue(req.Name)
+	r.PlacementController.Enqueue(string(controller.GetObjectKeyFromRequest(req)))
 	return ctrl.Result{}, nil
 }
 
-// SetupWithManager sets up the controller with the Manager.
-func (r *Reconciler) SetupWithManager(mgr ctrl.Manager) error {
+// SetupWithManagerForClusterResourcePlacement sets up the controller with the Manager.
+func (r *Reconciler) SetupWithManagerForClusterResourcePlacement(mgr ctrl.Manager) error {
 	return ctrl.NewControllerManagedBy(mgr).Named("clusterresourceplacement-watcher").
 		For(&fleetv1beta1.ClusterResourcePlacement{}).
 		WithEventFilter(predicate.GenerationChangedPredicate{}).
 		Complete(r)
 }
+
+// SetupWithManagerForResourcePlacement sets up the controller with the Manager.
+func (r *Reconciler) SetupWithManagerForResourcePlacement(mgr ctrl.Manager) error {
+	return ctrl.NewControllerManagedBy(mgr).Named("resourceplacement-watcher").
+		For(&fleetv1beta1.ResourcePlacement{}).
+		WithEventFilter(predicate.GenerationChangedPredicate{}).
+		Complete(r)
+}

pkg/controllers/clusterresourceplacementwatcher/watcher_integration_test.go

@@ -26,10 +26,17 @@ 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"
+	testCRPName   = "my-crp"
+	testRPName    = "my-rp"
+	testNamespace = "test-namespace"
+
+	eventuallyTimeout    = time.Second * 10
+	consistentlyDuration = time.Second * 10
+	interval             = time.Millisecond * 250
 )
 
 func clusterResourcePlacementForTest() *fleetv1beta1.ClusterResourcePlacement {
@@ -53,14 +60,30 @@ func clusterResourcePlacementForTest() *fleetv1beta1.ClusterResourcePlacement {
 	}
 }
 
+func resourcePlacementForTest() *fleetv1beta1.ResourcePlacement {
+	return &fleetv1beta1.ResourcePlacement{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      testRPName,
+			Namespace: testNamespace,
+		},
+		Spec: fleetv1beta1.PlacementSpec{
+			ResourceSelectors: []fleetv1beta1.ClusterResourceSelector{
+				{
+					Group:   corev1.GroupName,
+					Version: "v1",
+					Kind:    "ConfigMap",
+					LabelSelector: &metav1.LabelSelector{
+						MatchLabels: map[string]string{"env": "test"},
+					},
+				},
+			},
+			Policy: &fleetv1beta1.PlacementPolicy{},
+		},
+	}
+}
+
 // This container cannot be run in parallel with other ITs because it uses a shared fakePlacementController.
 var _ = Describe("Test ClusterResourcePlacement Watcher", Serial, func() {
-	const (
-		eventuallyTimeout    = time.Second * 10
-		consistentlyDuration = time.Second * 10
-		interval             = time.Millisecond * 250
-	)
-
 	var (
 		createdCRP = &fleetv1beta1.ClusterResourcePlacement{}
 	)
@@ -151,3 +174,52 @@ var _ = Describe("Test ClusterResourcePlacement Watcher", Serial, func() {
 		})
 	})
 })
+
+var _ = Describe("Test ResourcePlacement Watcher", Serial, func() {
+	var (
+		createdRP = &fleetv1beta1.ResourcePlacement{}
+		key       = controller.GetObjectKeyFromNamespaceName(testNamespace, testRPName)
+	)
+
+	BeforeEach(func() {
+		fakePlacementController.ResetQueue()
+
+		By("By creating a new resourcePlacement")
+		createdRP = resourcePlacementForTest()
+		Expect(k8sClient.Create(ctx, createdRP)).Should(Succeed())
+
+		By("By checking the placement queue before resetting")
+		// The event could arrive after the resetting, which causes the flakiness.
+		// It makes sure the queue is clear before proceed.
+		Eventually(func() bool {
+			return fakePlacementController.Key() == key
+		}, eventuallyTimeout, interval).Should(BeTrue(), "placementController should receive the RP namespaced name when creating RP")
+
+		By("By resetting the placement queue")
+		// Reset the queue to avoid the multiple create events triggered.
+		Consistently(func() error {
+			if fakePlacementController.Key() == key {
+				By("By finding the key and resetting the placement queue")
+				fakePlacementController.ResetQueue()
+			}
+			return nil
+		}, consistentlyDuration, interval).Should(Succeed(), "placementController queue should be stable empty after resetting")
+	})
+
+	Context("When deleting resourcePlacement", func() {
+		It("Should enqueue the event", func() {
+			By("By deleting rp")
+			Expect(k8sClient.Delete(ctx, createdRP)).Should(Succeed())
+
+			By("By checking rp")
+			Eventually(func() bool {
+				return errors.IsNotFound(k8sClient.Get(ctx, types.NamespacedName{Name: testRPName, Namespace: testNamespace}, createdRP))
+			}, eventuallyTimeout, interval).Should(BeTrue(), "rp should be deleted")
+
+			By("By checking placement controller queue")
+			Eventually(func() bool {
+				return fakePlacementController.Key() == key
+			}, eventuallyTimeout, interval).Should(BeTrue(), "placementController should receive the RP namespaced name")
+		})
+	})
+})