Repository: rancher/fleet

Issue:
ShardID is not taken into account for Watches statements
This was reported in https://github.com/rancher/fleet/issues/4183 and confirmed. 

The _Bundle_ reconciler is filtering its requests using:
```go
WithEventFilter(sharding.FilterByShardID(r.ShardID))
```
However, a Watches function is also being used, which monitors changes in Clusters and retrieves the Bundles that are deployed in those Clusters.
In that case, the request is being enqueued without filtering by ShardID, which can lead to a controller without an assigned ShardID processing that request.


More controllers might be affected by this.



Discussion:
@0xavi0: /backport v2.13.1

Solution patch:
diff --git a/internal/cmd/controller/reconciler/bundle_controller.go b/internal/cmd/controller/reconciler/bundle_controller.go
index 28f5af24..3daca014 100644
--- a/internal/cmd/controller/reconciler/bundle_controller.go
+++ b/internal/cmd/controller/reconciler/bundle_controller.go
@@ -99,26 +99,7 @@ func (r *BundleReconciler) SetupWithManager(mgr ctrl.Manager) error {
 		Watches(
 			// Fan out from bundledeployment to bundle, this is useful to update the
 			// bundle's status fields.
-			&fleet.BundleDeployment{},
-			handler.EnqueueRequestsFromMapFunc(func(ctx context.Context, a client.Object) []ctrl.Request {
-				bd := a.(*fleet.BundleDeployment)
-				labels := bd.GetLabels()
-				if labels == nil {
-					return nil
-				}
-
-				ns, name := target.BundleFromDeployment(labels)
-				if ns != "" && name != "" {
-					return []ctrl.Request{{
-						NamespacedName: types.NamespacedName{
-							Namespace: ns,
-							Name:      name,
-						},
-					}}
-				}
-
-				return nil
-			}),
+			&fleet.BundleDeployment{}, handler.EnqueueRequestsFromMapFunc(BundleDeploymentMapFunc(r)),
 			builder.WithPredicates(bundleDeploymentStatusChangedPredicate()),
 		).
 		Watches(
@@ -132,6 +113,9 @@ func (r *BundleReconciler) SetupWithManager(mgr ctrl.Manager) error {
 				}
 				requests := []ctrl.Request{}
 				for _, bundle := range bundlesToRefresh {
+					if !sharding.ShouldProcess(bundle, r.ShardID) {
+						continue
+					}
 					requests = append(requests, ctrl.Request{
 						NamespacedName: types.NamespacedName{
 							Namespace: bundle.Namespace,
@@ -1017,3 +1001,33 @@ func upper(op controllerutil.OperationResult) string {
 		return "Unknown"
 	}
 }
+
+// BundleDeploymentMapFunc returns a function that maps a BundleDeployment to a Bundle.
+func BundleDeploymentMapFunc(r *BundleReconciler) func(ctx context.Context, a client.Object) []reconcile.Request {
+	return func(ctx context.Context, a client.Object) []reconcile.Request {
+		bd, ok := a.(*fleet.BundleDeployment)
+		if !ok {
+			return nil
+		}
+
+		if !sharding.ShouldProcess(bd, r.ShardID) {
+			return nil
+		}
+		labels := bd.GetLabels()
+		if labels == nil {
+			return nil
+		}
+
+		ns, name := target.BundleFromDeployment(labels)
+		if ns != "" && name != "" {
+			return []reconcile.Request{{
+				NamespacedName: types.NamespacedName{
+					Namespace: ns,
+					Name:      name,
+				},
+			}}
+		}
+
+		return nil
+	}
+}
diff --git a/internal/cmd/controller/reconciler/bundledeployment_controller.go b/internal/cmd/controller/reconciler/bundledeployment_controller.go
index 1ea0a648..10bacd5f 100644
--- a/internal/cmd/controller/reconciler/bundledeployment_controller.go
+++ b/internal/cmd/controller/reconciler/bundledeployment_controller.go
@@ -138,17 +138,16 @@ func (r *BundleDeploymentReconciler) Reconcile(ctx context.Context, req ctrl.Req
 // status has changed, or the bundledeployment was created
 func bundleDeploymentStatusChangedPredicate() predicate.Funcs {
 	return predicate.Funcs{
-		CreateFunc: func(e event.CreateEvent) bool {
-			return true
-		},
+		CreateFunc: func(e event.CreateEvent) bool { return true },
 		UpdateFunc: func(e event.UpdateEvent) bool {
-			n := e.ObjectNew.(*fleet.BundleDeployment)
-			o := e.ObjectOld.(*fleet.BundleDeployment)
-			if n == nil || o == nil {
+			n, nOK := e.ObjectNew.(*fleet.BundleDeployment)
+			o, oOK := e.ObjectOld.(*fleet.BundleDeployment)
+			if !nOK || !oOK {
 				return false
 			}
 			return !n.DeletionTimestamp.IsZero() || !reflect.DeepEqual(n.Status, o.Status)
 		},
+		GenericFunc: func(e event.GenericEvent) bool { return false },
 	}
 }
 
diff --git a/pkg/sharding/sharding.go b/pkg/sharding/sharding.go
index 65cb3246..6a5198d0 100644
--- a/pkg/sharding/sharding.go
+++ b/pkg/sharding/sharding.go
@@ -16,6 +16,31 @@ const (
 	ShardingDefaultLabel string = "fleet.cattle.io/shard-default"
 )
 
+// ShouldProcess returns true if the given object should be processed by the shard
+// identified by shardID.
+//
+// Behavior:
+//   - If shardID == "", the shard is "unassigned" and should only handle objects
+//     *without* a shard label.
+//   - Otherwise, only objects with ShardingRefLabel == shardID are processed.
+func ShouldProcess(obj client.Object, shardID string) bool {
+	labels := obj.GetLabels()
+	if labels == nil {
+		// No labels at all
+		return shardID == ""
+	}
+
+	label, hasLabel := labels[ShardingRefLabel]
+
+	if shardID == "" {
+		// The "default" (unsharded) controller handles only unlabeled objects
+		return !hasLabel
+	}
+
+	// Otherwise, process only if the shard matches exactly
+	return hasLabel && label == shardID
+}
+
 // TypedFilterByShardID returns a predicate function that filters objects by the shard ID they reference
 func TypedFilterByShardID[T client.Object](shardID string) predicate.TypedFuncs[T] {
 	matchesLabel := func(o metav1.Object) bool {



Test patch:
diff --git a/internal/cmd/controller/reconciler/bundle_controller_test.go b/internal/cmd/controller/reconciler/bundle_controller_test.go
index 7a676661..8281ae58 100644
--- a/internal/cmd/controller/reconciler/bundle_controller_test.go
+++ b/internal/cmd/controller/reconciler/bundle_controller_test.go
@@ -10,12 +10,14 @@ import (
 	"strings"
 	"testing"
 
+	"github.com/google/go-cmp/cmp"
 	"github.com/rancher/fleet/internal/cmd/controller/errorutil"
 	"github.com/rancher/fleet/internal/cmd/controller/finalize"
 	"github.com/rancher/fleet/internal/cmd/controller/reconciler"
 	"github.com/rancher/fleet/internal/cmd/controller/target"
 	"github.com/rancher/fleet/internal/mocks"
 	fleetv1 "github.com/rancher/fleet/pkg/apis/fleet.cattle.io/v1alpha1"
+	"github.com/rancher/fleet/pkg/sharding"
 	"github.com/rancher/wrangler/v3/pkg/genericcondition"
 
 	"go.uber.org/mock/gomock"
@@ -1077,3 +1079,118 @@ func getBundleReadyCondition(b *fleetv1.Bundle) (genericcondition.GenericConditi
 	}
 	return genericcondition.GenericCondition{}, false
 }
+
+func TestBundleDeploymentMapFunc(t *testing.T) {
+	r := &reconciler.BundleReconciler{ShardID: "test-shard"}
+	mapFunc := reconciler.BundleDeploymentMapFunc(r)
+
+	testCases := []struct {
+		name     string
+		obj      client.Object
+		expected []reconcile.Request
+	}{
+		{
+			name: "Matching Shard ID",
+			obj: &fleetv1.BundleDeployment{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "bd-1",
+					Namespace: "cluster-ns",
+					Labels: map[string]string{
+						fleetv1.BundleLabel:          "my-bundle",
+						fleetv1.BundleNamespaceLabel: "fleet-ns",
+						sharding.ShardingRefLabel:    "test-shard",
+					},
+				},
+			},
+			expected: []reconcile.Request{
+				{NamespacedName: types.NamespacedName{Namespace: "fleet-ns", Name: "my-bundle"}},
+			},
+		},
+		{
+			name: "Non-matching Shard ID",
+			obj: &fleetv1.BundleDeployment{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "bd-2",
+					Namespace: "cluster-ns",
+					Labels: map[string]string{
+						fleetv1.BundleLabel:          "my-bundle",
+						fleetv1.BundleNamespaceLabel: "fleet-ns",
+						sharding.ShardingRefLabel:    "other-shard",
+					},
+				},
+			},
+			expected: nil,
+		},
+		{
+			name: "Default Shard, Object has no shard label",
+			obj: &fleetv1.BundleDeployment{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "bd-3",
+					Namespace: "cluster-ns",
+					Labels: map[string]string{
+						fleetv1.BundleLabel:          "my-bundle",
+						fleetv1.BundleNamespaceLabel: "fleet-ns",
+					},
+				},
+			},
+			expected: nil, // default shard is "", not "test-shard"
+		},
+		{
+			name: "Missing bundle labels",
+			obj: &fleetv1.BundleDeployment{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "bd-4",
+					Namespace: "cluster-ns",
+					Labels: map[string]string{
+						sharding.ShardingRefLabel: "test-shard",
+					},
+				},
+			},
+			expected: nil,
+		},
+		{
+			name: "Nil labels",
+			obj: &fleetv1.BundleDeployment{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "bd-5",
+					Namespace: "cluster-ns",
+				},
+			},
+			expected: nil,
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			reqs := mapFunc(context.Background(), tc.obj)
+			if diff := cmp.Diff(tc.expected, reqs); diff != "" {
+				t.Errorf("mismatch (-want +got):\n%s", diff)
+			}
+		})
+	}
+
+	t.Run("Default Shard ID", func(t *testing.T) {
+		r := &reconciler.BundleReconciler{ShardID: ""}
+		mapFunc := reconciler.BundleDeploymentMapFunc(r)
+
+		bd := &fleetv1.BundleDeployment{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      "bd-default",
+				Namespace: "cluster-ns",
+				Labels: map[string]string{
+					fleetv1.BundleLabel:          "my-bundle",
+					fleetv1.BundleNamespaceLabel: "fleet-ns",
+				},
+			},
+		}
+
+		expected := []reconcile.Request{
+			{NamespacedName: types.NamespacedName{Namespace: "fleet-ns", Name: "my-bundle"}},
+		}
+
+		reqs := mapFunc(context.Background(), bd)
+		if diff := cmp.Diff(expected, reqs); diff != "" {
+			t.Errorf("mismatch (-want +got):\n%s", diff)
+		}
+	})
+}
diff --git a/internal/cmd/controller/reconciler/bundledeployment_controller_test.go b/internal/cmd/controller/reconciler/bundledeployment_controller_test.go
new file mode 100644
index 00000000..215ac22e
--- /dev/null
+++ b/internal/cmd/controller/reconciler/bundledeployment_controller_test.go
@@ -0,0 +1,75 @@
+package reconciler
+
+import (
+	"testing"
+
+	fleetv1 "github.com/rancher/fleet/pkg/apis/fleet.cattle.io/v1alpha1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"sigs.k8s.io/controller-runtime/pkg/event"
+)
+
+func TestBundleDeploymentStatusChangedPredicate(t *testing.T) {
+	p := bundleDeploymentStatusChangedPredicate()
+
+	t.Run("Create", func(t *testing.T) {
+		e := event.CreateEvent{Object: &fleetv1.BundleDeployment{}}
+		if !p.Create(e) {
+			t.Error("expected true for create event")
+		}
+	})
+
+	t.Run("Delete", func(t *testing.T) {
+		// Delete is not defined, so it should be false
+		e := event.DeleteEvent{Object: &fleetv1.BundleDeployment{}}
+		if !p.Delete(e) {
+			t.Error("expected true for delete event")
+		}
+	})
+
+	t.Run("Generic", func(t *testing.T) {
+		// Generic is not defined, so it should be false
+		e := event.GenericEvent{Object: &fleetv1.BundleDeployment{}}
+		if p.Generic(e) {
+			t.Error("expected false for generic event")
+		}
+	})
+
+	t.Run("Update", func(t *testing.T) {
+		oldBD := &fleetv1.BundleDeployment{
+			Status: fleetv1.BundleDeploymentStatus{Ready: false},
+		}
+		newBD := oldBD.DeepCopy()
+
+		// No change
+		e := event.UpdateEvent{ObjectOld: oldBD, ObjectNew: newBD}
+		if p.Update(e) {
+			t.Error("should be false when status is identical")
+		}
+
+		// Status changed
+		newBD.Status.Ready = true
+		e = event.UpdateEvent{ObjectOld: oldBD, ObjectNew: newBD}
+		if !p.Update(e) {
+			t.Error("should be true when status changes")
+		}
+
+		// Deletion timestamp added
+		newBD = oldBD.DeepCopy()
+		now := metav1.Now()
+		newBD.DeletionTimestamp = &now
+		e = event.UpdateEvent{ObjectOld: oldBD, ObjectNew: newBD}
+		if !p.Update(e) {
+			t.Error("should be true when deletion timestamp is set")
+		}
+
+		// Nil objects
+		e = event.UpdateEvent{ObjectOld: nil, ObjectNew: newBD}
+		if p.Update(e) {
+			t.Error("should be false for nil old object")
+		}
+		e = event.UpdateEvent{ObjectOld: oldBD, ObjectNew: nil}
+		if p.Update(e) {
+			t.Error("should be false for nil new object")
+		}
+	})
+}