Use cache to retrieve WorkflowTemplates (argoproj#1534)

workflow/controller/controller.go

@@ -7,6 +7,8 @@ import (
 	"time"
 
 	wfv1 "github.com/argoproj/argo/pkg/apis/workflow/v1alpha1"
+	wfextv "github.com/argoproj/argo/pkg/client/informers/externalversions"
+	wfextvv1alpha1 "github.com/argoproj/argo/pkg/client/informers/externalversions/workflow/v1alpha1"
 	log "github.com/sirupsen/logrus"
 	apiv1 "k8s.io/api/core/v1"
 	apierr "k8s.io/apimachinery/pkg/api/errors"
@@ -54,20 +56,22 @@ type WorkflowController struct {
 	wfclientset   wfclientset.Interface
 
 	// datastructures to support the processing of workflows and workflow pods
-	wfInformer    cache.SharedIndexInformer
-	podInformer   cache.SharedIndexInformer
-	wfQueue       workqueue.RateLimitingInterface
-	podQueue      workqueue.RateLimitingInterface
-	completedPods chan string
-	gcPods        chan string // pods to be deleted depend on GC strategy
-	throttler     Throttler
-	wfDBctx       sqldb.DBRepository
+	wfInformer     cache.SharedIndexInformer
+	wftmplInformer wfextvv1alpha1.WorkflowTemplateInformer
+	podInformer    cache.SharedIndexInformer
+	wfQueue        workqueue.RateLimitingInterface
+	podQueue       workqueue.RateLimitingInterface
+	completedPods  chan string
+	gcPods         chan string // pods to be deleted depend on GC strategy
+	throttler      Throttler
+	wfDBctx        sqldb.DBRepository
 }
 
 const (
-	workflowResyncPeriod        = 20 * time.Minute
-	workflowMetricsResyncPeriod = 1 * time.Minute
-	podResyncPeriod             = 30 * time.Minute
+	workflowResyncPeriod         = 20 * time.Minute
+	workflowTemplateResyncPeriod = 20 * time.Minute
+	workflowMetricsResyncPeriod  = 1 * time.Minute
+	podResyncPeriod              = 30 * time.Minute
 )
 
 // NewWorkflowController instantiates a new WorkflowController
@@ -145,16 +149,20 @@ func (wfc *WorkflowController) Run(ctx context.Context, wfWorkers, podWorkers in
 
 	wfc.wfInformer = util.NewWorkflowInformer(wfc.restConfig, wfc.Config.Namespace, workflowResyncPeriod, wfc.tweakWorkflowlist)
 
+	informerFactory := wfextv.NewSharedInformerFactory(wfc.wfclientset, workflowTemplateResyncPeriod)
+	wfc.wftmplInformer = informerFactory.Argoproj().V1alpha1().WorkflowTemplates()
+
 	wfc.addWorkflowInformerHandler()
 	wfc.podInformer = wfc.newPodInformer()
 
 	go wfc.wfInformer.Run(ctx.Done())
+	go wfc.wftmplInformer.Informer().Run(ctx.Done())
 	go wfc.podInformer.Run(ctx.Done())
 	go wfc.podLabeler(ctx.Done())
 	go wfc.podGarbageCollector(ctx.Done())
 
 	// Wait for all involved caches to be synced, before processing items from the queue is started
-	for _, informer := range []cache.SharedIndexInformer{wfc.wfInformer, wfc.podInformer} {
+	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

workflow/controller/controller_test.go

@@ -2,20 +2,24 @@ package controller
 
 import (
 	"bytes"
+	"context"
 	"encoding/json"
 	"io"
 	"io/ioutil"
 	"testing"
+	"time"
 
 	"github.com/ghodss/yaml"
 	"github.com/stretchr/testify/assert"
 	apiv1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/kubernetes/fake"
+	"k8s.io/client-go/tools/cache"
 
 	wfv1 "github.com/argoproj/argo/pkg/apis/workflow/v1alpha1"
 	fakewfclientset "github.com/argoproj/argo/pkg/client/clientset/versioned/fake"
+	wfextv "github.com/argoproj/argo/pkg/client/informers/externalversions"
 	"github.com/argoproj/argo/workflow/config"
 )
 
@@ -42,13 +46,22 @@ spec:
 `
 
 func newController() *WorkflowController {
+	wfclientset := fakewfclientset.NewSimpleClientset()
+	informerFactory := wfextv.NewSharedInformerFactory(wfclientset, 10*time.Minute)
+	wftmplInformer := informerFactory.Argoproj().V1alpha1().WorkflowTemplates()
+	ctx := context.Background()
+	go wftmplInformer.Informer().Run(ctx.Done())
+	if !cache.WaitForCacheSync(ctx.Done(), wftmplInformer.Informer().HasSynced) {
+		panic("Timed out waiting for caches to sync")
+	}
 	return &WorkflowController{
 		Config: config.WorkflowControllerConfig{
 			ExecutorImage: "executor:latest",
 		},
-		kubeclientset: fake.NewSimpleClientset(),
-		wfclientset:   fakewfclientset.NewSimpleClientset(),
-		completedPods: make(chan string, 512),
+		kubeclientset:  fake.NewSimpleClientset(),
+		wfclientset:    wfclientset,
+		completedPods:  make(chan string, 512),
+		wftmplInformer: wftmplInformer,
 	}
 }
 

workflow/controller/operator.go

@@ -114,7 +114,7 @@ func newWorkflowOperationCtx(wf *wfv1.Workflow, wfc *WorkflowController) *wfOper
 		completedPods:      make(map[string]bool),
 		succeededPods:      make(map[string]bool),
 		deadline:           time.Now().UTC().Add(maxOperationTime),
-		tmplCtx:            templateresolution.NewContext(wfc.wfclientset, wf.Namespace, wf),
+		tmplCtx:            templateresolution.NewContext(wfc.wftmplInformer.Lister().WorkflowTemplates(wf.Namespace), wf),
 	}
 
 	if woc.wf.Status.Nodes == nil {

workflow/templateresolution/context.go

@@ -3,7 +3,7 @@ package templateresolution
 import (
 	"github.com/argoproj/argo/errors"
 	wfv1 "github.com/argoproj/argo/pkg/apis/workflow/v1alpha1"
-	wfclientset "github.com/argoproj/argo/pkg/client/clientset/versioned"
+	typed "github.com/argoproj/argo/pkg/client/clientset/versioned/typed/workflow/v1alpha1"
 	"github.com/argoproj/argo/workflow/common"
 	log "github.com/sirupsen/logrus"
 	apierr "k8s.io/apimachinery/pkg/api/errors"
@@ -13,22 +13,43 @@ import (
 // maxResolveDepth is the limit of template reference resolution.
 const maxResolveDepth int = 10
 
+// workflowTemplateInterfaceWrapper is an internal struct to wrap clientset.
+type workflowTemplateInterfaceWrapper struct {
+	clientset typed.WorkflowTemplateInterface
+}
+
+// Get retrieves the WorkflowTemplate of a given name.
+func (wrapper *workflowTemplateInterfaceWrapper) Get(name string) (*wfv1.WorkflowTemplate, error) {
+	return wrapper.clientset.Get(name, metav1.GetOptions{})
+}
+
+// WorkflowTemplateNamespaceLister helps get WorkflowTemplates.
+type WorkflowTemplateNamespacedGetter interface {
+	// Get retrieves the WorkflowTemplate from the indexer for a given name.
+	Get(name string) (*wfv1.WorkflowTemplate, error)
+}
+
 // Context is a context of template search.
 type Context struct {
-	// wfClientset is the clientset to get workflow templates.
-	wfClientset wfclientset.Interface
-	// namespace is the namespace of template search.
-	namespace string
+	// wftmplGetter is an interface to get WorkflowTemplates.
+	wftmplGetter WorkflowTemplateNamespacedGetter
 	// tmplBase is the base of local template search.
 	tmplBase wfv1.TemplateGetter
 }
 
 // NewContext returns new Context.
-func NewContext(wfClientset wfclientset.Interface, namespace string, tmplBase wfv1.TemplateGetter) *Context {
+func NewContext(wftmplGetter WorkflowTemplateNamespacedGetter, tmplBase wfv1.TemplateGetter) *Context {
+	return &Context{
+		wftmplGetter: wftmplGetter,
+		tmplBase:     tmplBase,
+	}
+}
+
+// NewContext returns new Context.
+func NewContextFromClientset(clientset typed.WorkflowTemplateInterface, tmplBase wfv1.TemplateGetter) *Context {
 	return &Context{
-		wfClientset: wfClientset,
-		namespace:   namespace,
-		tmplBase:    tmplBase,
+		wftmplGetter: &workflowTemplateInterfaceWrapper{clientset: clientset},
+		tmplBase:     tmplBase,
 	}
 }
 
@@ -43,7 +64,7 @@ func (ctx *Context) GetTemplateByName(name string) (*wfv1.Template, error) {
 
 // GetTemplateFromRef returns a template found by a given template ref.
 func (ctx *Context) GetTemplateFromRef(tmplRef *wfv1.TemplateRef) (*wfv1.Template, error) {
-	wftmpl, err := ctx.wfClientset.ArgoprojV1alpha1().WorkflowTemplates(ctx.namespace).Get(tmplRef.Name, metav1.GetOptions{})
+	wftmpl, err := ctx.wftmplGetter.Get(tmplRef.Name)
 	if err != nil {
 		if apierr.IsNotFound(err) {
 			return nil, errors.Errorf(errors.CodeNotFound, "workflow template %s not found", tmplRef.Name)
@@ -82,7 +103,7 @@ func (ctx *Context) GetTemplate(tmplHolder wfv1.TemplateHolder) (*wfv1.Template,
 func (ctx *Context) GetTemplateBase(tmplHolder wfv1.TemplateHolder) (wfv1.TemplateGetter, error) {
 	tmplRef := tmplHolder.GetTemplateRef()
 	if tmplRef != nil {
-		wftmpl, err := ctx.wfClientset.ArgoprojV1alpha1().WorkflowTemplates(ctx.namespace).Get(tmplRef.Name, metav1.GetOptions{})
+		wftmpl, err := ctx.wftmplGetter.Get(tmplRef.Name)
 		if err != nil && apierr.IsNotFound(err) {
 			return nil, errors.Errorf(errors.CodeNotFound, "workflow template %s not found", tmplRef.Name)
 		}
@@ -140,5 +161,5 @@ func (ctx *Context) resolveTemplateImpl(tmplHolder wfv1.TemplateHolder, depth in
 
 // WithTemplateBase creates new context with a wfv1.TemplateGetter.
 func (ctx *Context) WithTemplateBase(tmplBase wfv1.TemplateGetter) *Context {
-	return NewContext(ctx.wfClientset, ctx.namespace, tmplBase)
+	return NewContext(ctx.wftmplGetter, tmplBase)
 }

workflow/templateresolution/context_test.go

@@ -107,7 +107,7 @@ spec:
 func TestGetTemplateByName(t *testing.T) {
 	wfClientset := fakewfclientset.NewSimpleClientset()
 	wftmpl := unmarshalWftmpl(baseWorkflowTemplateYaml)
-	ctx := NewContext(wfClientset, metav1.NamespaceDefault, wftmpl)
+	ctx := NewContextFromClientset(wfClientset.ArgoprojV1alpha1().WorkflowTemplates(metav1.NamespaceDefault), wftmpl)
 
 	tmpl, err := ctx.GetTemplateByName("whalesay")
 	if !assert.NoError(t, err) {
@@ -131,7 +131,7 @@ func TestGetTemplateFromRef(t *testing.T) {
 		t.Fatal(err)
 	}
 	wftmpl := unmarshalWftmpl(baseWorkflowTemplateYaml)
-	ctx := NewContext(wfClientset, metav1.NamespaceDefault, wftmpl)
+	ctx := NewContextFromClientset(wfClientset.ArgoprojV1alpha1().WorkflowTemplates(metav1.NamespaceDefault), wftmpl)
 
 	// Get the template of existing template reference.
 	tmplRef := wfv1.TemplateRef{Name: "some-workflow-template", Template: "whalesay"}
@@ -164,7 +164,7 @@ func TestGetTemplate(t *testing.T) {
 		t.Fatal(err)
 	}
 	wftmpl := unmarshalWftmpl(baseWorkflowTemplateYaml)
-	ctx := NewContext(wfClientset, metav1.NamespaceDefault, wftmpl)
+	ctx := NewContextFromClientset(wfClientset.ArgoprojV1alpha1().WorkflowTemplates(metav1.NamespaceDefault), wftmpl)
 
 	// Get the template of existing template name.
 	tmplHolder := wfv1.Template{Template: "whalesay"}
@@ -211,7 +211,7 @@ func TestGetTemplateBase(t *testing.T) {
 		t.Fatal(err)
 	}
 	wftmpl := unmarshalWftmpl(baseWorkflowTemplateYaml)
-	ctx := NewContext(wfClientset, metav1.NamespaceDefault, wftmpl)
+	ctx := NewContextFromClientset(wfClientset.ArgoprojV1alpha1().WorkflowTemplates(metav1.NamespaceDefault), wftmpl)
 
 	// Get the template base of existing template name.
 	tmplHolder := wfv1.Template{Template: "whalesay"}
@@ -266,7 +266,7 @@ func TestResolveTemplate(t *testing.T) {
 		t.Fatal(err)
 	}
 	wftmpl := unmarshalWftmpl(baseWorkflowTemplateYaml)
-	ctx := NewContext(wfClientset, metav1.NamespaceDefault, wftmpl)
+	ctx := NewContextFromClientset(wfClientset.ArgoprojV1alpha1().WorkflowTemplates(metav1.NamespaceDefault), wftmpl)
 
 	// Get the template of template name.
 	tmplHolder := wfv1.Template{Template: "whalesay"}

workflow/validate/validate.go

@@ -85,8 +85,9 @@ func ValidateWorkflow(wfClientset wfclientset.Interface, namespace string, wf *w
 	if wf.Namespace != "" {
 		namespace = wf.Namespace
 	}
+
 	ctx := newTemplateValidationCtx(wfClientset, namespace, wf, opts)
-	tmplCtx := templateresolution.NewContext(wfClientset, namespace, wf)
+	tmplCtx := templateresolution.NewContextFromClientset(wfClientset.ArgoprojV1alpha1().WorkflowTemplates(namespace), wf)
 
 	err := validateWorkflowFieldNames(wf.Spec.Templates)
 	if err != nil {
@@ -159,7 +160,7 @@ func ValidateWorkflowTemplate(wfClientset wfclientset.Interface, namespace strin
 		namespace = wftmpl.Namespace
 	}
 	ctx := newTemplateValidationCtx(wfClientset, namespace, wftmpl, ValidateOpts{})
-	tmplCtx := templateresolution.NewContext(wfClientset, namespace, wftmpl)
+	tmplCtx := templateresolution.NewContextFromClientset(wfClientset.ArgoprojV1alpha1().WorkflowTemplates(namespace), wftmpl)
 
 	// Check if all templates can be resolved.
 	for _, template := range wftmpl.Spec.Templates {