add future time filter for log watchers when time across the year

Author: smileusd

pkg/systemlogmonitor/logwatchers/filelog/log_watcher.go

@@ -131,6 +131,13 @@ func (s *filelogWatcher) watchLoop() {
 			klog.V(5).Infof("Throwing away msg %q before start time: %v < %v", log.Message, log.Timestamp, s.startTime)
 			continue
 		}
+
+		// Discard messages after now, lots of log files are not record year. 1 min cover log write latency
+		if log.Timestamp.After(time.Now().Add(time.Minute)) {
+			klog.V(5).Infof("Throwing away msg %q after current time: %v < %v", log.Message, log.Timestamp, time.Now())
+			continue
+		}
+
 		s.logCh <- log
 	}
 }

pkg/systemlogmonitor/logwatchers/journald/log_watcher.go

@@ -130,6 +130,12 @@ func (j *journaldWatcher) watchLoop() {
 			continue
 		}
 
+		// Discard messages after now, the journal log is not record year. 1 min cover log write latency
+		if entry.RealtimeTimestamp > timeToJournalTimestamp(time.Now().Add(time.Minute)) {
+			klog.V(5).Infof("Throwing away msg %q after current time: %v < %v", entry.Fields[sdjournal.SD_JOURNAL_FIELD_MESSAGE], entry.RealtimeTimestamp, time.Now())
+			continue
+		}
+
 		j.logCh <- translate(entry)
 	}
 }

pkg/systemlogmonitor/logwatchers/kmsg/log_watcher_linux.go

@@ -115,6 +115,12 @@ func (k *kernelLogWatcher) watchLoop() {
 				continue
 			}
 
+			// Discard messages after now, lots of log files are not record year. 1 min cover log write latency
+			if msg.Timestamp.After(time.Now().Add(time.Minute)) {
+				klog.V(5).Infof("Throwing away msg %q after current time: %v < %v", msg.Message, msg.Timestamp, time.Now())
+				continue
+			}
+
 			k.logCh <- &logtypes.Log{
 				Message:   strings.TrimSpace(msg.Message),
 				Timestamp: msg.Timestamp,