fix(loki.source.kubernetes): Forward log lines with out-of-order timestamps

internal/component/loki/source/kubernetes/kubetail/tailer.go

@@ -266,9 +266,16 @@ func (t *tailer) tail(ctx context.Context, handler loki.EntryHandler) error {
 
 // processLogStream reads log lines from a reader and processes them.
 // It returns when the context is done, the stream ends, or an error occurs.
-func (t *tailer) processLogStream(ctx context.Context, stream io.ReadCloser, handler loki.EntryHandler, lastReadTime time.Time, positionsEnt positions.Entry, calc *rollingAverageCalculator) error {
+//
+// sinceTime is the time the stream was opened from. The API server applies
+// SinceTime with second precision, so the head of the stream can repeat lines
+// that were already shipped; those are the only lines skipped. Lines are never
+// compared against the previous line: the container runtime stamps stdout and
+// stderr independently, so a live stream is not ordered by timestamp.
+func (t *tailer) processLogStream(ctx context.Context, stream io.ReadCloser, handler loki.EntryHandler, sinceTime time.Time, positionsEnt positions.Entry, calc *rollingAverageCalculator) error {
 	ch := handler.Chan()
 	reader := bufio.NewReader(stream)
+	lastReadTime := sinceTime
 	for {
 		line, err := reader.ReadString('\n')
 
@@ -278,13 +285,11 @@ func (t *tailer) processLogStream(ctx context.Context, stream io.ReadCloser, han
 			calc.AddTimestamp(time.Now())
 
 			entryTimestamp, entryLine := parseKubernetesLog(line)
-			// Skip only if the timestamp is strictly before lastReadTime.
-			// This allows multiple log lines with the same timestamp to be processed,
-			// which is common in Windows containers that log rapidly.
-			if entryTimestamp.Before(lastReadTime) {
+			// Skip only if the timestamp is strictly before sinceTime, so that
+			// lines sharing the resume timestamp are still processed.
+			if entryTimestamp.Before(sinceTime) {
 				continue
 			}
-			lastReadTime = entryTimestamp
 
 			entry := loki.NewEntry(t.lset.Clone(), push.Entry{
 				Timestamp: entryTimestamp,
@@ -295,9 +300,14 @@ func (t *tailer) processLogStream(ctx context.Context, stream io.ReadCloser, han
 			case <-ctx.Done():
 				return nil
 			case ch <- entry:
-				// Save position after it's been sent over the channel.
-				t.opts.Positions.Put(positionsEnt.Path, positionsEnt.Labels, entryTimestamp.UnixMicro())
-				t.target.Report(entryTimestamp, nil)
+				// Save position after it's been sent over the channel. The saved
+				// position seeds sinceTime for the next tail, so it never moves
+				// backwards even when the stream is out of order.
+				if entryTimestamp.After(lastReadTime) {
+					lastReadTime = entryTimestamp
+				}
+				t.opts.Positions.Put(positionsEnt.Path, positionsEnt.Labels, lastReadTime.UnixMicro())
+				t.target.Report(lastReadTime, nil)
 			}
 		}
 

internal/component/loki/source/kubernetes/kubetail/tailer_test.go

@@ -99,6 +99,7 @@ func Test_processLogStream(t *testing.T) {
 		logLines           []string
 		lastReadTime       time.Time
 		expectLines        []string
+		expectLastEntry    time.Time
 	}{
 		{name: "duplicate timestamps are not discarded",
 			logLines: []string{
@@ -154,6 +155,32 @@ func Test_processLogStream(t *testing.T) {
 			expectLines:        []string{"line1\n", "line2\n", "line3\n", "line4\n"},
 			preserveMetaLabels: true,
 		},
+		{
+			// The container runtime stamps stdout and stderr independently, so a
+			// live stream can carry a line whose timestamp precedes the previous
+			// line's. It must still be forwarded, and the reported position must
+			// not move backwards.
+			name: "out-of-order line on a live stream is not discarded",
+			logLines: []string{
+				"2023-01-23T17:00:10.000000000Z line1\n",
+				"2023-01-23T17:00:11.011798230Z stderr_late_stamp\n",
+				"2023-01-23T17:00:11.011741569Z stdout_early_stamp\n",
+				"2023-01-23T17:00:12.000000000Z line4\n",
+			},
+			lastReadTime:    baseTime.Add(-1 * time.Second),
+			expectLines:     []string{"line1\n", "stderr_late_stamp\n", "stdout_early_stamp\n", "line4\n"},
+			expectLastEntry: baseTime.Add(2 * time.Second),
+		},
+		{
+			name: "out-of-order line older than the resume time is discarded",
+			logLines: []string{
+				"2023-01-23T17:00:09.999Z old_line\n",
+				"2023-01-23T17:00:10Z line1\n",
+				"2023-01-23T17:00:09.999Z old_line_again\n",
+			},
+			lastReadTime: baseTime,
+			expectLines:  []string{"line1\n"},
+		},
 	}
 
 	for _, tc := range tt {
@@ -221,6 +248,10 @@ func Test_processLogStream(t *testing.T) {
 
 			require.Equal(t, tc.expectLines, receivedLines, "received lines should match expected lines")
 
+			if !tc.expectLastEntry.IsZero() {
+				require.Eventually(t, func() bool { return target.LastEntry().Equal(tc.expectLastEntry) }, time.Second, 10*time.Millisecond, "LastEntry should be the latest timestamp seen, got %v", target.LastEntry())
+			}
+
 			if tc.preserveMetaLabels {
 				lbls := target.Labels()
 				require.Equal(t, "test", lbls.Get("job"))