fix: refactor runConfigMapWatcher to use Informers. Fixes #11657

workflow/controller/cache_gc.go

@@ -25,7 +25,7 @@ func init() {
 
 // syncAllCacheForGC syncs all cache for GC
 func (wfc *WorkflowController) syncAllCacheForGC(ctx context.Context) {
-	configMaps, err := wfc.configMapInformer.GetIndexer().ByIndex(indexes.ConfigMapLabelsIndex, common.LabelValueTypeConfigMapCache)
+	configMaps, err := wfc.cmInformer.GetIndexer().ByIndex(indexes.ConfigMapLabelsIndex, common.LabelValueTypeConfigMapCache)
 	if err != nil {
 		log.WithError(err).Error("Failed to get configmaps from informer")
 		return

workflow/controller/controller.go

@@ -19,9 +19,9 @@ import (
 	"golang.org/x/time/rate"
 	apiv1 "k8s.io/api/core/v1"
 	apierr "k8s.io/apimachinery/pkg/api/errors"
-	"k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"k8s.io/apimachinery/pkg/fields"
 	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/selection"
@@ -35,7 +35,6 @@ import (
 	typedv1 "k8s.io/client-go/kubernetes/typed/core/v1"
 	"k8s.io/client-go/rest"
 	"k8s.io/client-go/tools/cache"
-	apiwatch "k8s.io/client-go/tools/watch"
 	"k8s.io/client-go/util/retry"
 	"k8s.io/client-go/util/workqueue"
 
@@ -123,7 +122,8 @@ type WorkflowController struct {
 	wftmplInformer        wfextvv1alpha1.WorkflowTemplateInformer
 	cwftmplInformer       wfextvv1alpha1.ClusterWorkflowTemplateInformer
 	podInformer           cache.SharedIndexInformer
-	configMapInformer     cache.SharedIndexInformer
+	cmInformer            cache.SharedIndexInformer // configmaps of plugins, parameters, memoizations, etc
+	cmCtrlInformer        cache.SharedIndexInformer // controller's own configmap
 	wfQueue               workqueue.RateLimitingInterface
 	podCleanupQueue       workqueue.RateLimitingInterface // pods to be deleted or labelled depend on GC strategy
 	throttler             sync.Throttler
@@ -319,7 +319,11 @@ func (wfc *WorkflowController) Run(ctx context.Context, wfWorkers, workflowTTLWo
 	}
 	wfc.updateEstimatorFactory()
 
-	wfc.configMapInformer, err = wfc.newConfigMapInformer()
+	wfc.cmInformer, err = wfc.newConfigMapInformer()
+	if err != nil {
+		log.Fatal(err)
+	}
+	wfc.cmCtrlInformer, err = wfc.newConfigMapCtrlInformer(ctx)
 	if err != nil {
 		log.Fatal(err)
 	}
@@ -331,14 +335,11 @@ func (wfc *WorkflowController) Run(ctx context.Context, wfWorkers, workflowTTLWo
 		log.Fatal(err)
 	}
 
-	if os.Getenv("WATCH_CONTROLLER_SEMAPHORE_CONFIGMAPS") != "false" {
-		go wfc.runConfigMapWatcher(ctx.Done())
-	}
-
 	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.cmInformer.Run(ctx.Done())
+	go wfc.cmCtrlInformer.Run(ctx.Done())
 	go wfc.wfTaskSetInformer.Informer().Run(ctx.Done())
 	go wfc.artGCTaskInformer.Informer().Run(ctx.Done())
 	go wfc.taskResultInformer.Run(ctx.Done())
@@ -350,7 +351,8 @@ func (wfc *WorkflowController) Run(ctx context.Context, wfWorkers, workflowTTLWo
 		wfc.wfInformer.HasSynced,
 		wfc.wftmplInformer.Informer().HasSynced,
 		wfc.podInformer.HasSynced,
-		wfc.configMapInformer.HasSynced,
+		wfc.cmInformer.HasSynced,
+		wfc.cmCtrlInformer.HasSynced,
 		wfc.wfTaskSetInformer.Informer().HasSynced,
 		wfc.artGCTaskInformer.Informer().HasSynced,
 		wfc.taskResultInformer.HasSynced,
@@ -441,42 +443,9 @@ func (wfc *WorkflowController) initManagers(ctx context.Context) error {
 	return nil
 }
 
-func (wfc *WorkflowController) runConfigMapWatcher(stopCh <-chan struct{}) {
-	defer runtimeutil.HandleCrash(runtimeutil.PanicHandlers...)
-
-	ctx := context.Background()
-	retryWatcher, err := apiwatch.NewRetryWatcher("1", &cache.ListWatch{
-		WatchFunc: func(options metav1.ListOptions) (watch.Interface, error) {
-			return wfc.kubeclientset.CoreV1().ConfigMaps(wfc.managedNamespace).Watch(ctx, metav1.ListOptions{})
-		},
-	})
-	if err != nil {
-		panic(err)
-	}
-	defer retryWatcher.Stop()
-
-	for {
-		select {
-		case event := <-retryWatcher.ResultChan():
-			cm, ok := event.Object.(*apiv1.ConfigMap)
-			if !ok {
-				log.Errorf("invalid config map object received in config watcher. Ignored processing")
-				continue
-			}
-			log.Debugf("received config map %s/%s update", cm.Namespace, cm.Name)
-			if cm.GetName() == wfc.configController.GetName() && wfc.namespace == cm.GetNamespace() {
-				log.Infof("Received Workflow Controller config map %s/%s update", cm.Namespace, cm.Name)
-				wfc.UpdateConfig(ctx)
-			}
-			wfc.notifySemaphoreConfigUpdate(cm)
-		case <-stopCh:
-			return
-		}
-	}
-}
-
 // notifySemaphoreConfigUpdate will notify semaphore config update to pending workflows
 func (wfc *WorkflowController) notifySemaphoreConfigUpdate(cm *apiv1.ConfigMap) {
+	log.Debugf("received semaphore config map %s/%s update", cm.GetNamespace(), cm.GetName())
 	wfs, err := wfc.wfInformer.GetIndexer().ByIndex(indexes.SemaphoreConfigIndexName, fmt.Sprintf("%s/%s", cm.Namespace, cm.Name))
 	if err != nil {
 		log.Errorf("failed get the workflow from informer. %v", err)
@@ -1270,69 +1239,61 @@ func (wfc *WorkflowController) newPodInformer(ctx context.Context) (cache.Shared
 	return informer, nil
 }
 
+var watchControllerSemaphoreConfigMaps = os.Getenv("WATCH_CONTROLLER_SEMAPHORE_CONFIGMAPS") != "false"
+
 func (wfc *WorkflowController) newConfigMapInformer() (cache.SharedIndexInformer, error) {
-	indexInformer := v1.NewFilteredConfigMapInformer(wfc.kubeclientset, wfc.GetManagedNamespace(), 20*time.Minute, cache.Indexers{
+	indexInformer := v1.NewConfigMapInformer(wfc.kubeclientset, wfc.GetManagedNamespace(), 20*time.Minute, cache.Indexers{
 		indexes.ConfigMapLabelsIndex: indexes.ConfigMapIndexFunc,
-	}, func(opts *metav1.ListOptions) {
-		opts.LabelSelector = common.LabelKeyConfigMapType
 	})
+
 	log.WithField("executorPlugins", wfc.executorPlugins != nil).Info("Plugins")
-	if wfc.executorPlugins != nil {
-		_, err := indexInformer.AddEventHandler(cache.FilteringResourceEventHandler{
-			FilterFunc: func(obj interface{}) bool {
-				cm, err := meta.Accessor(obj)
-				if err != nil {
-					return false
-				}
-				return cm.GetLabels()[common.LabelKeyConfigMapType] == common.LabelValueTypeConfigMapExecutorPlugin
-			},
-			Handler: cache.ResourceEventHandlerFuncs{
-				AddFunc: func(obj interface{}) {
-					cm := obj.(*apiv1.ConfigMap)
-					p, err := plugin.FromConfigMap(cm)
-					if err != nil {
-						log.WithField("namespace", cm.GetNamespace()).
-							WithField("name", cm.GetName()).
-							WithError(err).
-							Error("failed to convert configmap to plugin")
-						return
-					}
-					if _, ok := wfc.executorPlugins[cm.GetNamespace()]; !ok {
-						wfc.executorPlugins[cm.GetNamespace()] = map[string]*spec.Plugin{}
-					}
-					wfc.executorPlugins[cm.GetNamespace()][cm.GetName()] = p
-					log.WithField("namespace", cm.GetNamespace()).
-						WithField("name", cm.GetName()).
-						Info("Executor plugin added")
-				},
-				UpdateFunc: func(_, obj interface{}) {
-					cm := obj.(*apiv1.ConfigMap)
-					p, err := plugin.FromConfigMap(cm)
-					if err != nil {
-						log.WithField("namespace", cm.GetNamespace()).
-							WithField("name", cm.GetName()).
-							WithError(err).
-							Error("failed to convert configmap to plugin")
-						return
-					}
 
-					wfc.executorPlugins[cm.GetNamespace()][cm.GetName()] = p
-					log.WithField("namespace", cm.GetNamespace()).
-						WithField("name", cm.GetName()).
-						Info("Executor plugin updated")
-				},
-				DeleteFunc: func(obj interface{}) {
-					key, _ := cache.DeletionHandlingMetaNamespaceKeyFunc(obj)
-					namespace, name, _ := cache.SplitMetaNamespaceKey(key)
-					delete(wfc.executorPlugins[namespace], name)
-					log.WithField("namespace", namespace).WithField("name", name).Info("Executor plugin removed")
-				},
-			},
-		})
-		if err != nil {
-			return nil, err
-		}
+	_, err := indexInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{
+		AddFunc: func(obj interface{}) {
+			cm := obj.(*apiv1.ConfigMap)
+			wfc.applyPluginCM(cm, "added")
+		},
+		UpdateFunc: func(_, obj interface{}) {
+			cm := obj.(*apiv1.ConfigMap)
+			wfc.applyPluginCM(cm, "updated")
+
+			if !watchControllerSemaphoreConfigMaps {
+				return
+			}
+
+			wfc.notifySemaphoreConfigUpdate(cm)
+		},
+		DeleteFunc: func(obj interface{}) {
+			cm := obj.(*apiv1.ConfigMap)
+			wfc.deletePluginCM(cm)
+		},
+	})
+
+	if err != nil {
+		return nil, err
+	}
+	return indexInformer, nil
+}
+
+func (wfc *WorkflowController) newConfigMapCtrlInformer(ctx context.Context) (cache.SharedIndexInformer, error) {
+	indexInformer := v1.NewFilteredConfigMapInformer(wfc.kubeclientset, wfc.GetNamespace(), 20*time.Minute, nil, func(opts *metav1.ListOptions) {
+		opts.FieldSelector = fields.OneTermEqualSelector(metav1.ObjectNameField, wfc.configController.GetName()).String() // only the controller configmap
+	})
+
+	if !watchControllerSemaphoreConfigMaps {
+		return indexInformer, nil
+	}
+
+	_, err := indexInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{
+		UpdateFunc: func(_, obj interface{}) {
+			cm := obj.(*apiv1.ConfigMap)
+			log.Infof("Received Workflow Controller config map %s/%s update", cm.GetNamespace(), cm.GetName())
+			wfc.UpdateConfig(ctx)
+		},
+	})
 
+	if err != nil {
+		return nil, err
 	}
 	return indexInformer, nil
 }
@@ -1355,11 +1316,52 @@ func (wfc *WorkflowController) setWorkflowDefaults(wf *wfv1.Workflow) error {
 	return nil
 }
 
+func (wfc *WorkflowController) GetNamespace() string {
+	return wfc.Config.Namespace
+}
+
 func (wfc *WorkflowController) GetManagedNamespace() string {
 	if wfc.managedNamespace != "" {
 		return wfc.managedNamespace
 	}
-	return wfc.Config.Namespace
+	return wfc.GetNamespace()
+}
+
+func (wfc *WorkflowController) isPluginCM(cm metav1.Object) bool {
+	return cm.GetLabels()[common.LabelKeyConfigMapType] == common.LabelValueTypeConfigMapExecutorPlugin
+}
+
+func (wfc *WorkflowController) applyPluginCM(cm *apiv1.ConfigMap, verb string) {
+	if !wfc.isPluginCM(cm) {
+		return
+	}
+
+	p, err := plugin.FromConfigMap(cm)
+	if err != nil {
+		log.WithField("namespace", cm.GetNamespace()).
+			WithField("name", cm.GetName()).
+			WithError(err).
+			Error("failed to convert configmap to plugin")
+		return
+	}
+	if _, ok := wfc.executorPlugins[cm.GetNamespace()]; !ok {
+		wfc.executorPlugins[cm.GetNamespace()] = map[string]*spec.Plugin{}
+	}
+	wfc.executorPlugins[cm.GetNamespace()][cm.GetName()] = p
+	log.WithField("namespace", cm.GetNamespace()).
+		WithField("name", cm.GetName()).
+		Infof("Executor plugin %s", verb)
+}
+
+func (wfc *WorkflowController) deletePluginCM(cm *apiv1.ConfigMap) {
+	if !wfc.isPluginCM(cm) {
+		return
+	}
+
+	key, _ := cache.DeletionHandlingMetaNamespaceKeyFunc(cm)
+	namespace, name, _ := cache.SplitMetaNamespaceKey(key)
+	delete(wfc.executorPlugins[namespace], name)
+	log.WithField("namespace", namespace).WithField("name", name).Info("Executor plugin removed")
 }
 
 func (wfc *WorkflowController) getMaxStackDepth() int {

workflow/controller/controller_test.go

@@ -322,7 +322,8 @@ func newController(options ...interface{}) (context.CancelFunc, *WorkflowControl
 		wfc.wftmplInformer = informerFactory.Argoproj().V1alpha1().WorkflowTemplates()
 		_ = wfc.addWorkflowInformerHandlers(ctx)
 		wfc.podInformer, _ = wfc.newPodInformer(ctx)
-		wfc.configMapInformer, _ = wfc.newConfigMapInformer()
+		wfc.cmInformer, _ = wfc.newConfigMapInformer()
+		wfc.cmCtrlInformer, _ = wfc.newConfigMapCtrlInformer(ctx)
 		wfc.createSynchronizationManager(ctx)
 		_ = wfc.initManagers(ctx)
 

workflow/controller/dag.go

@@ -439,7 +439,7 @@ func (woc *wfOperationCtx) executeDAGTask(ctx context.Context, dagCtx *dagContex
 			}
 		}
 
-		processedTmpl, err := common.ProcessArgs(tmpl, &task.Arguments, woc.globalParams, map[string]string{}, true, woc.wf.Namespace, woc.controller.configMapInformer.GetIndexer())
+		processedTmpl, err := common.ProcessArgs(tmpl, &task.Arguments, woc.globalParams, map[string]string{}, true, woc.wf.Namespace, woc.controller.cmInformer.GetIndexer())
 		if err != nil {
 			woc.markNodeError(node.Name, err)
 		}

workflow/controller/operator.go

@@ -639,7 +639,7 @@ func (woc *wfOperationCtx) setGlobalParameters(executionParameters wfv1.Argument
 	}
 	for _, param := range executionParameters.Parameters {
 		if param.ValueFrom != nil && param.ValueFrom.ConfigMapKeyRef != nil {
-			cmValue, err := common.GetConfigMapValue(woc.controller.configMapInformer.GetIndexer(), woc.wf.ObjectMeta.Namespace, param.ValueFrom.ConfigMapKeyRef.Name, param.ValueFrom.ConfigMapKeyRef.Key)
+			cmValue, err := common.GetConfigMapValue(woc.controller.cmInformer.GetIndexer(), woc.wf.ObjectMeta.Namespace, param.ValueFrom.ConfigMapKeyRef.Name, param.ValueFrom.ConfigMapKeyRef.Key)
 			if err != nil {
 				if param.ValueFrom.Default != nil {
 					woc.globalParams["workflow.parameters."+param.Name] = param.ValueFrom.Default.String()
@@ -1920,7 +1920,7 @@ func (woc *wfOperationCtx) executeTemplate(ctx context.Context, nodeName string,
 	}
 
 	// Inputs has been processed with arguments already, so pass empty arguments.
-	processedTmpl, err := common.ProcessArgs(resolvedTmpl, &args, woc.globalParams, localParams, false, woc.wf.Namespace, woc.controller.configMapInformer.GetIndexer())
+	processedTmpl, err := common.ProcessArgs(resolvedTmpl, &args, woc.globalParams, localParams, false, woc.wf.Namespace, woc.controller.cmInformer.GetIndexer())
 	if err != nil {
 		return woc.initializeNodeOrMarkError(node, nodeName, templateScope, orgTmpl, opts.boundaryID, opts.nodeFlag, err), err
 	}

workflow/controller/workflowpod.go

@@ -364,7 +364,7 @@ func (woc *wfOperationCtx) createWorkflowPod(ctx context.Context, nodeName strin
 		if tmpl.IsPodType() {
 			localParams[common.LocalVarPodName] = pod.Name
 		}
-		tmpl, err := common.ProcessArgs(tmpl, &wfv1.Arguments{}, woc.globalParams, localParams, false, woc.wf.Namespace, woc.controller.configMapInformer.GetIndexer())
+		tmpl, err := common.ProcessArgs(tmpl, &wfv1.Arguments{}, woc.globalParams, localParams, false, woc.wf.Namespace, woc.controller.cmInformer.GetIndexer())
 		if err != nil {
 			return nil, errors.Wrap(err, "", "Failed to substitute the PodSpecPatch variables")
 		}