Merge pull request #119375 from dgrisonnet/automated-cherry-pick-of-#…

…114237-#114236-#112334-upstream-release-1.26

Automated cherry pick of #114237: tools/events: retry on AlreadyExist for Series
#114236: tools/events: fix data race when emitting series
#112334: events: fix EventSeries starting count discrepancy

Kubernetes-commit: 694c7d3710afaafae8754356d86b35e93bb87658

tools/events/event_broadcaster.go

@@ -181,22 +181,24 @@ func (e *eventBroadcasterImpl) recordToSink(event *eventsv1.Event, clock clock.C
 					return nil
 				}
 				isomorphicEvent.Series = &eventsv1.EventSeries{
-					Count:            1,
+					Count:            2,
 					LastObservedTime: metav1.MicroTime{Time: clock.Now()},
 				}
-				return isomorphicEvent
+				// Make a copy of the Event to make sure that recording it
+				// doesn't mess with the object stored in cache.
+				return isomorphicEvent.DeepCopy()
 			}
 			e.eventCache[eventKey] = eventCopy
-			return eventCopy
+			// Make a copy of the Event to make sure that recording it doesn't
+			// mess with the object stored in cache.
+			return eventCopy.DeepCopy()
 		}()
 		if evToRecord != nil {
-			recordedEvent := e.attemptRecording(evToRecord)
-			if recordedEvent != nil {
-				recordedEventKey := getKey(recordedEvent)
-				e.mu.Lock()
-				defer e.mu.Unlock()
-				e.eventCache[recordedEventKey] = recordedEvent
-			}
+			// TODO: Add a metric counting the number of recording attempts
+			e.attemptRecording(evToRecord)
+			// We don't want the new recorded Event to be reflected in the
+			// client's cache because server-side mutations could mess with the
+			// aggregation mechanism used by the client.
 		}
 	}()
 }
