Enhance filter check in k8s event metricset (#29470) (#29482)

CHANGELOG.next.asciidoc

@@ -194,6 +194,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 - Groups same timestamp metric values to one event in the app_insights metricset. {pull}20403[20403]
 - Fix in rename processor to ingest metrics for `write.iops` to proper field instead of `write_iops` in rds metricset. {pull}28960[28960]
 - Use xpack.enabled on SM modules to write into .monitoring indices when using Metricbeat standalone {pull}28365[28365]
+- Enhance filter check in kubernetes event metricset. {pull}29470[29470]
 
 *Packetbeat*
 

libbeat/common/kubernetes/types.go

@@ -94,6 +94,11 @@ func Time(t *metav1.Time) time.Time {
 	return t.Time
 }
 
+// MicroTime extracts time from k8s.MicroTime type
+func MicroTime(t *metav1.MicroTime) time.Time {
+	return t.Time
+}
+
 // ContainerID parses the container ID to get the actual ID string
 func ContainerID(s PodContainerStatus) string {
 	cID, _ := ContainerIDWithRuntime(s)

metricbeat/module/kubernetes/event/event.go

@@ -108,8 +108,16 @@ func (m *MetricSet) Run(reporter mb.PushReporter) {
 	m.watcher.AddEventHandler(kubernetes.FilteringResourceEventHandler{
 		FilterFunc: func(obj interface{}) bool {
 			eve := obj.(*kubernetes.Event)
+			// if fields are null they are decoded to `0001-01-01 00:00:00 +0000 UTC`
+			// so we need to check if they are valid first
+			lastTimestampValid := !kubernetes.Time(&eve.LastTimestamp).IsZero()
+			eventTimeValid := !kubernetes.MicroTime(&eve.EventTime).IsZero()
 			// if skipOlder, skip events happened before watch
-			if m.skipOlder && kubernetes.Time(&eve.LastTimestamp).Before(now) {
+			if m.skipOlder && kubernetes.Time(&eve.LastTimestamp).Before(now) && lastTimestampValid {
+				return false
+			} else if m.skipOlder && kubernetes.MicroTime(&eve.EventTime).Before(now) && eventTimeValid {
+				// there might be cases that `LastTimestamp` is not a valid number so double check
+				// with `EventTime`
 				return false
 			}
 			return true