Check and/or add to sent cloud event cache in one operation

pkg/reconciler/events/cache/cache.go

@@ -31,29 +31,17 @@ type eventData struct {
 	Run *v1alpha1.Run `json:"run,omitempty"`
 }
 
-// AddEventSentToCache adds the particular object to cache marking it as sent
-func AddEventSentToCache(cacheClient *lru.Cache, event *cloudevents.Event) error {
-	if cacheClient == nil {
-		return errors.New("cache client is nil")
-	}
-	eventKey, err := EventKey(event)
-	if err != nil {
-		return err
-	}
-	cacheClient.Add(eventKey, nil)
-	return nil
-}
-
-// IsCloudEventSent checks if the event exists in the cache
-func IsCloudEventSent(cacheClient *lru.Cache, event *cloudevents.Event) (bool, error) {
+// ContainsOrAddCloudEvent checks if the event exists in the cache
+func ContainsOrAddCloudEvent(cacheClient *lru.Cache, event *cloudevents.Event) (bool, error) {
 	if cacheClient == nil {
 		return false, errors.New("cache client is nil")
 	}
 	eventKey, err := EventKey(event)
 	if err != nil {
 		return false, err
 	}
-	return cacheClient.Contains(eventKey), nil
+	isPresent, _ := cacheClient.ContainsOrAdd(eventKey, nil)
+	return isPresent, nil
 }
 
 // EventKey defines whether an event is considered different from another

pkg/reconciler/events/cache/cache_test.go

@@ -146,8 +146,8 @@ func TestAddCheckEvent(t *testing.T) {
 	for _, tc := range testcases {
 		t.Run(tc.name, func(t *testing.T) {
 			testCache, _ := lru.New(10)
-			AddEventSentToCache(testCache, tc.firstEvent)
-			found, _ := IsCloudEventSent(testCache, tc.secondEvent)
+			_, _ = ContainsOrAddCloudEvent(testCache, tc.firstEvent)
+			found, _ := ContainsOrAddCloudEvent(testCache, tc.secondEvent)
 			if d := cmp.Diff(tc.wantFound, found); d != "" {
 				t.Errorf("Cache check failure %s", diff.PrintWantGot(d))
 			}

pkg/reconciler/events/cloudevent/cloud_event_controller.go

@@ -149,7 +149,7 @@ func SendCloudEventWithRetries(ctx context.Context, object runtime.Object) error
 		logger.Debugf("Sending cloudevent of type %q", event.Type())
 		// In case of Run event, check cache if cloudevent is already sent
 		if isRun {
-			cloudEventSent, err := cache.IsCloudEventSent(cacheClient, event)
+			cloudEventSent, err := cache.ContainsOrAddCloudEvent(cacheClient, event)
 			if err != nil {
 				logger.Errorf("error while checking cache: %s", err)
 			}
@@ -163,15 +163,10 @@ func SendCloudEventWithRetries(ctx context.Context, object runtime.Object) error
 			recorder := controller.GetEventRecorder(ctx)
 			if recorder == nil {
 				logger.Warnf("No recorder in context, cannot emit error event")
+				return
 			}
 			recorder.Event(object, corev1.EventTypeWarning, "Cloud Event Failure", result.Error())
 		}
-		// In case of Run event, add to the cache to avoid duplicate events
-		if isRun {
-			if err := cache.AddEventSentToCache(cacheClient, event); err != nil {
-				logger.Errorf("error while adding sent event to cache: %s", err)
-			}
-		}
 	}()
 
 	return <-wasIn

test/events/events.go

@@ -26,6 +26,8 @@ import (
 // in the same order.
 func CheckEventsOrdered(t *testing.T, eventChan chan string, testName string, wantEvents []string) error {
 	t.Helper()
+	// Sleep 50ms to make sure events have delivered
+	time.Sleep(50 * time.Millisecond)
 	err := eventsFromChannel(eventChan, wantEvents)
 	if err != nil {
 		return fmt.Errorf("error in test %s: %v", testName, err)
@@ -37,6 +39,8 @@ func CheckEventsOrdered(t *testing.T, eventChan chan string, testName string, wa
 // were received via the given chan, in any order.
 func CheckEventsUnordered(t *testing.T, eventChan chan string, testName string, wantEvents []string) error {
 	t.Helper()
+	// Sleep 50ms to make sure events have delivered
+	time.Sleep(50 * time.Millisecond)
 	err := eventsFromChannelUnordered(eventChan, wantEvents)
 	if err != nil {
 		return fmt.Errorf("error in test %s: %v", testName, err)