deduplicate subsequent component events (#242)

Author: cbarbian-sap

internal/events/recorder.go

@@ -0,0 +1,78 @@
+/*
+SPDX-FileCopyrightText: 2024 SAP SE or an SAP affiliate company and component-operator-runtime contributors
+SPDX-License-Identifier: Apache-2.0
+*/
+
+package events
+
+import (
+	"fmt"
+	"sync"
+	"time"
+
+	"k8s.io/client-go/tools/record"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+)
+
+type DeduplicatingRecorder struct {
+	recorder record.EventRecorder
+	mutex    sync.Mutex
+	events   map[string]event
+}
+
+type event struct {
+	digest    string
+	timestamp time.Time
+}
+
+func NewDeduplicatingRecorder(recorder record.EventRecorder) *DeduplicatingRecorder {
+	return &DeduplicatingRecorder{
+		recorder: recorder,
+		events:   make(map[string]event),
+	}
+}
+
+func (r *DeduplicatingRecorder) Event(object client.Object, eventType string, reason string, message string) {
+	if r.isDuplicate(object, nil, eventType, reason, message) {
+		return
+	}
+	r.recorder.Event(object, eventType, reason, message)
+}
+
+func (r *DeduplicatingRecorder) Eventf(object client.Object, eventType string, reason string, messageFmt string, args ...any) {
+	if r.isDuplicate(object, nil, eventType, reason, fmt.Sprintf(messageFmt, args...)) {
+		return
+	}
+	r.recorder.Eventf(object, eventType, reason, messageFmt, args...)
+}
+
+func (r *DeduplicatingRecorder) AnnotatedEventf(object client.Object, annotations map[string]string, eventType string, reason string, messageFmt string, args ...any) {
+	if r.isDuplicate(object, annotations, eventType, reason, fmt.Sprintf(messageFmt, args...)) {
+		return
+	}
+	r.recorder.AnnotatedEventf(object, annotations, eventType, reason, messageFmt, args...)
+}
+
+func (r *DeduplicatingRecorder) isDuplicate(object client.Object, annotations map[string]string, eventType, reason, message string) bool {
+	uid := string(object.GetUID())
+	digest := calculateDigest(annotations, eventType, reason, message)
+	now := time.Now()
+	exp := time.Now().Add(-5 * time.Minute)
+
+	r.mutex.Lock()
+	defer r.mutex.Unlock()
+	for uid, event := range r.events {
+		if event.timestamp.Before(exp) {
+			delete(r.events, uid)
+		}
+	}
+	if r.events[uid].digest == digest {
+		return true
+	} else {
+		r.events[uid] = event{
+			digest:    digest,
+			timestamp: now,
+		}
+		return false
+	}
+}

internal/events/util.go

@@ -0,0 +1,31 @@
+/*
+SPDX-FileCopyrightText: 2024 SAP SE or an SAP affiliate company and component-operator-runtime contributors
+SPDX-License-Identifier: Apache-2.0
+*/
+
+package events
+
+import (
+	"crypto/sha256"
+	"encoding/hex"
+	"encoding/json"
+)
+
+// TODO: consolidate all the util files into an internal reuse package
+
+func must[T any](x T, err error) T {
+	if err != nil {
+		panic(err)
+	}
+	return x
+}
+
+func sha256hex(data []byte) string {
+	sum := sha256.Sum256(data)
+	return hex.EncodeToString(sum[:])
+}
+
+func calculateDigest(values ...any) string {
+	// note: this must() is ok because the input values are expected to be JSON values
+	return sha256hex(must(json.Marshal(values)))
+}

pkg/component/reconciler.go

@@ -41,6 +41,7 @@ import (
 
 	"github.com/sap/component-operator-runtime/internal/backoff"
 	"github.com/sap/component-operator-runtime/internal/clientfactory"
+	"github.com/sap/component-operator-runtime/internal/events"
 	"github.com/sap/component-operator-runtime/internal/metrics"
 	"github.com/sap/component-operator-runtime/pkg/cluster"
 	"github.com/sap/component-operator-runtime/pkg/manifests"
@@ -134,6 +135,7 @@ type Reconciler[T Component] struct {
 	groupVersionKind   schema.GroupVersionKind
 	controllerName     string
 	client             cluster.Client
+	eventRecorder      events.DeduplicatingRecorder
 	resourceGenerator  manifests.Generator
 	statusAnalyzer     status.StatusAnalyzer
 	options            ReconcilerOptions
@@ -328,9 +330,9 @@ func (r *Reconciler[T]) Reconcile(ctx context.Context, req ctrl.Request) (result
 		// such as the flux notfication recorder; should we therefore send the events asynchronously, or start synchronously and continue asynchronous
 		// after a little while?
 		if state == StateError {
-			r.client.EventRecorder().AnnotatedEventf(component, eventAnnotations, corev1.EventTypeWarning, reason, message)
+			r.eventRecorder.AnnotatedEventf(component, eventAnnotations, corev1.EventTypeWarning, reason, "%s", message)
 		} else {
-			r.client.EventRecorder().AnnotatedEventf(component, eventAnnotations, corev1.EventTypeNormal, reason, message)
+			r.eventRecorder.AnnotatedEventf(component, eventAnnotations, corev1.EventTypeNormal, reason, "%s", message)
 		}
 
 		if skipStatusUpdate {
@@ -617,6 +619,7 @@ func (r *Reconciler[T]) SetupWithManagerAndBuilder(mgr ctrl.Manager, blder *ctrl
 		}
 		r.client = clnt
 	}
+	r.eventRecorder = *events.NewDeduplicatingRecorder(r.client.EventRecorder())
 
 	component := newComponent[T]()
 	r.groupVersionKind, err = apiutil.GVKForObject(component, r.client.Scheme())