Skip to content

Commit 60a1567

Browse files
spiridonovshantanualsi
spiridonov
and
shantanualsi
authored
fix: duplicate lines with different structured metadata (#21039)
Co-authored-by: shantanualshi <shantanu.alshi@grafana.com>
1 parent 8764626 commit 60a1567

7 files changed

Lines changed: 159 additions & 41 deletions

File tree

‎pkg/chunkenc/symbols.go‎

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -27,6 +27,18 @@ type symbol struct {
2727

2828
type symbols []symbol
2929

30+
func (s symbols) Equal(other symbols) bool {
31+
if len(s) != len(other) {
32+
return false
33+
}
34+
for i := range s {
35+
if s[i] != other[i] {
36+
return false
37+
}
38+
}
39+
return true
40+
}
41+
3042
// symbolizer holds a collection of label names and values and assign symbols to them.
3143
// symbols are actually index numbers assigned based on when the entry is seen for the first time.
3244
type symbolizer struct {

‎pkg/chunkenc/unordered.go‎

Lines changed: 6 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -132,19 +132,20 @@ func (hb *unorderedHeadBlock) Append(ts int64, line string, structuredMetadata l
132132
}
133133
displaced := hb.rt.Add(e)
134134
if displaced[0] != nil {
135+
symbols, err := hb.symbolizer.Add(structuredMetadata)
136+
if err != nil {
137+
return false, err
138+
}
139+
135140
// While we support multiple entries at the same timestamp, we _do_ de-duplicate
136141
// entries at the same time with the same content, iterate through any existing
137142
// entries and ignore the line if we already have an entry with the same content
138143
for _, et := range displaced[0].(*nsEntries).entries {
139-
if et.line == line {
144+
if et.line == line && et.structuredMetadataSymbols.Equal(symbols) {
140145
e.entries = displaced[0].(*nsEntries).entries
141146
return true, nil
142147
}
143148
}
144-
symbols, err := hb.symbolizer.Add(structuredMetadata)
145-
if err != nil {
146-
return false, err
147-
}
148149

149150
e.entries = append(displaced[0].(*nsEntries).entries, nsEntry{line, symbols})
150151
} else {

‎pkg/chunkenc/unordered_test.go‎

Lines changed: 77 additions & 30 deletions
Original file line numberDiff line numberDiff line change
@@ -92,6 +92,7 @@ func Test_Unordered_InsertRetrieval(t *testing.T) {
9292
input, exp []entry
9393
dir logproto.Direction
9494
hasDup bool
95+
forFormat HeadBlockFmt
9596
}{
9697
{
9798
desc: "simple forward",
@@ -131,6 +132,50 @@ func Test_Unordered_InsertRetrieval(t *testing.T) {
131132
},
132133
dir: logproto.BACKWARD,
133134
},
135+
{
136+
desc: "different structured metadata forward",
137+
input: []entry{
138+
{0, "a", labels.EmptyLabels()}, {1, "b", labels.EmptyLabels()}, {0, "a", labels.FromStrings("a", "b")},
139+
},
140+
exp: []entry{
141+
{0, "a", labels.FromStrings("a", "b")}, {0, "a", labels.EmptyLabels()}, {1, "b", labels.EmptyLabels()},
142+
},
143+
forFormat: UnorderedWithStructuredMetadataHeadBlockFmt,
144+
},
145+
{
146+
desc: "different structured metadata backward",
147+
input: []entry{
148+
{0, "a", labels.EmptyLabels()}, {1, "b", labels.EmptyLabels()}, {0, "a", labels.FromStrings("a", "b")},
149+
},
150+
exp: []entry{
151+
{1, "b", labels.EmptyLabels()}, {0, "a", labels.EmptyLabels()}, {0, "a", labels.FromStrings("a", "b")},
152+
},
153+
dir: logproto.BACKWARD,
154+
forFormat: UnorderedWithStructuredMetadataHeadBlockFmt,
155+
},
156+
{
157+
desc: "different structured metadata forward",
158+
input: []entry{
159+
{0, "a", labels.EmptyLabels()}, {1, "b", labels.EmptyLabels()}, {0, "a", labels.FromStrings("a", "b")},
160+
},
161+
exp: []entry{
162+
{0, "a", labels.EmptyLabels()}, {1, "b", labels.EmptyLabels()},
163+
},
164+
hasDup: true,
165+
forFormat: UnorderedHeadBlockFmt,
166+
},
167+
{
168+
desc: "different structured metadata backward",
169+
input: []entry{
170+
{0, "a", labels.EmptyLabels()}, {1, "b", labels.EmptyLabels()}, {0, "a", labels.FromStrings("a", "b")},
171+
},
172+
exp: []entry{
173+
{1, "b", labels.EmptyLabels()}, {0, "a", labels.EmptyLabels()},
174+
},
175+
dir: logproto.BACKWARD,
176+
hasDup: true,
177+
forFormat: UnorderedHeadBlockFmt,
178+
},
134179
{
135180
desc: "ts collision forward",
136181
input: []entry{
@@ -153,7 +198,7 @@ func Test_Unordered_InsertRetrieval(t *testing.T) {
153198
{
154199
desc: "ts remove exact dupe forward",
155200
input: []entry{
156-
{0, "a", labels.EmptyLabels()}, {0, "b", labels.EmptyLabels()}, {1, "c", labels.EmptyLabels()}, {0, "b", labels.FromStrings("a", "b")},
201+
{0, "a", labels.EmptyLabels()}, {0, "b", labels.EmptyLabels()}, {1, "c", labels.EmptyLabels()}, {0, "b", labels.EmptyLabels()},
157202
},
158203
exp: []entry{
159204
{0, "a", labels.EmptyLabels()}, {0, "b", labels.EmptyLabels()}, {1, "c", labels.EmptyLabels()},
@@ -164,7 +209,7 @@ func Test_Unordered_InsertRetrieval(t *testing.T) {
164209
{
165210
desc: "ts remove exact dupe backward",
166211
input: []entry{
167-
{0, "a", labels.EmptyLabels()}, {0, "b", labels.EmptyLabels()}, {1, "c", labels.EmptyLabels()}, {0, "b", labels.FromStrings("a", "b")},
212+
{0, "a", labels.EmptyLabels()}, {0, "b", labels.EmptyLabels()}, {1, "c", labels.EmptyLabels()}, {0, "b", labels.EmptyLabels()},
168213
},
169214
exp: []entry{
170215
{1, "c", labels.EmptyLabels()}, {0, "b", labels.EmptyLabels()}, {0, "a", labels.EmptyLabels()},
@@ -178,38 +223,40 @@ func Test_Unordered_InsertRetrieval(t *testing.T) {
178223
UnorderedHeadBlockFmt,
179224
UnorderedWithStructuredMetadataHeadBlockFmt,
180225
} {
181-
t.Run(format.String(), func(t *testing.T) {
182-
hb := newUnorderedHeadBlock(format, newSymbolizer())
183-
dup := false
184-
for _, e := range tc.input {
185-
tmpdup, err := hb.Append(e.t, e.s, e.structuredMetadata)
186-
if !dup { // only set dup if it's not already true
187-
if tmpdup { // can't examine duplicates until we start getting all the data
188-
dup = true
226+
if tc.forFormat == 0 || tc.forFormat == format {
227+
t.Run(format.String(), func(t *testing.T) {
228+
hb := newUnorderedHeadBlock(format, newSymbolizer())
229+
dup := false
230+
for _, e := range tc.input {
231+
tmpdup, err := hb.Append(e.t, e.s, e.structuredMetadata)
232+
if !dup { // only set dup if it's not already true
233+
if tmpdup { // can't examine duplicates until we start getting all the data
234+
dup = true
235+
}
189236
}
237+
require.Nil(t, err)
190238
}
191-
require.Nil(t, err)
192-
}
193-
require.Equal(t, tc.hasDup, dup)
194-
195-
itr := hb.Iterator(
196-
context.Background(),
197-
tc.dir,
198-
0,
199-
math.MaxInt64,
200-
noopStreamPipeline,
201-
)
202-
203-
expected := make([]entry, len(tc.exp))
204-
copy(expected, tc.exp)
205-
if format < UnorderedWithStructuredMetadataHeadBlockFmt {
206-
for i := range expected {
207-
expected[i].structuredMetadata = labels.EmptyLabels()
239+
require.Equal(t, tc.hasDup, dup)
240+
241+
itr := hb.Iterator(
242+
context.Background(),
243+
tc.dir,
244+
0,
245+
math.MaxInt64,
246+
noopStreamPipeline,
247+
)
248+
249+
expected := make([]entry, len(tc.exp))
250+
copy(expected, tc.exp)
251+
if format < UnorderedWithStructuredMetadataHeadBlockFmt {
252+
for i := range expected {
253+
expected[i].structuredMetadata = labels.EmptyLabels()
254+
}
208255
}
209-
}
210256

211-
iterEq(t, expected, itr)
212-
})
257+
iterEq(t, expected, itr)
258+
})
259+
}
213260
}
214261
})
215262
}

‎pkg/ingester/stream.go‎

Lines changed: 23 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,7 @@ import (
1616
"go.opentelemetry.io/otel/attribute"
1717
"go.opentelemetry.io/otel/trace"
1818

19+
pushtypes "github.com/grafana/loki/pkg/push"
1920
"github.com/grafana/loki/v3/pkg/chunkenc"
2021
"github.com/grafana/loki/v3/pkg/distributor/writefailures"
2122
"github.com/grafana/loki/v3/pkg/ingester/wal"
@@ -34,8 +35,9 @@ import (
3435
var ErrEntriesExist = errors.New("duplicate push - entries already exist")
3536

3637
type line struct {
37-
ts time.Time
38-
content string
38+
ts time.Time
39+
content string
40+
structuredMetadata pushtypes.LabelsAdapter
3941
}
4042

4143
type stream struct {
@@ -369,6 +371,7 @@ func (s *stream) storeEntries(ctx context.Context, entries []logproto.Entry, usa
369371
s.entryCt++
370372
s.lastLine.ts = entries[i].Timestamp
371373
s.lastLine.content = entries[i].Line
374+
s.lastLine.structuredMetadata = entries[i].StructuredMetadata
372375
if s.highestTs.Before(entries[i].Timestamp) {
373376
s.highestTs = entries[i].Timestamp
374377
}
@@ -417,7 +420,9 @@ func (s *stream) validateEntries(ctx context.Context, entries []logproto.Entry,
417420
//
418421
// NOTE: it's still possible for duplicates to be appended if a stream is
419422
// deleted from inactivity.
420-
if entries[i].Timestamp.Equal(lastLine.ts) && entries[i].Line == lastLine.content {
423+
if entries[i].Timestamp.Equal(lastLine.ts) &&
424+
entries[i].Line == lastLine.content &&
425+
labelsEqual(entries[i].StructuredMetadata, lastLine.structuredMetadata) {
421426
continue
422427
}
423428

@@ -447,6 +452,7 @@ func (s *stream) validateEntries(ctx context.Context, entries []logproto.Entry,
447452

448453
lastLine.ts = entries[i].Timestamp
449454
lastLine.content = entries[i].Line
455+
lastLine.structuredMetadata = entries[i].StructuredMetadata
450456
if highestTs.Before(entries[i].Timestamp) {
451457
highestTs = entries[i].Timestamp
452458
}
@@ -669,3 +675,17 @@ func headBlockType(chunkfmt byte, unorderedWrites bool) chunkenc.HeadBlockFmt {
669675
}
670676
return chunkenc.OrderedHeadBlockFmt
671677
}
678+
679+
func labelsEqual(a, b pushtypes.LabelsAdapter) bool {
680+
if len(a) != len(b) {
681+
return false
682+
}
683+
684+
for i := range a {
685+
if a[i].Name != b[i].Name || a[i].Value != b[i].Value {
686+
return false
687+
}
688+
}
689+
690+
return true
691+
}

‎pkg/iter/entry_iterator.go‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -121,7 +121,7 @@ func (i *mergeEntryIterator) fillBuffer() {
121121

122122
// We support multiple entries with the same timestamp, and we want to
123123
// preserve their original order.
124-
// Entries with identical timestamp and line are removed as duplicates.
124+
// Entries with identical timestamp, line, and metadata are removed as duplicates.
125125
for {
126126
next := i.tree.Winner()
127127
entry := next.At()
@@ -139,7 +139,7 @@ func (i *mergeEntryIterator) fillBuffer() {
139139

140140
var dupe bool
141141
for _, t := range previous {
142-
if t.Line == entry.Line {
142+
if t.Equal(entry) {
143143
i.stats.AddDuplicates(1)
144144
dupe = true
145145
break

‎pkg/iter/entry_iterator_test.go‎

Lines changed: 38 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -922,3 +922,41 @@ func TestDedupeMergeEntryIterator(t *testing.T) {
922922
require.Equal(t, []string{"0", "2", "1", "3"}, lines)
923923
}
924924
}
925+
926+
func TestMergeIteratorNoDedupDifferentStructuredMetadata(t *testing.T) {
927+
// Entries with the same timestamp and line but different structured metadata
928+
// must NOT be deduplicated — they are distinct entries.
929+
const labels = `{app="foo"}`
930+
ts := time.Unix(1, 0)
931+
line := "same message"
932+
933+
entry1 := logproto.Entry{
934+
Timestamp: ts,
935+
Line: line,
936+
StructuredMetadata: []logproto.LabelAdapter{{Name: "trace_id", Value: "aaa"}},
937+
}
938+
entry2 := logproto.Entry{
939+
Timestamp: ts,
940+
Line: line,
941+
StructuredMetadata: []logproto.LabelAdapter{{Name: "trace_id", Value: "bbb"}},
942+
}
943+
944+
it := NewMergeEntryIterator(context.Background(),
945+
[]EntryIterator{
946+
NewStreamIterator(logproto.Stream{Labels: labels, Hash: hashLabels(labels), Entries: []logproto.Entry{entry1}}),
947+
NewStreamIterator(logproto.Stream{Labels: labels, Hash: hashLabels(labels), Entries: []logproto.Entry{entry2}}),
948+
}, logproto.FORWARD)
949+
950+
var got []logproto.Entry
951+
for it.Next() {
952+
got = append(got, it.At())
953+
}
954+
require.NoError(t, it.Err())
955+
956+
require.Len(t, got, 2, "entries with different structured metadata must not be deduplicated")
957+
require.Equal(t, ts, got[0].Timestamp)
958+
require.Equal(t, line, got[0].Line)
959+
require.Equal(t, ts, got[1].Timestamp)
960+
require.Equal(t, line, got[1].Line)
961+
require.NotEqual(t, got[0].StructuredMetadata, got[1].StructuredMetadata)
962+
}

‎tools/dev/loki-tsdb-storage-s3/config/loki.yaml‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,4 @@
1-
auth_enabled: true
1+
auth_enabled: true
22
common:
33
compactor_address: http://compactor:8006
44
chunk_store_config:

0 commit comments

Comments
 (0)