feat: Collect duplicate log line metrics

pkg/chunkenc/interface.go

@@ -21,6 +21,7 @@ var (
 	ErrInvalidSize     = errors.New("invalid size")
 	ErrInvalidFlag     = errors.New("invalid flag")
 	ErrInvalidChecksum = errors.New("invalid chunk checksum")
+	ErrDuplicateEntry  = errors.New("duplicate entry")
 )
 
 type errTooFarBehind struct {
@@ -48,6 +49,31 @@ func IsOutOfOrderErr(err error) bool {
 	return err == ErrOutOfOrder || IsErrTooFarBehind(err)
 }
 
+type errDuplicateEntry struct {
+	// original timestamp of the entry itself.
+	entryTs time.Time
+
+	// string representation of the stream for the entry
+	stream string
+}
+
+func IsErrDuplicateEntry(err error) bool {
+	_, ok := err.(*errDuplicateEntry)
+	return ok
+}
+
+func ErrDuplicateLogEntry(entryTs time.Time, stream string) error {
+	return &errDuplicateEntry{entryTs: entryTs, stream: stream}
+}
+
+func (m *errDuplicateEntry) Error() string {
+	return fmt.Sprintf("entry is a duplicate, entry timestamp is: %s, stream information is: %s", m.entryTs.Format(time.RFC3339), m.stream)
+}
+
+func IsDuplicateEntryErr(err error) bool {
+	return err == ErrDuplicateEntry || IsErrDuplicateEntry(err)
+}
+
 // Encoding is the identifier for a chunk encoding.
 type Encoding byte
 

pkg/chunkenc/interface_test.go

@@ -37,3 +37,13 @@ func TestIsOutOfOrderErr(t *testing.T) {
 		require.Equal(t, true, IsOutOfOrderErr(err))
 	}
 }
+
+func TestIsDuplicateEntryErr(t *testing.T) {
+	if !IsDuplicateEntryErr(ErrDuplicateEntry) {
+		t.Errorf("IsDuplicateEntryErr() = false, want true for ErrDuplicateEntry")
+	}
+
+	if IsDuplicateEntryErr(ErrOutOfOrder) {
+		t.Errorf("IsDuplicateEntryErr() = true, want false for errors other than ErrDuplicateEntry")
+	}
+}

pkg/chunkenc/unordered.go

