diff --git a/internal/component/loki/source/kubernetes/kubetail/tailer.go b/internal/component/loki/source/kubernetes/kubetail/tailer.go index 9257a50eff5..12c60917bb3 100644 --- a/internal/component/loki/source/kubernetes/kubetail/tailer.go +++ b/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) } } diff --git a/internal/component/loki/source/kubernetes/kubetail/tailer_test.go b/internal/component/loki/source/kubernetes/kubetail/tailer_test.go index d94b8d1344c..d8891ac1b52 100644 --- a/internal/component/loki/source/kubernetes/kubetail/tailer_test.go +++ b/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"))