diff --git a/pkg/chunkenc/memchunk.go b/pkg/chunkenc/memchunk.go index 01bf110f945ae..ec4c816d1dff7 100644 --- a/pkg/chunkenc/memchunk.go +++ b/pkg/chunkenc/memchunk.go @@ -1401,6 +1401,7 @@ func newBufferedIterator(ctx context.Context, pool compression.ReaderPool, b []b pool: pool, format: format, symbolizer: symbolizer, + currStructuredMetadata: structuredMetadataPool.Get().(labels.Labels), } } @@ -1443,14 +1444,14 @@ func (si *bufferedIterator) moveNext() (int64, []byte, labels.Labels, bool) { if err != nil { if err != io.EOF { si.err = err - return 0, nil, nil, false + return 0, nil, labels.EmptyLabels(), false } if si.readBufValid == 0 { // Got EOF and no data in the buffer. - return 0, nil, nil, false + return 0, nil, labels.EmptyLabels(), false } if si.readBufValid == lastAttempt { // Got EOF and could not parse same data last time. si.err = fmt.Errorf("invalid data in chunk") - return 0, nil, nil, false + return 0, nil, labels.EmptyLabels(), false } } var l uint64 @@ -1465,7 +1466,7 @@ func (si *bufferedIterator) moveNext() (int64, []byte, labels.Labels, bool) { if lineSize >= maxLineLength { si.err = fmt.Errorf("line too long %d, maximum %d", lineSize, maxLineLength) - return 0, nil, nil, false + return 0, nil, labels.EmptyLabels(), false } // If the buffer is not yet initialize or too small, we get a new one. if si.buf == nil || lineSize > cap(si.buf) { @@ -1476,7 +1477,7 @@ func (si *bufferedIterator) moveNext() (int64, []byte, labels.Labels, bool) { si.buf = BytesBufferPool.Get(lineSize).([]byte) if lineSize > cap(si.buf) { si.err = fmt.Errorf("could not get a line buffer of size %d, actual %d", lineSize, cap(si.buf)) - return 0, nil, nil, false + return 0, nil, labels.EmptyLabels(), false } } si.buf = si.buf[:lineSize] @@ -1496,7 +1497,7 @@ func (si *bufferedIterator) moveNext() (int64, []byte, labels.Labels, bool) { continue } si.err = err - return 0, nil, nil, false + return 0, nil, labels.EmptyLabels(), false } } @@ -1505,7 +1506,7 @@ func (si *bufferedIterator) moveNext() (int64, []byte, labels.Labels, bool) { if si.format < ChunkFormatV4 { si.stats.AddDecompressedBytes(decompressedBytes) si.stats.AddDecompressedLines(1) - return ts, si.buf[:lineSize], nil, true + return ts, si.buf[:lineSize], labels.EmptyLabels(), true } lastAttempt = 0 @@ -1516,14 +1517,14 @@ func (si *bufferedIterator) moveNext() (int64, []byte, labels.Labels, bool) { if err != nil { if err != io.EOF { si.err = err - return 0, nil, nil, false + return 0, nil, labels.EmptyLabels(), false } if si.readBufValid == 0 { // Got EOF and no data in the buffer. - return 0, nil, nil, false + return 0, nil, labels.EmptyLabels(), false } if si.readBufValid == lastAttempt { // Got EOF and could not parse same data last time. si.err = fmt.Errorf("invalid data in chunk") - return 0, nil, nil, false + return 0, nil, labels.EmptyLabels(), false } } var l uint64 @@ -1573,7 +1574,7 @@ func (si *bufferedIterator) moveNext() (int64, []byte, labels.Labels, bool) { si.symbolsBuf = SymbolsPool.Get(nSymbols).([]symbol) if nSymbols > cap(si.symbolsBuf) { si.err = fmt.Errorf("could not get a symbols matrix of size %d, actual %d", nSymbols, cap(si.symbolsBuf)) - return 0, nil, nil, false + return 0, nil, labels.EmptyLabels(), false } } @@ -1589,14 +1590,14 @@ func (si *bufferedIterator) moveNext() (int64, []byte, labels.Labels, bool) { if err != nil { if err != io.EOF { si.err = err - return 0, nil, nil, false + return 0, nil, labels.EmptyLabels(), false } if si.readBufValid == 0 { // Got EOF and no data in the buffer. - return 0, nil, nil, false + return 0, nil, labels.EmptyLabels(), false } if si.readBufValid == lastAttempt { // Got EOF and could not parse same data last time. si.err = fmt.Errorf("invalid data in chunk") - return 0, nil, nil, false + return 0, nil, labels.EmptyLabels(), false } } sName, nWidth = binary.Uvarint(si.readBuf[:si.readBufValid]) @@ -1615,7 +1616,8 @@ func (si *bufferedIterator) moveNext() (int64, []byte, labels.Labels, bool) { si.stats.AddDecompressedStructuredMetadataBytes(decompressedStructuredMetadataBytes) si.stats.AddDecompressedBytes(decompressedBytes + decompressedStructuredMetadataBytes) - return ts, si.buf[:lineSize], si.symbolizer.Lookup(si.symbolsBuf[:nSymbols], si.currStructuredMetadata), true + labelsBuilder := log.NewBufferedLabelsBuilder(si.currStructuredMetadata) + return ts, si.buf[:lineSize], si.symbolizer.Lookup(si.symbolsBuf[:nSymbols], labelsBuilder), true } func (si *bufferedIterator) Err() error { return si.err } @@ -1644,9 +1646,9 @@ func (si *bufferedIterator) close() { si.symbolsBuf = nil } - if si.currStructuredMetadata != nil { + if !si.currStructuredMetadata.IsEmpty() { structuredMetadataPool.Put(si.currStructuredMetadata) // nolint:staticcheck - si.currStructuredMetadata = nil + si.currStructuredMetadata = labels.EmptyLabels() // TODO: maybe a reset would be better } si.origBytes = nil diff --git a/pkg/chunkenc/memchunk_test.go b/pkg/chunkenc/memchunk_test.go index a399844101bde..91178fc976bc9 100644 --- a/pkg/chunkenc/memchunk_test.go +++ b/pkg/chunkenc/memchunk_test.go @@ -1120,6 +1120,7 @@ func BenchmarkHeadBlockIterator(b *testing.B) { for iter.Next() { _ = iter.At() } + iter.Close() } }) } diff --git a/pkg/chunkenc/symbols.go b/pkg/chunkenc/symbols.go index f5d3310921abe..2d0f5c6faf505 100644 --- a/pkg/chunkenc/symbols.go +++ b/pkg/chunkenc/symbols.go @@ -13,6 +13,7 @@ import ( "github.com/prometheus/prometheus/model/labels" "github.com/grafana/loki/v3/pkg/compression" + "github.com/grafana/loki/v3/pkg/logql/log" "github.com/grafana/loki/v3/pkg/util" ) @@ -58,16 +59,18 @@ func (s *symbolizer) Reset() { // Add adds new labels pairs to the collection and returns back a symbol for each existing and new label pair func (s *symbolizer) Add(lbls labels.Labels) symbols { - if len(lbls) == 0 { + if lbls.IsEmpty() { return nil } - syms := make([]symbol, len(lbls)) + syms := make([]symbol, 0, lbls.Len()) - for i, label := range lbls { - syms[i].Name = s.add(label.Name) - syms[i].Value = s.add(label.Value) - } + lbls.Range(func(label labels.Label) { + syms = append(syms, symbol{ + Name: s.add(label.Name), + Value: s.add(label.Value), + }) + }) return syms } @@ -97,20 +100,20 @@ func (s *symbolizer) add(lbl string) uint32 { } // Lookup coverts and returns labels pairs for the given symbols -func (s *symbolizer) Lookup(syms symbols, buf labels.Labels) labels.Labels { +func (s *symbolizer) Lookup(syms symbols, buf *log.BufferedLabelsBuilder) labels.Labels { if len(syms) == 0 { - return nil + return labels.EmptyLabels() } + if buf == nil { - buf = structuredMetadataPool.Get().(labels.Labels) + structuredMetadata := structuredMetadataPool.Get().(labels.Labels) + buf = log.NewBufferedLabelsBuilder(structuredMetadata) } - buf = buf[:0] - for _, symbol := range syms { - buf = append(buf, labels.Label{Name: s.lookup(symbol.Name), Value: s.lookup(symbol.Value)}) + buf.Add(labels.Label{Name: s.lookup(symbol.Name), Value: s.lookup(symbol.Value)}) } - return buf + return buf.Labels() } func (s *symbolizer) lookup(idx uint32) string { diff --git a/pkg/chunkenc/symbols_test.go b/pkg/chunkenc/symbols_test.go index 1f286d7b56d50..872ec7c5e915d 100644 --- a/pkg/chunkenc/symbols_test.go +++ b/pkg/chunkenc/symbols_test.go @@ -29,22 +29,11 @@ func TestSymbolizer(t *testing.T) { { name: "no duplicate labels", labelsToAdd: []labels.Labels{ - { - labels.Label{ - Name: "foo", - Value: "bar", - }, - }, - { - labels.Label{ - Name: "fizz", - Value: "buzz", - }, - labels.Label{ - Name: "ping", - Value: "pong", - }, - }, + labels.FromStrings("foo", "bar"), + labels.FromStrings( + "fizz", "buzz", + "ping", "pong", + ), }, expectedSymbols: []symbols{ { @@ -71,30 +60,15 @@ func TestSymbolizer(t *testing.T) { { name: "with duplicate labels", labelsToAdd: []labels.Labels{ - { - labels.Label{ - Name: "foo", - Value: "bar", - }, - { - Name: "bar", - Value: "foo", - }, - }, - { - labels.Label{ - Name: "foo", - Value: "bar", - }, - labels.Label{ - Name: "fizz", - Value: "buzz", - }, - labels.Label{ - Name: "ping", - Value: "pong", - }, - }, + labels.FromStrings( + "foo", "bar", + "bar", "foo", + ), + labels.FromStrings( + "foo", "bar", + "fizz", "buzz", + "ping", "pong", + ), }, expectedSymbols: []symbols{ { @@ -108,14 +82,14 @@ func TestSymbolizer(t *testing.T) { }, }, { - symbol{ - Name: 0, - Value: 1, - }, symbol{ Name: 2, Value: 3, }, + symbol{ + Name: 1, + Value: 0, + }, symbol{ Name: 4, Value: 5, @@ -130,10 +104,10 @@ func TestSymbolizer(t *testing.T) { for _, encoding := range testEncodings { t.Run(fmt.Sprintf("%s - %s", tc.name, encoding), func(t *testing.T) { s := newSymbolizer() - for i, labels := range tc.labelsToAdd { - symbols := s.Add(labels) + for i, lbls := range tc.labelsToAdd { + symbols := s.Add(lbls) require.Equal(t, tc.expectedSymbols[i], symbols) - require.Equal(t, labels, s.Lookup(symbols, nil)) + require.Equal(t, lbls, s.Lookup(symbols, nil)) } // Test that Lookup returns empty labels if no symbols are provided. @@ -144,8 +118,7 @@ func TestSymbolizer(t *testing.T) { Value: 0, }, }, nil) - require.Equal(t, "", ret[0].Name) - require.Equal(t, "", ret[0].Value) + require.Equal(t, `{""=""}`, ret.String()) } require.Equal(t, tc.expectedNumLabels, len(s.labels)) diff --git a/pkg/chunkenc/unordered.go b/pkg/chunkenc/unordered.go index c3d46ba5a84c2..041cdcea087e4 100644 --- a/pkg/chunkenc/unordered.go +++ b/pkg/chunkenc/unordered.go @@ -115,7 +115,7 @@ func (e *nsEntries) ValueAtDimension(_ uint64) int64 { func (hb *unorderedHeadBlock) Append(ts int64, line string, structuredMetadata labels.Labels) (bool, error) { if hb.format < UnorderedWithStructuredMetadataHeadBlockFmt { // structuredMetadata must be ignored for the previous head block formats - structuredMetadata = nil + structuredMetadata = labels.EmptyLabels() } // This is an allocation hack. The rangetree lib does not // support the ability to pass a "mutate" function during an insert @@ -155,7 +155,7 @@ func (hb *unorderedHeadBlock) Append(ts int64, line string, structuredMetadata l } hb.size += len(line) - hb.size += len(structuredMetadata) * 2 * 4 // 4 bytes per label and value pair as structuredMetadataSymbols + hb.size += structuredMetadata.Len() * 2 * 4 // 4 bytes per label and value pair as structuredMetadataSymbols hb.lines++ return false, nil @@ -163,9 +163,9 @@ func (hb *unorderedHeadBlock) Append(ts int64, line string, structuredMetadata l func metaLabelsLen(metaLabels labels.Labels) int { length := 0 - for _, label := range metaLabels { + metaLabels.Range(func(label labels.Label) { length += len(label.Name) + len(label.Value) - } + }) return length } @@ -252,14 +252,15 @@ func (hb *unorderedHeadBlock) Iterator(ctx context.Context, direction logproto.D // cutting of blocks. streams := map[string]*logproto.Stream{} baseHash := pipeline.BaseLabels().Hash() - var structuredMetadata labels.Labels + structuredMetadata := structuredMetadataPool.Get().(labels.Labels) + labelsBuilder := log.NewBufferedLabelsBuilder(structuredMetadata) _ = hb.forEntries( ctx, direction, mint, maxt, func(statsCtx *stats.Context, ts int64, line string, structuredMetadataSymbols symbols) error { - structuredMetadata = hb.symbolizer.Lookup(structuredMetadataSymbols, structuredMetadata) + structuredMetadata = hb.symbolizer.Lookup(structuredMetadataSymbols, labelsBuilder) newLine, parsedLbs, matches := pipeline.ProcessString(ts, line, structuredMetadata) if !matches { return nil @@ -298,9 +299,7 @@ func (hb *unorderedHeadBlock) Iterator(ctx context.Context, direction logproto.D } return iter.EntryIteratorWithClose(iter.NewStreamsIterator(streamsResult, direction), func() error { - if structuredMetadata != nil { - structuredMetadataPool.Put(structuredMetadata) // nolint:staticcheck - } + structuredMetadataPool.Put(structuredMetadata) // nolint:staticcheck return nil }) } @@ -313,14 +312,15 @@ func (hb *unorderedHeadBlock) SampleIterator( ) iter.SampleIterator { series := map[string]*logproto.Series{} setQueryReferencedStructuredMetadata := false - var structuredMetadata labels.Labels + structuredMetadata := structuredMetadataPool.Get().(labels.Labels) + labelsBuilder := log.NewBufferedLabelsBuilder(structuredMetadata) _ = hb.forEntries( ctx, logproto.FORWARD, mint, maxt, func(statsCtx *stats.Context, ts int64, line string, structuredMetadataSymbols symbols) error { - structuredMetadata = hb.symbolizer.Lookup(structuredMetadataSymbols, structuredMetadata) + structuredMetadata = hb.symbolizer.Lookup(structuredMetadataSymbols, labelsBuilder) for _, extractor := range extractor { value, lbls, ok := extractor.ProcessString(ts, line, structuredMetadata) @@ -372,9 +372,7 @@ func (hb *unorderedHeadBlock) SampleIterator( for _, s := range series { SamplesPool.Put(s.Samples) } - if structuredMetadata != nil { - structuredMetadataPool.Put(structuredMetadata) // nolint:staticcheck - } + structuredMetadataPool.Put(structuredMetadata) // nolint:staticcheck return nil }) } diff --git a/pkg/chunkenc/unordered_test.go b/pkg/chunkenc/unordered_test.go index f1483cedfe546..36c8612fe4f4c 100644 --- a/pkg/chunkenc/unordered_test.go +++ b/pkg/chunkenc/unordered_test.go @@ -22,11 +22,15 @@ import ( func iterEq(t *testing.T, exp []entry, got iter.EntryIterator) { var i int for got.Next() { - require.Equal(t, logproto.Entry{ + expected := logproto.Entry{ Timestamp: time.Unix(0, exp[i].t), Line: exp[i].s, StructuredMetadata: logproto.FromLabelsToLabelAdapters(exp[i].structuredMetadata), - }, got.At()) + } + if exp[i].structuredMetadata.IsEmpty() { + expected.StructuredMetadata = nil + } + require.Equal(t, expected, got.At()) require.Equal(t, exp[i].structuredMetadata.String(), got.Labels()) i++ } @@ -36,7 +40,7 @@ func iterEq(t *testing.T, exp []entry, got iter.EntryIterator) { func Test_forEntriesEarlyReturn(t *testing.T) { hb := newUnorderedHeadBlock(UnorderedHeadBlockFmt, newSymbolizer()) for i := 0; i < 10; i++ { - dup, err := hb.Append(int64(i), fmt.Sprint(i), labels.Labels{{Name: "i", Value: fmt.Sprint(i)}}) + dup, err := hb.Append(int64(i), fmt.Sprint(i), labels.FromStrings("i", fmt.Sprint(i))) require.False(t, dup) require.Nil(t, err) } @@ -94,67 +98,67 @@ func Test_Unordered_InsertRetrieval(t *testing.T) { { desc: "simple forward", input: []entry{ - {0, "a", nil}, {1, "b", nil}, {2, "c", labels.Labels{{Name: "a", Value: "b"}}}, + {0, "a", labels.EmptyLabels()}, {1, "b", labels.EmptyLabels()}, {2, "c", labels.FromStrings("a", "b")}, }, exp: []entry{ - {0, "a", nil}, {1, "b", nil}, {2, "c", labels.Labels{{Name: "a", Value: "b"}}}, + {0, "a", labels.EmptyLabels()}, {1, "b", labels.EmptyLabels()}, {2, "c", labels.FromStrings("a", "b")}, }, }, { desc: "simple backward", input: []entry{ - {0, "a", nil}, {1, "b", nil}, {2, "c", labels.Labels{{Name: "a", Value: "b"}}}, + {0, "a", labels.EmptyLabels()}, {1, "b", labels.EmptyLabels()}, {2, "c", labels.FromStrings("a", "b")}, }, exp: []entry{ - {2, "c", labels.Labels{{Name: "a", Value: "b"}}}, {1, "b", nil}, {0, "a", nil}, + {2, "c", labels.FromStrings("a", "b")}, {1, "b", labels.EmptyLabels()}, {0, "a", labels.EmptyLabels()}, }, dir: logproto.BACKWARD, }, { desc: "unordered forward", input: []entry{ - {1, "b", nil}, {0, "a", nil}, {2, "c", labels.Labels{{Name: "a", Value: "b"}}}, + {1, "b", labels.EmptyLabels()}, {0, "a", labels.EmptyLabels()}, {2, "c", labels.FromStrings("a", "b")}, }, exp: []entry{ - {0, "a", nil}, {1, "b", nil}, {2, "c", labels.Labels{{Name: "a", Value: "b"}}}, + {0, "a", labels.EmptyLabels()}, {1, "b", labels.EmptyLabels()}, {2, "c", labels.FromStrings("a", "b")}, }, }, { desc: "unordered backward", input: []entry{ - {1, "b", nil}, {0, "a", nil}, {2, "c", labels.Labels{{Name: "a", Value: "b"}}}, + {1, "b", labels.EmptyLabels()}, {0, "a", labels.EmptyLabels()}, {2, "c", labels.FromStrings("a", "b")}, }, exp: []entry{ - {2, "c", labels.Labels{{Name: "a", Value: "b"}}}, {1, "b", nil}, {0, "a", nil}, + {2, "c", labels.FromStrings("a", "b")}, {1, "b", labels.EmptyLabels()}, {0, "a", labels.EmptyLabels()}, }, dir: logproto.BACKWARD, }, { desc: "ts collision forward", input: []entry{ - {0, "a", labels.Labels{{Name: "a", Value: "b"}}}, {0, "b", labels.Labels{{Name: "a", Value: "b"}}}, {1, "c", nil}, + {0, "a", labels.FromStrings("a", "b")}, {0, "b", labels.FromStrings("a", "b")}, {1, "c", labels.EmptyLabels()}, }, exp: []entry{ - {0, "a", labels.Labels{{Name: "a", Value: "b"}}}, {0, "b", labels.Labels{{Name: "a", Value: "b"}}}, {1, "c", nil}, + {0, "a", labels.FromStrings("a", "b")}, {0, "b", labels.FromStrings("a", "b")}, {1, "c", labels.EmptyLabels()}, }, }, { desc: "ts collision backward", input: []entry{ - {0, "a", labels.Labels{{Name: "a", Value: "b"}}}, {0, "b", nil}, {1, "c", nil}, + {0, "a", labels.FromStrings("a", "b")}, {0, "b", labels.EmptyLabels()}, {1, "c", labels.EmptyLabels()}, }, exp: []entry{ - {1, "c", nil}, {0, "b", nil}, {0, "a", labels.Labels{{Name: "a", Value: "b"}}}, + {1, "c", labels.EmptyLabels()}, {0, "b", labels.EmptyLabels()}, {0, "a", labels.FromStrings("a", "b")}, }, dir: logproto.BACKWARD, }, { desc: "ts remove exact dupe forward", input: []entry{ - {0, "a", nil}, {0, "b", nil}, {1, "c", nil}, {0, "b", labels.Labels{{Name: "a", Value: "b"}}}, + {0, "a", labels.EmptyLabels()}, {0, "b", labels.EmptyLabels()}, {1, "c", labels.EmptyLabels()}, {0, "b", labels.FromStrings("a", "b")}, }, exp: []entry{ - {0, "a", nil}, {0, "b", nil}, {1, "c", nil}, + {0, "a", labels.EmptyLabels()}, {0, "b", labels.EmptyLabels()}, {1, "c", labels.EmptyLabels()}, }, dir: logproto.FORWARD, hasDup: true, @@ -162,10 +166,10 @@ func Test_Unordered_InsertRetrieval(t *testing.T) { { desc: "ts remove exact dupe backward", input: []entry{ - {0, "a", nil}, {0, "b", nil}, {1, "c", nil}, {0, "b", labels.Labels{{Name: "a", Value: "b"}}}, + {0, "a", labels.EmptyLabels()}, {0, "b", labels.EmptyLabels()}, {1, "c", labels.EmptyLabels()}, {0, "b", labels.FromStrings("a", "b")}, }, exp: []entry{ - {1, "c", nil}, {0, "b", nil}, {0, "a", nil}, + {1, "c", labels.EmptyLabels()}, {0, "b", labels.EmptyLabels()}, {0, "a", labels.EmptyLabels()}, }, dir: logproto.BACKWARD, hasDup: true, @@ -202,7 +206,7 @@ func Test_Unordered_InsertRetrieval(t *testing.T) { copy(expected, tc.exp) if format < UnorderedWithStructuredMetadataHeadBlockFmt { for i := range expected { - expected[i].structuredMetadata = nil + expected[i].structuredMetadata = labels.EmptyLabels() } } @@ -226,10 +230,10 @@ func Test_UnorderedBoundedIter(t *testing.T) { mint: 1, maxt: 4, input: []entry{ - {0, "a", nil}, {1, "b", labels.Labels{{Name: "a", Value: "b"}}}, {2, "c", nil}, {3, "d", nil}, {4, "e", nil}, + {0, "a", labels.EmptyLabels()}, {1, "b", labels.FromStrings("a", "b")}, {2, "c", labels.EmptyLabels()}, {3, "d", labels.EmptyLabels()}, {4, "e", labels.EmptyLabels()}, }, exp: []entry{ - {1, "b", labels.Labels{{Name: "a", Value: "b"}}}, {2, "c", nil}, {3, "d", nil}, + {1, "b", labels.FromStrings("a", "b")}, {2, "c", labels.EmptyLabels()}, {3, "d", labels.EmptyLabels()}, }, }, { @@ -237,10 +241,10 @@ func Test_UnorderedBoundedIter(t *testing.T) { mint: 1, maxt: 4, input: []entry{ - {0, "a", nil}, {1, "b", labels.Labels{{Name: "a", Value: "b"}}}, {2, "c", nil}, {3, "d", nil}, {4, "e", nil}, + {0, "a", labels.EmptyLabels()}, {1, "b", labels.FromStrings("a", "b")}, {2, "c", labels.EmptyLabels()}, {3, "d", labels.EmptyLabels()}, {4, "e", labels.EmptyLabels()}, }, exp: []entry{ - {3, "d", nil}, {2, "c", nil}, {1, "b", labels.Labels{{Name: "a", Value: "b"}}}, + {3, "d", labels.EmptyLabels()}, {2, "c", labels.EmptyLabels()}, {1, "b", labels.FromStrings("a", "b")}, }, dir: logproto.BACKWARD, }, @@ -249,10 +253,10 @@ func Test_UnorderedBoundedIter(t *testing.T) { mint: 1, maxt: 4, input: []entry{ - {0, "a", nil}, {2, "c", nil}, {1, "b", labels.Labels{{Name: "a", Value: "b"}}}, {4, "e", nil}, {3, "d", nil}, + {0, "a", labels.EmptyLabels()}, {2, "c", labels.EmptyLabels()}, {1, "b", labels.FromStrings("a", "b")}, {4, "e", labels.EmptyLabels()}, {3, "d", labels.EmptyLabels()}, }, exp: []entry{ - {1, "b", labels.Labels{{Name: "a", Value: "b"}}}, {2, "c", nil}, {3, "d", nil}, + {1, "b", labels.FromStrings("a", "b")}, {2, "c", labels.EmptyLabels()}, {3, "d", labels.EmptyLabels()}, }, }, } { @@ -281,7 +285,7 @@ func Test_UnorderedBoundedIter(t *testing.T) { copy(expected, tc.exp) if format < UnorderedWithStructuredMetadataHeadBlockFmt { for i := range expected { - expected[i].structuredMetadata = nil + expected[i].structuredMetadata = labels.EmptyLabels() } } @@ -296,14 +300,14 @@ func TestHeadBlockInterop(t *testing.T) { unordered, ordered := newUnorderedHeadBlock(UnorderedHeadBlockFmt, nil), &headBlock{} unorderedWithStructuredMetadata := newUnorderedHeadBlock(UnorderedWithStructuredMetadataHeadBlockFmt, newSymbolizer()) for i := 0; i < 100; i++ { - metaLabels := labels.Labels{{Name: "foo", Value: fmt.Sprint(99 - i)}} + metaLabels := labels.FromStrings("foo", fmt.Sprint(99-i)) dup, err := unordered.Append(int64(99-i), fmt.Sprint(99-i), metaLabels) require.False(t, dup) require.Nil(t, err) dup, err = unorderedWithStructuredMetadata.Append(int64(99-i), fmt.Sprint(99-i), metaLabels) require.False(t, dup) require.Nil(t, err) - dup, err = ordered.Append(int64(i), fmt.Sprint(i), labels.Labels{{Name: "foo", Value: fmt.Sprint(i)}}) + dup, err = ordered.Append(int64(i), fmt.Sprint(i), labels.FromStrings("foo", fmt.Sprint(i))) require.False(t, dup) require.Nil(t, err) } @@ -424,7 +428,7 @@ func BenchmarkHeadBlockWrites(b *testing.B) { var structuredMetadata labels.Labels if withStructuredMetadata { - structuredMetadata = labels.Labels{{Name: "foo", Value: fmt.Sprint(ts)}} + structuredMetadata = labels.FromStrings("foo", fmt.Sprint(ts)) } writes = append(writes, entry{ @@ -706,7 +710,7 @@ func TestReorderAcrossBlocks(t *testing.T) { from, to := c.Bounds() require.Nil(t, c.Close()) - itr, err := c.Iterator(context.Background(), from, to.Add(time.Nanosecond), logproto.FORWARD, log.NewNoopPipeline().ForStream(nil)) + itr, err := c.Iterator(context.Background(), from, to.Add(time.Nanosecond), logproto.FORWARD, log.NewNoopPipeline().ForStream(labels.EmptyLabels())) require.Nil(t, err) exp := []entry{ @@ -731,7 +735,7 @@ func TestReorderAcrossBlocks(t *testing.T) { } func Test_HeadIteratorHash(t *testing.T) { - lbs := labels.Labels{labels.Label{Name: "foo", Value: "bar"}} + lbs := labels.FromStrings("foo", "bar") countEx, err := log.NewLineSampleExtractor(log.CountExtractor, nil, nil, false, false) require.NoError(t, err) bytesEx, err := log.NewLineSampleExtractor(log.BytesExtractor, nil, nil, false, false) @@ -743,7 +747,7 @@ func Test_HeadIteratorHash(t *testing.T) { "ordered": &headBlock{}, } { t.Run(fmt.Sprintf("%s SampleIterator", name), func(t *testing.T) { - dup, err := b.Append(1, "foo", labels.Labels{{Name: "foo", Value: "bar"}}) + dup, err := b.Append(1, "foo", labels.FromStrings("foo", "bar")) require.False(t, dup) require.NoError(t, err) eit := b.Iterator(context.Background(), logproto.BACKWARD, 0, 2, log.NewNoopPipeline().ForStream(lbs)) @@ -759,7 +763,7 @@ func Test_HeadIteratorHash(t *testing.T) { }) t.Run(fmt.Sprintf("%s SampleIterator with multiple extractors", name), func(t *testing.T) { - dup, err := b.Append(1, "bar", labels.Labels{{Name: "bar", Value: "foo"}}) + dup, err := b.Append(1, "bar", labels.FromStrings("bar", "foo")) require.False(t, dup) require.NoError(t, err) eit := b.Iterator( diff --git a/pkg/chunkenc/variants.go b/pkg/chunkenc/variants.go index f68088b49d828..9f17d2b8a8dbb 100644 --- a/pkg/chunkenc/variants.go +++ b/pkg/chunkenc/variants.go @@ -2,10 +2,8 @@ package chunkenc import ( "context" - "sort" "github.com/cespare/xxhash/v2" - "github.com/prometheus/prometheus/model/labels" "github.com/grafana/loki/v3/pkg/compression" "github.com/grafana/loki/v3/pkg/iter" @@ -86,25 +84,6 @@ func (e *multiExtractorSampleBufferedIterator) Next() bool { return false } -func flattenLabels(buf labels.Labels, many ...labels.Labels) labels.Labels { - var size int - for _, lbls := range many { - size += len(lbls) - } - - if buf == nil || cap(buf) < size { - buf = make(labels.Labels, 0, size) - } else { - buf = buf[:0] - } - - for _, lbls := range many { - buf = append(buf, lbls...) - } - sort.Sort(buf) - return buf -} - func (e *multiExtractorSampleBufferedIterator) Close() error { for _, extractor := range e.extractors { if extractor.ReferencedStructuredMetadata() { diff --git a/pkg/compactor/retention/expiration_test.go b/pkg/compactor/retention/expiration_test.go index 154b2eebc845a..3f7d65c6e0164 100644 --- a/pkg/compactor/retention/expiration_test.go +++ b/pkg/compactor/retention/expiration_test.go @@ -163,8 +163,8 @@ func TestTenantsRetention_RetentionPeriodFor(t *testing.T) { }, }) - require.Equal(t, time.Duration(sevenDays), tr.RetentionPeriodFor("1", nil)) - require.Equal(t, time.Duration(oneDay), tr.RetentionPeriodFor("1", labels.Labels{labels.Label{Name: "foo", Value: "bar"}})) + require.Equal(t, time.Duration(sevenDays), tr.RetentionPeriodFor("1", labels.EmptyLabels())) + require.Equal(t, time.Duration(oneDay), tr.RetentionPeriodFor("1", labels.FromStrings("foo", "bar"))) } func Test_expirationChecker_Expired_zeroValue(t *testing.T) { diff --git a/pkg/compactor/retention/series_test.go b/pkg/compactor/retention/series_test.go index 04577d540b3df..2ec590958b6a4 100644 --- a/pkg/compactor/retention/series_test.go +++ b/pkg/compactor/retention/series_test.go @@ -4,18 +4,19 @@ import ( "sort" "testing" + "github.com/prometheus/prometheus/model/labels" "github.com/stretchr/testify/require" ) func Test_UserSeries(t *testing.T) { m := newUserSeriesMap() - m.Add([]byte(`series1`), []byte(`user1`), nil) - m.Add([]byte(`series1`), []byte(`user1`), nil) - m.Add([]byte(`series1`), []byte(`user2`), nil) - m.Add([]byte(`series2`), []byte(`user1`), nil) - m.Add([]byte(`series2`), []byte(`user1`), nil) - m.Add([]byte(`series2`), []byte(`user2`), nil) + m.Add([]byte(`series1`), []byte(`user1`), labels.EmptyLabels()) + m.Add([]byte(`series1`), []byte(`user1`), labels.EmptyLabels()) + m.Add([]byte(`series1`), []byte(`user2`), labels.EmptyLabels()) + m.Add([]byte(`series2`), []byte(`user1`), labels.EmptyLabels()) + m.Add([]byte(`series2`), []byte(`user1`), labels.EmptyLabels()) + m.Add([]byte(`series2`), []byte(`user2`), labels.EmptyLabels()) keys := []string{} diff --git a/pkg/compactor/retention/util_test.go b/pkg/compactor/retention/util_test.go index f30713fc042d9..420acf5618860 100644 --- a/pkg/compactor/retention/util_test.go +++ b/pkg/compactor/retention/util_test.go @@ -201,9 +201,9 @@ func (t *table) Put(chk chunk.Chunk) { func (t *table) GetChunks(userID string, from, through model.Time, metric labels.Labels) []chunk.Chunk { var chunks []chunk.Chunk var matchers []*labels.Matcher - for _, l := range metric { + metric.Range(func(l labels.Label) { matchers = append(matchers, labels.MustNewMatcher(labels.MatchEqual, l.Name, l.Value)) - } + }) for seriesID := range t.chunks[userID] { for _, chk := range t.chunks[userID][seriesID] { diff --git a/pkg/logql/log/labels.go b/pkg/logql/log/labels.go index 3d741b62b0478..846f600a9218d 100644 --- a/pkg/logql/log/labels.go +++ b/pkg/logql/log/labels.go @@ -831,3 +831,22 @@ func (i internedStringSet) Get(data []byte, createNew func() (string, bool)) (st }{s: newStr, ok: ok} return newStr, ok } + +// BufferedLabelsBuilder is a simple builder that uses a label buffer passed in. +// It is used to avoid allocations when building labels. +type BufferedLabelsBuilder struct { + buf labels.Labels +} + +func NewBufferedLabelsBuilder(labels labels.Labels) *BufferedLabelsBuilder { + return &BufferedLabelsBuilder{buf: labels[:0]} +} + +func (b *BufferedLabelsBuilder) Add(label labels.Label) { + b.buf = append(b.buf, label) +} + +func (b *BufferedLabelsBuilder) Labels() labels.Labels { + //slices.SortFunc(b.buf, func(a, b labels.Label) int { return strings.Compare(a.Name, b.Name) }) + return b.buf +}