@@ -135,7 +135,7 @@ func (hb *unorderedHeadBlock) Append(ts int64, line string, structuredMetadata l
 		for _, et := range displaced[0].(*nsEntries).entries {
 			if et.line == line {
 				e.entries = displaced[0].(*nsEntries).entries
-				return nil
+				return ErrDuplicateEntry
 			}
 		}
 		e.entries = append(displaced[0].(*nsEntries).entries, nsEntry{line, hb.symbolizer.Add(structuredMetadata)})

pkg/chunkenc/unordered_test.go

@@ -86,6 +86,7 @@ func Test_Unordered_InsertRetrieval(t *testing.T) {
 		desc       string
 		input, exp []entry
 		dir        logproto.Direction
+		hasDup     bool
 	}{
 		{
 			desc: "simple forward",
@@ -152,7 +153,8 @@ func Test_Unordered_InsertRetrieval(t *testing.T) {
 			exp: []entry{
 				{0, "a", nil}, {0, "b", nil}, {1, "c", nil},
 			},
-			dir: logproto.FORWARD,
+			dir:    logproto.FORWARD,
+			hasDup: true,
 		},
 		{
 			desc: "ts remove exact dupe backward",
@@ -162,7 +164,8 @@ func Test_Unordered_InsertRetrieval(t *testing.T) {
 			exp: []entry{
 				{1, "c", nil}, {0, "b", nil}, {0, "a", nil},
 			},
-			dir: logproto.BACKWARD,
+			dir:    logproto.BACKWARD,
+			hasDup: true,
 		},
 	} {
 		t.Run(tc.desc, func(t *testing.T) {
@@ -173,7 +176,14 @@ func Test_Unordered_InsertRetrieval(t *testing.T) {
 				t.Run(format.String(), func(t *testing.T) {
 					hb := newUnorderedHeadBlock(format, newSymbolizer())
 					for _, e := range tc.input {
-						require.Nil(t, hb.Append(e.t, e.s, e.structuredMetadata))
+						err := hb.Append(e.t, e.s, e.structuredMetadata)
+						if tc.hasDup {
+							if err != nil && err != ErrDuplicateEntry {
+								require.Equal(t, err, ErrDuplicateEntry)
+							}
+						} else {
+							require.Nil(t, err)
+						}
 					}
 
 					itr := hb.Iterator(

pkg/distributor/writefailures/manager.go

@@ -39,7 +39,8 @@ func (m *Manager) Log(tenantID string, err error) {
 		return
 	}
 
-	if !m.tenantCfgs.LimitedLogPushErrors(tenantID) {
+	if !m.tenantCfgs.LimitedLogPushErrors(tenantID) &&
+		!m.tenantCfgs.LogDuplicateStreamInfo(tenantID) {
 		return
 	}
 

pkg/ingester/instance.go

@@ -320,7 +320,7 @@ func (i *instance) createStream(ctx context.Context, pushReqStream logproto.Stre
 		return nil, fmt.Errorf("failed to create stream: %w", err)
 	}
 
-	s := newStream(chunkfmt, headfmt, i.cfg, i.limiter, i.instanceID, fp, sortedLabels, i.limiter.UnorderedWrites(i.instanceID), i.streamRateCalculator, i.metrics, i.writeFailures)
+	s := newStream(chunkfmt, headfmt, i.cfg, i.limiter, i.instanceID, fp, sortedLabels, i.limiter.UnorderedWrites(i.instanceID), i.streamRateCalculator, i.metrics, i.writeFailures, i.configs)
 
 	// record will be nil when replaying the wal (we don't want to rewrite wal entries as we replay them).
 	if record != nil {
@@ -358,7 +358,7 @@ func (i *instance) createStreamByFP(ls labels.Labels, fp model.Fingerprint) (*st
 		return nil, fmt.Errorf("failed to create stream for fingerprint: %w", err)
 	}
 
-	s := newStream(chunkfmt, headfmt, i.cfg, i.limiter, i.instanceID, fp, sortedLabels, i.limiter.UnorderedWrites(i.instanceID), i.streamRateCalculator, i.metrics, i.writeFailures)
+	s := newStream(chunkfmt, headfmt, i.cfg, i.limiter, i.instanceID, fp, sortedLabels, i.limiter.UnorderedWrites(i.instanceID), i.streamRateCalculator, i.metrics, i.writeFailures, i.configs)
 
 	i.streamsCreatedTotal.Inc()
 	memoryStreams.WithLabelValues(i.instanceID).Inc()

pkg/ingester/instance_test.go

@@ -308,7 +308,7 @@ func setupTestStreams(t *testing.T) (*instance, time.Time, int) {
 		require.NoError(t, err)
 		chunkfmt, headfmt, err := instance.chunkFormatAt(minTs(&testStream))
 		require.NoError(t, err)
-		chunk := newStream(chunkfmt, headfmt, cfg, limiter, "fake", 0, nil, true, NewStreamRateCalculator(), NilMetrics, nil).NewChunk()
+		chunk := newStream(chunkfmt, headfmt, cfg, limiter, "fake", 0, nil, true, NewStreamRateCalculator(), NilMetrics, nil, nil).NewChunk()
 		for _, entry := range testStream.Entries {
 			err = chunk.Append(&entry)
 			require.NoError(t, err)
@@ -567,7 +567,7 @@ func Benchmark_instance_addNewTailer(b *testing.B) {
 
 	b.Run("addTailersToNewStream", func(b *testing.B) {
 		for n := 0; n < b.N; n++ {
-			inst.addTailersToNewStream(newStream(chunkfmt, headfmt, nil, limiter, "fake", 0, lbs, true, NewStreamRateCalculator(), NilMetrics, nil))
+			inst.addTailersToNewStream(newStream(chunkfmt, headfmt, nil, limiter, "fake", 0, lbs, true, NewStreamRateCalculator(), NilMetrics, nil, nil))
 		}
 	})
 }

pkg/ingester/stream.go

@@ -8,6 +8,8 @@ import (
 	"sync"
 	"time"
 
+	"github.com/grafana/loki/v3/pkg/runtime"
+
 	"github.com/go-kit/log/level"
 	"github.com/grafana/dskit/httpgrpc"
 	"github.com/opentracing/opentracing-go"
@@ -78,6 +80,8 @@ type stream struct {
 
 	chunkFormat          byte
 	chunkHeadBlockFormat chunkenc.HeadBlockFmt
+
+	configs *runtime.TenantConfigs
 }
 
 type chunkDesc struct {
@@ -107,6 +111,7 @@ func newStream(
 	streamRateCalculator *StreamRateCalculator,
 	metrics *ingesterMetrics,
 	writeFailures *writefailures.Manager,
+	configs *runtime.TenantConfigs,
 ) *stream {
 	hashNoShard, _ := labels.HashWithoutLabels(make([]byte, 0, 1024), ShardLbName)
 	return &stream{
@@ -126,6 +131,8 @@ func newStream(
 		writeFailures:        writeFailures,
 		chunkFormat:          chunkFormat,
 		chunkHeadBlockFormat: headBlockFmt,
+
+		configs: configs,
 	}
 }
 
@@ -334,13 +341,23 @@ func (s *stream) storeEntries(ctx context.Context, entries []logproto.Entry, usa
 
 		chunk.lastUpdated = time.Now()
 		if err := chunk.chunk.Append(&entries[i]); err != nil {
-			invalid = append(invalid, entryWithError{&entries[i], err})
-			if chunkenc.IsOutOfOrderErr(err) {
-				s.writeFailures.Log(s.tenant, err)
-				outOfOrderSamples++
-				outOfOrderBytes += len(entries[i].Line)
+			if chunkenc.IsDuplicateEntryErr(err) {
+				if s.configs.LogDuplicateMetrics(s.tenant) {
+					s.reportDuplicateMetrics(len(entries[i].Line))
+				}
+				if s.configs.LogDuplicateStreamInfo(s.tenant) {
+					err = chunkenc.ErrDuplicateLogEntry(entries[i].Timestamp, s.labelsString)
+					s.writeFailures.Log(s.tenant, err)
+				}
+			} else {
+				invalid = append(invalid, entryWithError{&entries[i], err})
+				if chunkenc.IsOutOfOrderErr(err) {
+					s.writeFailures.Log(s.tenant, err)
+					outOfOrderSamples++
+					outOfOrderBytes += len(entries[i].Line)
+				}
+				continue
 			}
-			continue
 		}
 
 		s.entryCt++
@@ -380,6 +397,13 @@ func (s *stream) validateEntries(ctx context.Context, entries []logproto.Entry,
 		// NOTE: it's still possible for duplicates to be appended if a stream is
 		// deleted from inactivity.
 		if entries[i].Timestamp.Equal(lastLine.ts) && entries[i].Line == lastLine.content {
+			if s.configs.LogDuplicateMetrics(s.tenant) {
+				s.reportDuplicateMetrics(len(entries[i].Line))
+			}
+			if s.configs.LogDuplicateStreamInfo(s.tenant) {
+				err := chunkenc.ErrDuplicateLogEntry(entries[i].Timestamp, s.labelsString)
+				s.writeFailures.Log(s.tenant, err)
+			}
 			continue
 		}
 
@@ -456,6 +480,10 @@ func (s *stream) reportMetrics(ctx context.Context, outOfOrderSamples, outOfOrde
 	}
 }
 
+func (s *stream) reportDuplicateMetrics(duplicateLogLineBytes int) {
+	validation.DuplicateLogEntries.WithLabelValues(validation.DiscardedBytesTotal, s.tenant).Add(float64(duplicateLogLineBytes))
+}
+
 func (s *stream) cutChunk(ctx context.Context) *chunkDesc {
 	if sp := opentracing.SpanFromContext(ctx); sp != nil {
 		sp.LogKV("event", "stream started to cut chunk")