Merge pull request #151 from limhawjia/refactor-informer-manager

refactor(informer-manager): use gvk as keys and support TweakListOptions

pkg/apis/core/v1alpha1/extensions_federatedtypeconfig.go

@@ -55,6 +55,15 @@ func (f *FederatedTypeConfig) GetSourceTypeGVR() schema.GroupVersionResource {
 	}
 }
 
+func (f *FederatedTypeConfig) GetSourceTypeGVK() schema.GroupVersionKind {
+	apiResource := f.GetSourceType()
+	return schema.GroupVersionKind{
+		Group:   apiResource.Group,
+		Version: apiResource.Version,
+		Kind:    apiResource.Kind,
+	}
+}
+
 func (f *FederatedTypeConfig) GetStatusCollectionEnabled() bool {
 	return f.Spec.StatusCollection != nil
 }

pkg/controllers/util/federatedinformer.go

@@ -289,28 +289,6 @@ func NewFederatedInformer(
 	return federatedInformer, err
 }
 
-func IsClusterReady(clusterStatus *fedcorev1a1.FederatedClusterStatus) bool {
-	for _, condition := range clusterStatus.Conditions {
-		if condition.Type == fedcorev1a1.ClusterReady {
-			if condition.Status == corev1.ConditionTrue {
-				return true
-			}
-		}
-	}
-	return false
-}
-
-func IsClusterJoined(clusterStatus *fedcorev1a1.FederatedClusterStatus) bool {
-	for _, condition := range clusterStatus.Conditions {
-		if condition.Type == fedcorev1a1.ClusterJoined {
-			if condition.Status == corev1.ConditionTrue {
-				return true
-			}
-		}
-	}
-	return false
-}
-
 type informer struct {
 	controller cache.Controller
 	store      cache.Store

pkg/util/cluster/util.go

@@ -0,0 +1,46 @@
+/*
+Copyright 2023 The KubeAdmiral 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 cluster
+
+import (
+	corev1 "k8s.io/api/core/v1"
+
+	fedcorev1a1 "github.com/kubewharf/kubeadmiral/pkg/apis/core/v1alpha1"
+)
+
+
+func IsClusterReady(clusterStatus *fedcorev1a1.FederatedClusterStatus) bool {
+	for _, condition := range clusterStatus.Conditions {
+		if condition.Type == fedcorev1a1.ClusterReady {
+			if condition.Status == corev1.ConditionTrue {
+				return true
+			}
+		}
+	}
+	return false
+}
+
+func IsClusterJoined(clusterStatus *fedcorev1a1.FederatedClusterStatus) bool {
+	for _, condition := range clusterStatus.Conditions {
+		if condition.Type == fedcorev1a1.ClusterJoined {
+			if condition.Status == corev1.ConditionTrue {
+				return true
+			}
+		}
+	}
+	return false
+}

pkg/util/informermanager/common.go

@@ -0,0 +1,25 @@
+/*
+Copyright 2023 The KubeAdmiral 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 informermanager
+
+import fedcorev1a1 "github.com/kubewharf/kubeadmiral/pkg/apis/core/v1alpha1"
+
+// RegisterOncePredicate can be used to as an EventHandlerGenerator predicate
+// to generate and register event handlers exactly once for each FTC.
+func RegisterOncePredicate(old, _ *fedcorev1a1.FederatedTypeConfig) bool {
+	return old == nil
+}

pkg/util/informermanager/federatedinformermanager.go

@@ -24,6 +24,7 @@ import (
 	"sync"
 
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/client-go/dynamic"
@@ -34,8 +35,9 @@ import (
 	fedcorev1a1 "github.com/kubewharf/kubeadmiral/pkg/apis/core/v1alpha1"
 	fedcorev1a1informers "github.com/kubewharf/kubeadmiral/pkg/client/informers/externalversions/core/v1alpha1"
 	fedcorev1a1listers "github.com/kubewharf/kubeadmiral/pkg/client/listers/core/v1alpha1"
-	"github.com/kubewharf/kubeadmiral/pkg/controllers/util"
+	clusterutil "github.com/kubewharf/kubeadmiral/pkg/util/cluster"
 	"github.com/kubewharf/kubeadmiral/pkg/util/logging"
+	"github.com/kubewharf/kubeadmiral/pkg/util/managedlabel"
 )
 
 type federatedInformerManager struct {
@@ -83,7 +85,7 @@ func NewFederatedInformerManager(
 	clusterInformer.Informer().AddEventHandler(cache.FilteringResourceEventHandler{
 		FilterFunc: func(obj interface{}) bool {
 			cluster := obj.(*fedcorev1a1.FederatedCluster)
-			return util.IsClusterJoined(&cluster.Status)
+			return clusterutil.IsClusterJoined(&cluster.Status)
 		},
 		Handler: cache.ResourceEventHandlerFuncs{
 			AddFunc:    func(obj interface{}) { manager.enqueue(obj) },
@@ -129,7 +131,7 @@ func (m *federatedInformerManager) worker(ctx context.Context) {
 		m.queue.AddRateLimited(key)
 		return
 	}
-	if apierrors.IsNotFound(err) || !util.IsClusterJoined(&cluster.Status) {
+	if apierrors.IsNotFound(err) || !clusterutil.IsClusterJoined(&cluster.Status) {
 		if err := m.processClusterDeletion(ctx, name); err != nil {
 			logger.Error(err, "Failed to process FederatedCluster, will retry")
 			m.queue.AddRateLimited(key)
@@ -185,7 +187,19 @@ func (m *federatedInformerManager) processCluster(
 			return fmt.Errorf("failed to get client for cluster %s: %w", clusterName, err), true
 		}
 
-		manager := NewInformerManager(clusterClient, m.ftcInformer)
+		manager := NewInformerManager(
+			clusterClient,
+			m.ftcInformer,
+			func(opts *metav1.ListOptions) {
+				selector := &metav1.LabelSelector{}
+				metav1.AddLabelToSelector(
+					selector,
+					managedlabel.ManagedByKubeAdmiralLabelKey,
+					managedlabel.ManagedByKubeAdmiralLabelValue,
+				)
+				opts.LabelSelector = metav1.FormatLabelSelector(selector)
+			},
+		)
 
 		ctx, cancel := context.WithCancel(ctx)
 		for _, generator := range m.eventHandlerGenerators {
@@ -268,7 +282,7 @@ func (m *federatedInformerManager) GetFederatedTypeConfigLister() fedcorev1a1lis
 }
 
 func (m *federatedInformerManager) GetResourceLister(
-	gvr schema.GroupVersionResource,
+	gvk schema.GroupVersionKind,
 	cluster string,
 ) (lister cache.GenericLister, informerSynced cache.InformerSynced, exists bool) {
 	m.lock.RLock()
@@ -279,7 +293,7 @@ func (m *federatedInformerManager) GetResourceLister(
 		return nil, nil, false
 	}
 
-	return manager.GetResourceLister(gvr)
+	return manager.GetResourceLister(gvk)
 }
 
 func (m *federatedInformerManager) HasSynced() bool {

pkg/util/informermanager/federatedinformermanager_test.go

@@ -182,11 +182,11 @@ func TestFederatedInformerManager(t *testing.T) {
 		// 2. Verify that listers for existing FTCs and clusters are eventually available
 
 		for _, ftc := range defaultFTCs {
-			gvr := ftc.GetSourceTypeGVR()
+			gvk := ftc.GetSourceTypeGVK()
 
 			for _, cluster := range defaultClusters {
 				g.Eventually(func(g gomega.Gomega) {
-					lister, informerSynced, exists := manager.GetResourceLister(gvr, cluster.Name)
+					lister, informerSynced, exists := manager.GetResourceLister(gvk, cluster.Name)
 
 					g.Expect(exists).To(gomega.BeTrue())
 					g.Expect(lister).ToNot(gomega.BeNil())
@@ -197,17 +197,17 @@ func TestFederatedInformerManager(t *testing.T) {
 
 		// 3. Verify that the lister for non-existent FTCs or clusters are not available
 
-		lister, informerSynced, exists := manager.GetResourceLister(common.DaemonSetGVR, "cluster-1")
+		lister, informerSynced, exists := manager.GetResourceLister(daemonsetGVK, "cluster-1")
 		g.Expect(exists).To(gomega.BeFalse())
 		g.Expect(lister).To(gomega.BeNil())
 		g.Expect(informerSynced).To(gomega.BeNil())
 
-		lister, informerSynced, exists = manager.GetResourceLister(common.DeploymentGVR, "cluster-4")
+		lister, informerSynced, exists = manager.GetResourceLister(deploymentGVK, "cluster-4")
 		g.Expect(exists).To(gomega.BeFalse())
 		g.Expect(lister).To(gomega.BeNil())
 		g.Expect(informerSynced).To(gomega.BeNil())
 
-		lister, informerSynced, exists = manager.GetResourceLister(common.DaemonSetGVR, "cluster-4")
+		lister, informerSynced, exists = manager.GetResourceLister(daemonsetGVK, "cluster-4")
 		g.Expect(exists).To(gomega.BeFalse())
 		g.Expect(lister).To(gomega.BeNil())
 		g.Expect(informerSynced).To(gomega.BeNil())
@@ -245,13 +245,13 @@ func TestFederatedInformerManager(t *testing.T) {
 		}()
 
 		ftc := daemonsetFTC
-		gvr := ftc.GetSourceTypeGVR()
+		gvk := ftc.GetSourceTypeGVK()
 
 		// 2. Verify that listers for daemonsets FTCs is not available at the start
 
 		g.Consistently(func(g gomega.Gomega) {
 			for _, cluster := range defaultClusters {
-				lister, informerSynced, exists := manager.GetResourceLister(common.DeploymentGVR, cluster.Name)
+				lister, informerSynced, exists := manager.GetResourceLister(deploymentGVK, cluster.Name)
 				g.Expect(exists).To(gomega.BeFalse())
 				g.Expect(lister).To(gomega.BeNil())
 				g.Expect(informerSynced).To(gomega.BeNil())
@@ -267,7 +267,7 @@ func TestFederatedInformerManager(t *testing.T) {
 
 		g.Eventually(func(g gomega.Gomega) {
 			for _, cluster := range defaultClusters {
-				lister, informerSynced, exists := manager.GetResourceLister(gvr, cluster.Name)
+				lister, informerSynced, exists := manager.GetResourceLister(gvk, cluster.Name)
 				g.Expect(exists).To(gomega.BeTrue())
 				g.Expect(lister).ToNot(gomega.BeNil())
 				g.Expect(informerSynced()).To(gomega.BeTrue())
@@ -308,9 +308,9 @@ func TestFederatedInformerManager(t *testing.T) {
 
 		g.Consistently(func(g gomega.Gomega) {
 			for _, ftc := range defaultFTCs {
-				gvr := ftc.GetSourceTypeGVR()
+				gvk := ftc.GetSourceTypeGVK()
 
-				lister, informerSynced, exists := manager.GetResourceLister(gvr, cluster.Name)
+				lister, informerSynced, exists := manager.GetResourceLister(gvk, cluster.Name)
 				g.Expect(exists).To(gomega.BeFalse())
 				g.Expect(lister).To(gomega.BeNil())
 				g.Expect(informerSynced).To(gomega.BeNil())
@@ -326,9 +326,9 @@ func TestFederatedInformerManager(t *testing.T) {
 
 		g.Eventually(func(g gomega.Gomega) {
 			for _, ftc := range defaultFTCs {
-				gvr := ftc.GetSourceTypeGVR()
+				gvk := ftc.GetSourceTypeGVK()
 
-				lister, informerSynced, exists := manager.GetResourceLister(gvr, cluster.Name)
+				lister, informerSynced, exists := manager.GetResourceLister(gvk, cluster.Name)
 				g.Expect(exists).To(gomega.BeTrue())
 				g.Expect(lister).ToNot(gomega.BeNil())
 				g.Expect(informerSynced()).To(gomega.BeTrue())