[CHAOS-76][FIX] Replace per-disruption caches with shared caches to reduce Watch connections (#1082)

perf(watchers): share caches across disruptions

  Each active Disruption opened two Watch connections to the API
  server — one per-disruption chaos pod cache and one per-disruption
  target pod/node cache. At scale this caused goroutine bloat,
  memory pressure, and excessive load on the API server.

  Replace per-disruption chaos pod caches with a single
  SharedChaosPodHandler registered on the manager's existing
  informer. Replace per-disruption target caches with a
  NamespaceCachePool so disruptions in the same namespace share
  one cache. Watch connections drop from O(disruptions) to
  O(namespaces).

  Also fixes stale handler accumulation on shared informers
  (source.Kind discards its registration), relabel-out reconcile
  missed by the old enqueue path, and duplicate HA events emitted
  by standby replicas.

  Jira: CHAOS-7

Author: aymericDD

controllers/disruption_controller.go

@@ -35,7 +35,6 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/types"
-	kubeinformers "k8s.io/client-go/informers"
 	"k8s.io/client-go/tools/record"
 	"k8s.io/client-go/util/workqueue"
 	ctrl "sigs.k8s.io/controller-runtime"
@@ -713,7 +712,7 @@ func (r *DisruptionReconciler) createChaosPods(ctx context.Context, instance *ch
 	case chaostypes.DisruptionLevelPod:
 		pod := corev1.Pod{}
 
-		if err = r.Client.Get(ctx, types.NamespacedName{Namespace: instance.Namespace, Name: target}, &pod); err != nil {
+		if err = r.APIReader.Get(ctx, types.NamespacedName{Namespace: instance.Namespace, Name: target}, &pod); err != nil {
 			return fmt.Errorf("error getting target to inject: %w", err)
 		}
 
@@ -1090,7 +1089,7 @@ func (r *DisruptionReconciler) getSelectorMatchingTargets(instance *chaosv1beta1
 	// select either pods or nodes depending on the disruption level
 	switch instance.Spec.Level {
 	case chaostypes.DisruptionLevelPod:
-		pods, totalCount, err := r.TargetSelector.GetMatchingPodsOverTotalPods(r.Client, instance)
+		pods, totalCount, err := r.TargetSelector.GetMatchingPodsOverTotalPods(r.APIReader, instance)
 		if err != nil {
 			return nil, 0, fmt.Errorf("can't get pods matching the given label selector: %w", err)
 		}
@@ -1101,7 +1100,7 @@ func (r *DisruptionReconciler) getSelectorMatchingTargets(instance *chaosv1beta1
 
 		totalAvailableTargetsCount = totalCount
 	case chaostypes.DisruptionLevelNode:
-		nodes, totalCount, err := r.TargetSelector.GetMatchingNodesOverTotalNodes(r.Client, instance)
+		nodes, totalCount, err := r.TargetSelector.GetMatchingNodesOverTotalNodes(r.APIReader, instance)
 		if err != nil {
 			return nil, 0, fmt.Errorf("can't get nodes matching the given label selector: %w", err)
 		}
@@ -1190,15 +1189,15 @@ func (r *DisruptionReconciler) recordEventOnTarget(ctx context.Context, instance
 	case chaostypes.DisruptionLevelPod:
 		p := &corev1.Pod{}
 
-		if err := r.Client.Get(ctx, types.NamespacedName{Namespace: instance.Namespace, Name: target}, p); err != nil {
+		if err := r.APIReader.Get(ctx, types.NamespacedName{Namespace: instance.Namespace, Name: target}, p); err != nil {
 			r.log.Errorw("event failed to be registered on target", tagutil.ErrorKey, err, tagutil.TargetNameKey, target)
 		}
 
 		o = p
 	case chaostypes.DisruptionLevelNode:
 		n := &corev1.Node{}
 
-		if err := r.Client.Get(ctx, types.NamespacedName{Name: target}, n); err != nil {
+		if err := r.APIReader.Get(ctx, types.NamespacedName{Name: target}, n); err != nil {
 			r.log.Errorw("event failed to be registered on target", tagutil.ErrorKey, err, tagutil.TargetNameKey, target)
 		}
 
@@ -1211,7 +1210,7 @@ func (r *DisruptionReconciler) recordEventOnTarget(ctx context.Context, instance
 }
 
 // SetupWithManager setups the current reconciler with the given manager
-func (r *DisruptionReconciler) SetupWithManager(mgr ctrl.Manager, kubeInformerFactory kubeinformers.SharedInformerFactory) (controller.Controller, error) {
+func (r *DisruptionReconciler) SetupWithManager(mgr ctrl.Manager) (controller.Controller, error) {
 	podToDisruption := func(ctx context.Context, d *corev1.Pod) []reconcile.Request {
 		// podtoDisruption is a function that maps pods to disruptions. it is meant to be used as an event handler on a pod informer
 		// this function should safely return an empty list of requests to reconcile if the object we receive is not actually a chaos pod
@@ -1232,12 +1231,38 @@ func (r *DisruptionReconciler) SetupWithManager(mgr ctrl.Manager, kubeInformerFa
 		return []reconcile.Request{{NamespacedName: types.NamespacedName{Name: name, Namespace: namespace}}}
 	}
 
-	return ctrl.NewControllerManagedBy(mgr).
+	cont, err := ctrl.NewControllerManagedBy(mgr).
 		For(&chaosv1beta1.Disruption{}).
 		WithOptions(controller.Options{RateLimiter: workqueue.NewTypedItemExponentialFailureRateLimiter[reconcile.Request](time.Second, time.Hour)}).
 		WatchesRawSource(source.Kind(mgr.GetCache(), &corev1.Pod{}, handler.TypedEnqueueRequestsFromMapFunc(podToDisruption))).
 		WithEventFilter(chaosEventsPredicate()).
 		Build(r)
+	if err != nil {
+		return nil, err
+	}
+
+	// Register a single shared chaos pod handler on the manager's pod informer.
+	// This replaces per-disruption ChaosPodWatcher caches: all chaos pod events are
+	// handled by one shared handler that dispatches by reading disruption labels from
+	// the pod at event time.
+	podInformer, err := mgr.GetCache().GetInformer(context.Background(), &corev1.Pod{})
+	if err != nil {
+		return nil, fmt.Errorf("SetupWithManager: failed to get pod informer: %w", err)
+	}
+
+	sharedHandler := watchers.NewSharedChaosPodHandler(
+		r.APIReader,
+		r.Recorder,
+		r.BaseLog,
+		watchers.NewWatcherMetricsAdapter(r.MetricsSink, r.BaseLog),
+		mgr.Elected(), // gate event/metric emission on leadership in HA deployments
+	)
+
+	if _, err := podInformer.AddEventHandler(sharedHandler); err != nil {
+		return nil, fmt.Errorf("SetupWithManager: failed to register shared chaos pod handler: %w", err)
+	}
+
+	return cont, nil
 }
 
 // chaosEventsPredicate determines if given event is a chaos related one or not

main.go

@@ -49,6 +49,7 @@ import (
 	"k8s.io/client-go/tools/cache"
 
 	ctrl "sigs.k8s.io/controller-runtime"
+	ctrlcache "sigs.k8s.io/controller-runtime/pkg/cache"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	metricsserver "sigs.k8s.io/controller-runtime/pkg/metrics/server"
 	"sigs.k8s.io/controller-runtime/pkg/webhook"
@@ -132,6 +133,19 @@ func main() {
 			Port:    cfg.Controller.Webhook.Port,
 			CertDir: cfg.Controller.Webhook.CertDir,
 		}),
+		Cache: ctrlcache.Options{
+			DefaultTransform: ctrlcache.TransformStripManagedFields(),
+			// Restrict pod watch to the chaos namespace only. The controller only
+			// needs pod events for chaos pods (all in ChaosNamespace). Target pods
+			// in user namespaces are read via APIReader on the reconcile path.
+			ByObject: map[client.Object]ctrlcache.ByObject{
+				&corev1.Pod{}: {
+					Namespaces: map[string]ctrlcache.Config{
+						cfg.Injector.ChaosNamespace: {},
+					},
+				},
+			},
+		},
 	})
 	if err != nil {
 		logger.Fatalw("unable to start manager", tags.ErrorKey, err)
@@ -204,6 +218,7 @@ func main() {
 
 	chaosPodService, err := services.NewChaosPodService(services.ChaosPodServiceConfig{
 		Client:         mgr.GetClient(),
+		Reader:         mgr.GetAPIReader(),
 		ChaosNamespace: cfg.Injector.ChaosNamespace,
 		TargetSelector: targetSelector,
 		Injector: services.ChaosPodServiceInjectorConfig{
@@ -244,9 +259,8 @@ func main() {
 	}
 
 	informerClient := kubernetes.NewForConfigOrDie(ctrl.GetConfigOrDie())
-	kubeInformerFactory := kubeinformers.NewSharedInformerFactoryWithOptions(informerClient, time.Hour*24, kubeinformers.WithNamespace(cfg.Injector.ChaosNamespace))
 
-	cont, err := disruptionReconciler.SetupWithManager(mgr, kubeInformerFactory)
+	cont, err := disruptionReconciler.SetupWithManager(mgr)
 	if err != nil {
 		logger.Fatalw("unable to create controller", tags.ControllerKey, chaosv1beta1.DisruptionKind, tags.ErrorKey, err)
 	}
@@ -257,6 +271,7 @@ func main() {
 		Reader:         mgr.GetAPIReader(),
 		Recorder:       disruptionReconciler.Recorder,
 		ChaosNamespace: cfg.Injector.ChaosNamespace,
+		CachePool:      watchers.NewNamespaceCachePool(ctrl.GetConfigOrDie()),
 	}
 	watcherFactory := watchers.NewWatcherFactory(watchersFactoryConfig)
 	disruptionReconciler.DisruptionsWatchersManager = watchers.NewDisruptionsWatchersManager(cont, watcherFactory, mgr.GetAPIReader())
@@ -286,8 +301,9 @@ func main() {
 
 	defer cancel()
 
-	stopCh := make(chan struct{})
-	kubeInformerFactory.Start(stopCh)
+	// Buffered so the send at mgr.Start failure doesn't block when rollout informers
+	// are disabled and nobody is reading the channel.
+	stopCh := make(chan struct{}, 1)
 
 	go disruptionReconciler.ReportMetrics(ctx)
 

services/chaospod.go

@@ -80,6 +80,7 @@ type ChaosPodServiceInjectorConfig struct {
 // ChaosPodServiceConfig contains configuration options for the chaosPodService.
 type ChaosPodServiceConfig struct {
 	Client           client.Client                 // Kubernetes client for interacting with the API server.
+	Reader           client.Reader                 // Uncached reader for target pods in user namespaces.
 	ChaosNamespace   string                        // Namespace where chaos-related resources are located.
 	TargetSelector   targetselector.TargetSelector // Target selector for selecting target pods.
 	Injector         ChaosPodServiceInjectorConfig // Configuration options for the injector.
@@ -110,6 +111,13 @@ func NewChaosPodService(config ChaosPodServiceConfig) (ChaosPodService, error) {
 		return nil, fmt.Errorf("you must provide a non nil Kubernetes client")
 	}
 
+	// Default Reader to Client for backwards compatibility: callers that do not set
+	// Reader explicitly still get a working implementation (cached reads) rather than
+	// a nil-pointer panic during TargetIsHealthy / HandleOrphanedChaosPods.
+	if config.Reader == nil {
+		config.Reader = config.Client
+	}
+
 	return &chaosPodService{
 		config: config,
 	}, nil
@@ -141,7 +149,7 @@ func (m *chaosPodService) HandleChaosPodTermination(ctx context.Context, disrupt
 	target := chaosPod.Labels[chaostypes.TargetLabel]
 
 	// Check if the target of the disruption is healthy (running and ready).
-	if err := m.config.TargetSelector.TargetIsHealthy(target, m.config.Client, disruption); err != nil {
+	if err := m.config.TargetSelector.TargetIsHealthy(target, m.config.Reader, disruption); err != nil {
 		// return the error unless we have a specific reason to ignore it.
 		if !apierrors.IsNotFound(err) && chaosPodAllowedErrors.isNotAllowed(strings.ToLower(err.Error())) {
 			return false, err
@@ -412,7 +420,7 @@ func (m *chaosPodService) HandleOrphanedChaosPods(ctx context.Context, req ctrl.
 		logger.Infow("checking if we can clean up orphaned chaos pod", tagutil.ChaosPodNameKey, pod.Name, tagutil.TargetNameKey, target)
 
 		// if target doesn't exist, we can try to clean up the chaos pod
-		if err = m.config.Client.Get(ctx, types.NamespacedName{Name: target, Namespace: req.Namespace}, &p); apierrors.IsNotFound(err) {
+		if err = m.config.Reader.Get(ctx, types.NamespacedName{Name: target, Namespace: req.Namespace}, &p); apierrors.IsNotFound(err) {
 			logger.Warnw("orphaned chaos pod detected, will attempt to delete", tagutil.ChaosPodNameKey, pod.Name)
 
 			if err = m.removeFinalizerForChaosPod(ctx, &pod); err != nil {

services/chaospod_test.go

@@ -88,6 +88,9 @@ var _ = Describe("Chaos Pod Service", func() {
 		if chaosPodServiceConfig.Client == nil {
 			chaosPodServiceConfig.Client = k8sClientMock
 		}
+		if chaosPodServiceConfig.Reader == nil {
+			chaosPodServiceConfig.Reader = k8sClientMock
+		}
 
 		// Action
 		chaosPodService, err = services.NewChaosPodService(chaosPodServiceConfig)

targetselector/running_target_selector.go

@@ -33,7 +33,7 @@ func NewRunningTargetSelector(controllerEnableSafeguards bool, controllerNodeNam
 }
 
 // GetMatchingPodsOverTotalPods returns a pods list containing all running pods matching the given label selector and namespace and the count of pods matching the selector
-func (r runningTargetSelector) GetMatchingPodsOverTotalPods(c client.Client, instance *chaosv1beta1.Disruption) (*corev1.PodList, int, error) {
+func (r runningTargetSelector) GetMatchingPodsOverTotalPods(c client.Reader, instance *chaosv1beta1.Disruption) (*corev1.PodList, int, error) {
 	// get parsed selector
 	selector, err := GetLabelSelectorFromInstance(instance)
 	if err != nil {
@@ -115,7 +115,7 @@ podLoop:
 }
 
 // GetMatchingNodesOverTotalNodes returns a nodes list containing all nodes matching the given label selector and the count of nodes matching the selector
-func (r runningTargetSelector) GetMatchingNodesOverTotalNodes(c client.Client, instance *chaosv1beta1.Disruption) (*corev1.NodeList, int, error) {
+func (r runningTargetSelector) GetMatchingNodesOverTotalNodes(c client.Reader, instance *chaosv1beta1.Disruption) (*corev1.NodeList, int, error) {
 	// get parsed selector
 	selector, err := GetLabelSelectorFromInstance(instance)
 	if err != nil {
@@ -174,7 +174,7 @@ nodeLoop:
 }
 
 // TargetIsHealthy returns an error if the given target is unhealthy or does not exist
-func (r runningTargetSelector) TargetIsHealthy(target string, c client.Client, instance *chaosv1beta1.Disruption) error {
+func (r runningTargetSelector) TargetIsHealthy(target string, c client.Reader, instance *chaosv1beta1.Disruption) error {
 	switch instance.Spec.Level {
 	case chaostypes.DisruptionLevelPod:
 		var p corev1.Pod

targetselector/target_selector.go

@@ -18,13 +18,13 @@ import (
 // TargetSelector is an interface for applying network disruptions to a Kubernetes Cluster
 type TargetSelector interface {
 	// GetMatchingPodsOverTotalPods Returns list of matching ready and untargeted pods and number of total pods
-	GetMatchingPodsOverTotalPods(c client.Client, instance *chaosv1beta1.Disruption) (*corev1.PodList, int, error)
+	GetMatchingPodsOverTotalPods(c client.Reader, instance *chaosv1beta1.Disruption) (*corev1.PodList, int, error)
 
 	// GetMatchingNodesOverTotalNodes Returns list of matching ready and untargeted nodes and number of total nodes
-	GetMatchingNodesOverTotalNodes(c client.Client, instance *chaosv1beta1.Disruption) (*corev1.NodeList, int, error)
+	GetMatchingNodesOverTotalNodes(c client.Reader, instance *chaosv1beta1.Disruption) (*corev1.NodeList, int, error)
 
 	// TargetIsHealthy Returns an error if the given target is unhealthy or does not exist
-	TargetIsHealthy(target string, c client.Client, instance *chaosv1beta1.Disruption) error
+	TargetIsHealthy(target string, c client.Reader, instance *chaosv1beta1.Disruption) error
 }
 
 // ValidateLabelSelector assert label selector matches valid grammar, avoids CORE-414