fixup: Fix race issues in leader relection

Signed-off-by: rjanakiraman <[email protected]>
kcp-dev · Jun 26, 2024 · 02c2045 · 02c2045
1 parent e907f8c
commit 02c2045
Show file tree

Hide file tree

Showing 16 changed files with 338 additions and 194 deletions.
diff --git a/pkg/permissionclaim/permissionclaim_labeler.go b/pkg/permissionclaim/permissionclaim_labeler.go
@@ -60,10 +60,6 @@ func NewLabeler(
 	apiBindingInformer apisv1alpha1informers.APIBindingClusterInformer,
 	apiExportInformer, globalAPIExportInformer apisv1alpha1informers.APIExportClusterInformer,
 ) *Labeler {
-	indexers.AddIfNotPresentOrDie(apiExportInformer.Informer().GetIndexer(), cache.Indexers{
-		indexers.ByLogicalClusterPathAndName: indexers.IndexByLogicalClusterPathAndName,
-	})
-
 	return &Labeler{
 		listAPIBindingsAcceptingClaimedGroupResource: func(clusterName logicalcluster.Name, groupResource schema.GroupResource) ([]*apisv1alpha1.APIBinding, error) {
 			indexKey := indexers.ClusterAndGroupResourceValue(clusterName, groupResource)
@@ -148,3 +144,13 @@ func (l *Labeler) LabelsFor(ctx context.Context, cluster logicalcluster.Name, gr
 
 	return labels, nil
 }
+
+/*
+installIndexers method is added, as we are re-electing the controllers
+as the informers are always running, we cannot call this in multiple places.
+*/
+func InstallIndexers(apiExportInformer apisv1alpha1informers.APIExportClusterInformer) {
+	indexers.AddIfNotPresentOrDie(apiExportInformer.Informer().GetIndexer(), cache.Indexers{
+		indexers.ByLogicalClusterPathAndName: indexers.IndexByLogicalClusterPathAndName,
+	})
+}
diff --git a/pkg/reconciler/apis/apibinding/apibinding_controller.go b/pkg/reconciler/apis/apibinding/apibinding_controller.go
@@ -164,21 +164,6 @@ func NewController(
 
 	logger := logging.WithReconciler(klog.Background(), ControllerName)
 
-	// APIBinding indexers
-	indexers.AddIfNotPresentOrDie(apiBindingInformer.Informer().GetIndexer(), cache.Indexers{
-		indexers.APIBindingsByAPIExport: indexers.IndexAPIBindingByAPIExport,
-	})
-
-	// APIExport indexers
-	indexers.AddIfNotPresentOrDie(apiExportInformer.Informer().GetIndexer(), cache.Indexers{
-		indexers.ByLogicalClusterPathAndName: indexers.IndexByLogicalClusterPathAndName,
-		indexAPIExportsByAPIResourceSchema:   indexAPIExportsByAPIResourceSchemasFunc,
-	})
-	indexers.AddIfNotPresentOrDie(globalAPIExportInformer.Informer().GetIndexer(), cache.Indexers{
-		indexers.ByLogicalClusterPathAndName: indexers.IndexByLogicalClusterPathAndName,
-		indexAPIExportsByAPIResourceSchema:   indexAPIExportsByAPIResourceSchemasFunc,
-	})
-
 	// APIBinding handlers
 	_, _ = apiBindingInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
 		AddFunc: func(obj interface{}) { c.enqueueAPIBinding(objOrTombstone[*apisv1alpha1.APIBinding](obj), logger, "") },
@@ -484,3 +469,28 @@ func (c *controller) process(ctx context.Context, key string) (bool, error) {
 
 	return requeue, utilerrors.NewAggregate(errs)
 }
+
+/*
+installIndexers method is added, as we are re-electing the controllers
+as the informers are always running, we cannot call this in multiple places.
+*/
+func InstallIndexers(
+	apiBindingInformer apisv1alpha1informers.APIBindingClusterInformer,
+	apiExportInformer apisv1alpha1informers.APIExportClusterInformer,
+	globalAPIExportInformer apisv1alpha1informers.APIExportClusterInformer,
+) {
+	// APIBinding indexers
+	indexers.AddIfNotPresentOrDie(apiBindingInformer.Informer().GetIndexer(), cache.Indexers{
+		indexers.APIBindingsByAPIExport: indexers.IndexAPIBindingByAPIExport,
+	})
+
+	// APIExport indexers
+	indexers.AddIfNotPresentOrDie(apiExportInformer.Informer().GetIndexer(), cache.Indexers{
+		indexers.ByLogicalClusterPathAndName: indexers.IndexByLogicalClusterPathAndName,
+		indexAPIExportsByAPIResourceSchema:   indexAPIExportsByAPIResourceSchemasFunc,
+	})
+	indexers.AddIfNotPresentOrDie(globalAPIExportInformer.Informer().GetIndexer(), cache.Indexers{
+		indexers.ByLogicalClusterPathAndName: indexers.IndexByLogicalClusterPathAndName,
+		indexAPIExportsByAPIResourceSchema:   indexAPIExportsByAPIResourceSchemasFunc,
+	})
+}
diff --git a/pkg/reconciler/apis/apiexport/apiexport_controller.go b/pkg/reconciler/apis/apiexport/apiexport_controller.go
@@ -121,14 +121,6 @@ func NewController(
 		commit: committer.NewCommitter[*APIExport, Patcher, *APIExportSpec, *APIExportStatus](kcpClusterClient.ApisV1alpha1().APIExports()),
 	}
 
-	indexers.AddIfNotPresentOrDie(
-		apiExportInformer.Informer().GetIndexer(),
-		cache.Indexers{
-			indexers.APIExportByIdentity: indexers.IndexAPIExportByIdentity,
-			indexers.APIExportBySecret:   indexers.IndexAPIExportBySecret,
-		},
-	)
-
 	_, _ = apiExportInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
 		AddFunc: func(obj interface{}) {
 			c.enqueueAPIExport(obj.(*apisv1alpha1.APIExport))
@@ -338,3 +330,17 @@ func (c *controller) process(ctx context.Context, key string) error {
 
 	return utilerrors.NewAggregate(errs)
 }
+
+/*
+installIndexers method is added, as we are re-electing the controllers
+as the informers are always running, we cannot call this in multiple places.
+*/
+func InstallIndexers(apiExportInformer apisv1alpha1informers.APIExportClusterInformer) {
+	indexers.AddIfNotPresentOrDie(
+		apiExportInformer.Informer().GetIndexer(),
+		cache.Indexers{
+			indexers.APIExportByIdentity: indexers.IndexAPIExportByIdentity,
+			indexers.APIExportBySecret:   indexers.IndexAPIExportBySecret,
+		},
+	)
+}
diff --git a/pkg/reconciler/apis/apiexportendpointslice/apiexportendpointslice_controller.go b/pkg/reconciler/apis/apiexportendpointslice/apiexportendpointslice_controller.go
@@ -100,14 +100,6 @@ func NewController(
 		commit:                                committer.NewCommitter[*APIExportEndpointSlice, Patcher, *APIExportEndpointSliceSpec, *APIExportEndpointSliceStatus](kcpClusterClient.ApisV1alpha1().APIExportEndpointSlices()),
 	}
 
-	indexers.AddIfNotPresentOrDie(globalAPIExportClusterInformer.Informer().GetIndexer(), cache.Indexers{
-		indexers.ByLogicalClusterPathAndName: indexers.IndexByLogicalClusterPathAndName,
-	})
-
-	indexers.AddIfNotPresentOrDie(apiExportEndpointSliceClusterInformer.Informer().GetIndexer(), cache.Indexers{
-		indexAPIExportEndpointSliceByAPIExport: indexAPIExportEndpointSliceByAPIExportFunc,
-	})
-
 	_, _ = apiExportEndpointSliceClusterInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
 		AddFunc: func(obj interface{}) {
 			c.enqueueAPIExportEndpointSlice(obj)
@@ -158,10 +150,6 @@ func NewController(
 		},
 	)
 
-	indexers.AddIfNotPresentOrDie(apiExportEndpointSliceClusterInformer.Informer().GetIndexer(), cache.Indexers{
-		indexAPIExportEndpointSlicesByPartition: indexAPIExportEndpointSlicesByPartitionFunc,
-	})
-
 	return c, nil
 }
 
@@ -389,3 +377,21 @@ func filterShardEvent(oldObj, newObj interface{}) bool {
 	}
 	return false
 }
+
+/*
+installIndexers method is added, as we are re-electing the controllers
+as the informers are always running, we cannot call this in multiple places.
+*/
+func InstallIndexers(globalAPIExportClusterInformer apisinformers.APIExportClusterInformer, apiExportEndpointSliceClusterInformer apisinformers.APIExportEndpointSliceClusterInformer) {
+	indexers.AddIfNotPresentOrDie(globalAPIExportClusterInformer.Informer().GetIndexer(), cache.Indexers{
+		indexers.ByLogicalClusterPathAndName: indexers.IndexByLogicalClusterPathAndName,
+	})
+
+	indexers.AddIfNotPresentOrDie(apiExportEndpointSliceClusterInformer.Informer().GetIndexer(), cache.Indexers{
+		indexAPIExportEndpointSliceByAPIExport: indexAPIExportEndpointSliceByAPIExportFunc,
+	})
+
+	indexers.AddIfNotPresentOrDie(apiExportEndpointSliceClusterInformer.Informer().GetIndexer(), cache.Indexers{
+		indexAPIExportEndpointSlicesByPartition: indexAPIExportEndpointSlicesByPartitionFunc,
+	})
+}
diff --git a/pkg/reconciler/apis/crdcleanup/crdcleanup_controller.go b/pkg/reconciler/apis/crdcleanup/crdcleanup_controller.go
@@ -71,13 +71,6 @@ func NewController(
 		},
 	}
 
-	indexers.AddIfNotPresentOrDie(
-		apiBindingInformer.Informer().GetIndexer(),
-		cache.Indexers{
-			indexers.APIBindingByBoundResourceUID: indexers.IndexAPIBindingByBoundResourceUID,
-		},
-	)
-
 	_, _ = crdInformer.Informer().AddEventHandler(cache.FilteringResourceEventHandler{
 		FilterFunc: func(obj interface{}) bool {
 			crd := obj.(*apiextensionsv1.CustomResourceDefinition)
@@ -252,3 +245,16 @@ func (c *controller) process(ctx context.Context, key string) error {
 
 	return nil
 }
+
+/*
+installIndexers method is added, as we are re-electing the controllers
+as the informers are always running, we cannot call this in multiple places.
+*/
+func InstallIndexers(apiBindingInformer apisv1alpha1informers.APIBindingClusterInformer) {
+	indexers.AddIfNotPresentOrDie(
+		apiBindingInformer.Informer().GetIndexer(),
+		cache.Indexers{
+			indexers.APIBindingByBoundResourceUID: indexers.IndexAPIBindingByBoundResourceUID,
+		},
+	)
+}
diff --git a/pkg/reconciler/apis/extraannotationsync/apibindingannotation_controller.go b/pkg/reconciler/apis/extraannotationsync/apibindingannotation_controller.go
@@ -106,14 +106,6 @@ func NewController(
 
 	logger := logging.WithReconciler(klog.Background(), ControllerName)
 
-	indexers.AddIfNotPresentOrDie(apiExportInformer.Informer().GetIndexer(), cache.Indexers{
-		indexers.ByLogicalClusterPathAndName: indexers.IndexByLogicalClusterPathAndName,
-	})
-
-	indexers.AddIfNotPresentOrDie(apiBindingInformer.Informer().GetIndexer(), cache.Indexers{
-		indexers.APIBindingsByAPIExport: indexers.IndexAPIBindingByAPIExport,
-	})
-
 	_, _ = apiExportInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
 		AddFunc:    func(obj interface{}) { c.enqueueAPIExport(obj, logger) },
 		UpdateFunc: func(_, obj interface{}) { c.enqueueAPIExport(obj, logger) },
@@ -303,3 +295,17 @@ func syncExtraAnnotationPatch(a1, a2 map[string]string) ([]byte, error) {
 
 	return json.Marshal(patch)
 }
+
+/*
+installIndexers method is added, as we are re-electing the controllers
+as the informers are always running, we cannot call this in multiple places.
+*/
+func InstallIndexers(apiExportInformer apisinformers.APIExportClusterInformer, apiBindingInformer apisinformers.APIBindingClusterInformer) {
+	indexers.AddIfNotPresentOrDie(apiExportInformer.Informer().GetIndexer(), cache.Indexers{
+		indexers.ByLogicalClusterPathAndName: indexers.IndexByLogicalClusterPathAndName,
+	})
+
+	indexers.AddIfNotPresentOrDie(apiBindingInformer.Informer().GetIndexer(), cache.Indexers{
+		indexers.APIBindingsByAPIExport: indexers.IndexAPIBindingByAPIExport,
+	})
+}
diff --git a/pkg/reconciler/apis/permissionclaimlabel/permissionclaimlabel_controller.go b/pkg/reconciler/apis/permissionclaimlabel/permissionclaimlabel_controller.go
@@ -78,10 +78,6 @@ func NewController(
 		commit: committer.NewCommitter[*APIBinding, Patcher, *APIBindingSpec, *APIBindingStatus](kcpClusterClient.ApisV1alpha1().APIBindings()),
 	}
 
-	indexers.AddIfNotPresentOrDie(apiExportInformer.Informer().GetIndexer(), cache.Indexers{
-		indexers.ByLogicalClusterPathAndName: indexers.IndexByLogicalClusterPathAndName,
-	})
-
 	_, _ = apiBindingInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
 		AddFunc: func(obj interface{}) { c.enqueueAPIBinding(obj, logger) },
 		UpdateFunc: func(_, newObj interface{}) {
@@ -215,3 +211,21 @@ func (c *controller) process(ctx context.Context, key string) error {
 
 	return utilerrors.NewAggregate(errs)
 }
+
+/*
+installIndexers method is added, as we are re-electing the controllers
+as the informers are always running, we cannot call this in multiple places.
+*/
+func InstallIndexers(apiExportInformer apisv1alpha1informers.APIExportClusterInformer, apiBindingInformer apisv1alpha1informers.APIBindingClusterInformer) {
+	indexers.AddIfNotPresentOrDie(apiExportInformer.Informer().GetIndexer(), cache.Indexers{
+		indexers.ByLogicalClusterPathAndName: indexers.IndexByLogicalClusterPathAndName,
+	})
+
+	if err := apiBindingInformer.Informer().GetIndexer().AddIndexers(
+		cache.Indexers{
+			indexers.APIBindingByClusterAndAcceptedClaimedGroupResources: indexers.IndexAPIBindingByClusterAndAcceptedClaimedGroupResources,
+		},
+	); err != nil {
+		panic(err)
+	}
+}
diff --git a/pkg/reconciler/apis/permissionclaimlabel/permissionclaimlabel_resource_controller.go b/pkg/reconciler/apis/permissionclaimlabel/permissionclaimlabel_resource_controller.go
@@ -31,11 +31,9 @@ import (
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
 	"k8s.io/apimachinery/pkg/util/wait"
-	"k8s.io/client-go/tools/cache"
 	"k8s.io/client-go/util/workqueue"
 	"k8s.io/klog/v2"
 
-	"github.com/kcp-dev/kcp/pkg/indexers"
 	"github.com/kcp-dev/kcp/pkg/informer"
 	"github.com/kcp-dev/kcp/pkg/logging"
 	"github.com/kcp-dev/kcp/pkg/permissionclaim"
@@ -55,14 +53,6 @@ func NewResourceController(
 	apiBindingInformer apisv1alpha1informers.APIBindingClusterInformer,
 	apiExportInformer, globalAPIExportInformer apisv1alpha1informers.APIExportClusterInformer,
 ) (*resourceController, error) {
-	if err := apiBindingInformer.Informer().GetIndexer().AddIndexers(
-		cache.Indexers{
-			indexers.APIBindingByClusterAndAcceptedClaimedGroupResources: indexers.IndexAPIBindingByClusterAndAcceptedClaimedGroupResources,
-		},
-	); err != nil {
-		return nil, err
-	}
-
 	c := &resourceController{
 		queue:                  workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), ResourceControllerName),
 		kcpClusterClient:       kcpClusterClient,

diff --git a/pkg/reconciler/cache/labelclusterrolebindings/labelclusterrolebinding_controller.go b/pkg/reconciler/cache/labelclusterrolebindings/labelclusterrolebinding_controller.go
@@ -83,10 +83,6 @@ func NewController(
 		commit: committer.NewStatuslessCommitter[*rbacv1.ClusterRoleBinding, rbacclientv1.ClusterRoleBindingInterface](kubeClusterClient.RbacV1().ClusterRoleBindings(), committer.ShallowCopy[rbacv1.ClusterRoleBinding]),
 	}
 
-	indexers.AddIfNotPresentOrDie(clusterRoleBindingInformer.Informer().GetIndexer(), cache.Indexers{
-		labelclusterroles.ClusterRoleBindingByClusterRoleName: labelclusterroles.IndexClusterRoleBindingByClusterRoleName,
-	})
-
 	_, _ = clusterRoleBindingInformer.Informer().AddEventHandler(cache.FilteringResourceEventHandler{
 		FilterFunc: replication.IsNoSystemClusterName,
 		Handler: cache.ResourceEventHandlerFuncs{
@@ -278,3 +274,13 @@ func (c *controller) process(ctx context.Context, key string) (bool, error) {
 
 	return requeue, utilerrors.NewAggregate(errs)
 }
+
+/*
+installIndexers method is added, as we are re-electing the controllers
+as the informers are always running, we cannot call this in multiple places.
+*/
+func InstallIndexers(clusterRoleBindingInformer kcprbacinformers.ClusterRoleBindingClusterInformer) {
+	indexers.AddIfNotPresentOrDie(clusterRoleBindingInformer.Informer().GetIndexer(), cache.Indexers{
+		labelclusterroles.ClusterRoleBindingByClusterRoleName: labelclusterroles.IndexClusterRoleBindingByClusterRoleName,
+	})
+}
diff --git a/pkg/reconciler/cache/labelclusterroles/labelclusterrole_controller.go b/pkg/reconciler/cache/labelclusterroles/labelclusterrole_controller.go
@@ -82,10 +82,6 @@ func NewController(
 		commit: committer.NewStatuslessCommitter[*rbacv1.ClusterRole, rbacclientv1.ClusterRoleInterface](kubeClusterClient.RbacV1().ClusterRoles(), committer.ShallowCopy[rbacv1.ClusterRole]),
 	}
 
-	indexers.AddIfNotPresentOrDie(clusterRoleBindingInformer.Informer().GetIndexer(), cache.Indexers{
-		ClusterRoleBindingByClusterRoleName: IndexClusterRoleBindingByClusterRoleName,
-	})
-
 	_, _ = clusterRoleInformer.Informer().AddEventHandler(cache.FilteringResourceEventHandler{
 		FilterFunc: replication.IsNoSystemClusterName,
 		Handler: cache.ResourceEventHandlerFuncs{
@@ -277,3 +273,13 @@ func (c *controller) process(ctx context.Context, key string) (bool, error) {
 
 	return requeue, utilerrors.NewAggregate(errs)
 }
+
+/*
+installIndexers method is added, as we are re-electing the controllers
+as the informers are always running, we cannot call this in multiple places.
+*/
+func InstallIndexers(clusterRoleBindingInformer kcprbacinformers.ClusterRoleBindingClusterInformer) {
+	indexers.AddIfNotPresentOrDie(clusterRoleBindingInformer.Informer().GetIndexer(), cache.Indexers{
+		ClusterRoleBindingByClusterRoleName: IndexClusterRoleBindingByClusterRoleName,
+	})
+}