decouple informer cache population and event handling

Signed-off-by: Wei Weng <Wei.Weng@microsoft.com>

Author: Wei Weng

cmd/hubagent/workload/setup.go

@@ -496,7 +496,23 @@ func SetupControllers(ctx context.Context, wg *sync.WaitGroup, mgr ctrl.Manager,
 	}
 	resourceChangeController := controller.NewController(resourceChangeControllerName, controller.ClusterWideKeyFunc, rcr.Reconcile, rateLimiter)
 
+	// Set up the InformerPopulator that runs on ALL pods (leader and followers)
+	// This ensures all pods have synced informer caches for webhook validation
+	klog.Info("Setting up informer populator")
+	informerPopulator := &resourcewatcher.InformerPopulator{
+		DiscoveryClient: discoverClient,
+		RESTMapper:      mgr.GetRESTMapper(),
+		InformerManager: dynamicInformerManager,
+		ResourceConfig:  resourceConfig,
+	}
+
+	if err := mgr.Add(informerPopulator); err != nil {
+		klog.ErrorS(err, "Failed to setup informer populator")
+		return err
+	}
+
 	// Set up a runner that starts all the custom controllers we created above
+	// This runs ONLY on the leader and adds event handlers to the informers created by InformerPopulator
 	resourceChangeDetector := &resourcewatcher.ChangeDetector{
 		DiscoveryClient: discoverClient,
 		RESTMapper:      mgr.GetRESTMapper(),

pkg/resourcewatcher/change_dector.go

@@ -23,7 +23,6 @@ import (
 	"golang.org/x/sync/errgroup"
 	"k8s.io/apimachinery/pkg/api/meta"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
-	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/client-go/discovery"
 	"k8s.io/client-go/tools/cache"
@@ -137,43 +136,28 @@ func (d *ChangeDetector) discoverAPIResourcesLoop(ctx context.Context, period ti
 	}, period)
 }
 
