feat: add config to skip sending workflow events. Fixes #13042 (#13746)

config/config.go

@@ -26,6 +26,9 @@ type Config struct {
 	// NodeEvents configures how node events are emitted
 	NodeEvents NodeEvents `json:"nodeEvents,omitempty"`
 
+	// WorkflowEvents configures how workflow events are emitted
+	WorkflowEvents WorkflowEvents `json:"workflowEvents,omitempty"`
+
 	// Executor holds container customizations for the executor to use when running pods
 	Executor *apiv1.Container `json:"executor,omitempty"`
 

config/workflow_events.go

@@ -0,0 +1,9 @@
+package config
+
+type WorkflowEvents struct {
+	Enabled *bool `json:"enabled,omitempty"`
+}
+
+func (e WorkflowEvents) IsEnabled() bool {
+	return e.Enabled == nil || *e.Enabled
+}

config/workflow_events_test.go

@@ -0,0 +1,14 @@
+package config
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"k8s.io/utils/ptr"
+)
+
+func TestWorkflowEvents_IsEnabled(t *testing.T) {
+	assert.True(t, WorkflowEvents{}.IsEnabled())
+	assert.False(t, WorkflowEvents{Enabled: ptr.To(false)}.IsEnabled())
+	assert.True(t, WorkflowEvents{Enabled: ptr.To(true)}.IsEnabled())
+}

docs/workflow-controller-configmap.yaml

@@ -41,6 +41,13 @@ data:
   nodeEvents: |
     enabled: true
 
+  # Whether or not to emit events on workflow status changes. These can take a up a lot of space in
+  # k8s (typically etcd), see nodeEvents above.
+  # This config item allows you to disable this.
+  # (since v3.6)
+  workflowEvents: |
+    enabled: true
+
   # uncomment following 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

workflow/controller/operator.go

@@ -2355,13 +2355,15 @@ func (woc *wfOperationCtx) markWorkflowPhase(ctx context.Context, phase wfv1.Wor
 		if _, ok := woc.wf.ObjectMeta.Labels[common.LabelKeyCompleted]; !ok {
 			woc.wf.ObjectMeta.Labels[common.LabelKeyCompleted] = "false"
 		}
-		switch phase {
-		case wfv1.WorkflowRunning:
-			woc.eventRecorder.Event(woc.wf, apiv1.EventTypeNormal, "WorkflowRunning", "Workflow Running")
-		case wfv1.WorkflowSucceeded:
-			woc.eventRecorder.Event(woc.wf, apiv1.EventTypeNormal, "WorkflowSucceeded", "Workflow completed")
-		case wfv1.WorkflowFailed, wfv1.WorkflowError:
-			woc.eventRecorder.Event(woc.wf, apiv1.EventTypeWarning, "WorkflowFailed", message)
+		if woc.controller.Config.WorkflowEvents.IsEnabled() {
+			switch phase {
+			case wfv1.WorkflowRunning:
+				woc.eventRecorder.Event(woc.wf, apiv1.EventTypeNormal, "WorkflowRunning", "Workflow Running")
+			case wfv1.WorkflowSucceeded:
+				woc.eventRecorder.Event(woc.wf, apiv1.EventTypeNormal, "WorkflowSucceeded", "Workflow completed")
+			case wfv1.WorkflowFailed, wfv1.WorkflowError:
+				woc.eventRecorder.Event(woc.wf, apiv1.EventTypeWarning, "WorkflowFailed", message)
+			}
 		}
 	}
 	if woc.wf.Status.StartedAt.IsZero() && phase != wfv1.WorkflowPending {