feat(controller): Add a shared index informer for ConfigMaps (#6644)

Signed-off-by: Yuan Tang <[email protected]>

workflow/controller/controller.go

@@ -9,6 +9,8 @@ import (
  "syscall"
  "time"
 
+ v1 "k8s.io/client-go/informers/core/v1"
+
  "github.com/argoproj/pkg/errors"
  syncpkg "github.com/argoproj/pkg/sync"
  log "github.com/sirupsen/logrus"
@@ -94,6 +96,7 @@ type WorkflowController struct {
  wftmplInformer wfextvv1alpha1.WorkflowTemplateInformer
  cwftmplInformer wfextvv1alpha1.ClusterWorkflowTemplateInformer
  podInformer cache.SharedIndexInformer
+ configMapInformer cache.SharedIndexInformer
  wfQueue workqueue.RateLimitingInterface
  podQueue workqueue.RateLimitingInterface
  podCleanupQueue workqueue.RateLimitingInterface // pods to be deleted or labelled depend on GC strategy
@@ -215,6 +218,8 @@ func (wfc *WorkflowController) Run(ctx context.Context, wfWorkers, workflowTTLWo
  wfc.podInformer = wfc.newPodInformer(ctx)
  wfc.updateEstimatorFactory()
 
+ wfc.configMapInformer = wfc.newConfigMapInformer()
+
  // Create Synchronization Manager
  wfc.createSynchronizationManager(ctx)
  // init managers: throttler and SynchronizationManager
@@ -227,6 +232,7 @@ func (wfc *WorkflowController) Run(ctx context.Context, wfWorkers, workflowTTLWo
  go wfc.wfInformer.Run(ctx.Done())
  go wfc.wftmplInformer.Informer().Run(ctx.Done())
  go wfc.podInformer.Run(ctx.Done())
+ go wfc.configMapInformer.Run(ctx.Done())
  go wfc.wfTaskSetInformer.Informer().Run(ctx.Done())
 
  // Wait for all involved caches to be synced, before processing items from the queue is started
@@ -1062,6 +1068,14 @@ func (wfc *WorkflowController) newPodInformer(ctx context.Context) cache.SharedI
  return informer
 }
 
+func (wfc *WorkflowController) newConfigMapInformer() cache.SharedIndexInformer {
+ return v1.NewFilteredConfigMapInformer(wfc.kubeclientset, wfc.GetManagedNamespace(), 20*time.Minute, cache.Indexers{
+ indexes.ConfigMapLabelsIndex: indexes.ConfigMapIndexFunc,
+ }, func(opts *metav1.ListOptions) {
+ opts.LabelSelector = indexes.ConfigMapTypeLabel
+ })
+}
+
 // call this func whenever the configuration changes, or when the workflow informer changes
 func (wfc *WorkflowController) updateEstimatorFactory() {
  wfc.estimatorFactory = estimation.NewEstimatorFactory(wfc.wfInformer, wfc.hydrator, wfc.wfArchive)

workflow/controller/indexes/configmap_index.go

@@ -0,0 +1,16 @@
+package indexes
+
+import (
+ corev1 "k8s.io/api/core/v1"
+)
+
+const ConfigMapTypeLabel = "workflows.argoproj.io/configmap-type"
+
+func ConfigMapIndexFunc(obj interface{}) ([]string, error) {
+ cm, ok := obj.(*corev1.ConfigMap)
+
+ if !ok {
+ return nil, nil
+ }
+ return []string{cm.GetLabels()[ConfigMapTypeLabel]}, nil
+}

workflow/controller/indexes/configmap_index_test.go

@@ -0,0 +1,25 @@
+package indexes
+
+import (
+ "testing"
+
+ metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+
+ "github.com/stretchr/testify/assert"
+ corev1 "k8s.io/api/core/v1"
+)
+
+func TestConfigMapIndexFunc(t *testing.T) {
+ t.Run("NoLabel", func(t *testing.T) {
+ values, err := ConfigMapIndexFunc(&corev1.ConfigMap{})
+ assert.NoError(t, err)
+ assert.Equal(t, []string{""}, values)
+ })
+ t.Run("HasLabel", func(t *testing.T) {
+ values, err := ConfigMapIndexFunc(&corev1.ConfigMap{
+ ObjectMeta: metav1.ObjectMeta{Labels: map[string]string{ConfigMapTypeLabel: "cache"}},
+ })
+ assert.NoError(t, err)
+ assert.ElementsMatch(t, values, []string{"cache"})
+ })
+}

workflow/controller/indexes/indexes.go

@@ -13,6 +13,7 @@ const (
  WorkflowTemplateIndex = "workflowtemplate"
  WorkflowPhaseIndex = "workflow.phase"
  PodPhaseIndex = "pod.phase"
+ ConfigMapLabelsIndex = "configmap.labels"
  ConditionsIndex = "status.conditions"
  SemaphoreConfigIndexName = "bySemaphoreConfigMap"
  UIDIndex = "uid"