fix: controller performance optimization

Signed-off-by: jhu02 <[email protected]>
keikoproj · Feb 15, 2022 · a2c927f · a2c927f
1 parent e259f63
commit a2c927f
Show file tree

Hide file tree

Showing 2 changed files with 31 additions and 9 deletions.
diff --git a/controllers/addon_controller.go b/controllers/addon_controller.go
@@ -27,12 +27,12 @@ import (
 	batchv1beta1 "k8s.io/api/batch/v1beta1"
 	v1 "k8s.io/api/core/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/dynamic"
-	"k8s.io/client-go/informers"
+	"k8s.io/client-go/informers/internalinterfaces"
 	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/tools/cache"
 	"k8s.io/client-go/tools/record"
 	"k8s.io/client-go/util/retry"
 	ctrl "sigs.k8s.io/controller-runtime"
@@ -43,11 +43,11 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 	"sigs.k8s.io/controller-runtime/pkg/source"
 
-	wfv1 "github.com/argoproj/argo-workflows/v3/pkg/apis/workflow/v1alpha1"
 	addonmgrv1alpha1 "github.com/keikoproj/addon-manager/api/v1alpha1"
 	"github.com/keikoproj/addon-manager/pkg/addon"
 	"github.com/keikoproj/addon-manager/pkg/common"
 	"github.com/keikoproj/addon-manager/pkg/workflows"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
 
 const (
@@ -68,8 +68,7 @@ var (
 		&appsv1.ReplicaSet{TypeMeta: metav1.TypeMeta{Kind: "ReplicaSet", APIVersion: "apps/v1"}},
 		&appsv1.StatefulSet{TypeMeta: metav1.TypeMeta{Kind: "StatefulSet", APIVersion: "apps/v1"}},
 	}
-	finalizerName      = "delete.addonmgr.keikoproj.io"
-	generatedInformers informers.SharedInformerFactory
+	finalizerName = "delete.addonmgr.keikoproj.io"
 )
 
 // AddonReconciler reconciles a Addon object
@@ -193,15 +192,21 @@ func New(mgr manager.Manager, stopChan <-chan struct{}) (controller.Controller,
 		return nil, err
 	}
 
-	if err := c.Watch(&source.Kind{Type: &wfv1.Workflow{}}, &handler.EnqueueRequestForOwner{
+	sharedInforms := NewWorkflowInformer(r.dynClient, managedNameSpace, workflowResyncPeriod, cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc},
+		internalinterfaces.TweakListOptionsFunc(func(x *metav1.ListOptions) {
+			r := InstanceIDRequirement("addon-manager-workflow-controller")
+			x.LabelSelector = r.String()
+		}),
+	)
+	if err := c.Watch(&source.Informer{Informer: sharedInforms}, &handler.EnqueueRequestForOwner{
 		IsController: true,
 		OwnerType:    &addonmgrv1alpha1.Addon{},
 	}); err != nil {
 		return nil, err
 	}
 
 	// Watch workflows only in managed namespace
-	wfInforms := NewWfInformers(NewWorkflowInformer(r.dynClient, managedNameSpace, workflowResyncPeriod, nil, nil), stopChan)
+	wfInforms := NewWfInformers(sharedInforms, stopChan)
 	err = mgr.Add(wfInforms)
 	if err != nil {
 		return nil, fmt.Errorf("failed to start workflowinformers")

diff --git a/controllers/controller.go b/controllers/controller.go
@@ -7,16 +7,19 @@ import (
 
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/apimachinery/pkg/selection"
 	"k8s.io/apimachinery/pkg/watch"
 	"k8s.io/client-go/dynamic"
 	"k8s.io/client-go/informers/internalinterfaces"
 	"k8s.io/client-go/tools/cache"
 )
 
 const (
-	workflowResyncPeriod = 20 * time.Minute
+	workflowResyncPeriod         = 20 * time.Minute
+	LabelKeyControllerInstanceID = "workflows.argoproj.io/controller-instanceid"
 )
 
 type WfInformers struct {
@@ -42,7 +45,7 @@ func (wfinfo *WfInformers) Start(ctx context.Context) error {
 	return nil
 }
 
-func NewWorkflowInformer(dclient dynamic.Interface, ns string, resyncPeriod time.Duration, tweakListOptions internalinterfaces.TweakListOptionsFunc, indexers cache.Indexers) cache.SharedIndexInformer {
+func NewWorkflowInformer(dclient dynamic.Interface, ns string, resyncPeriod time.Duration, indexers cache.Indexers, tweakListOptions internalinterfaces.TweakListOptionsFunc) cache.SharedIndexInformer {
 	resource := schema.GroupVersionResource{
 		Group:    "argoproj.io",
 		Version:  "v1alpha1",
@@ -81,3 +84,17 @@ func NewFilteredUnstructuredInformer(resource schema.GroupVersionResource, clien
 		indexers,
 	)
 }
+
+func InstanceIDRequirement(instanceID string) labels.Requirement {
+	var instanceIDReq *labels.Requirement
+	var err error
+	if instanceID != "" {
+		instanceIDReq, err = labels.NewRequirement(LabelKeyControllerInstanceID, selection.Equals, []string{instanceID})
+	} else {
+		instanceIDReq, err = labels.NewRequirement(LabelKeyControllerInstanceID, selection.DoesNotExist, nil)
+	}
+	if err != nil {
+		panic(err)
+	}
+	return *instanceIDReq
+}