bundle controller shard filtering based on cluster changes

[adibrastegarnia]

Is there an existing issue for this?

• I have searched the existing issues

Current Behavior

Previously we had a discussion about bundle controller was not filtering bundles correctly based on shard ID and this PR tried to address it #4282 (#4275). however, I don't think this PR solves the issue completely.

Let’s assume I labeled my GitRepo and Bundle as shard-0, while the cluster resource (which is independent of these objects) is labeled as shard-1. If a GitRepo is created first and a cluster is added later, the cluster change triggers the bundle controller. However, since the cluster is in shard-1, the controller cannot find any bundles to reconcile because the bundles generated by the GitRepo belong to shard-0.

To resolve this, I believe WithEventFilter(sharding.FilterByShardID(r.ShardID)) should be removed so it no longer applies to all resources. Instead, we should apply shard filtering specifically to the Bundle and BundleDeployment watchers. We should allow all clusters to trigger the watch, then filter the bundles based on the shard ID associated with the cluster that triggered the watch.

cc @0xavi0

Expected Behavior

No response

Steps To Reproduce

No response

Environment

- Architecture:
- Fleet Version:
- Cluster:
  - Provider:
  - Options:
  - Kubernetes Version:

Logs

Anything else?

No response

[johnjcool]

Could this be the right fix for this regression? After applying it, BundleDeployments are created correctly for new clusters in our setup.

1 file changed, 5 insertions(+), 1 deletion(-)

diff --git a/internal/cmd/controller/reconciler/bundle_controller.go b/internal/cmd/controller/reconciler/bundle_controller.go
index 0e842da1..54b109c8 100644
--- a/internal/cmd/controller/reconciler/bundle_controller.go
+++ b/internal/cmd/controller/reconciler/bundle_controller.go
@@ -126,7 +126,7 @@ func (r *BundleReconciler) SetupWithManager(mgr ctrl.Manager) error {
                        }),
                        builder.WithPredicates(clusterChangedPredicate()),
                ).
-               WithEventFilter(sharding.FilterByShardID(r.ShardID)).
+               // WithEventFilter(sharding.FilterByShardID(r.ShardID)).
                WithOptions(controller.Options{MaxConcurrentReconciles: r.Workers}).
                Complete(r)
 }
@@ -146,6 +146,10 @@ func (r *BundleReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctr
                return ctrl.Result{}, client.IgnoreNotFound(err)
        }
 