-// discoverResources goes through all the api resources in the cluster and create informers on selected types
+// discoverResources goes through all the api resources in the cluster and adds event handlers to informers
 func (d *ChangeDetector) discoverResources(dynamicResourceEventHandler cache.ResourceEventHandler) {
-	newResources, err := d.getWatchableResources()
+	newResources, err := getWatchableResources(d.DiscoveryClient)
 	var dynamicResources []informer.APIResourceMeta
 	if err != nil {
 		klog.ErrorS(err, "Failed to get all the api resources from the cluster")
 	}
 	for _, res := range newResources {
 		// all the static resources are disabled by default
-		if d.shouldWatchResource(res.GroupVersionResource) {
+		if shouldWatchResource(res.GroupVersionResource, d.RESTMapper, d.ResourceConfig) {
 			dynamicResources = append(dynamicResources, res)
 		}
 	}
-	d.InformerManager.AddDynamicResources(dynamicResources, dynamicResourceEventHandler, err == nil)
-	// this will start the newly added informers if there is any
-	d.InformerManager.Start()
-}
 
-// gvrDisabled returns whether GroupVersionResource is disabled.
-func (d *ChangeDetector) shouldWatchResource(gvr schema.GroupVersionResource) bool {
-	// By default, all of the APIs are allowed.
-	if d.ResourceConfig == nil {
-		return true
+	// On the leader, add event handlers to informers that were already created by InformerPopulator
+	// The informers exist on all pods, but only the leader adds handlers and processes events
+	for _, res := range dynamicResources {
+		d.InformerManager.AddEventHandlerToInformer(res.GroupVersionResource, dynamicResourceEventHandler)
 	}
 
-	gvks, err := d.RESTMapper.KindsFor(gvr)
-	if err != nil {
-		klog.ErrorS(err, "gvr transform failed", "gvr", gvr.String())
-		return false
-	}
-	for _, gvk := range gvks {
-		if d.ResourceConfig.IsResourceDisabled(gvk) {
-			klog.V(4).InfoS("Skip watch resource", "group version kind", gvk.String())
-			return false
-		}
-	}
-	return true
+	// this will start the newly added informers if there is any
+	d.InformerManager.Start()
 }
 
 // dynamicResourceFilter filters out resources that we don't want to watch

pkg/resourcewatcher/informer_populator.go

@@ -0,0 +1,114 @@
+/*
+Copyright 2025 The KubeFleet Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package resourcewatcher
+
+import (
+	"context"
+	"time"
+
+	"k8s.io/apimachinery/pkg/api/meta"
+	"k8s.io/apimachinery/pkg/util/wait"
+	"k8s.io/client-go/discovery"
+	"k8s.io/klog/v2"
+	"sigs.k8s.io/controller-runtime/pkg/manager"
+
+	"github.com/kubefleet-dev/kubefleet/pkg/utils"
+	"github.com/kubefleet-dev/kubefleet/pkg/utils/informer"
+)
+
+const (
+	// informerPopulatorDiscoveryPeriod is how often the InformerPopulator rediscovers API resources
+	informerPopulatorDiscoveryPeriod = 30 * time.Second
+)
+
+// make sure that our InformerPopulator implements controller runtime interfaces
+var (
+	_ manager.Runnable               = &InformerPopulator{}
+	_ manager.LeaderElectionRunnable = &InformerPopulator{}
+)
+
+// InformerPopulator discovers API resources and creates informers for them WITHOUT adding event handlers.
+// This allows follower pods to have synced informer caches for webhook validation while the leader's
+// ChangeDetector adds event handlers and runs controllers.
+type InformerPopulator struct {
+	// DiscoveryClient is used to do resource discovery.
+	DiscoveryClient discovery.DiscoveryInterface
+
+	// RESTMapper is used to convert between GVK and GVR
+	RESTMapper meta.RESTMapper
+
+	// InformerManager manages all the dynamic informers created by the discovery client
+	InformerManager informer.Manager
+
+	// ResourceConfig contains all the API resources that we won't select based on the allowed or skipped propagating APIs option.
+	ResourceConfig *utils.ResourceConfig
+}
+
+// Start runs the informer populator, discovering resources and creating informers.
+// This runs on ALL pods (leader and followers) to ensure all have synced caches.
+func (p *InformerPopulator) Start(ctx context.Context) error {
+	klog.InfoS("Starting the informer populator")
+	defer klog.InfoS("The informer populator is stopped")
+
+	// Run initial discovery to create informers
+	p.discoverAndCreateInformers()
+
+	// Wait for initial cache sync
+	p.InformerManager.WaitForCacheSync()
+	klog.InfoS("Informer populator: initial cache sync complete")
+
+	// Continue discovering resources periodically to handle CRD installations
+	wait.UntilWithContext(ctx, func(ctx context.Context) {
+		p.discoverAndCreateInformers()
+	}, informerPopulatorDiscoveryPeriod)
+
+	return nil
+}
+
+// discoverAndCreateInformers discovers API resources and creates informers WITHOUT adding event handlers
+func (p *InformerPopulator) discoverAndCreateInformers() {
+	newResources, err := getWatchableResources(p.DiscoveryClient)
+	if err != nil {
+		klog.ErrorS(err, "Failed to get all the api resources from the cluster")
+	}
+
+	var resourcesToPopulate []informer.APIResourceMeta
+	for _, res := range newResources {
+		if shouldWatchResource(res.GroupVersionResource, p.RESTMapper, p.ResourceConfig) {
+			resourcesToPopulate = append(resourcesToPopulate, res)
+		}
+	}
+
+	// Create informers directly without adding event handlers.
+	// This avoids adding any event handlers on follower pods
+	for _, res := range resourcesToPopulate {
+		p.InformerManager.CreateInformerForResource(res)
+	}
+
+	// Start any newly created informers
+	p.InformerManager.Start()
+
+	if err == nil {
+		klog.V(2).InfoS("Informer populator: discovered resources", "count", len(resourcesToPopulate))
+	}
+}
+
+// NeedLeaderElection implements LeaderElectionRunnable interface.
+// Returns false so this runs on ALL pods (leader and followers).
+func (p *InformerPopulator) NeedLeaderElection() bool {
+	return false
+}