fix(events): Adds config flag. Reduce number of dupe events emitted. (a…

…rgoproj#3205)

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"`

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
+}

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())
+}

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

manifests/cluster-install/workflow-controller-rbac/workflow-controller-clusterrole.yaml

@@ -80,6 +80,7 @@ rules:
  - events
  verbs:
  - create
+ - patch
 - apiGroups:
  - "policy"
  resources:

manifests/install.yaml

@@ -250,6 +250,7 @@ rules:
  - events
  verbs:
  - create
+ - patch
 - apiGroups:
  - policy
  resources:

manifests/namespace-install.yaml

@@ -163,6 +163,7 @@ rules:
  - events
  verbs:
  - create
+ - patch
 - apiGroups:
  - policy
  resources:

manifests/namespace-install/workflow-controller-rbac/workflow-controller-role.yaml

@@ -84,6 +84,7 @@ rules:
  - events
  verbs:
  - create
+ - patch
 - apiGroups:
  - "policy"
  resources:

manifests/quick-start-minimal.yaml

@@ -163,6 +163,7 @@ rules:
  - events
  verbs:
  - create
+ - patch
 - apiGroups:
  - policy
  resources:

manifests/quick-start-mysql.yaml

@@ -163,6 +163,7 @@ rules:
  - events
  verbs:
  - create
+ - patch
 - apiGroups:
  - policy
  resources:

manifests/quick-start-postgres.yaml

@@ -163,6 +163,7 @@ rules:
  - events
  verbs:
  - create
+ - patch
 - apiGroups:
  - policy
  resources:

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)

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() {