@@ -248,6 +250,14 @@ func recordEvent(sink EventSink, event *eventsv1.Event) (*eventsv1.Event, bool)
 		return nil, false
 	case *errors.StatusError:
 		if errors.IsAlreadyExists(err) {
+			// If we tried to create an Event from an EventSerie, it means that
+			// the original Patch request failed because the Event we were
+			// trying to patch didn't exist. If the creation failed because the
+			// Event now exists, it is safe to retry.  This occurs when a new
+			// Event is emitted twice in a very short period of time.
+			if isEventSeries {
+				return nil, true
+			}
 			klog.V(5).Infof("Server rejected event '%#v': '%v' (will not retry!)", event, err)
 		} else {
 			klog.Errorf("Server rejected event '%#v': '%v' (will not retry!)", event, err)

tools/events/event_broadcaster_test.go

@@ -0,0 +1,103 @@
+/*
+Copyright 2022 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package events
+
+import (
+	"context"
+	"reflect"
+	"testing"
+
+	eventsv1 "k8s.io/api/events/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/util/diff"
+	"k8s.io/client-go/kubernetes/fake"
+)
+
+func TestRecordEventToSink(t *testing.T) {
+	nonIsomorphicEvent := eventsv1.Event{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "test",
+			Namespace: metav1.NamespaceDefault,
+		},
+		Series: nil,
+	}
+
+	isomorphicEvent := *nonIsomorphicEvent.DeepCopy()
+	isomorphicEvent.Series = &eventsv1.EventSeries{Count: 2}
+
+	testCases := []struct {
+		name                  string
+		eventsToRecord        []eventsv1.Event
+		expectedRecordedEvent eventsv1.Event
+	}{
+		{
+			name: "record one Event",
+			eventsToRecord: []eventsv1.Event{
+				nonIsomorphicEvent,
+			},
+			expectedRecordedEvent: nonIsomorphicEvent,
+		},
+		{
+			name: "record one Event followed by an isomorphic one",
+			eventsToRecord: []eventsv1.Event{
+				nonIsomorphicEvent,
+				isomorphicEvent,
+			},
+			expectedRecordedEvent: isomorphicEvent,
+		},
+		{
+			name: "record one isomorphic Event before the original",
+			eventsToRecord: []eventsv1.Event{
+				isomorphicEvent,
+				nonIsomorphicEvent,
+			},
+			expectedRecordedEvent: isomorphicEvent,
+		},
+		{
+			name: "record one isomorphic Event without one already existing",
+			eventsToRecord: []eventsv1.Event{
+				isomorphicEvent,
+			},
+			expectedRecordedEvent: isomorphicEvent,
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			kubeClient := fake.NewSimpleClientset()
+			eventSink := &EventSinkImpl{Interface: kubeClient.EventsV1()}
+
+			for _, ev := range tc.eventsToRecord {
+				recordEvent(eventSink, &ev)
+			}
+
+			recordedEvents, err := kubeClient.EventsV1().Events(metav1.NamespaceDefault).List(context.TODO(), metav1.ListOptions{})
+			if err != nil {
+				t.Errorf("expected to be able to list Events from fake client")
+			}
+
+			if len(recordedEvents.Items) != 1 {
+				t.Errorf("expected one Event to be recorded, found: %d", len(recordedEvents.Items))
+			}
+
+			recordedEvent := recordedEvents.Items[0]
+			if !reflect.DeepEqual(recordedEvent, tc.expectedRecordedEvent) {
+				t.Errorf("expected to have recorded Event: %#+v, got: %#+v\n diff: %s", tc.expectedRecordedEvent, recordedEvent, diff.ObjectReflectDiff(tc.expectedRecordedEvent, recordedEvent))
+			}
+		})
+	}
+}

tools/events/eventseries_test.go

@@ -17,6 +17,7 @@ limitations under the License.
 package events
 
 import (
+	"context"
 	"strconv"
 	"testing"
 	"time"
@@ -29,6 +30,7 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	k8sruntime "k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/util/wait"
+	fake "k8s.io/client-go/kubernetes/fake"
 	"k8s.io/client-go/kubernetes/scheme"
 	restclient "k8s.io/client-go/rest"
 	ref "k8s.io/client-go/tools/reference"
@@ -106,7 +108,7 @@ func TestEventSeriesf(t *testing.T) {
 	nonIsomorphicEvent := expectedEvent.DeepCopy()
 	nonIsomorphicEvent.Action = "stopped"
 
-	expectedEvent.Series = &eventsv1.EventSeries{Count: 1}
+	expectedEvent.Series = &eventsv1.EventSeries{Count: 2}
 	table := []struct {
 		regarding    k8sruntime.Object
 		related      k8sruntime.Object
@@ -185,6 +187,44 @@ func TestEventSeriesf(t *testing.T) {
 	close(stopCh)
 }
 
+// TestEventSeriesWithEventSinkImplRace verifies that when Events are emitted to
+// an EventSink consecutively there is no data race.  This test is meant to be
+// run with the `-race` option.
+func TestEventSeriesWithEventSinkImplRace(t *testing.T) {
+	kubeClient := fake.NewSimpleClientset()
+
+	eventSink := &EventSinkImpl{Interface: kubeClient.EventsV1()}
+	eventBroadcaster := NewBroadcaster(eventSink)
+
+	stopCh := make(chan struct{})
+	eventBroadcaster.StartRecordingToSink(stopCh)
+
+	recorder := eventBroadcaster.NewRecorder(scheme.Scheme, "test")
+
+	recorder.Eventf(&v1.ObjectReference{}, nil, v1.EventTypeNormal, "reason", "action", "", "")
+	recorder.Eventf(&v1.ObjectReference{}, nil, v1.EventTypeNormal, "reason", "action", "", "")
+
+	err := wait.PollImmediate(100*time.Millisecond, 5*time.Second, func() (done bool, err error) {
+		events, err := kubeClient.EventsV1().Events(metav1.NamespaceDefault).List(context.TODO(), metav1.ListOptions{})
+		if err != nil {
+			return false, err
+		}
+
+		if len(events.Items) != 1 {
+			return false, nil
+		}
+
+		if events.Items[0].Series == nil {
+			return false, nil
+		}
+
+		return true, nil
+	})
+	if err != nil {
+		t.Fatal("expected that 2 identical Eventf calls would result in the creation of an Event with a Serie")
+	}
+}
+
 func validateEvent(messagePrefix string, expectedUpdate bool, actualEvent *eventsv1.Event, expectedEvent *eventsv1.Event, t *testing.T) {
 	recvEvent := *actualEvent