🐛 Fix the issue of event loss caused by time filtering. (#1101)

* skipp the expired event

Signed-off-by: myan <[email protected]>

* fix the ut

Signed-off-by: myan <[email protected]>

* integration

Signed-off-by: myan <[email protected]>

---------

Signed-off-by: myan <[email protected]>

Author: yanmxa

agent/pkg/status/controller/filter/time_filter.go

@@ -22,6 +22,7 @@ var (
 	eventTimeCache         = make(map[string]time.Time)
 	lastEventTimeCache     = make(map[string]time.Time)
 	eventTimeCacheInterval = 5 * time.Second
+	deltaDuration          = 3 * time.Second
 )
 
 // CacheTime cache the latest time
@@ -38,7 +39,10 @@ func Newer(key string, val time.Time) bool {
 	if !ok {
 		return true
 	}
-	return val.After(old)
+
+	// add noise to the time filter to ensure that events occurring very close together in time are not discarded
+	older := old.Add(-deltaDuration)
+	return val.After(older)
 }
 
 // LaunchTimeFilter start a goroutine periodically sync the time filter cache to configMap

agent/pkg/status/controller/filter/time_filter_test.go

@@ -141,4 +141,14 @@ func TestTimeFilter(t *testing.T) {
 	assert.Nil(t, err)
 	assert.True(t, cachedTime.Equal(expiredTime))
 	cancel()
+
+	fmt.Println(">> verify5: don't lose events with similar time")
+	similiarTime := time.Now().Add(10 * time.Second)
+	assert.True(t, Newer(eventType, similiarTime))
+	CacheTime(eventType, similiarTime)
+	assert.True(t, Newer(eventType, similiarTime))
+	CacheTime(eventType, similiarTime.Add(2*time.Second))
+	assert.True(t, Newer(eventType, similiarTime))
+	CacheTime(eventType, similiarTime.Add(5*time.Second))
+	assert.False(t, Newer(eventType, similiarTime))
 }

agent/pkg/status/controller/policies/status_event_emitter.go

@@ -9,6 +9,7 @@ import (
 	cloudevents "github.com/cloudevents/sdk-go/v2"
 	"github.com/go-logr/logr"
 	corev1 "k8s.io/api/core/v1"
+	"k8s.io/klog"
 	policiesv1 "open-cluster-management.io/governance-policy-propagator/api/v1"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
@@ -102,6 +103,7 @@ func (h *statusEventEmitter) Update(obj client.Object) bool {
 			for _, evt := range detail.History {
 				// if the event time is older thant the filter cached sent event time, then skip it
 				if !filter.Newer(h.name, evt.LastTimestamp.Time) {
+					klog.Infof("skip the expired event: %s", evt.EventName)
 					continue
 				}
 

test/integration/agent/status/localpolicy_event_test.go

@@ -91,7 +91,7 @@ var _ = Describe("LocalPolicyEventEmitter", Ordered, func() {
 		Expect(err).Should(Succeed())
 
 		name := strings.Replace(string(enum.LocalRootPolicyEventType), enum.EventTypePrefix, "", -1)
-		filter.CacheTime(name, cachedRootPolicyEvent.CreationTimestamp.Time.Add(5*time.Second))
+		filter.CacheTime(name, cachedRootPolicyEvent.CreationTimestamp.Time.Add(9*time.Second)) // the delta is 3 seconds
 
 		By("Create a expired event")
 		expiredEvent := &corev1.Event{