Repository: argoproj/argo (go)

Issue:
Make the controller tolerant to malformed resources
Currently, if you `kubectl apply -f malformed-cronworkflow.yaml` (or workflow template or cluster workflow template) we cannot list those resources, see #3666.

The workaround is delete the malformed resource. However, in a multi-tenant environment, this means a single malformed resource could other user's workflows.

# Analysis

@jessesuen you asked about whether or not having malformed resources (i.e. kubectl applied) would impact the controller. I loaded the malformed resources (1x cron workflow, 1x template template, 1x cluster workflow template, 1x workflow) and I could still create workflows directly.

Problems:

* I created another valid cron workflow - this did not schedule. 
* I could not create a workflow from a valid workflow template
* I could not create a workflow from a valid cluster workflow template.

These resolved after deleting the resource - I did not need to restart the container.

# Diagnostics

```
 controller | E0805 11:42:30.478792   33530 reflector.go:153] pkg/mod/k8s.io/client-go@v0.17.5/tools/cache/reflector.go:105: Failed to list *v1alpha1.CronWorkflow: v1alpha1.CronWorkflowList.Items: []v1alpha1.CronWorkflow: v1alpha1.CronWorkflow.Spec: v1alpha1.CronWorkflowSpec.WorkflowSpec: v1alpha1.WorkflowSpec.Arguments: v1alpha1.Arguments.Parameters: []v1alpha1.Parameter: decode slice: expect [ or n, but found {, error found in #10 byte of ...|ameters":{"someParam|..., bigger context ...|spec":{"workflowSpec":{"arguments":{"parameters":{"someParam":"Hello, world!"}}}}}],"kind":"CronWork|...
 controller | E0805 11:42:30.505177   33530 reflector.go:153] pkg/mod/k8s.io/client-go@v0.17.5/tools/cache/reflector.go:105: Failed to list *v1alpha1.WorkflowTemplate: v1alpha1.WorkflowTemplateList.Items: v1alpha1.WorkflowTemplates: v1alpha1.WorkflowTemplate.Spec: v1alpha1.WorkflowTemplateSpec.WorkflowSpec: Arguments: v1alpha1.Arguments.Parameters: []v1alpha1.Parameter: decode slice: expect [ or n, but found {, error found in #10 byte of ...|ameters":{"someParam|..., bigger context ...|e81e75032120"},"spec":{"arguments":{"parameters":{"someParam":"Hello, world!"}}}}],"kind":"WorkflowT|...
```

https://github.com/kubernetes/kubernetes/issues/57705
https://github.com/argoproj/argo/issues/632

