Repository: argoproj/argo
Pull request #3205 (2020-06-10 17:45:38)

fix(events): Adds config flag. Reduce number of dupe events emitted.
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).

# Changes

* Adds a new flag to disable node events. I'm hoping we won't need this, but just in case.
* Use `EventRecorder` to batch events.

# Testing

I've not added an e2e test for the flag, because this is not default behaviour. So this must be tested manually.

alexec: @sarabala1979 do you think you might be able to review this in the next day or two please?
sonarcloud[bot]: Kudos, SonarCloud Quality Gate passed!

[<img src='https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/bug.png' alt='Bug' width='16' height='16' />](https://sonarcloud.io/project/issues?id=argoproj_argo&pullRequest=3205&resolved=false&types=BUG) [<img src='https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A.png' alt='A' width='16' height='16' />](https://sonarcloud.io/project/issues?id=argoproj_argo&pullRequest=3205&resolved=false&types=BUG) [0 Bugs](https://sonarcloud.io/project/issues?id=argoproj_argo&pullRequest=3205&resolved=false&types=BUG)  
[<img src='https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/vulnerability.png' alt='Vulnerability' width='16' height='16' />](https://sonarcloud.io/project/issues?id=argoproj_argo&pullRequest=3205&resolved=false&types=VULNERABILITY) [<img src='https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A.png' alt='A' width='16' height='16' />](https://sonarcloud.io/project/issues?id=argoproj_argo&pullRequest=3205&resolved=false&types=VULNERABILITY) [0 Vulnerabilities](https://sonarcloud.io/project/issues?id=argoproj_argo&pullRequest=3205&resolved=false&types=VULNERABILITY) (and [<img src='https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/security_hotspot.png' alt='Security Hotspot' width='16' height='16' />](https://sonarcloud.io/project/issues?id=argoproj_argo&pullRequest=3205&resolved=false&types=SECURITY_HOTSPOT) [0 Security Hotspots](https://sonarcloud.io/project/issues?id=argoproj_argo&pullRequest=3205&resolved=false&types=SECURITY_HOTSPOT) to review)  
[<img src='https://sonarsource.github.io/sonarcloud-github-static-resources/v2/common/code_smell.png' alt='Code Smell' width='16' height='16' />](https://sonarcloud.io/project/issues?id=argoproj_argo&pullRequest=3205&resolved=false&types=CODE_SMELL) [<img src='https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/RatingBadge/A.png' alt='A' width='16' height='16' />](https://sonarcloud.io/project/issues?id=argoproj_argo&pullRequest=3205&resolved=false&types=CODE_SMELL) [0 Code Smells](https://sonarcloud.io/project/issues?id=argoproj_argo&pullRequest=3205&resolved=false&types=CODE_SMELL)

[<img src='https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/CoverageChart/60.png' alt='68.4%' width='16' height='16' />](https://sonarcloud.io/component_measures?id=argoproj_argo&pullRequest=3205&metric=new_coverage&view=list) [68.4% Coverage](https://sonarcloud.io/component_measures?id=argoproj_argo&pullRequest=3205&metric=new_coverage&view=list)  
[<img src='https://sonarsource.github.io/sonarcloud-github-static-resources/v2/checks/Duplications/3.png' alt='0.0%' width='16' height='16' />](https://sonarcloud.io/component_measures?id=argoproj_argo&pullRequest=3205&metric=new_duplicated_lines_density&view=list) [0.0% Duplication](https://sonarcloud.io/component_measures?id=argoproj_argo&pullRequest=3205&metric=new_duplicated_lines_density&view=list)



Patch:
diff --git a/config/config.go b/config/config.go
index 94deeefce..f55ad6ce6 100644
--- a/config/config.go
+++ b/config/config.go
@@ -14,6 +14,9 @@ type Config struct {
 	// SSO in settings for single-sign on
 	SSO sso.Config `json:"sso,omitempty"`
 
+	// NodeEvents configures how node events are omitted
+	NodeEvents NodeEvents `json:"nodeEvents,omitempty"`
+
 	// ExecutorImage is the image name of the executor to use when running pods
 	// DEPRECATED: use --executor-image flag to workflow-controller instead
 	ExecutorImage string `json:"executorImage,omitempty"`
diff --git a/config/node_events.go b/config/node_events.go
new file mode 100644
index 000000000..51c76e263
--- /dev/null
+++ b/config/node_events.go
@@ -0,0 +1,9 @@
+package config
+
+type NodeEvents struct {
+	Enabled *bool `json:"enabled,omitempty"`
+}
+
+func (e NodeEvents) IsEnabled() bool {
+	return e.Enabled == nil || *e.Enabled
+}
diff --git a/docs/workflow-controller-configmap.yaml b/docs/workflow-controller-configmap.yaml
index 837188ee6..118224c21 100644
--- a/docs/workflow-controller-configmap.yaml
+++ b/docs/workflow-controller-configmap.yaml
@@ -18,6 +18,14 @@ data:
     # (available since Argo v2.3)
     parallelism: 10
 
+    # Whether or not to emit events on node completion. These can take a up a lot of space in
+    # k8s (typically etcd) resulting in errors when trying to create new events:
+    # "Unable to create audit event: etcdserver: mvcc: database space exceeded"
+    # This config item allows you to disable this.
+    # (since v2.9)
+    nodeEvents:
+      enabled: true
+
     # uncomment flowing lines if workflow controller runs in a different k8s cluster with the 
     # workflow workloads, or needs to communicate with the k8s apiserver using an out-of-cluster
     # kubeconfig secret
diff --git a/manifests/cluster-install/workflow-controller-rbac/workflow-controller-clusterrole.yaml b/manifests/cluster-install/workflow-controller-rbac/workflow-controller-clusterrole.yaml
index e5e78d239..169168378 100644
--- a/manifests/cluster-install/workflow-controller-rbac/workflow-controller-clusterrole.yaml
+++ b/manifests/cluster-install/workflow-controller-rbac/workflow-controller-clusterrole.yaml
@@ -80,6 +80,7 @@ rules:
   - events
   verbs:
   - create
+  - patch
 - apiGroups:
     - "policy"
   resources:
diff --git a/manifests/install.yaml b/manifests/install.yaml
index 1dcf2702b..36811d84c 100644
--- a/manifests/install.yaml
+++ b/manifests/install.yaml
@@ -250,6 +250,7 @@ rules:
   - events
   verbs:
   - create
+  - patch
 - apiGroups:
   - policy
   resources:
diff --git a/manifests/namespace-install.yaml b/manifests/namespace-install.yaml
index 47f2b70be..30a7eb1b3 100644
--- a/manifests/namespace-install.yaml
+++ b/manifests/namespace-install.yaml
@@ -163,6 +163,7 @@ rules:
   - events
   verbs:
   - create
+  - patch
 - apiGroups:
   - policy
   resources:
diff --git a/manifests/namespace-install/workflow-controller-rbac/workflow-controller-role.yaml b/manifests/namespace-install/workflow-controller-rbac/workflow-controller-role.yaml
index 9677d76a2..5556ab739 100644
--- a/manifests/namespace-install/workflow-controller-rbac/workflow-controller-role.yaml
+++ b/manifests/namespace-install/workflow-controller-rbac/workflow-controller-role.yaml
@@ -84,6 +84,7 @@ rules:
   - events
   verbs:
   - create
+  - patch
 - apiGroups:
     - "policy"
   resources:
diff --git a/manifests/quick-start-minimal.yaml b/manifests/quick-start-minimal.yaml
index 75ad7b2e8..8f41294d5 100644
--- a/manifests/quick-start-minimal.yaml
+++ b/manifests/quick-start-minimal.yaml
@@ -163,6 +163,7 @@ rules:
   - events
   verbs:
   - create
+  - patch
 - apiGroups:
   - policy
   resources:
diff --git a/manifests/quick-start-mysql.yaml b/manifests/quick-start-mysql.yaml
index 01e887fbd..5e53202d4 100644
--- a/manifests/quick-start-mysql.yaml
+++ b/manifests/quick-start-mysql.yaml
@@ -163,6 +163,7 @@ rules:
   - events
   verbs:
   - create
+  - patch
 - apiGroups:
   - policy
   resources:
diff --git a/manifests/quick-start-postgres.yaml b/manifests/quick-start-postgres.yaml
index 7e7bef89f..7390dd522 100644
--- a/manifests/quick-start-postgres.yaml
+++ b/manifests/quick-start-postgres.yaml
@@ -163,6 +163,7 @@ rules:
   - events
   verbs:
   - create
+  - patch
 - apiGroups:
   - policy
   resources:
diff --git a/util/argo/audit_logger.go b/util/argo/audit_logger.go
deleted file mode 100644
index 141349fe0..000000000
--- a/util/argo/audit_logger.go
+++ /dev/null
@@ -1,123 +0,0 @@
-package argo
-
-import (
-	"fmt"
-	"time"
-
-	log "github.com/sirupsen/logrus"
-	corev1 "k8s.io/api/core/v1"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/client-go/kubernetes"
-
-	wfv1 "github.com/argoproj/argo/pkg/apis/workflow/v1alpha1"
-	"github.com/argoproj/argo/workflow/common"
-)
-
-type AuditLogger struct {
-	kIf       kubernetes.Interface
-	component string
-	ns        string
-}
-
-type EventInfo struct {
-	Type   string
-	Reason string
-}
-
-const (
-	EventReasonWorkflowRunning       = "WorkflowRunning"
-	EventReasonWorkflowSucceeded     = "WorkflowSucceeded"
-	EventReasonWorkflowFailed        = "WorkflowFailed"
-	EventReasonWorkflowTimedOut      = "WorkflowTimedOut"
-	EventReasonWorkflowNodeSucceeded = "WorkflowNodeSucceeded"
-	EventReasonWorkflowNodeFailed    = "WorkflowNodeFailed"
-	EventReasonWorkflowNodeError     = "WorkflowNodeError"
-)
-
-func (l *AuditLogger) logEvent(workflow *wfv1.Workflow, info EventInfo, message string, annotations map[string]string) {
-	logCtx := log.WithFields(log.Fields{
-		"type":     info.Type,
-		"reason":   info.Reason,
-		"workflow": workflow.Name,
-	})
-	t := metav1.Time{Time: time.Now()}
-	event := corev1.Event{
-		ObjectMeta: metav1.ObjectMeta{
-			Name:        fmt.Sprintf("%v.%x", workflow.Name, t.UnixNano()),
-			Annotations: annotations,
-		},
-		Source: corev1.EventSource{
-			Component: l.component,
-		},
-		InvolvedObject: corev1.ObjectReference{
-			Kind:            wfv1.WorkflowSchemaGroupVersionKind.Kind,
-			APIVersion:      wfv1.SchemeGroupVersion.String(),
-			Name:            workflow.Name,
-			Namespace:       workflow.Namespace,
-			ResourceVersion: workflow.ResourceVersion,
-			UID:             workflow.UID,
-		},
-		FirstTimestamp: t,
-		LastTimestamp:  t,
-		Count:          1,
-		Message:        message,
-		Type:           info.Type,
-		Reason:         info.Reason,
-	}
-	logCtx.WithField("event", event).Debug()
-	_, err := l.kIf.CoreV1().Events(workflow.Namespace).Create(&event)
-	if err != nil {
-		logCtx.Errorf("Unable to create audit event: %v", err)
-		return
-	}
-}
-
-func (l *AuditLogger) logWorkflowEvent(workflow *wfv1.Workflow, info EventInfo, message string, annotations map[string]string) {
-	l.logEvent(workflow, info, message, annotations)
-}
-
-func (l *AuditLogger) LogWorkflowEvent(workflow *wfv1.Workflow, info EventInfo, message string) {
-	l.logWorkflowEvent(workflow, info, message, nil)
-}
-
-func (l *AuditLogger) LogWorkflowNodeEvent(workflow *wfv1.Workflow, node *wfv1.NodeStatus) {
-	l.logWorkflowEvent(
-		workflow,
-		EventInfo{Type: eventType(node.Phase), Reason: nodePhaseReason(node.Phase)},
-		nodeMessage(node),
-		map[string]string{
-			common.AnnotationKeyNodeType: string(node.Type),
-			common.AnnotationKeyNodeName: node.Name,
-		})
-}
-
-func eventType(phase wfv1.NodePhase) string {
-	return map[wfv1.NodePhase]string{
-		wfv1.NodeError:     corev1.EventTypeWarning,
-		wfv1.NodeFailed:    corev1.EventTypeWarning,
-		wfv1.NodeSucceeded: corev1.EventTypeNormal,
-	}[phase]
-}
-func nodePhaseReason(phase wfv1.NodePhase) string {
-	return map[wfv1.NodePhase]string{
-		wfv1.NodeError:     EventReasonWorkflowNodeError,
-		wfv1.NodeFailed:    EventReasonWorkflowNodeFailed,
-		wfv1.NodeSucceeded: EventReasonWorkflowNodeSucceeded,
-	}[phase]
-}
-
-func nodeMessage(node *wfv1.NodeStatus) string {
-	message := fmt.Sprintf("%v node %s", node.Phase, node.Name)
-	if node.Message != "" {
-		message = message + ": " + node.Message
-	}
-	return message
-}
-
-func NewAuditLogger(ns string, kIf kubernetes.Interface, component string) *AuditLogger {
-	return &AuditLogger{
-		ns:        ns,
-		kIf:       kIf,
-		component: component,
-	}
-}
diff --git a/workflow/controller/controller.go b/workflow/controller/controller.go
index 0824cefba..e6d49b67d 100644
--- a/workflow/controller/controller.go
+++ b/workflow/controller/controller.go
@@ -22,8 +22,11 @@ import (
 	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/apimachinery/pkg/watch"
 	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/kubernetes/scheme"
+	typedcorev1 "k8s.io/client-go/kubernetes/typed/core/v1"
 	"k8s.io/client-go/rest"
 	"k8s.io/client-go/tools/cache"
+	"k8s.io/client-go/tools/record"
 	"k8s.io/client-go/util/workqueue"
 	"upper.io/db.v3/lib/sqlbuilder"
 
@@ -81,6 +84,7 @@ type WorkflowController struct {
 	hydrator              hydrator.Interface
 	wfArchive             sqldb.WorkflowArchive
 	metrics               metrics.Metrics
+	eventRecorder         record.EventRecorder
 }
 
 const (
@@ -102,6 +106,10 @@ func NewWorkflowController(
 	containerRuntimeExecutor,
 	configMap string,
 ) *WorkflowController {
+	eventBroadcaster := record.NewBroadcaster()
+	eventBroadcaster.StartLogging(log.Debugf)
+	eventBroadcaster.StartRecordingToSink(&typedcorev1.EventSinkImpl{Interface: kubeclientset.CoreV1().Events(namespace)})
+	eventRecorder := eventBroadcaster.NewRecorder(scheme.Scheme, apiv1.EventSource{Component: "workflow-controller"})
 	wfc := WorkflowController{
 		restConfig:                 restConfig,
 		kubeclientset:              kubeclientset,
@@ -116,6 +124,7 @@ func NewWorkflowController(
 		configController:           config.NewController(namespace, configMap, kubeclientset),
 		completedPods:              make(chan string, 512),
 		gcPods:                     make(chan string, 512),
+		eventRecorder:              eventRecorder,
 	}
 	wfc.throttler = NewThrottler(0, wfc.wfQueue)
 	wfc.UpdateConfig()
diff --git a/workflow/controller/operator.go b/workflow/controller/operator.go
index 3cc105f26..2cf2611ee 100644
--- a/workflow/controller/operator.go
+++ b/workflow/controller/operator.go
@@ -25,6 +25,7 @@ import (
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/client-go/tools/cache"
+	"k8s.io/client-go/tools/record"
 	"k8s.io/utils/pointer"
 	"sigs.k8s.io/yaml"
 
@@ -34,7 +35,6 @@ import (
 	wfv1 "github.com/argoproj/argo/pkg/apis/workflow/v1alpha1"
 	"github.com/argoproj/argo/pkg/client/clientset/versioned/typed/workflow/v1alpha1"
 	"github.com/argoproj/argo/util"
-	"github.com/argoproj/argo/util/argo"
 	"github.com/argoproj/argo/util/resource"
 	"github.com/argoproj/argo/util/retry"
 	"github.com/argoproj/argo/workflow/common"
@@ -81,8 +81,7 @@ type wfOperationCtx struct {
 	// workflowDeadline is the deadline which the workflow is expected to complete before we
 	// terminate the workflow.
 	workflowDeadline *time.Time
-	// auditLogger is the argo audit logger
-	auditLogger *argo.AuditLogger
+	eventRecorder    record.EventRecorder
 	// preExecutionNodePhases contains the phases of all the nodes before the current operation. Necessary to infer
 	// changes in phase for metric emission
 	preExecutionNodePhases map[string]wfv1.NodePhase
@@ -132,7 +131,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),
-		auditLogger:            argo.NewAuditLogger(wf.ObjectMeta.Namespace, wfc.kubeclientset, wf.ObjectMeta.Name),
+		eventRecorder:          wfc.eventRecorder,
 		preExecutionNodePhases: make(map[string]wfv1.NodePhase),
 	}
 
@@ -194,10 +193,10 @@ func (woc *wfOperationCtx) operate() {
 		if err != nil {
 			msg := fmt.Sprintf("Unable to create PDB resource for workflow, %s error: %s", woc.wf.Name, err)
 			woc.markWorkflowFailed(msg)
-			woc.auditLogger.LogWorkflowEvent(woc.wf, argo.EventInfo{Type: apiv1.EventTypeWarning, Reason: argo.EventReasonWorkflowFailed}, msg)
+			woc.eventRecorder.Event(woc.wf, apiv1.EventTypeWarning, "WorkflowFailed", msg)
 			return
 		}
-		woc.auditLogger.LogWorkflowEvent(woc.wf, argo.EventInfo{Type: apiv1.EventTypeNormal, Reason: argo.EventReasonWorkflowRunning}, "Workflow Running")
+		woc.eventRecorder.Event(woc.wf, apiv1.EventTypeNormal, "WorkflowRunning", "Workflow Running")
 		validateOpts := validate.ValidateOpts{ContainerRuntimeExecutor: woc.controller.GetContainerRuntimeExecutor()}
 		wftmplGetter := templateresolution.WrapWorkflowTemplateInterface(woc.controller.wfclientset.ArgoprojV1alpha1().WorkflowTemplates(woc.wf.Namespace))
 		cwftmplGetter := templateresolution.WrapClusterWorkflowTemplateInterface(woc.controller.wfclientset.ArgoprojV1alpha1().ClusterWorkflowTemplates())
@@ -207,7 +206,7 @@ func (woc *wfOperationCtx) operate() {
 		if err != nil {
 			msg := fmt.Sprintf("invalid spec: %s", err.Error())
 			woc.markWorkflowFailed(msg)
-			woc.auditLogger.LogWorkflowEvent(woc.wf, argo.EventInfo{Type: apiv1.EventTypeWarning, Reason: argo.EventReasonWorkflowFailed}, msg)
+			woc.eventRecorder.Event(woc.wf, apiv1.EventTypeWarning, "WorkflowFailed", msg)
 			return
 		}
 		// If we received conditions during validation (such as SpecWarnings), add them to the Workflow object
@@ -232,7 +231,7 @@ func (woc *wfOperationCtx) operate() {
 		}
 		if err != nil {
 			woc.log.Errorf("%s error: %+v", woc.wf.ObjectMeta.Name, err)
-			woc.auditLogger.LogWorkflowEvent(woc.wf, argo.EventInfo{Type: apiv1.EventTypeWarning, Reason: argo.EventReasonWorkflowTimedOut}, "Workflow timed out")
+			woc.eventRecorder.Event(woc.wf, apiv1.EventTypeWarning, "WorkflowTimedOut", "Workflow timed out")
 			// TODO: we need to re-add to the workqueue, but should happen in caller
 			return
 		}
@@ -265,7 +264,7 @@ func (woc *wfOperationCtx) operate() {
 			msg := fmt.Sprintf("Failed to load artifact repository configMap: %+v", err)
 			woc.log.Errorf(msg)
 			woc.markWorkflowError(err, true)
-			woc.auditLogger.LogWorkflowEvent(woc.wf, argo.EventInfo{Type: apiv1.EventTypeWarning, Reason: argo.EventReasonWorkflowFailed}, msg)
+			woc.eventRecorder.Event(woc.wf, apiv1.EventTypeWarning, "WorkflowFailed", msg)
 			return
 		}
 	}
@@ -283,7 +282,7 @@ func (woc *wfOperationCtx) operate() {
 		msg := fmt.Sprintf("%s pvc create error: %+v", woc.wf.ObjectMeta.Name, err)
 		woc.log.Errorf(msg)
 		woc.markWorkflowError(err, true)
-		woc.auditLogger.LogWorkflowEvent(woc.wf, argo.EventInfo{Type: apiv1.EventTypeWarning, Reason: argo.EventReasonWorkflowFailed}, msg)
+		woc.eventRecorder.Event(woc.wf, apiv1.EventTypeWarning, "WorkflowFailed", msg)
 		return
 	}
 
@@ -294,9 +293,9 @@ func (woc *wfOperationCtx) operate() {
 		woc.log.Errorf(msg)
 		switch err {
 		case ErrDeadlineExceeded:
-			woc.auditLogger.LogWorkflowEvent(woc.wf, argo.EventInfo{Type: apiv1.EventTypeWarning, Reason: argo.EventReasonWorkflowTimedOut}, msg)
+			woc.eventRecorder.Event(woc.wf, apiv1.EventTypeWarning, "WorkflowTimedOut", msg)
 		default:
-			woc.auditLogger.LogWorkflowEvent(woc.wf, argo.EventInfo{Type: apiv1.EventTypeWarning, Reason: argo.EventReasonWorkflowFailed}, msg)
+			woc.eventRecorder.Event(woc.wf, apiv1.EventTypeWarning, "WorkflowFailed", msg)
 		}
 		return
 	}
@@ -359,7 +358,7 @@ func (woc *wfOperationCtx) operate() {
 		// markCompletion so that we can retry PVC deletion (TODO: use workqueue.ReAdd())
 		// This error phase may be cleared if a subsequent delete attempt is successful.
 		woc.markWorkflowError(err, false)
-		woc.auditLogger.LogWorkflowEvent(woc.wf, argo.EventInfo{Type: apiv1.EventTypeWarning, Reason: argo.EventReasonWorkflowFailed}, msg)
+		woc.eventRecorder.Event(woc.wf, apiv1.EventTypeWarning, "WorkflowFailed", msg)
 		return
 	}
 
@@ -379,17 +378,17 @@ func (woc *wfOperationCtx) operate() {
 			// if main workflow succeeded, but the exit node was unsuccessful
 			// the workflow is now considered unsuccessful.
 			woc.markWorkflowPhase(onExitNode.Phase, true, onExitNode.Message)
-			woc.auditLogger.LogWorkflowEvent(woc.wf, argo.EventInfo{Type: apiv1.EventTypeWarning, Reason: argo.EventReasonWorkflowFailed}, onExitNode.Message)
+			woc.eventRecorder.Event(woc.wf, apiv1.EventTypeWarning, "WorkflowFailed", onExitNode.Message)
 		} else {
 			woc.markWorkflowSuccess()
-			woc.auditLogger.LogWorkflowEvent(woc.wf, argo.EventInfo{Type: apiv1.EventTypeNormal, Reason: argo.EventReasonWorkflowSucceeded}, "Workflow completed")
+			woc.eventRecorder.Event(woc.wf, apiv1.EventTypeNormal, "WorkflowSucceeded", "Workflow completed")
 		}
 	case wfv1.NodeFailed:
 		woc.markWorkflowFailed(workflowMessage)
-		woc.auditLogger.LogWorkflowEvent(woc.wf, argo.EventInfo{Type: apiv1.EventTypeWarning, Reason: argo.EventReasonWorkflowFailed}, workflowMessage)
+		woc.eventRecorder.Event(woc.wf, apiv1.EventTypeWarning, "WorkflowFailed", workflowMessage)
 	case wfv1.NodeError:
 		woc.markWorkflowPhase(wfv1.NodeError, true, workflowMessage)
-		woc.auditLogger.LogWorkflowEvent(woc.wf, argo.EventInfo{Type: apiv1.EventTypeWarning, Reason: argo.EventReasonWorkflowFailed}, workflowMessage)
+		woc.eventRecorder.Event(woc.wf, apiv1.EventTypeWarning, "WorkflowFailed", workflowMessage)
 	default:
 		// NOTE: we should never make it here because if the the node was 'Running'
 		// we should have returned earlier.
@@ -1738,7 +1737,27 @@ func (woc *wfOperationCtx) markNodePhase(nodeName string, phase wfv1.NodePhase,
 }
 
 func (woc *wfOperationCtx) onNodeComplete(node *wfv1.NodeStatus) {
-	woc.auditLogger.LogWorkflowNodeEvent(woc.wf, node)
+	if !woc.controller.Config.NodeEvents.IsEnabled() {
+		return
+	}
+	message := fmt.Sprintf("%v node %s", node.Phase, node.Name)
+	if node.Message != "" {
+		message = message + ": " + node.Message
+	}
+	eventType := apiv1.EventTypeWarning
+	if node.Phase == wfv1.NodeSucceeded {
+		eventType = apiv1.EventTypeNormal
+	}
+	woc.eventRecorder.AnnotatedEventf(
+		woc.wf,
+		map[string]string{
+			common.AnnotationKeyNodeType: string(node.Type),
+			common.AnnotationKeyNodeName: node.Name,
+		},
+		eventType,
+		fmt.Sprintf("WorkflowNode%s", node.Phase),
+		message,
+	)
 }
 
 // markNodeError is a convenience method to mark a node with an error and set the message from the error


Test patch:
diff --git a/config/node_events_test.go b/config/node_events_test.go
new file mode 100644
index 000000000..394e343a2
--- /dev/null
+++ b/config/node_events_test.go
@@ -0,0 +1,14 @@
+package config
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"k8s.io/utils/pointer"
+)
+
+func TestNodeEvents_IsEnabled(t *testing.T) {
+	assert.True(t, NodeEvents{}.IsEnabled())
+	assert.False(t, NodeEvents{Enabled: pointer.BoolPtr(false)}.IsEnabled())
+	assert.True(t, NodeEvents{Enabled: pointer.BoolPtr(true)}.IsEnabled())
+}
\ No newline at end of file
diff --git a/test/e2e/fixtures/then.go b/test/e2e/fixtures/then.go
index 91362fb9e..f87273e36 100644
--- a/test/e2e/fixtures/then.go
+++ b/test/e2e/fixtures/then.go
@@ -2,9 +2,9 @@ package fixtures
 
 import (
 	"testing"
+	"time"
 
 	log "github.com/sirupsen/logrus"
-	"github.com/stretchr/testify/assert"
 	apiv1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/client-go/kubernetes"
@@ -85,36 +85,34 @@ func (t *Then) ExpectWorkflowList(listOptions metav1.ListOptions, block func(t *
 	return t
 }
 
-func (t *Then) expectAuditEvents(block func(*testing.T, []apiv1.Event)) *Then {
-	eventList, err := t.kubeClient.CoreV1().Events(Namespace).List(metav1.ListOptions{})
+func (t *Then) ExpectAuditEvents(blocks ...func(*testing.T, apiv1.Event)) *Then {
+	eventList, err := t.kubeClient.CoreV1().Events(Namespace).Watch(metav1.ListOptions{})
 	if err != nil {
 		t.t.Fatal(err)
 	}
-	var events []apiv1.Event
-	for _, e := range eventList.Items {
-		if e.Namespace == Namespace && e.InvolvedObject.Kind == workflow.WorkflowKind {
-			events = append(events, e)
+	ticker := time.NewTicker(5 * time.Second)
+	defer ticker.Stop()
+	for len(blocks) > 0 {
+		select {
+		case <-ticker.C:
+			t.t.Fatal("timeout waiting for events")
+		case event := <-eventList.ResultChan():
+			e, ok := event.Object.(*apiv1.Event)
+			if !ok {
+				t.t.Fatal("event is not an event")
+			}
+			if e.InvolvedObject.Name == t.workflowName && e.Namespace == Namespace && e.InvolvedObject.Kind == workflow.WorkflowKind {
+				blocks[0](t.t, *e)
+				blocks = blocks[1:]
+				if t.t.Failed() {
+					t.t.FailNow()
+				}
+			}
 		}
 	}
-	log.WithFields(log.Fields{"event": events}).Debug("Events")
-	block(t.t, events)
-	if t.t.Failed() {
-		t.t.FailNow()
-	}
 	return t
 }
 
-func (t *Then) ExpectAuditEvent(f func(apiv1.Event) bool) *Then {
-	return t.expectAuditEvents(func(t *testing.T, events []apiv1.Event) {
-		for _, item := range events {
-			if f(item) {
-				return
-			}
-		}
-		assert.Fail(t, "did not see expected event")
-	})
-}
-
 func (t *Then) RunCli(args []string, block func(t *testing.T, output string, err error)) *Then {
 	output, err := runCli("../../dist/argo", append([]string{"-n", Namespace}, args...)...)
 	block(t.t, output, err)
diff --git a/test/e2e/functional_test.go b/test/e2e/functional_test.go
index 25b69119b..a68a3c91e 100644
--- a/test/e2e/functional_test.go
+++ b/test/e2e/functional_test.go
@@ -2,7 +2,6 @@ package e2e
 
 import (
 	"regexp"
-	"strings"
 	"testing"
 	"time"
 
@@ -11,10 +10,8 @@ import (
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 
-	"github.com/argoproj/argo/pkg/apis/workflow"
 	wfv1 "github.com/argoproj/argo/pkg/apis/workflow/v1alpha1"
 	"github.com/argoproj/argo/test/e2e/fixtures"
-	"github.com/argoproj/argo/util/argo"
 )
 
 type FunctionalSuite struct {
@@ -187,20 +184,23 @@ func (s *FunctionalSuite) TestEventOnNodeFail() {
 		Workflow("@expectedfailures/failed-step-event.yaml").
 		When().
 		SubmitWorkflow().
-		WaitForWorkflow(30 * time.Second).
+		WaitForWorkflow(30*time.Second).
 		Then().
-		ExpectAuditEvent(func(e corev1.Event) bool {
-			return strings.HasPrefix(e.InvolvedObject.Name, "failed-step-event-") &&
-				e.Reason == argo.EventReasonWorkflowFailed &&
-				e.Message == "failed with exit code 1"
-		}).
-		ExpectAuditEvent(func(e corev1.Event) bool {
-			return e.InvolvedObject.Kind == workflow.WorkflowKind &&
-				e.Reason == argo.EventReasonWorkflowNodeFailed &&
-				strings.HasPrefix(e.Message, "Failed node failed-step-event-") &&
-				e.Annotations["workflows.argoproj.io/node-type"] == "Pod" &&
-				strings.Contains(e.Annotations["workflows.argoproj.io/node-name"], "failed-step-event-")
-		})
+		ExpectAuditEvents(
+			func(t *testing.T, e corev1.Event) {
+				assert.Equal(t, "WorkflowRunning", e.Reason)
+			},
+			func(t *testing.T, e corev1.Event) {
+				assert.Equal(t, e.Reason, "WorkflowNodeFailed")
+				assert.Contains(t, e.Message, "Failed node failed-step-event-")
+				assert.Equal(t, e.Annotations["workflows.argoproj.io/node-type"], "Pod")
+				assert.Contains(t, e.Annotations["workflows.argoproj.io/node-name"], "failed-step-event-")
+			},
+			func(t *testing.T, e corev1.Event) {
+				assert.Equal(t, "WorkflowFailed", e.Reason)
+				assert.Equal(t, "failed with exit code 1", e.Message)
+			},
+		)
 }
 
 func (s *FunctionalSuite) TestEventOnWorkflowSuccess() {
@@ -209,20 +209,23 @@ func (s *FunctionalSuite) TestEventOnWorkflowSuccess() {
 		Workflow("@functional/success-event.yaml").
 		When().
 		SubmitWorkflow().
-		WaitForWorkflow(60 * time.Second).
+		WaitForWorkflow(60*time.Second).
 		Then().
-		ExpectAuditEvent(func(e corev1.Event) bool {
-			return strings.HasPrefix(e.InvolvedObject.Name, "success-event-") &&
-				e.Reason == argo.EventReasonWorkflowSucceeded &&
-				e.Message == "Workflow completed"
-		}).
-		ExpectAuditEvent(func(e corev1.Event) bool {
-			return e.InvolvedObject.Kind == workflow.WorkflowKind &&
-				e.Reason == argo.EventReasonWorkflowNodeSucceeded &&
-				strings.HasPrefix(e.Message, "Succeeded node success-event-") &&
-				e.Annotations["workflows.argoproj.io/node-type"] == "Pod" &&
-				strings.Contains(e.Annotations["workflows.argoproj.io/node-name"], "success-event-")
-		})
+		ExpectAuditEvents(
+			func(t *testing.T, e corev1.Event) {
+				assert.Equal(t, "WorkflowRunning", e.Reason)
+			},
+			func(t *testing.T, e corev1.Event) {
+				assert.Equal(t, "WorkflowNodeSucceeded", e.Reason)
+				assert.Contains(t, e.Message, "Succeeded node success-event-")
+				assert.Equal(t, "Pod", e.Annotations["workflows.argoproj.io/node-type"])
+				assert.Contains(t, e.Annotations["workflows.argoproj.io/node-name"], "success-event-")
+			},
+			func(t *testing.T, e corev1.Event) {
+				assert.Equal(t, "WorkflowSucceeded", e.Reason)
+				assert.Equal(t, "Workflow completed", e.Message)
+			},
+		)
 }
 
 func (s *FunctionalSuite) TestEventOnPVCFail() {
@@ -231,13 +234,17 @@ func (s *FunctionalSuite) TestEventOnPVCFail() {
 		Workflow("@expectedfailures/volumes-pvc-fail-event.yaml").
 		When().
 		SubmitWorkflow().
-		WaitForWorkflow(120 * time.Second).
+		WaitForWorkflow(120*time.Second).
 		Then().
-		ExpectAuditEvent(func(e corev1.Event) bool {
-			return strings.HasPrefix(e.InvolvedObject.Name, "volumes-pvc-fail-event-") &&
-				e.Reason == argo.EventReasonWorkflowFailed &&
-				strings.Contains(e.Message, "pvc create error")
-		})
+		ExpectAuditEvents(
+			func(t *testing.T, e corev1.Event) {
+				assert.Equal(t, "WorkflowRunning", e.Reason)
+			},
+			func(t *testing.T, e corev1.Event) {
+				assert.Equal(t, "WorkflowFailed", e.Reason)
+				assert.Contains(t, e.Message, "pvc create error")
+			},
+		)
 }
 
 func (s *FunctionalSuite) TestArtifactRepositoryRef() {
diff --git a/util/argo/audit_logger_test.go b/util/argo/audit_logger_test.go
deleted file mode 100644
index 6ad7ec1ef..000000000
--- a/util/argo/audit_logger_test.go
+++ /dev/null
@@ -1,63 +0,0 @@
-package argo
-
-import (
-	"testing"
-
-	"github.com/stretchr/testify/assert"
-	corev1 "k8s.io/api/core/v1"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/client-go/kubernetes/fake"
-
-	wfv1 "github.com/argoproj/argo/pkg/apis/workflow/v1alpha1"
-)
-
-func TestAuditLogger_logEvent(t *testing.T) {
-	kube := fake.NewSimpleClientset()
-	l := AuditLogger{kIf: kube, component: "my-component"}
-	l.LogWorkflowNodeEvent(
-		&wfv1.Workflow{
-			ObjectMeta: metav1.ObjectMeta{Namespace: "my-ns", Name: "my-wf", UID: "my-uid", ResourceVersion: "1234"},
-		},
-		&wfv1.NodeStatus{Type: wfv1.NodeTypePod, Phase: wfv1.NodeSucceeded, Name: "my-node", Message: "my message"},
-	)
-	list, err := kube.CoreV1().Events("my-ns").List(metav1.ListOptions{})
-	if assert.NoError(t, err) && assert.Len(t, list.Items, 1) {
-		e := list.Items[0]
-		assert.NotEmpty(t, e.Name)
-		assert.Len(t, e.Annotations, 2)
-		assert.Equal(t, "my-node", e.Annotations["workflows.argoproj.io/node-name"])
-		assert.Equal(t, "Pod", e.Annotations["workflows.argoproj.io/node-type"])
-		assert.Equal(t, "my-component", e.Source.Component)
-		assert.Equal(t, corev1.ObjectReference{
-			Kind:            "Workflow",
-			Namespace:       "my-ns",
-			Name:            "my-wf",
-			UID:             "my-uid",
-			APIVersion:      "argoproj.io/v1alpha1",
-			ResourceVersion: "1234",
-		}, e.InvolvedObject)
-		assert.NotEmpty(t, e.FirstTimestamp)
-		assert.NotEmpty(t, e.LastTimestamp)
-		assert.Equal(t, int32(1), e.Count)
-		assert.Equal(t, "Succeeded node my-node: my message", e.Message)
-		assert.Equal(t, "Normal", e.Type)
-		assert.Equal(t, "WorkflowNodeSucceeded", e.Reason)
-	}
-}
-
-func Test_eventType(t *testing.T) {
-	assert.Equal(t, corev1.EventTypeWarning, eventType(wfv1.NodeError))
-	assert.Equal(t, corev1.EventTypeWarning, eventType(wfv1.NodeFailed))
-	assert.Equal(t, corev1.EventTypeNormal, eventType(wfv1.NodeSucceeded))
-}
-
-func Test_nodePhaseReason(t *testing.T) {
-	assert.Equal(t, EventReasonWorkflowNodeError, nodePhaseReason(wfv1.NodeError))
-	assert.Equal(t, EventReasonWorkflowNodeFailed, nodePhaseReason(wfv1.NodeFailed))
-	assert.Equal(t, EventReasonWorkflowNodeSucceeded, nodePhaseReason(wfv1.NodeSucceeded))
-}
-
-func Test_nodeMessage(t *testing.T) {
-	assert.Equal(t, "Succeeded node my-node", nodeMessage(&wfv1.NodeStatus{Phase: wfv1.NodeSucceeded, Name: "my-node"}))
-	assert.Equal(t, "Succeeded node my-node: my-message", nodeMessage(&wfv1.NodeStatus{Phase: wfv1.NodeSucceeded, Name: "my-node", Message: "my-message"}))
-}
diff --git a/workflow/controller/controller_test.go b/workflow/controller/controller_test.go
index afbace04f..185f7eedc 100644
--- a/workflow/controller/controller_test.go
+++ b/workflow/controller/controller_test.go
@@ -14,6 +14,7 @@ import (
 	"k8s.io/client-go/kubernetes/fake"
 	k8stesting "k8s.io/client-go/testing"
 	"k8s.io/client-go/tools/cache"
+	"k8s.io/client-go/tools/record"
 	"k8s.io/client-go/util/workqueue"
 	"sigs.k8s.io/yaml"
 
@@ -121,7 +122,7 @@ func newController(objects ...runtime.Object) (context.CancelFunc, *WorkflowCont
 		},
 		kubeclientset:   kube,
 		wfclientset:     wfclientset,
-		completedPods:   make(chan string, 512),
+		completedPods:   make(chan string, 16),
 		wfInformer:      wfInformer,
 		wftmplInformer:  wftmplInformer,
 		cwftmplInformer: cwftmplInformer,
@@ -130,6 +131,7 @@ func newController(objects ...runtime.Object) (context.CancelFunc, *WorkflowCont
 		wfArchive:       sqldb.NullWorkflowArchive,
 		hydrator:        hydratorfake.Noop,
 		metrics:         metrics.New(metrics.ServerConfig{}, metrics.ServerConfig{}),
+		eventRecorder:   record.NewFakeRecorder(16),
 	}
 	return cancel, controller
 }
diff --git a/workflow/controller/operator_test.go b/workflow/controller/operator_test.go
index 31cb386e5..f6a617710 100644
--- a/workflow/controller/operator_test.go
+++ b/workflow/controller/operator_test.go
@@ -13,12 +13,12 @@ import (
 	apiv1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/tools/record"
 	"sigs.k8s.io/yaml"
 
 	"github.com/argoproj/argo/config"
 	wfv1 "github.com/argoproj/argo/pkg/apis/workflow/v1alpha1"
 	"github.com/argoproj/argo/test"
-	"github.com/argoproj/argo/util/argo"
 	"github.com/argoproj/argo/workflow/common"
 	hydratorfake "github.com/argoproj/argo/workflow/hydrator/fake"
 	"github.com/argoproj/argo/workflow/util"
@@ -2730,54 +2730,18 @@ func TestEventInvalidSpec(t *testing.T) {
 	assert.NoError(t, err)
 	woc := newWorkflowOperationCtx(wf, controller)
 	woc.operate()
-	events, err := controller.kubeclientset.CoreV1().Events("").List(metav1.ListOptions{})
-	assert.NoError(t, err)
-	assert.Equal(t, 2, len(events.Items))
-	runningEvent := events.Items[0]
-	assert.Equal(t, argo.EventReasonWorkflowRunning, runningEvent.Reason)
-	invalidSpecEvent := events.Items[1]
-	assert.Equal(t, argo.EventReasonWorkflowFailed, invalidSpecEvent.Reason)
-	assert.Equal(t, "invalid spec: template name '123' undefined", invalidSpecEvent.Message)
+	events := getEvents(controller, 2)
+	assert.Equal(t, "Normal WorkflowRunning Workflow Running", events[0])
+	assert.Equal(t, "Warning WorkflowFailed invalid spec: template name '123' undefined", events[1])
 }
 
-var timeout = `
-apiVersion: argoproj.io/v1alpha1
-kind: Workflow
-metadata:
-  name: timeout-template
-spec:
-  entrypoint: sleep
-  activeDeadlineSeconds: 1
-  templates:
-  - name: sleep
-    container:
-      image: alpine:latest
-      command: [sh, -c, sleep 10]
-`
-
-func TestEventTimeout(t *testing.T) {
-	// Test whether a WorkflowTimedOut event is emitted in case of timeout
-	cancel, controller := newController()
-	defer cancel()
-	wfcset := controller.wfclientset.ArgoprojV1alpha1().Workflows("")
-	wf := unmarshalWF(timeout)
-	wf, err := wfcset.Create(wf)
-	assert.NoError(t, err)
-	woc := newWorkflowOperationCtx(wf, controller)
-	woc.operate()
-	makePodsPhase(t, apiv1.PodRunning, controller.kubeclientset, wf.ObjectMeta.Namespace)
-	wf, err = wfcset.Get(wf.ObjectMeta.Name, metav1.GetOptions{})
-	assert.NoError(t, err)
-	woc = newWorkflowOperationCtx(wf, controller)
-	time.Sleep(10 * time.Second)
-	woc.operate()
-	events, err := controller.kubeclientset.CoreV1().Events("").List(metav1.ListOptions{})
-	assert.NoError(t, err)
-	assert.Equal(t, 2, len(events.Items))
-	runningEvent := events.Items[0]
-	assert.Equal(t, argo.EventReasonWorkflowRunning, runningEvent.Reason)
-	timeoutEvent := events.Items[1]
-	assert.Equal(t, argo.EventReasonWorkflowFailed, timeoutEvent.Reason)
+func getEvents(controller *WorkflowController, num int) []string {
+	c := controller.eventRecorder.(*record.FakeRecorder).Events
+	events := make([]string, num)
+	for i := 0; i < num; i++ {
+		events[i] = <-c
+	}
+	return events
 }
 
 var failLoadArtifactRepoCm = `
@@ -2812,14 +2776,9 @@ func TestEventFailArtifactRepoCm(t *testing.T) {
 	assert.NoError(t, err)
 	woc := newWorkflowOperationCtx(wf, controller)
 	woc.operate()
-	events, err := controller.kubeclientset.CoreV1().Events("").List(metav1.ListOptions{})
-	assert.NoError(t, err)
-	assert.Equal(t, 2, len(events.Items))
-	runningEvent := events.Items[0]
-	assert.Equal(t, argo.EventReasonWorkflowRunning, runningEvent.Reason)
-	failEvent := events.Items[1]
-	assert.Equal(t, argo.EventReasonWorkflowFailed, failEvent.Reason)
-	assert.Equal(t, "Failed to load artifact repository configMap: failed to find artifactory ref {,}/artifact-repository#config", failEvent.Message)
+	events := getEvents(controller, 2)
+	assert.Equal(t, "Normal WorkflowRunning Workflow Running", events[0])
+	assert.Equal(t, "Warning WorkflowFailed Failed to load artifact repository configMap: failed to find artifactory ref {,}/artifact-repository#config", events[1])
 }
 
 var pdbwf = `