app/vlagent/kubernetescollector: fix Kubernetes system logs timestamps parsing

app/vlagent/kubernetescollector/processor.go

@@ -12,6 +12,7 @@ import (
 	"time"
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/fasttime"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/flagutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
 	"github.com/valyala/fastjson"
@@ -252,7 +253,9 @@ func parseLogRowContent(p *logstorage.JSONParser, data []byte) (int64, bool) {
 
 		return timestamp, true
 	case 'I', 'W', 'E', 'F':
-		timestamp, fields, ok := tryParseKlog(p.Fields, bytesutil.ToUnsafeString(data))
+		ts := fasttime.UnixTimestamp()
+		current := time.Unix(int64(ts), 0).UTC()
+		timestamp, fields, ok := tryParseKlog(p.Fields, bytesutil.ToUnsafeString(data), current)
 		if !ok {
 			return 0, false
 		}
@@ -265,7 +268,7 @@ func parseLogRowContent(p *logstorage.JSONParser, data []byte) (int64, bool) {
 
 // tryParseKlog parses the given string in Kubernetes Log format and returns the parsed fields.
 // See https://github.com/kubernetes/klog/
-func tryParseKlog(dst []logstorage.Field, src string) (int64, []logstorage.Field, bool) {
+func tryParseKlog(dst []logstorage.Field, src string, current time.Time) (int64, []logstorage.Field, bool) {
 	if len(src) < len("I0101 00:00:00.000000 1 p:1] m") {
 		return 0, nil, false
 	}
@@ -282,7 +285,11 @@ func tryParseKlog(dst []logstorage.Field, src string) (int64, []logstorage.Field
 		return 0, nil, false
 	}
 	src = src[len("0102 15:04:05.000000"):]
-	t = t.AddDate(time.Now().Year(), 0, 0)
+	t = t.AddDate(current.Year(), 0, 0)
+	if t.Add(-time.Hour * 24).After(current) {
+		// Adjust time to the previous year.
+		t = t.AddDate(-1, 0, 0)
+	}
 	timestamp := t.UnixNano()
 
 	// Remove trailing spaces.

app/vlagent/kubernetescollector/processor_test.go

@@ -109,10 +109,12 @@ func TestProcessor(t *testing.T) {
 }
 
 func TestParseKlog(t *testing.T) {
+	current := time.Date(1971, time.December, 20, 0, 0, 0, 0, time.UTC)
+
 	f := func(src, fieldsExpected string, timestampExpected int64) {
 		t.Helper()
 
-		timestamp, fields, ok := tryParseKlog(nil, src)
+		timestamp, fields, ok := tryParseKlog(nil, src, current)
 		if !ok {
 			t.Fatalf("cannot parse klog line %q", src)
 		}
@@ -130,51 +132,45 @@ func TestParseKlog(t *testing.T) {
 	// Parse simple line
 	in := `I1215 07:34:12.017826       94 serving.go:374] foobar`
 	want := `{"level":"INFO","thread_id":"94","source_line":"serving.go:374","_msg":"foobar"}`
-	timestampExpected := expectedKlogTimestamp("1215", "07:34:12.017826")
+	timestampExpected := int64(61630452017826000)
 	f(in, want, timestampExpected)
 
 	// Parse multiple words
 	in = `I1215 07:34:12.017826       24 serving.go:374] Generated self-signed cert (/tmp/apiserver.crt, /tmp/apiserver.key)`
 	want = `{"level":"INFO","thread_id":"24","source_line":"serving.go:374","_msg":"Generated self-signed cert (/tmp/apiserver.crt, /tmp/apiserver.key)"}`
-	timestampExpected = expectedKlogTimestamp("1215", "07:34:12.017826")
+	timestampExpected = 61630452017826000
 	f(in, want, timestampExpected)
 
 	// Parse key="value" pair
 	in = `I1215 07:34:11.695645       42 controller.go:824] "Starting provisioner controller" component="rancher.io/local-path_local-path-provisioner-5cf85fd84d-bf8vk_626b5057-e081-4b71-9a19-5e371ae0211b"`
 	want = `{"level":"INFO","thread_id":"42","source_line":"controller.go:824","_msg":"Starting provisioner controller","component":"rancher.io/local-path_local-path-provisioner-5cf85fd84d-bf8vk_626b5057-e081-4b71-9a19-5e371ae0211b"}`
-	timestampExpected = expectedKlogTimestamp("1215", "07:34:11.695645")
+	timestampExpected = 61630451695645000
 	f(in, want, timestampExpected)
 
 	// Parse key="value" pairs
 	in = `I1215 10:34:26.907803       1 server.go:191] "Failed probe" probe="metric-storage-ready" err="no metrics to serve"`
 	want = `{"level":"INFO","thread_id":"1","source_line":"server.go:191","_msg":"Failed probe","probe":"metric-storage-ready","err":"no metrics to serve"}`
-	timestampExpected = expectedKlogTimestamp("1215", "10:34:26.907803")
+	timestampExpected = 61641266907803000
 	f(in, want, timestampExpected)
 
 	// Parse quoted msg without additional fields
 	in = `I1215 07:34:12.324492       1234 tlsconfig.go:240] "Starting DynamicServingCertificateController"`
 	want = `{"level":"INFO","thread_id":"1234","source_line":"tlsconfig.go:240","_msg":"Starting DynamicServingCertificateController"}`
-	timestampExpected = expectedKlogTimestamp("1215", "07:34:12.324492")
+	timestampExpected = 61630452324492000
 	f(in, want, timestampExpected)
-}
 
-// expectedKlogTimestamp calculates the expected klog timestamp for the given month/day and time.
-// This matches the logic in tryParseKlog which uses time.Now().Year() to determine the year.
-func expectedKlogTimestamp(monthDay, timeStr string) int64 {
-	s := monthDay + " " + timeStr
-	t, err := time.ParseInLocation("0102 15:04:05.000000", s, time.UTC)
-	if err != nil {
-		panic(fmt.Sprintf("failed to parse timestamp %q: %s", s, err))
-	}
-	t = t.AddDate(time.Now().Year(), 0, 0)
-	return t.UnixNano()
+	// Adjust time to the previous year
+	in = `I1221 00:00:00.000001       1234 main.go:1] foo`
+	want = `{"level":"INFO","thread_id":"1234","source_line":"main.go:1","_msg":"foo"}`
+	timestampExpected = 30585600000001000
+	f(in, want, timestampExpected)
 }
 
 func TestParseKlogFailure(t *testing.T) {
 	f := func(src string) {
 		t.Helper()
 
-		_, fields, ok := tryParseKlog(nil, src)
+		_, fields, ok := tryParseKlog(nil, src, time.Now())
 		if ok {
 			got := logstorage.MarshalFieldsToJSON(nil, fields)
 			t.Fatalf("unexpected success; got\n%s", got)

docs/victorialogs/CHANGELOG.md

@@ -37,6 +37,7 @@ according to the following docs:
 * BUGFIX: [LogsQL](https://docs.victoriametrics.com/victorialogs/logsql/): use local time zone for the VictoriaLogs server when the [`day_range`](https://docs.victoriametrics.com/victorialogs/logsql/#day-range-filter) or [`week_range`](https://docs.victoriametrics.com/victorialogs/logsql/#week-range-filter) filter doesn't contain explicitly specified `offset ...` suffix. This aligns with the behaviour when the timezone information is missing in the [`_time` filter](https://docs.victoriametrics.com/victorialogs/logsql/#time-filter).
 * BUGFIX: [web UI](https://docs.victoriametrics.com/victorialogs/querying/#web-ui): fix bars width calculation and visual misalignment relative to time axis. See [#900](https://github.com/VictoriaMetrics/VictoriaLogs/issues/900).
 * BUGFIX: [metrics](https://docs.victoriametrics.com/victorialogs/metrics/): fix `vl_http_errors_total{path="..."}` metric name mismatch for `/internal/select/*` endpoints (it was exposed as `vl_http_request_errors_total{path="..."}`). See [#1005](https://github.com/VictoriaMetrics/VictoriaLogs/pull/1005).
+* BUGFIX: [Kubernetes Collector](https://docs.victoriametrics.com/victorialogs/vlagent/#collect-kubernetes-pod-logs): properly parse [Kubernetes system log](https://kubernetes.io/docs/concepts/cluster-administration/system-logs/) timestamps. This prevents future logs issues that occurred during the New Year transition.
 * BUGFIX: [Kubernetes Collector](https://docs.victoriametrics.com/victorialogs/vlagent/#collect-kubernetes-pod-logs): add additional file fingerprint validation when resuming from a stale checkpoint file. This prevents `cannot parse Container Runtime Interface log line` errors that could occur if multiple log rotations happened while vlagent was down.
 
 ## [v1.43.1](https://github.com/VictoriaMetrics/VictoriaLogs/releases/tag/v1.43.1)