Pull request:
fix(controller): Tolerate malformed resources. Fixes #3677
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] 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. 
* [x] My organization is added to [USERS.md](https://github.com/argoproj/argo/blob/master/USERS.md).

Fixes #3677

# Summary

This PR replaces the standard Kubernetes informer with a dynamic informer, where the code ignores malformed resources.

 # Changes

The approach differs slightly for workflow templates and cron workflows. Cron is simpler to directly replace the informer with our own informer, keeping the changes simpler. With workflow templates, the code wraps the informers in a "tolerant" translation layer that ignores malformed resources for list operations.

# Testing

This is tested using e2e tests. 

Patch:
diff --git a/cmd/workflow-controller/main.go b/cmd/workflow-controller/main.go
index f0eea34dc..73bd19022 100644
--- a/cmd/workflow-controller/main.go
+++ b/cmd/workflow-controller/main.go
@@ -7,6 +7,7 @@ import (
 	"time"
 
 	"github.com/argoproj/pkg/cli"
+	"github.com/argoproj/pkg/errors"
 	kubecli "github.com/argoproj/pkg/kube/cli"
 	"github.com/argoproj/pkg/stats"
 	log "github.com/sirupsen/logrus"
@@ -78,7 +79,8 @@ func NewRootCommand() *cobra.Command {
 			}
 
 			// start a controller on instances of our custom resource
-			wfController := controller.NewWorkflowController(config, kubeclientset, wfclientset, namespace, managedNamespace, executorImage, executorImagePullPolicy, containerRuntimeExecutor, configMap)
+			wfController, err := controller.NewWorkflowController(config, kubeclientset, wfclientset, namespace, managedNamespace, executorImage, executorImagePullPolicy, containerRuntimeExecutor, configMap)
+			errors.CheckError(err)
 			ctx, cancel := context.WithCancel(context.Background())
 			defer cancel()
 
diff --git a/pkg/apis/workflow/register.go b/pkg/apis/workflow/register.go
index 066bf1593..08649ff20 100644
--- a/pkg/apis/workflow/register.go
+++ b/pkg/apis/workflow/register.go
@@ -3,6 +3,7 @@ package workflow
 // Workflow constants
 const (
 	Group                            string = "argoproj.io"
+	Version                          string = "v1alpha1"
 	WorkflowKind                     string = "Workflow"
 	WorkflowSingular                 string = "workflow"
 	WorkflowPlural                   string = "workflows"
diff --git a/workflow/controller/controller.go b/workflow/controller/controller.go
index 790b40dbb..ef344ddb1 100644
--- a/workflow/controller/controller.go
+++ b/workflow/controller/controller.go
@@ -21,6 +21,7 @@ import (
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/apimachinery/pkg/watch"
+	"k8s.io/client-go/dynamic"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/rest"
 	"k8s.io/client-go/tools/cache"
@@ -33,11 +34,11 @@ import (
 	"github.com/argoproj/argo/persist/sqldb"
 	wfv1 "github.com/argoproj/argo/pkg/apis/workflow/v1alpha1"
 	wfclientset "github.com/argoproj/argo/pkg/client/clientset/versioned"
-	wfextv "github.com/argoproj/argo/pkg/client/informers/externalversions"
 	wfextvv1alpha1 "github.com/argoproj/argo/pkg/client/informers/externalversions/workflow/v1alpha1"
 	authutil "github.com/argoproj/argo/util/auth"
 	"github.com/argoproj/argo/workflow/common"
 	controllercache "github.com/argoproj/argo/workflow/controller/cache"
+	"github.com/argoproj/argo/workflow/controller/informer"
 	"github.com/argoproj/argo/workflow/controller/pod"
 	"github.com/argoproj/argo/workflow/cron"
 	"github.com/argoproj/argo/workflow/hydrator"
@@ -65,9 +66,10 @@ type WorkflowController struct {
 	containerRuntimeExecutor   string
 
 	// restConfig is used by controller to send a SIGUSR1 to the wait sidecar using remotecommand.NewSPDYExecutor().
-	restConfig    *rest.Config
-	kubeclientset kubernetes.Interface
-	wfclientset   wfclientset.Interface
+	restConfig       *rest.Config
+	kubeclientset    kubernetes.Interface
+	dynamicInterface dynamic.Interface
+	wfclientset      wfclientset.Interface
 
 	// datastructures to support the processing of workflows and workflow pods
 	wfInformer            cache.SharedIndexInformer
@@ -98,10 +100,16 @@ const (
 )
 
 // NewWorkflowController instantiates a new WorkflowController
-func NewWorkflowController(restConfig *rest.Config, kubeclientset kubernetes.Interface, wfclientset wfclientset.Interface, namespace string, managedNamespace string, executorImage, executorImagePullPolicy, containerRuntimeExecutor, configMap string) *WorkflowController {
+func NewWorkflowController(restConfig *rest.Config, kubeclientset kubernetes.Interface, wfclientset wfclientset.Interface, namespace, managedNamespace, executorImage, executorImagePullPolicy, containerRuntimeExecutor, configMap string) (*WorkflowController, error) {
+	dynamicInterface, err := dynamic.NewForConfig(restConfig)
+	if err != nil {
+		return nil, err
+	}
+
 	wfc := WorkflowController{
 		restConfig:                 restConfig,
 		kubeclientset:              kubeclientset,
+		dynamicInterface:           dynamicInterface,
 		wfclientset:                wfclientset,
 		namespace:                  namespace,
 		managedNamespace:           managedNamespace,
@@ -125,7 +133,7 @@ func NewWorkflowController(restConfig *rest.Config, kubeclientset kubernetes.Int
 	wfc.throttler.SetParallelism(wfc.getParallelism())
 	wfc.podQueue = workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "pod_queue")
 
-	return &wfc
+	return &wfc, nil
 }
 
 // RunTTLController runs the workflow TTL controller
@@ -138,7 +146,7 @@ func (wfc *WorkflowController) runTTLController(ctx context.Context) {
 }
 
 func (wfc *WorkflowController) runCronController(ctx context.Context) {
-	cronController := cron.NewCronController(wfc.wfclientset, wfc.restConfig, wfc.namespace, wfc.GetManagedNamespace(), wfc.Config.InstanceID, wfc.metrics)
+	cronController := cron.NewCronController(wfc.wfclientset, wfc.restConfig, wfc.dynamicInterface, wfc.namespace, wfc.GetManagedNamespace(), wfc.Config.InstanceID, wfc.metrics)
 	cronController.Run(ctx)
 }
 
@@ -151,7 +159,7 @@ func (wfc *WorkflowController) Run(ctx context.Context, wfWorkers, podWorkers in
 	log.Infof("Workers: workflow: %d, pod: %d", wfWorkers, podWorkers)
 
 	wfc.wfInformer = util.NewWorkflowInformer(wfc.restConfig, wfc.GetManagedNamespace(), workflowResyncPeriod, wfc.tweakListOptions)
-	wfc.wftmplInformer = wfc.newWorkflowTemplateInformer()
+	wfc.wftmplInformer = informer.NewTolerantWorkflowTemplateInformer(wfc.dynamicInterface, workflowTemplateResyncPeriod, wfc.namespace)
 
 	wfc.addWorkflowInformerHandlers()
 	wfc.podInformer = wfc.newPodInformer()
@@ -243,7 +251,7 @@ func (wfc *WorkflowController) createClusterWorkflowTemplateInformer(ctx context
 	errors.CheckError(err)
 
 	if cwftGetAllowed && cwftListAllowed && cwftWatchAllowed {
-		wfc.cwftmplInformer = wfc.newClusterWorkflowTemplateInformer()
+		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")
@@ -775,14 +783,6 @@ func (wfc *WorkflowController) setWorkflowDefaults(wf *wfv1.Workflow) error {
 	return nil
 }
 
-func (wfc *WorkflowController) newWorkflowTemplateInformer() wfextvv1alpha1.WorkflowTemplateInformer {
-	return wfextv.NewSharedInformerFactoryWithOptions(wfc.wfclientset, workflowTemplateResyncPeriod, wfextv.WithNamespace(wfc.GetManagedNamespace())).Argoproj().V1alpha1().WorkflowTemplates()
-}
-
-func (wfc *WorkflowController) newClusterWorkflowTemplateInformer() wfextvv1alpha1.ClusterWorkflowTemplateInformer {
-	return wfextv.NewSharedInformerFactoryWithOptions(wfc.wfclientset, clusterWorkflowTemplateResyncPeriod).Argoproj().V1alpha1().ClusterWorkflowTemplates()
-}
-
 func (wfc *WorkflowController) GetManagedNamespace() string {
 	if wfc.managedNamespace != "" {
 		return wfc.managedNamespace
diff --git a/workflow/controller/informer/cluster_workflow_template_convert.go b/workflow/controller/informer/cluster_workflow_template_convert.go
new file mode 100644
index 000000000..959474bdf
--- /dev/null
+++ b/workflow/controller/informer/cluster_workflow_template_convert.go
@@ -0,0 +1,33 @@
+package informer
+
+import (
+	"fmt"
+	"reflect"
+
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"k8s.io/apimachinery/pkg/runtime"
+
+	wfv1 "github.com/argoproj/argo/pkg/apis/workflow/v1alpha1"
+)
+
+// this function always tries to return a value, even if it is badly formed
+func objectToClusterWorkflowTemplate(object runtime.Object) (*wfv1.ClusterWorkflowTemplate, error) {
+	v := &wfv1.ClusterWorkflowTemplate{}
+	un, ok := object.(*unstructured.Unstructured)
+	if !ok {
+		return v, fmt.Errorf("malformed cluster workflow template: expected \"*unstructured.Unstructured\", got \"%s\"", reflect.TypeOf(object).Name())
+	}
+	err := runtime.DefaultUnstructuredConverter.FromUnstructured(un.Object, v)
+	if err != nil {
+		return v, fmt.Errorf("malformed cluster workflow template \"%s\": %w", un.GetName(), err)
+	}
+	return v, nil
+}
+
+func objectsToClusterWorkflowTemplates(list []runtime.Object) []*wfv1.ClusterWorkflowTemplate {
+	ret := make([]*wfv1.ClusterWorkflowTemplate, len(list))
+	for i, object := range list {
+		ret[i], _ = objectToClusterWorkflowTemplate(object)
+	}
+	return ret
+}
diff --git a/workflow/controller/informer/tolerant_cluster_workflow_template_informer.go b/workflow/controller/informer/tolerant_cluster_workflow_template_informer.go
new file mode 100644
index 000000000..6cf5d82d6
--- /dev/null
+++ b/workflow/controller/informer/tolerant_cluster_workflow_template_informer.go
@@ -0,0 +1,34 @@
+package informer
+
+import (
+	"time"
+
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/client-go/dynamic"
+	"k8s.io/client-go/dynamic/dynamicinformer"
+	"k8s.io/client-go/informers"
+	"k8s.io/client-go/tools/cache"
+
+	"github.com/argoproj/argo/pkg/apis/workflow"
+	extwfv1 "github.com/argoproj/argo/pkg/client/informers/externalversions/workflow/v1alpha1"
+	"github.com/argoproj/argo/pkg/client/listers/workflow/v1alpha1"
+)
+
+type tolerantClusterWorkflowTemplateInformer struct {
+	delegate informers.GenericInformer
+}
+
+// a drop-in replacement for `extwfv1.ClusterWorkflowTemplateInformer` that ignores malformed resources
+func NewTolerantClusterWorkflowTemplateInformer(dynamicInterface dynamic.Interface, defaultResync time.Duration) extwfv1.ClusterWorkflowTemplateInformer {
+	return &tolerantClusterWorkflowTemplateInformer{delegate: dynamicinformer.NewFilteredDynamicSharedInformerFactory(dynamicInterface, defaultResync, "", func(options *metav1.ListOptions) {}).
+		ForResource(schema.GroupVersionResource{Group: workflow.Group, Version: workflow.Version, Resource: workflow.ClusterWorkflowTemplatePlural})}
+}
+
+func (t *tolerantClusterWorkflowTemplateInformer) Informer() cache.SharedIndexInformer {
+	return t.delegate.Informer()
+}
+
+func (t *tolerantClusterWorkflowTemplateInformer) Lister() v1alpha1.ClusterWorkflowTemplateLister {
+	return &tolerantClusterWorkflowTemplateLister{delegate: t.delegate.Lister()}
+}
diff --git a/workflow/controller/informer/tolerant_cluster_workflow_template_lister.go b/workflow/controller/informer/tolerant_cluster_workflow_template_lister.go
new file mode 100644
index 000000000..e16fcbd4f
--- /dev/null
+++ b/workflow/controller/informer/tolerant_cluster_workflow_template_lister.go
@@ -0,0 +1,25 @@
+package informer
+
+import (
+	v1Label "k8s.io/apimachinery/pkg/labels"
+	"k8s.io/client-go/tools/cache"
+
+	wfv1 "github.com/argoproj/argo/pkg/apis/workflow/v1alpha1"
+)
+
+type tolerantClusterWorkflowTemplateLister struct {
+	delegate cache.GenericLister
+}
+
+func (t *tolerantClusterWorkflowTemplateLister) List(selector v1Label.Selector) ([]*wfv1.ClusterWorkflowTemplate, error) {
+	list, err := t.delegate.List(selector)
+	return objectsToClusterWorkflowTemplates(list), err
+}
+
+func (t *tolerantClusterWorkflowTemplateLister) Get(name string) (*wfv1.ClusterWorkflowTemplate, error) {
+	object, err := t.delegate.Get(name)
+	if err != nil {
+		return nil, err
+	}
+	return objectToClusterWorkflowTemplate(object)
+}
diff --git a/workflow/controller/informer/tolerant_workflow_template_informer.go b/workflow/controller/informer/tolerant_workflow_template_informer.go
new file mode 100644
index 000000000..10a72439a
--- /dev/null
+++ b/workflow/controller/informer/tolerant_workflow_template_informer.go
@@ -0,0 +1,34 @@
+package informer
+
+import (
+	"time"
+
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/client-go/dynamic"
+	"k8s.io/client-go/dynamic/dynamicinformer"
+	"k8s.io/client-go/informers"
+	"k8s.io/client-go/tools/cache"
+
+	"github.com/argoproj/argo/pkg/apis/workflow"
+	extwfv1 "github.com/argoproj/argo/pkg/client/informers/externalversions/workflow/v1alpha1"
+	"github.com/argoproj/argo/pkg/client/listers/workflow/v1alpha1"
+)
+
+type tolerantWorkflowTemplateInformer struct {
+	delegate informers.GenericInformer
+}
+
+// a drop-in replacement for `extwfv1.WorkflowTemplateInformer` that ignores malformed resources
+func NewTolerantWorkflowTemplateInformer(dynamicInterface dynamic.Interface, defaultResync time.Duration, namespace string) extwfv1.WorkflowTemplateInformer {
+	return &tolerantWorkflowTemplateInformer{delegate: dynamicinformer.NewFilteredDynamicSharedInformerFactory(dynamicInterface, defaultResync, namespace, func(options *metav1.ListOptions) {}).
+		ForResource(schema.GroupVersionResource{Group: workflow.Group, Version: workflow.Version, Resource: workflow.WorkflowTemplatePlural})}
+}
+
+func (t *tolerantWorkflowTemplateInformer) Informer() cache.SharedIndexInformer {
+	return t.delegate.Informer()
+}
+
+func (t *tolerantWorkflowTemplateInformer) Lister() v1alpha1.WorkflowTemplateLister {
+	return &tolerantWorkflowTemplateLister{delegate: t.delegate.Lister()}
+}
diff --git a/workflow/controller/informer/tolerant_workflow_template_lister.go b/workflow/controller/informer/tolerant_workflow_template_lister.go
new file mode 100644
index 000000000..f954b20fa
--- /dev/null
+++ b/workflow/controller/informer/tolerant_workflow_template_lister.go
@@ -0,0 +1,24 @@
+package informer
+
+import (
+	v1Label "k8s.io/apimachinery/pkg/labels"
+	"k8s.io/client-go/tools/cache"
+
+	wfv1 "github.com/argoproj/argo/pkg/apis/workflow/v1alpha1"
+	"github.com/argoproj/argo/pkg/client/listers/workflow/v1alpha1"
+)
+
+type tolerantWorkflowTemplateLister struct {
+	delegate cache.GenericLister
+}
+
+var _ v1alpha1.WorkflowTemplateLister = &tolerantWorkflowTemplateLister{}
+
+func (t *tolerantWorkflowTemplateLister) List(selector v1Label.Selector) ([]*wfv1.WorkflowTemplate, error) {
+	list, err := t.delegate.List(selector)
+	return objectsToWorkflowTemplates(list), err
+}
+
+func (t *tolerantWorkflowTemplateLister) WorkflowTemplates(namespace string) v1alpha1.WorkflowTemplateNamespaceLister {
+	return &tolerantWorkflowTemplateNamespaceLister{delegate: t.delegate.ByNamespace(namespace)}
+}
diff --git a/workflow/controller/informer/tolerant_workflow_template_namespace_lister.co.go b/workflow/controller/informer/tolerant_workflow_template_namespace_lister.co.go
new file mode 100644
index 000000000..c719c0075
--- /dev/null
+++ b/workflow/controller/informer/tolerant_workflow_template_namespace_lister.co.go
@@ -0,0 +1,28 @@
+package informer
+
+import (
+	v1Label "k8s.io/apimachinery/pkg/labels"
+	"k8s.io/client-go/tools/cache"
+
+	wfv1 "github.com/argoproj/argo/pkg/apis/workflow/v1alpha1"
+	"github.com/argoproj/argo/pkg/client/listers/workflow/v1alpha1"
+)
+
+type tolerantWorkflowTemplateNamespaceLister struct {
+	delegate cache.GenericNamespaceLister
+}
+
+var _ v1alpha1.WorkflowTemplateNamespaceLister = &tolerantWorkflowTemplateNamespaceLister{}
+
+func (t *tolerantWorkflowTemplateNamespaceLister) Get(name string) (*wfv1.WorkflowTemplate, error) {
+	object, err := t.delegate.Get(name)
+	if err != nil {
+		return nil, err
+	}
+	return objectToWorkflowTemplate(object)
+}
+
+func (t *tolerantWorkflowTemplateNamespaceLister) List(selector v1Label.Selector) ([]*wfv1.WorkflowTemplate, error) {
+	list, err := t.delegate.List(selector)
+	return objectsToWorkflowTemplates(list), err
+}
diff --git a/workflow/controller/informer/workflow_template_convert.go b/workflow/controller/informer/workflow_template_convert.go
new file mode 100644
index 000000000..fde6acf39
--- /dev/null
+++ b/workflow/controller/informer/workflow_template_convert.go
@@ -0,0 +1,32 @@
+package informer
+
+import (
+	"fmt"
+	"reflect"
+
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"k8s.io/apimachinery/pkg/runtime"
+
+	wfv1 "github.com/argoproj/argo/pkg/apis/workflow/v1alpha1"
+)
+
+func objectToWorkflowTemplate(object runtime.Object) (*wfv1.WorkflowTemplate, error) {
+	v := &wfv1.WorkflowTemplate{}
+	un, ok := object.(*unstructured.Unstructured)
+	if !ok {
+		return v, fmt.Errorf("malformed workflow template: expected \"*unstructured.Unstructured\", got \"%s\"", reflect.TypeOf(object).Name())
+	}
+	err := runtime.DefaultUnstructuredConverter.FromUnstructured(un.Object, v)
+	if err != nil {
+		return v, fmt.Errorf("malformed workflow template \"%s/%s\": %w", un.GetNamespace(), un.GetName(), err)
+	}
+	return v, nil
+}
+
+func objectsToWorkflowTemplates(list []runtime.Object) []*wfv1.WorkflowTemplate {
+	ret := make([]*wfv1.WorkflowTemplate, len(list))
+	for i, object := range list {
+		ret[i], _ = objectToWorkflowTemplate(object)
+	}
+	return ret
+}
diff --git a/workflow/cron/controller.go b/workflow/cron/controller.go
index 57b7a35e9..426a99eb7 100644
--- a/workflow/cron/controller.go
+++ b/workflow/cron/controller.go
@@ -3,6 +3,7 @@ package cron
 import (
 	"context"
 	"fmt"
+	"reflect"
 	"sync"
 	"time"
 
@@ -12,16 +13,20 @@ import (
 	"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/runtime/schema"
 	"k8s.io/apimachinery/pkg/selection"
 	"k8s.io/apimachinery/pkg/util/wait"
+	"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"
 
+	"github.com/argoproj/argo/pkg/apis/workflow"
 	"github.com/argoproj/argo/pkg/apis/workflow/v1alpha1"
 	"github.com/argoproj/argo/pkg/client/clientset/versioned"
-	"github.com/argoproj/argo/pkg/client/informers/externalversions"
-	extv1alpha1 "github.com/argoproj/argo/pkg/client/informers/externalversions/workflow/v1alpha1"
 	"github.com/argoproj/argo/workflow/common"
 	"github.com/argoproj/argo/workflow/metrics"
 	"github.com/argoproj/argo/workflow/util"
@@ -39,9 +44,10 @@ type Controller struct {
 	wfInformer         cache.SharedIndexInformer
 	wfLister           util.WorkflowLister
 	wfQueue            workqueue.RateLimitingInterface
-	cronWfInformer     extv1alpha1.CronWorkflowInformer
+	cronWfInformer     informers.GenericInformer
 	cronWfQueue        workqueue.RateLimitingInterface
 	restConfig         *rest.Config
+	dynamicInterface   dynamic.Interface
 	metrics            *metrics.Metrics
 }
 
@@ -51,14 +57,7 @@ const (
 	cronWorkflowWorkflowWorkers = 2
 )
 
-func NewCronController(
-	wfclientset versioned.Interface,
-	restConfig *rest.Config,
-	namespace string,
-	managedNamespace string,
-	instanceId string,
-	metrics *metrics.Metrics,
-) *Controller {
+func NewCronController(wfclientset versioned.Interface, restConfig *rest.Config, dynamicInterface dynamic.Interface, namespace string, managedNamespace string, instanceId string, metrics *metrics.Metrics) *Controller {
 	return &Controller{
 		wfClientset:        wfclientset,
 		namespace:          namespace,
@@ -66,6 +65,7 @@ func NewCronController(
 		instanceId:         instanceId,
 		cron:               cron.New(),
 		restConfig:         restConfig,
+		dynamicInterface:   dynamicInterface,
 		nameEntryIDMap:     make(map[string]cron.EntryID),
 		nameEntryIDMapLock: &sync.Mutex{},
 		wfQueue:            workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "wf_cron_queue"),
@@ -82,10 +82,9 @@ func (cc *Controller) Run(ctx context.Context) {
 		log.Infof("...with InstanceID: %s", cc.instanceId)
 	}
 
-	cc.cronWfInformer = externalversions.NewSharedInformerFactoryWithOptions(cc.wfClientset, cronWorkflowResyncPeriod, externalversions.WithNamespace(cc.managedNamespace),
-		externalversions.WithTweakListOptions(func(options *v1.ListOptions) {
-			cronWfInformerListOptionsFunc(options, cc.instanceId)
-		})).Argoproj().V1alpha1().CronWorkflows()
+	cc.cronWfInformer = dynamicinformer.NewFilteredDynamicSharedInformerFactory(cc.dynamicInterface, cronWorkflowResyncPeriod, cc.managedNamespace, func(options *v1.ListOptions) {
+		cronWfInformerListOptionsFunc(options, cc.instanceId)
+	}).ForResource(schema.GroupVersionResource{Group: workflow.Group, Version: workflow.Version, Resource: workflow.CronWorkflowPlural})
 	cc.addCronWorkflowInformerHandler()
 
 	cc.wfInformer = util.NewWorkflowInformer(cc.restConfig, cc.managedNamespace, cronWorkflowResyncPeriod, func(options *v1.ListOptions) {
@@ -141,9 +140,15 @@ func (cc *Controller) processNextCronItem() bool {
 		return true
 	}
 
-	cronWf, ok := obj.(*v1alpha1.CronWorkflow)
+	un, ok := obj.(*unstructured.Unstructured)
 	if !ok {
-		log.Warnf("Key '%s' in index is not a CronWorkflow", key)
+		log.WithField("key", key).Errorf("malformed cluster workflow template: expected *unstructured.Unstructured, got %s", reflect.TypeOf(obj).Name())
+		return true
+	}
+	cronWf := &v1alpha1.CronWorkflow{}
+	err = runtime.DefaultUnstructuredConverter.FromUnstructured(un.Object, cronWf)
+	if err != nil {
+		log.WithError(err).WithField("key", key).Error("malformed cron workflow: could not convert from unstructured")
 		return true
 	}
 
@@ -151,7 +156,7 @@ func (cc *Controller) processNextCronItem() bool {
 
 	err = cronWorkflowOperationCtx.validateCronWorkflow()
 	if err != nil {
-		log.WithError(err).Error("invalid cron workflow")
+		log.WithError(err).WithField("key", key).Error("invalid cron workflow")
 		return true
 	}
 


Test patch:
diff --git a/test/e2e/fixtures/e2e_suite.go b/test/e2e/fixtures/e2e_suite.go
index e08f45eaa..a9c83a3da 100644
--- a/test/e2e/fixtures/e2e_suite.go
+++ b/test/e2e/fixtures/e2e_suite.go
@@ -96,6 +96,7 @@ func (s *E2ESuite) countWorkflows() int {
 }
 
 func (s *E2ESuite) DeleteResources(label string) {
+
 	// delete all cron workflows
 	cronList, err := s.cronClient.List(metav1.ListOptions{LabelSelector: label})
 	s.CheckError(err)
diff --git a/test/e2e/fixtures/given.go b/test/e2e/fixtures/given.go
index 29d8daef5..3fe5822bd 100644
--- a/test/e2e/fixtures/given.go
+++ b/test/e2e/fixtures/given.go
@@ -5,6 +5,7 @@ import (
 	"strings"
 	"testing"
 
+	"github.com/stretchr/testify/assert"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/client-go/kubernetes"
 	"sigs.k8s.io/yaml"
@@ -135,14 +136,13 @@ func (g *Given) CronWorkflow(text string) *Given {
 	return g
 }
 
-func (g *Given) RunCli(args []string, block func(t *testing.T, output string, err error)) *Given {
+var NoError = func(t *testing.T, output string, err error) {
+	assert.NoError(t, err, output)
+}
 
+func (g *Given) Exec(name string, args []string, block func(t *testing.T, output string, err error)) *Given {
 	g.t.Helper()
-	if !strings.Contains(strings.Join(args, ","), "-n") {
-		args = append([]string{"-n", Namespace}, args...)
-	}
-	output, err := runCli("../../dist/argo", args...)
-
+	output, err := Exec(name, args...)
 	block(g.t, output, err)
 	if g.t.Failed() {
 		g.t.FailNow()
@@ -150,6 +150,10 @@ func (g *Given) RunCli(args []string, block func(t *testing.T, output string, er
 	return g
 }
 
+func (g *Given) RunCli(args []string, block func(t *testing.T, output string, err error)) *Given {
+	return g.Exec("../../dist/argo", append([]string{"-n", Namespace}, args...), block)
+}
+
 func (g *Given) ClusterWorkflowTemplate(text string) *Given {
 	g.t.Helper()
 	cwfTemplate := &wfv1.ClusterWorkflowTemplate{}
diff --git a/test/e2e/fixtures/then.go b/test/e2e/fixtures/then.go
index 001c184c2..216c4d2b7 100644
--- a/test/e2e/fixtures/then.go
+++ b/test/e2e/fixtures/then.go
@@ -121,7 +121,7 @@ func (t *Then) ExpectAuditEvents(blocks ...func(*testing.T, apiv1.Event)) *Then
 
 func (t *Then) RunCli(args []string, block func(t *testing.T, output string, err error)) *Then {
 	t.t.Helper()
-	output, err := runCli("../../dist/argo", append([]string{"-n", Namespace}, args...)...)
+	output, err := Exec("../../dist/argo", append([]string{"-n", Namespace}, args...)...)
 	block(t.t, output, err)
 	if t.t.Failed() {
 		t.t.FailNow()
diff --git a/test/e2e/fixtures/util.go b/test/e2e/fixtures/util.go
index 3c9d105a1..cca312845 100644
--- a/test/e2e/fixtures/util.go
+++ b/test/e2e/fixtures/util.go
@@ -15,7 +15,7 @@ import (
 	"k8s.io/client-go/kubernetes/scheme"
 )
 
-func runCli(name string, args ...string) (string, error) {
+func Exec(name string, args ...string) (string, error) {
 	cmd := exec.Command(name, args...)
 	cmd.Env = os.Environ()
 	log.Info(cmd.String())
diff --git a/test/e2e/fixtures/when.go b/test/e2e/fixtures/when.go
index 76948c89d..1d66bff40 100644
--- a/test/e2e/fixtures/when.go
+++ b/test/e2e/fixtures/when.go
@@ -4,7 +4,6 @@ import (
 	"testing"
 	"time"
 
-	"github.com/argoproj/pkg/humanize"
 	log "github.com/sirupsen/logrus"
 	corev1 "k8s.io/api/core/v1"
 	apierr "k8s.io/apimachinery/pkg/api/errors"
@@ -198,7 +197,7 @@ func (w *When) WaitForWorkflowName(workflowName string, timeout time.Duration) *
 
 func (w *When) Wait(timeout time.Duration) *When {
 	w.t.Helper()
-	log.Infof("Waiting for %s", humanize.Duration(timeout))
+	log.Infof("Waiting for %v", timeout)
 	time.Sleep(timeout)
 	log.Infof("Done waiting")
 	return w
@@ -225,7 +224,7 @@ func (w *When) And(block func()) *When {
 
 func (w *When) RunCli(args []string, block func(t *testing.T, output string, err error)) *When {
 	w.t.Helper()
-	output, err := runCli("../../dist/argo", append([]string{"-n", Namespace}, args...)...)
+	output, err := Exec("../../dist/argo", append([]string{"-n", Namespace}, args...)...)
 	block(w.t, output, err)
 	if w.t.Failed() {
 		w.t.FailNow()
diff --git a/test/e2e/malformed_resources_test.go b/test/e2e/malformed_resources_test.go
new file mode 100644
index 000000000..9487cdd6d
--- /dev/null
+++ b/test/e2e/malformed_resources_test.go
@@ -0,0 +1,116 @@
+// +build e2e
+
+package e2e
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/suite"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+
+	wfv1 "github.com/argoproj/argo/pkg/apis/workflow/v1alpha1"
+	"github.com/argoproj/argo/test/e2e/fixtures"
+	"github.com/argoproj/argo/workflow/common"
+)
+
+type MalformedResourcesSuite struct {
+	fixtures.E2ESuite
+}
+
+func (s *MalformedResourcesSuite) AfterTest(suiteName, testName string) {
+
+	// delete any malformed items first, as they'll break later clean-up
+	_, err := fixtures.Exec("kubectl", "-n", fixtures.Namespace, "delete", "workflows,workflowtemplates,clusterworkflowtemplates,cronworkflows", "-l", "argo-e2e=malformed")
+	s.CheckError(err)
+
+	s.E2ESuite.AfterTest(suiteName, testName)
+}
+
+func (s *MalformedResourcesSuite) TestMalformedWorkflow() {
+	s.Given().
+		Exec("kubectl", []string{"apply", "-f", "testdata/malformed/malformed-workflow.yaml"}, fixtures.NoError).
+		WorkflowName("malformed").
+		When().
+		// it is not possible to wait for this to finish, because it is malformed
+		Wait(15 * time.Second).
+		Then().
+		ExpectWorkflow(func(t *testing.T, metadata *metav1.ObjectMeta, status *wfv1.WorkflowStatus) {
+			assert.Equal(t, "malformed", metadata.Name)
+			assert.Equal(t, wfv1.NodeFailed, status.Phase)
+		})
+}
+
+func (s *MalformedResourcesSuite) TestMalformedCronWorkflow() {
+	s.Given().
+		Exec("kubectl", []string{"apply", "-f", "testdata/malformed/malformed-cronworkflow.yaml"}, fixtures.NoError).
+		Exec("kubectl", []string{"apply", "-f", "testdata/wellformed/wellformed-cronworkflow.yaml"}, fixtures.NoError).
+		When().
+		WaitForWorkflow(1*time.Minute + 15*time.Second).
+		Then().
+		ExpectWorkflow(func(t *testing.T, metadata *metav1.ObjectMeta, status *wfv1.WorkflowStatus) {
+			assert.Equal(t, "wellformed", metadata.Labels[common.LabelKeyCronWorkflow])
+			assert.Equal(t, wfv1.NodeSucceeded, status.Phase)
+		})
+}
+
+func (s *MalformedResourcesSuite) TestMalformedWorkflowTemplate() {
+	s.Given().
+		Exec("kubectl", []string{"apply", "-f", "testdata/malformed/malformed-workflowtemplate.yaml"}, fixtures.NoError).
+		Exec("kubectl", []string{"apply", "-f", "testdata/wellformed/wellformed-workflowtemplate.yaml"}, fixtures.NoError).
+		Exec("kubectl", []string{"apply", "-f", "testdata/wellformed/wellformed-workflow-with-workflow-template-ref.yaml"}, fixtures.NoError).
+		When().
+		WaitForWorkflow(15 * time.Second).
+		Then().
+		ExpectWorkflow(func(t *testing.T, metadata *metav1.ObjectMeta, status *wfv1.WorkflowStatus) {
+			assert.Equal(t, "wellformed", metadata.Name)
+			assert.Equal(t, wfv1.NodeSucceeded, status.Phase)
+		})
+}
+
+func (s *MalformedResourcesSuite) TestMalformedWorkflowTemplateRef() {
+	s.Given().
+		Exec("kubectl", []string{"apply", "-f", "testdata/malformed/malformed-workflowtemplate.yaml"}, fixtures.NoError).
+		Exec("kubectl", []string{"apply", "-f", "testdata/wellformed/wellformed-workflow-with-malformed-workflow-template-ref.yaml"}, fixtures.NoError).
+		When().
+		WaitForWorkflow(15 * time.Second).
+		Then().
+		ExpectWorkflow(func(t *testing.T, metadata *metav1.ObjectMeta, status *wfv1.WorkflowStatus) {
+			assert.Equal(t, "wellformed", metadata.Name)
+			assert.Equal(t, wfv1.NodeError, status.Phase)
+			assert.Contains(t, status.Message, "malformed workflow template")
+		})
+}
+
+func (s *MalformedResourcesSuite) TestMalformedClusterWorkflowTemplate() {
+	s.Given().
+		Exec("kubectl", []string{"apply", "-f", "testdata/malformed/malformed-clusterworkflowtemplate.yaml"}, fixtures.NoError).
+		Exec("kubectl", []string{"apply", "-f", "testdata/wellformed/wellformed-clusterworkflowtemplate.yaml"}, fixtures.NoError).
+		Exec("kubectl", []string{"apply", "-f", "testdata/wellformed/wellformed-workflow-with-cluster-workflow-template-ref.yaml"}, fixtures.NoError).
+		When().
+		WaitForWorkflow(15 * time.Second).
+		Then().
+		ExpectWorkflow(func(t *testing.T, metadata *metav1.ObjectMeta, status *wfv1.WorkflowStatus) {
+			assert.Equal(t, "wellformed", metadata.Name)
+			assert.Equal(t, wfv1.NodeSucceeded, status.Phase)
+		})
+}
+
+func (s *MalformedResourcesSuite) TestMalformedClusterWorkflowTemplateRef() {
+	s.Given().
+		Exec("kubectl", []string{"apply", "-f", "testdata/malformed/malformed-clusterworkflowtemplate.yaml"}, fixtures.NoError).
+		Exec("kubectl", []string{"apply", "-f", "testdata/wellformed/wellformed-workflow-with-malformed-cluster-workflow-template-ref.yaml"}, fixtures.NoError).
+		When().
+		WaitForWorkflow(15 * time.Second).
+		Then().
+		ExpectWorkflow(func(t *testing.T, metadata *metav1.ObjectMeta, status *wfv1.WorkflowStatus) {
+			assert.Equal(t, "wellformed", metadata.Name)
+			assert.Equal(t, wfv1.NodeError, status.Phase)
+			assert.Contains(t, status.Message, "malformed cluster workflow template")
+		})
+}
+
+func TestMalformedResourcesSuite(t *testing.T) {
+	suite.Run(t, new(MalformedResourcesSuite))
+}
diff --git a/test/e2e/testdata/malformed/malformed-clusterworkflowtemplate.yaml b/test/e2e/testdata/malformed/malformed-clusterworkflowtemplate.yaml
new file mode 100644
index 000000000..6108cbdb7
--- /dev/null
+++ b/test/e2e/testdata/malformed/malformed-clusterworkflowtemplate.yaml
@@ -0,0 +1,10 @@
+apiVersion: argoproj.io/v1alpha1
+kind: ClusterWorkflowTemplate
+metadata:
+  name: malformed
+  labels:
+    argo-e2e: malformed
+spec:
+  arguments:
+    parameters:
+      someParam: "Hello, world!"
diff --git a/test/e2e/testdata/malformed/malformed-cronworkflow.yaml b/test/e2e/testdata/malformed/malformed-cronworkflow.yaml
new file mode 100644
index 000000000..97d810938
--- /dev/null
+++ b/test/e2e/testdata/malformed/malformed-cronworkflow.yaml
@@ -0,0 +1,11 @@
+apiVersion: argoproj.io/v1alpha1
+kind: CronWorkflow
+metadata:
+  name: malformed
+  labels:
+    argo-e2e: malformed
+spec:
+  workflowSpec:
+    arguments:
+      parameters:
+        someParam: "Hello, world!"
diff --git a/test/e2e/testdata/malformed/malformed-workflow.yaml b/test/e2e/testdata/malformed/malformed-workflow.yaml
new file mode 100644
index 000000000..92ccc98c7
--- /dev/null
+++ b/test/e2e/testdata/malformed/malformed-workflow.yaml
@@ -0,0 +1,10 @@
+apiVersion: argoproj.io/v1alpha1
+kind: Workflow
+metadata:
+  name: malformed
+  labels:
+    argo-e2e: malformed
+spec:
+  arguments:
+    parameters:
+      someParam: "Hello, world!"
diff --git a/test/e2e/testdata/malformed/malformed-workflowtemplate.yaml b/test/e2e/testdata/malformed/malformed-workflowtemplate.yaml
new file mode 100644
index 000000000..72e42470f
--- /dev/null
+++ b/test/e2e/testdata/malformed/malformed-workflowtemplate.yaml
@@ -0,0 +1,10 @@
+apiVersion: argoproj.io/v1alpha1
+kind: WorkflowTemplate
+metadata:
+  name: malformed
+  labels:
+    argo-e2e: malformed
+spec:
+  arguments:
+    parameters:
+      someParam: "Hello, world!"
diff --git a/test/e2e/testdata/wellformed/wellformed-clusterworkflowtemplate.yaml b/test/e2e/testdata/wellformed/wellformed-clusterworkflowtemplate.yaml
new file mode 100644
index 000000000..37462535c
--- /dev/null
+++ b/test/e2e/testdata/wellformed/wellformed-clusterworkflowtemplate.yaml
@@ -0,0 +1,12 @@
+apiVersion: argoproj.io/v1alpha1
+kind: ClusterWorkflowTemplate
+metadata:
+  name: wellformed
+  labels:
+    argo-e2e: "true"
+spec:
+  entrypoint: main
+  templates:
+    - name: main
+      container:
+        image: argoproj/argosay:v2
\ No newline at end of file
diff --git a/test/e2e/testdata/wellformed/wellformed-cronworkflow.yaml b/test/e2e/testdata/wellformed/wellformed-cronworkflow.yaml
new file mode 100644
index 000000000..68643db42
--- /dev/null
+++ b/test/e2e/testdata/wellformed/wellformed-cronworkflow.yaml
@@ -0,0 +1,17 @@
+apiVersion: argoproj.io/v1alpha1
+kind: CronWorkflow
+metadata:
+  name: wellformed
+  labels:
+    argo-e2e: "true"
+spec:
+  schedule: "* * * * *"
+  workflowMetadata:
+    labels:
+      argo-e2e: "true"
+  workflowSpec:
+    entrypoint: main
+    templates:
+      - name: main
+        container:
+          image: argoproj/argosay:v2
\ No newline at end of file
diff --git a/test/e2e/testdata/wellformed/wellformed-workflow-with-cluster-workflow-template-ref.yaml b/test/e2e/testdata/wellformed/wellformed-workflow-with-cluster-workflow-template-ref.yaml
new file mode 100644
index 000000000..c838df3dc
--- /dev/null
+++ b/test/e2e/testdata/wellformed/wellformed-workflow-with-cluster-workflow-template-ref.yaml
@@ -0,0 +1,10 @@
+apiVersion: argoproj.io/v1alpha1
+kind: Workflow
+metadata:
+  name: wellformed
+  labels:
+    argo-e2e: "true"
+spec:
+  workflowTemplateRef:
+    name: wellformed
+    clusterScope: true
\ No newline at end of file
diff --git a/test/e2e/testdata/wellformed/wellformed-workflow-with-malformed-cluster-workflow-template-ref.yaml b/test/e2e/testdata/wellformed/wellformed-workflow-with-malformed-cluster-workflow-template-ref.yaml
new file mode 100644
index 000000000..380ce231a
--- /dev/null
+++ b/test/e2e/testdata/wellformed/wellformed-workflow-with-malformed-cluster-workflow-template-ref.yaml
@@ -0,0 +1,10 @@
+apiVersion: argoproj.io/v1alpha1
+kind: Workflow
+metadata:
+  name: wellformed
+  labels:
+    argo-e2e: "true"
+spec:
+  workflowTemplateRef:
+    name: malformed
+    clusterScope: true
\ No newline at end of file
diff --git a/test/e2e/testdata/wellformed/wellformed-workflow-with-malformed-workflow-template-ref.yaml b/test/e2e/testdata/wellformed/wellformed-workflow-with-malformed-workflow-template-ref.yaml
new file mode 100644
index 000000000..bdb8715aa
--- /dev/null
+++ b/test/e2e/testdata/wellformed/wellformed-workflow-with-malformed-workflow-template-ref.yaml
@@ -0,0 +1,9 @@
+apiVersion: argoproj.io/v1alpha1
+kind: Workflow
+metadata:
+  name: wellformed
+  labels:
+    argo-e2e: "true"
+spec:
+  workflowTemplateRef:
+    name: malformed
\ No newline at end of file
diff --git a/test/e2e/testdata/wellformed/wellformed-workflow-with-workflow-template-ref.yaml b/test/e2e/testdata/wellformed/wellformed-workflow-with-workflow-template-ref.yaml
new file mode 100644
index 000000000..f3de2926d
--- /dev/null
+++ b/test/e2e/testdata/wellformed/wellformed-workflow-with-workflow-template-ref.yaml
@@ -0,0 +1,9 @@
+apiVersion: argoproj.io/v1alpha1
+kind: Workflow
+metadata:
+  name: wellformed
+  labels:
+    argo-e2e: "true"
+spec:
+  workflowTemplateRef:
+    name: wellformed
\ No newline at end of file
diff --git a/test/e2e/testdata/wellformed/wellformed-workflowtemplate.yaml b/test/e2e/testdata/wellformed/wellformed-workflowtemplate.yaml
new file mode 100644
index 000000000..f198d4226
--- /dev/null
+++ b/test/e2e/testdata/wellformed/wellformed-workflowtemplate.yaml
@@ -0,0 +1,12 @@
+apiVersion: argoproj.io/v1alpha1
+kind: WorkflowTemplate
+metadata:
+  name: wellformed
+  labels:
+    argo-e2e: "true"
+spec:
+  entrypoint: main
+  templates:
+    - name: main
+      container:
+        image: argoproj/argosay:v2
\ No newline at end of file
diff --git a/workflow/controller/controller_test.go b/workflow/controller/controller_test.go
index 4bb31d323..1d8ce8a50 100644
--- a/workflow/controller/controller_test.go
+++ b/workflow/controller/controller_test.go
@@ -7,6 +7,7 @@ import (
 
 	"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"
 
 	"github.com/stretchr/testify/assert"
@@ -24,6 +25,7 @@ import (
 	"github.com/argoproj/argo/persist/sqldb"
 	wfv1 "github.com/argoproj/argo/pkg/apis/workflow/v1alpha1"
 	fakewfclientset "github.com/argoproj/argo/pkg/client/clientset/versioned/fake"
+	"github.com/argoproj/argo/pkg/client/clientset/versioned/scheme"
 	wfextv "github.com/argoproj/argo/pkg/client/informers/externalversions"
 	"github.com/argoproj/argo/test"
 	controllercache "github.com/argoproj/argo/workflow/controller/cache"
@@ -135,6 +137,7 @@ func newController(objects ...runtime.Object) (context.CancelFunc, *WorkflowCont
 			ExecutorImage: "executor:latest",
 		},
 		kubeclientset:        kube,
+		dynamicInterface:     dynamicfake.NewSimpleDynamicClient(scheme.Scheme),
 		wfclientset:          wfclientset,
 		completedPods:        make(chan string, 16),
 		wfInformer:           wfInformer,
diff --git a/workflow/controller/informer/cluster_workflow_template_convert_test.go b/workflow/controller/informer/cluster_workflow_template_convert_test.go
new file mode 100644
index 000000000..2a007327d
--- /dev/null
+++ b/workflow/controller/informer/cluster_workflow_template_convert_test.go
@@ -0,0 +1,39 @@
+package informer
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	corev1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"k8s.io/apimachinery/pkg/runtime"
+
+	wfv1 "github.com/argoproj/argo/pkg/apis/workflow/v1alpha1"
+)
+
+func Test_objectToClusterWorkflowTemplate(t *testing.T) {
+	t.Run("NotUnstructured", func(t *testing.T) {
+		v, err := objectToClusterWorkflowTemplate(&corev1.Status{})
+		assert.EqualError(t, err, "malformed cluster workflow template: expected \"*unstructured.Unstructured\", got \"\"")
+		assert.NotNil(t, v)
+	})
+	t.Run("MalformedClusterWorkflowTemplate", func(t *testing.T) {
+		v, err := objectToClusterWorkflowTemplate(&unstructured.Unstructured{Object: map[string]interface{}{
+			"metadata": map[string]interface{}{"name": "my-name"},
+			"spec":     "ops",
+		}})
+		assert.EqualError(t, err, "malformed cluster workflow template \"my-name\": cannot restore struct from: string")
+		if assert.NotNil(t, v) {
+			assert.Equal(t, "my-name", v.Name)
+		}
+	})
+	t.Run("ClusterWorkflowTemplate", func(t *testing.T) {
+		v, err := objectToClusterWorkflowTemplate(&unstructured.Unstructured{})
+		assert.NoError(t, err)
+		assert.Equal(t, &wfv1.ClusterWorkflowTemplate{}, v)
+	})
+}
+
+func Test_objectsToClusterWorkflowTemplates(t *testing.T) {
+	assert.Len(t, objectsToClusterWorkflowTemplates([]runtime.Object{&corev1.Status{}, &unstructured.Unstructured{}}), 2)
+}
diff --git a/workflow/controller/informer/workflow_template_convert_test.go b/workflow/controller/informer/workflow_template_convert_test.go
new file mode 100644
index 000000000..f7da862b3
--- /dev/null
+++ b/workflow/controller/informer/workflow_template_convert_test.go
@@ -0,0 +1,40 @@
+package informer
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	corev1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"k8s.io/apimachinery/pkg/runtime"
+
+	wfv1 "github.com/argoproj/argo/pkg/apis/workflow/v1alpha1"
+)
+
+func Test_objectToWorkflowTemplate(t *testing.T) {
+	t.Run("NotUnstructured", func(t *testing.T) {
+		v, err := objectToWorkflowTemplate(&corev1.Status{})
+		assert.EqualError(t, err, "malformed workflow template: expected \"*unstructured.Unstructured\", got \"\"")
+		assert.NotNil(t, v)
+	})
+	t.Run("MalformedWorkflowTemplate", func(t *testing.T) {
+		v, err := objectToWorkflowTemplate(&unstructured.Unstructured{Object: map[string]interface{}{
+			"metadata": map[string]interface{}{"namespace": "my-ns", "name": "my-name"},
+			"spec":     "ops",
+		}})
+		assert.EqualError(t, err, "malformed workflow template \"my-ns/my-name\": cannot restore struct from: string")
+		if assert.NotNil(t, v) {
+			assert.Equal(t, "my-ns", v.Namespace)
+			assert.Equal(t, "my-name", v.Name)
+		}
+	})
+	t.Run("WorkflowTemplate", func(t *testing.T) {
+		v, err := objectToWorkflowTemplate(&unstructured.Unstructured{})
+		assert.NoError(t, err)
+		assert.Equal(t, &wfv1.WorkflowTemplate{}, v)
+	})
+}
+
+func Test_objectsToWorkflowTemplates(t *testing.T) {
+	assert.Len(t, objectsToWorkflowTemplates([]runtime.Object{&corev1.Status{}, &unstructured.Unstructured{}}), 2)
+}