Merge branch 'main' into admonitionFormat

pkg/logql/log/fmt.go

@@ -222,8 +222,12 @@ func (lf *LineFormatter) Process(ts int64, line []byte, lbs *LabelsBuilder) ([]b
 	lf.currentTs = ts
 
 	// map now is taking from a pool
-	m := lbs.Map()
-	defer smp.Put(m)
+	m, ret := lbs.Map()
+	defer func() {
+		if ret { // if we return the base map from the labels builder we should not put it back in the pool
+			smp.Put(m)
+		}
+	}()
 	if err := lf.Template.Execute(lf.buf, m); err != nil {
 		lbs.SetErr(errTemplateFormat)
 		lbs.SetErrorDetails(err.Error())

pkg/logql/log/fmt_test.go

@@ -2,6 +2,7 @@ package log
 
 import (
 	"fmt"
+	"sync"
 	"testing"
 	"time"
 
@@ -923,3 +924,48 @@ func TestInvalidUnixTimes(t *testing.T) {
 	_, err = unixToTime("464")
 	require.Error(t, err)
 }
+
+func TestMapPoolPanic(_ *testing.T) {
+	wg := sync.WaitGroup{}
+	wg.Add(1)
+	wgFinished := sync.WaitGroup{}
+
+	ls := labels.FromStrings("cluster", "us-central-0")
+	builder := NewBaseLabelsBuilder().ForLabels(ls, ls.Hash())
+	// this specific line format was part of the query that first alerted us to the panic caused by map pooling in the label/line formatter Process functions
+	tmpl := `[1m{{if .level }}{{alignRight 5 .level}}{{else if .severity}}{{alignRight 5 .severity}}{{end}}[0m [90m[{{alignRight 10 .resources_service_instance_id}}{{if .attributes_thread_name}}/{{alignRight 20 .attributes_thread_name}}{{else if eq "java" .resources_telemetry_sdk_language }}                    {{end}}][0m [36m{{if .instrumentation_scope_name }}{{alignRight 40 .instrumentation_scope_name}}{{end}}[0m {{.body}} {{if .traceid}} [37m[3m[traceid={{.traceid}}]{{end}}`
+	a := newMustLineFormatter(tmpl)
+	a.Process(0,
+		[]byte("logger=sqlstore.metrics traceID=XXXXXXXXXXXXXXXXXXXXXXXXXXXX t=2024-01-04T23:58:47.696779826Z level=debug msg=\"query finished\" status=success elapsedtime=1.523571ms sql=\"some SQL query\" error=null"),
+		builder,
+	)
+
+	for i := 0; i < 100; i++ {
+		wgFinished.Add(1)
+		go func() {
+			wg.Wait()
+			a := newMustLineFormatter(tmpl)
+			a.Process(0,
+				[]byte("logger=sqlstore.metrics traceID=XXXXXXXXXXXXXXXXXXXXXXXXXXXX t=2024-01-04T23:58:47.696779826Z level=debug msg=\"query finished\" status=success elapsedtime=1.523571ms sql=\"some SQL query\" error=null"),
+				builder,
+			)
+			wgFinished.Done()
+		}()
+	}
+	for i := 0; i < 100; i++ {
+		wgFinished.Add(1)
+		j := i
+		go func() {
+			wg.Wait()
+			m := smp.Get()
+			for k, v := range m {
+				m[k] = fmt.Sprintf("%s%d", v, j)
+			}
+			smp.Put(m)
+			wgFinished.Done()
+
+		}()
+	}
+	wg.Done()
+	wgFinished.Wait()
+}

pkg/logql/log/labels.go

@@ -495,22 +495,21 @@ func (b *LabelsBuilder) IntoMap(m map[string]string) {
 	}
 }
 
-func (b *LabelsBuilder) Map() map[string]string {
+func (b *LabelsBuilder) Map() (map[string]string, bool) {
 	if !b.hasDel() && !b.hasAdd() && !b.HasErr() {
 		if b.baseMap == nil {
 			b.baseMap = b.base.Map()
 		}
-		return b.baseMap
+		return b.baseMap, false
 	}
 	b.buf = b.UnsortedLabels(b.buf)
 	// todo should we also cache maps since limited by the result ?
 	// Maps also don't create a copy of the labels.
 	res := smp.Get()
-	clear(res)
 	for _, l := range b.buf {
 		res[l.Name] = l.Value
 	}
-	return res
+	return res, true
 }
 
 // LabelsResult returns the LabelsResult from the builder.