+       if !sharding.ShouldProcess(bundle, r.ShardID) {
+               return ctrl.Result{}, nil
+       }
+
        if bundle.Labels[fleet.RepoLabel] != "" {
                logger = logger.WithValues(
                        "gitrepo", bundle.Labels[fleet.RepoLabel],

[adibrastegarnia]

This is how we think it should be fixed. I don't think we want to enqueue the object and then filter it instead filtering should be done per object (bundle watch, BD watch, and then clusters)

func (r *BundleReconciler) SetupWithManager(mgr ctrl.Manager) error {
	return ctrl.NewControllerManagedBy(mgr).
		For(&fleet.Bundle{}, builder.WithPredicates(sharding.FilterByShardID(r.ShardID))).
		Watches(
			// Fan out from bundledeployment to bundle
			&fleet.BundleDeployment{}, handler.EnqueueRequestsFromMapFunc(BundleDeploymentMapFunc(r)),
			builder.WithPredicates(
				bundleDeploymentStatusChangedPredicate(),
				sharding.FilterByShardID(r.ShardID),
			),
		).
		Watches(
			// Fan out from cluster to bundle
			// NOTE: No shard filter here - all controllers see all cluster events
			// but only enqueue bundles for their shard
			&fleet.Cluster{},
			handler.EnqueueRequestsFromMapFunc(func(ctx context.Context, a client.Object) []ctrl.Request {
				cluster := a.(*fleet.Cluster)
                        func (r *BundleReconciler) SetupWithManager(mgr ctrl.Manager) error {
			}),
			builder.WithPredicates(predicate.ResourceVersionChangedPredicate{}),
		).

		WithOptions(controller.Options{MaxConcurrentReconciles: r.Workers}).
		Complete(r)

[johnjcool]

yes it works.

// SetupWithManager sets up the controller with the Manager.
func (r *BundleReconciler) SetupWithManager(mgr ctrl.Manager) error {
	return ctrl.NewControllerManagedBy(mgr).
		For(&fleet.Bundle{},
			builder.WithPredicates(
				// do not trigger for bundle status changes (except for cache sync)
				predicate.Or(
					TypedResourceVersionUnchangedPredicate[client.Object]{},
					predicate.GenerationChangedPredicate{},
					predicate.AnnotationChangedPredicate{},
					predicate.LabelChangedPredicate{},
				),
				// additionally, only watch bundles for this shard
				sharding.FilterByShardID(r.ShardID),
			),
		).
		// Note: Maybe improve with WatchesMetadata, does it have access to labels?
		Watches(
			// Fan out from bundledeployment to bundle, this is useful to update the
			// bundle's status fields.
			&fleet.BundleDeployment{}, handler.EnqueueRequestsFromMapFunc(BundleDeploymentMapFunc(r)),
			builder.WithPredicates(
				bundleDeploymentStatusChangedPredicate(),
				sharding.FilterByShardID(r.ShardID),
			),
		).
		Watches(
			// Fan out from cluster to bundle, this is useful for targeting and templating.
			&fleet.Cluster{},
			handler.EnqueueRequestsFromMapFunc(func(ctx context.Context, a client.Object) []ctrl.Request {
				cluster := a.(*fleet.Cluster)
				bundlesToRefresh, _, err := r.Query.BundlesForCluster(ctx, cluster)
				if err != nil {
					return nil
				}
				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,
							Name:      bundle.Name,
						},
					})
				}

				return requests
			}),
			builder.WithPredicates(clusterChangedPredicate()),
		).
		// WithEventFilter(sharding.FilterByShardID(r.ShardID)).
		WithOptions(controller.Options{MaxConcurrentReconciles: r.Workers}).
		Complete(r)
}

[johnjcool]

Could this also be relevant here?

fleet/internal/cmd/controller/reconciler/schedule_controller.go

Lines 49 to 65 in 1708279

	// SetupWithManager sets up the controller with the Manager.
	func (r *ScheduleReconciler) SetupWithManager(mgr ctrl.Manager) error {
	return ctrl.NewControllerManagedBy(mgr).
	For(&fleet.Schedule{},
	builder.WithPredicates(
	predicate.GenerationChangedPredicate{},
	),
	).
	Watches(
	&fleet.Cluster{},
	handler.EnqueueRequestsFromMapFunc(r.mapClustersToSchedules),
	builder.WithPredicates(clusterChangedPredicate()),
	).
	WithEventFilter(sharding.FilterByShardID(r.ShardID)).
	WithOptions(controller.Options{MaxConcurrentReconciles: r.Workers}).
	Complete(r)
	}

[adibrastegarnia]

Could this also be relevant here?

fleet/internal/cmd/controller/reconciler/schedule_controller.go

Lines 49 to 65 in 1708279

// SetupWithManager sets up the controller with the Manager.
func (r *ScheduleReconciler) SetupWithManager(mgr ctrl.Manager) error {
return ctrl.NewControllerManagedBy(mgr).
For(&fleet.Schedule{},
builder.WithPredicates(
predicate.GenerationChangedPredicate{},
),
).
Watches(
&fleet.Cluster{},
handler.EnqueueRequestsFromMapFunc(r.mapClustersToSchedules),
builder.WithPredicates(clusterChangedPredicate()),
).
WithEventFilter(sharding.FilterByShardID(r.ShardID)).
WithOptions(controller.Options{MaxConcurrentReconciles: r.Workers}).
Complete(r)
}

I haven't looked at that part but any controllers which pretty much has the same pattern could have the same issue specially when they are watching objects that are different but somehow dependent but and they can get different shard-id.

[kkaempf]

Looks like @johnjcool wants to open a pull request 🥺