Repository: argoproj/argo
Pull request #4453 (2020-11-03 19:48:52)

chore(controller): Refactor and tidy up
Signed-off-by: Alex Collins <alex_collins@intuit.com>

Checklist:

* [x] Either (a) I've created an [enhancement proposal](https://github.com/argoproj/argo/issues/new/choose) and discussed it with the community, (b) this is a bug fix, or (c) this is a chore.
* [x] The title of the PR is (a) [conventional](https://www.conventionalcommits.org/en/v1.0.0/), (b) states what changed, and (c) suffixes the related issues number. E.g. `"fix(controller): Updates such and such. Fixes #1234"`.  
* [x] My organization is added to [USERS.md](https://github.com/argoproj/argo/blob/master/USERS.md).
* [x] I've signed the CLA.
* [x] I have written unit and/or e2e tests for my change. PRs without these are unlikely to be merged.
* [x] My builds are green. Try syncing with master if they are not. 


alexec: @sarabala1979 as requested in throttler PR, I have separated out these changes.

Patch:
diff --git a/workflow/controller/controller.go b/workflow/controller/controller.go
index eedca6c1f..3610455bd 100644
--- a/workflow/controller/controller.go
+++ b/workflow/controller/controller.go
@@ -158,7 +158,7 @@ func (wfc *WorkflowController) runTTLController(ctx context.Context) {
 }
 
 func (wfc *WorkflowController) runCronController(ctx context.Context) {
-	cronController := cron.NewCronController(wfc.wfclientset, wfc.restConfig, wfc.dynamicInterface, wfc.namespace, wfc.GetManagedNamespace(), wfc.Config.InstanceID, wfc.metrics, wfc.eventRecorderManager)
+	cronController := cron.NewCronController(wfc.wfclientset, wfc.dynamicInterface, wfc.namespace, wfc.GetManagedNamespace(), wfc.Config.InstanceID, wfc.metrics, wfc.eventRecorderManager)
 	cronController.Run(ctx)
 }
 
@@ -196,13 +196,8 @@ func (wfc *WorkflowController) Run(ctx context.Context, wfWorkers, podWorkers in
 	go wfc.runCronController(ctx)
 	go wfc.metrics.RunServer(ctx)
 
-	// Wait for all involved caches to be synced, before processing items from the queue is started
-	for _, informer := range []cache.SharedIndexInformer{wfc.wfInformer, wfc.wftmplInformer.Informer(), wfc.podInformer} {
-		if !cache.WaitForCacheSync(ctx.Done(), informer.HasSynced) {
-			log.Error("Timed out waiting for caches to sync")
-			return
-		}
-	}
+	wfc.createClusterWorkflowTemplateInformer(ctx)
+	wfc.waitForCacheSync(ctx)
 
 	// Create Synchronization Manager
 	err := wfc.createSynchronizationManager()
@@ -210,8 +205,6 @@ func (wfc *WorkflowController) Run(ctx context.Context, wfWorkers, podWorkers in
 		panic(err)
 	}
 
-	wfc.createClusterWorkflowTemplateInformer(ctx)
-
 	for i := 0; i < wfWorkers; i++ {
 		go wait.Until(wfc.runWorker, time.Second, ctx.Done())
 	}
@@ -221,6 +214,18 @@ func (wfc *WorkflowController) Run(ctx context.Context, wfWorkers, podWorkers in
 	<-ctx.Done()
 }
 
+func (wfc *WorkflowController) waitForCacheSync(ctx context.Context) {
+	// Wait for all involved caches to be synced, before processing items from the queue is started
+	if !cache.WaitForCacheSync(ctx.Done(), wfc.wfInformer.HasSynced, wfc.wftmplInformer.Informer().HasSynced, wfc.podInformer.HasSynced) {
+		panic("Timed out waiting for caches to sync")
+	}
+	if wfc.cwftmplInformer != nil {
+		if !cache.WaitForCacheSync(ctx.Done(), wfc.cwftmplInformer.Informer().HasSynced) {
+			panic("Timed out waiting for caches to sync")
+		}
+	}
+}
+
 // Create and initialize the Synchronization Manager
 func (wfc *WorkflowController) createSynchronizationManager() error {
 	getSyncLimit := func(lockKey string) (int, error) {
@@ -274,10 +279,6 @@ func (wfc *WorkflowController) createClusterWorkflowTemplateInformer(ctx context
 	if cwftGetAllowed && cwftListAllowed && cwftWatchAllowed {
 		wfc.cwftmplInformer = informer.NewTolerantClusterWorkflowTemplateInformer(wfc.dynamicInterface, clusterWorkflowTemplateResyncPeriod)
 		go wfc.cwftmplInformer.Informer().Run(ctx.Done())
-		if !cache.WaitForCacheSync(ctx.Done(), wfc.cwftmplInformer.Informer().HasSynced) {
-			log.Error("Timed out waiting for caches to sync")
-			return
-		}
 	} else {
 		log.Warnf("Controller doesn't have RBAC access for ClusterWorkflowTemplates")
 	}
@@ -782,9 +783,7 @@ func (wfc *WorkflowController) archiveWorkflowAux(obj interface{}) error {
 }
 
 func (wfc *WorkflowController) newWorkflowPodWatch() *cache.ListWatch {
-	c := wfc.kubeclientset.CoreV1().RESTClient()
-	resource := "pods"
-	namespace := wfc.GetManagedNamespace()
+	c := wfc.kubeclientset.CoreV1().Pods(wfc.GetManagedNamespace())
 	// completed=false
 	incompleteReq, _ := labels.NewRequirement(common.LabelKeyCompleted, selection.Equals, []string{"false"})
 	labelSelector := labels.NewSelector().
@@ -793,20 +792,12 @@ func (wfc *WorkflowController) newWorkflowPodWatch() *cache.ListWatch {
 
 	listFunc := func(options metav1.ListOptions) (runtime.Object, error) {
 		options.LabelSelector = labelSelector.String()
-		req := c.Get().
-			Namespace(namespace).
-			Resource(resource).
-			VersionedParams(&options, metav1.ParameterCodec)
-		return req.Do().Get()
+		return c.List(options)
 	}
 	watchFunc := func(options metav1.ListOptions) (watch.Interface, error) {
 		options.Watch = true
 		options.LabelSelector = labelSelector.String()
-		req := c.Get().
-			Namespace(namespace).
-			Resource(resource).
-			VersionedParams(&options, metav1.ParameterCodec)
-		return req.Watch()
+		return c.Watch(options)
 	}
 	return &cache.ListWatch{ListFunc: listFunc, WatchFunc: watchFunc}
 }
diff --git a/workflow/cron/controller.go b/workflow/cron/controller.go
index f961a0282..96b2d88ec 100644
--- a/workflow/cron/controller.go
+++ b/workflow/cron/controller.go
@@ -20,7 +20,6 @@ import (
 	"k8s.io/client-go/dynamic"
 	"k8s.io/client-go/dynamic/dynamicinformer"
 	"k8s.io/client-go/informers"
-	"k8s.io/client-go/rest"
 	"k8s.io/client-go/tools/cache"
 	"k8s.io/client-go/util/workqueue"
 
@@ -45,7 +44,6 @@ type Controller struct {
 	wfQueue              workqueue.RateLimitingInterface
 	cronWfInformer       informers.GenericInformer
 	cronWfQueue          workqueue.RateLimitingInterface
-	restConfig           *rest.Config
 	dynamicInterface     dynamic.Interface
 	metrics              *metrics.Metrics
 	eventRecorderManager events.EventRecorderManager
@@ -56,7 +54,7 @@ const (
 	cronWorkflowWorkers      = 8
 )
 
-func NewCronController(wfclientset versioned.Interface, restConfig *rest.Config, dynamicInterface dynamic.Interface, namespace string, managedNamespace string, instanceId string, metrics *metrics.Metrics, eventRecorderManager events.EventRecorderManager) *Controller {
+func NewCronController(wfclientset versioned.Interface, dynamicInterface dynamic.Interface, namespace string, managedNamespace string, instanceId string, metrics *metrics.Metrics, eventRecorderManager events.EventRecorderManager) *Controller {
 	return &Controller{
 		wfClientset:          wfclientset,
 		namespace:            namespace,
@@ -64,7 +62,6 @@ func NewCronController(wfclientset versioned.Interface, restConfig *rest.Config,
 		instanceId:           instanceId,
 		cron:                 newCronFacade(),
 		keyLock:              sync.NewKeyLock(),
-		restConfig:           restConfig,
 		dynamicInterface:     dynamicInterface,
 		wfQueue:              workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "wf_cron_queue"),
 		cronWfQueue:          workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "cron_wf_queue"),


Test patch:
diff --git a/workflow/controller/controller_test.go b/workflow/controller/controller_test.go
index f29a9ab12..30d660ee5 100644
--- a/workflow/controller/controller_test.go
+++ b/workflow/controller/controller_test.go
@@ -3,13 +3,13 @@ package controller
 import (
 	"context"
 	"testing"
-	"time"
 
 	"github.com/argoproj/pkg/sync"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/labels"
 	dynamicfake "k8s.io/client-go/dynamic/fake"
 	"k8s.io/client-go/tools/record"
+	"k8s.io/utils/pointer"
 
 	"github.com/stretchr/testify/assert"
 	authorizationv1 "k8s.io/api/authorization/v1"
@@ -19,7 +19,6 @@ import (
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/kubernetes/fake"
 	k8stesting "k8s.io/client-go/testing"
-	"k8s.io/client-go/tools/cache"
 	"k8s.io/client-go/util/workqueue"
 	"sigs.k8s.io/yaml"
 
@@ -36,6 +35,8 @@ import (
 	"github.com/argoproj/argo/workflow/events"
 	hydratorfake "github.com/argoproj/argo/workflow/hydrator/fake"
 	"github.com/argoproj/argo/workflow/metrics"
+	workflowsync "github.com/argoproj/argo/workflow/sync"
+	"github.com/argoproj/argo/workflow/util"
 )
 
 var helloWorldWf = `
@@ -121,85 +122,111 @@ func (t testEventRecorderManager) Get(string) record.EventRecorder {
 
 var _ events.EventRecorderManager = &testEventRecorderManager{}
 
-func newController(objects ...runtime.Object) (context.CancelFunc, *WorkflowController) {
+func newController(options ...interface{}) (context.CancelFunc, *WorkflowController) {
+	// get all the objects and add to the fake
+	var objects []runtime.Object
+	var uns []runtime.Object
+	for _, opt := range options {
+		switch v := opt.(type) {
+		// special case for workflows must be unstructured
+		case *wfv1.Workflow:
+			un, err := util.ToUnstructured(v)
+			if err != nil {
+				panic(err)
+			}
+			uns = append(uns, un)
+			objects = append(objects, v)
+		case runtime.Object:
+			objects = append(objects, v)
+		}
+	}
+
 	wfclientset := fakewfclientset.NewSimpleClientset(objects...)
-	informerFactory := wfextv.NewSharedInformerFactory(wfclientset, 10*time.Minute)
-	wfInformer := cache.NewSharedIndexInformer(nil, nil, 0, nil)
-	wftmplInformer := informerFactory.Argoproj().V1alpha1().WorkflowTemplates()
-	cwftmplInformer := informerFactory.Argoproj().V1alpha1().ClusterWorkflowTemplates()
+	dynamicClient := dynamicfake.NewSimpleDynamicClient(scheme.Scheme, uns...)
+	informerFactory := wfextv.NewSharedInformerFactory(wfclientset, 0)
 	ctx, cancel := context.WithCancel(context.Background())
-	go wftmplInformer.Informer().Run(ctx.Done())
-	go cwftmplInformer.Informer().Run(ctx.Done())
-	if !cache.WaitForCacheSync(ctx.Done(), wftmplInformer.Informer().HasSynced) {
-		panic("Timed out waiting for caches to sync")
-	}
-	if !cache.WaitForCacheSync(ctx.Done(), cwftmplInformer.Informer().HasSynced) {
-		panic("Timed out waiting for caches to sync")
-	}
 	kube := fake.NewSimpleClientset()
-	controller := &WorkflowController{
-		Config: config.Config{
-			ExecutorImage: "executor:latest",
-		},
+	wfc := &WorkflowController{
+		Config:               config.Config{ExecutorImage: "executor:latest"},
 		kubeclientset:        kube,
-		dynamicInterface:     dynamicfake.NewSimpleDynamicClient(scheme.Scheme),
+		dynamicInterface:     dynamicClient,
 		wfclientset:          wfclientset,
 		completedPods:        make(chan string, 16),
-		wfInformer:           wfInformer,
-		wftmplInformer:       wftmplInformer,
-		cwftmplInformer:      cwftmplInformer,
-		wfQueue:              workqueue.NewRateLimitingQueue(workqueue.DefaultControllerRateLimiter()),
-		podQueue:             workqueue.NewRateLimitingQueue(workqueue.DefaultControllerRateLimiter()),
 		workflowKeyLock:      sync.NewKeyLock(),
 		wfArchive:            sqldb.NullWorkflowArchive,
 		hydrator:             hydratorfake.Noop,
 		estimatorFactory:     estimation.DummyEstimatorFactory,
-		metrics:              metrics.New(metrics.ServerConfig{}, metrics.ServerConfig{}),
 		eventRecorderManager: &testEventRecorderManager{eventRecorder: record.NewFakeRecorder(16)},
 		archiveLabelSelector: labels.Everything(),
 		cacheFactory:         controllercache.NewCacheFactory(kube, "default"),
 	}
-	controller.podInformer = controller.newPodInformer()
-	return cancel, controller
+
+	for _, opt := range options {
+		switch v := opt.(type) {
+		// any post-processing
+		case func(workflowController *WorkflowController):
+			v(wfc)
+		}
+	}
+
+	// always compare to NewWorkflowController to see what this block of code should be doing
+	{
+		wfc.metrics = metrics.New(metrics.ServerConfig{}, metrics.ServerConfig{})
+		wfc.wfQueue = workqueue.NewRateLimitingQueue(workqueue.DefaultControllerRateLimiter())
+		wfc.throttler = workflowsync.NewThrottler(0, wfc.wfQueue)
+		wfc.throttler.SetParallelism(wfc.getParallelism())
+		wfc.podQueue = workqueue.NewRateLimitingQueue(workqueue.DefaultControllerRateLimiter())
+	}
+
+	// always compare to WorkflowController.Run to see what this block of code should be doing
+	{
+		wfc.wfInformer = util.NewWorkflowInformer(dynamicClient, "", 0, wfc.tweakListOptions, indexers)
+		wfc.wftmplInformer = informerFactory.Argoproj().V1alpha1().WorkflowTemplates()
+		wfc.addWorkflowInformerHandlers()
+		wfc.podInformer = wfc.newPodInformer()
+		go wfc.wfInformer.Run(ctx.Done())
+		go wfc.wftmplInformer.Informer().Run(ctx.Done())
+		go wfc.podInformer.Run(ctx.Done())
+		wfc.cwftmplInformer = informerFactory.Argoproj().V1alpha1().ClusterWorkflowTemplates()
+		go wfc.cwftmplInformer.Informer().Run(ctx.Done())
+		wfc.waitForCacheSync(ctx)
+	}
+	return cancel, wfc
 }
 
 func newControllerWithDefaults() (context.CancelFunc, *WorkflowController) {
-	cancel, controller := newController()
-	myBool := true
-	controller.Config.WorkflowDefaults = &wfv1.Workflow{
-		Spec: wfv1.WorkflowSpec{
-			HostNetwork: &myBool,
-		},
-	}
+	cancel, controller := newController(func(controller *WorkflowController) {
+		controller.Config.WorkflowDefaults = &wfv1.Workflow{
+			Spec: wfv1.WorkflowSpec{HostNetwork: pointer.BoolPtr(true)},
+		}
+	})
 	return cancel, controller
 }
 
 func newControllerWithComplexDefaults() (context.CancelFunc, *WorkflowController) {
-	cancel, controller := newController()
-	myBool := true
-	var ten int32 = 10
-	var seven int32 = 10
-	controller.Config.WorkflowDefaults = &wfv1.Workflow{
-		ObjectMeta: metav1.ObjectMeta{
-			Annotations: map[string]string{
-				"annotation": "value",
-			},
-			Labels: map[string]string{
-				"label": "value",
+	cancel, controller := newController(func(controller *WorkflowController) {
+		controller.Config.WorkflowDefaults = &wfv1.Workflow{
+			ObjectMeta: metav1.ObjectMeta{
+				Annotations: map[string]string{
+					"annotation": "value",
+				},
+				Labels: map[string]string{
+					"label": "value",
+				},
 			},
-		},
-		Spec: wfv1.WorkflowSpec{
-			HostNetwork:        &myBool,
-			Entrypoint:         "good_entrypoint",
-			ServiceAccountName: "my_service_account",
-			TTLStrategy: &wfv1.TTLStrategy{
-				SecondsAfterCompletion: &ten,
-				SecondsAfterSuccess:    &ten,
-				SecondsAfterFailure:    &ten,
+			Spec: wfv1.WorkflowSpec{
+				HostNetwork:        pointer.BoolPtr(true),
+				Entrypoint:         "good_entrypoint",
+				ServiceAccountName: "my_service_account",
+				TTLStrategy: &wfv1.TTLStrategy{
+					SecondsAfterCompletion: pointer.Int32Ptr(10),
+					SecondsAfterSuccess:    pointer.Int32Ptr(10),
+					SecondsAfterFailure:    pointer.Int32Ptr(10),
+				},
+				TTLSecondsAfterFinished: pointer.Int32Ptr(7),
 			},
-			TTLSecondsAfterFinished: &seven,
-		},
-	}
+		}
+	})
 	return cancel, controller
 }
 
diff --git a/workflow/controller/operator_template_scope_test.go b/workflow/controller/operator_template_scope_test.go
index 4e231eac1..6be76a078 100644
--- a/workflow/controller/operator_template_scope_test.go
+++ b/workflow/controller/operator_template_scope_test.go
@@ -75,13 +75,11 @@ func TestTemplateScope(t *testing.T) {
 
 	cancel, controller := newController(wf, wftmpl1, wftmpl2)
 	defer cancel()
-	wfcset := controller.wfclientset.ArgoprojV1alpha1().Workflows("default")
 
 	woc := newWorkflowOperationCtx(wf, controller)
 	woc.operate()
 
-	wf, err := wfcset.Get(wf.Name, metav1.GetOptions{})
-	assert.NoError(t, err)
+	wf = woc.wf
 
 	node := findNodeByName(wf.Status.Nodes, "test-template-scope")
 	if assert.NotNil(t, node, "Node %s not found", "test-templte-scope") {