fix: sanatize structured metadata at query time

pkg/logql/log/pipeline.go

@@ -67,7 +67,7 @@ func (n *noopPipeline) ForStream(labels labels.Labels) StreamPipeline {
 	}
 	n.mu.RUnlock()
 
-	sp := &noopStreamPipeline{n.baseBuilder.ForLabels(labels, h)}
+	sp := &noopStreamPipeline{n.baseBuilder.ForLabels(labels, h), make([]int, 0, 10)}
 
 	n.mu.Lock()
 	defer n.mu.Unlock()
@@ -92,7 +92,8 @@ func IsNoopPipeline(p Pipeline) bool {
 }
 
 type noopStreamPipeline struct {
-	builder *LabelsBuilder
+	builder    *LabelsBuilder
+	offsetsBuf []int
 }
 
 func (n noopStreamPipeline) ReferencedStructuredMetadata() bool {
@@ -101,6 +102,9 @@ func (n noopStreamPipeline) ReferencedStructuredMetadata() bool {
 
 func (n noopStreamPipeline) Process(_ int64, line []byte, structuredMetadata ...labels.Label) ([]byte, LabelsResult, bool) {
 	n.builder.Reset()
+	for i, lb := range structuredMetadata {
+		structuredMetadata[i].Name = replaceChars(lb.Name, n.offsetsBuf)
+	}
 	n.builder.Add(StructuredMetadataLabel, structuredMetadata...)
 	return line, n.builder.LabelsResult(), true
 }
@@ -176,12 +180,13 @@ func NewPipeline(stages []Stage) Pipeline {
 }
 
 type streamPipeline struct {
-	stages  []Stage
-	builder *LabelsBuilder
+	stages     []Stage
+	builder    *LabelsBuilder
+	offsetsBuf []int
 }
 
 func NewStreamPipeline(stages []Stage, labelsBuilder *LabelsBuilder) StreamPipeline {
-	return &streamPipeline{stages, labelsBuilder}
+	return &streamPipeline{stages, labelsBuilder, make([]int, 0, 10)}
 }
 
 func (p *pipeline) ForStream(labels labels.Labels) StreamPipeline {
@@ -220,6 +225,11 @@ func (p *streamPipeline) ReferencedStructuredMetadata() bool {
 func (p *streamPipeline) Process(ts int64, line []byte, structuredMetadata ...labels.Label) ([]byte, LabelsResult, bool) {
 	var ok bool
 	p.builder.Reset()
+
+	for i, lb := range structuredMetadata {
+		structuredMetadata[i].Name = replaceChars(lb.Name, p.offsetsBuf)
+	}
+
 	p.builder.Add(StructuredMetadataLabel, structuredMetadata...)
 
 	for _, s := range p.stages {
@@ -381,3 +391,32 @@ func unsafeGetBytes(s string) []byte {
 func unsafeGetString(buf []byte) string {
 	return *((*string)(unsafe.Pointer(&buf)))
 }
+
+func replaceChars(str string, offsets []int) string {
+	offsets = offsets[:0]
+	for i, r := range str {
+		if !isDigit(r) && !isAlpha(r) {
+			offsets = append(offsets, i)
+		}
+	}
+
+	if len(offsets) > 0 {
+		runes := []rune(str)
+		for _, offset := range offsets {
+			runes[offset] = '_'
+		}
+
+		return string(runes)
+	}
+
+	return str
+}
+
+func isDigit(r rune) bool {
+	return '0' <= r && r <= '9'
+}
+
+// isAlpha reports whether r is an alphabetic or underscore.
+func isAlpha(r rune) bool {
+	return r == '_' || ('a' <= r && r <= 'z') || ('A' <= r && r <= 'Z')
+}

pkg/logql/log/pipeline_test.go

@@ -56,6 +56,18 @@
 	require.Equal(t, expectedLabelsResults.String(), lbr.String())
 	require.Equal(t, true, matches)
 
+	// test structured metadata with disallowed label names
+	structuredMetadata = append(labels.FromStrings("y", "1", "z", "2"), labels.Label{Name: "zsomething-bad", Value: "foo"})
+	expectedStructuredMetadata := append(labels.FromStrings("y", "1", "z", "2"), labels.Label{Name: "zsomething_bad", Value: "foo"})
+	expectedLabelsResults = append(lbs, expectedStructuredMetadata...)
+
+	l, lbr, matches = pipeline.ForStream(lbs).Process(0, []byte(""), structuredMetadata...)
+	require.Equal(t, []byte(""), l)
+	require.Equal(t, NewLabelsResult(expectedLabelsResults.String(), expectedLabelsResults.Hash(), lbs, expectedStructuredMetadata, labels.EmptyLabels()), lbr)
+	require.Equal(t, expectedLabelsResults.Hash(), lbr.Hash())
+	require.Equal(t, expectedLabelsResults.String(), lbr.String())
+	require.Equal(t, true, matches)
+
 	pipeline.Reset()
 	require.Len(t, pipeline.cache, 0)
 }
@@ -171,6 +183,17 @@
 	require.Equal(t, nil, lbr)
 	require.Equal(t, false, matches)
 
+	// test structured metadata with disallowed label names
+	withBadLabel := append(structuredMetadata, labels.Label{Name: "zsomething-bad", Value: "foo"})
+	expectedStructuredMetadata := append(structuredMetadata, labels.Label{Name: "zsomething_bad", Value: "foo"})
+	expectedLabelsResults = append(lbs, expectedStructuredMetadata...)
+
+	l, lbr, matches = p.ForStream(lbs).Process(0, []byte(""), withBadLabel...)
+	require.Equal(t, NewLabelsResult(expectedLabelsResults.String(), expectedLabelsResults.Hash(), lbs, expectedStructuredMetadata, labels.EmptyLabels()), lbr)
+	require.Equal(t, expectedLabelsResults.Hash(), lbr.Hash())
+	require.Equal(t, expectedLabelsResults.String(), lbr.String())
+	require.Equal(t, true, matches)
+
 	// Reset caches
 	p.baseBuilder.del = []string{"foo", "bar"}
 	p.baseBuilder.add = [numValidCategories]labels.Labels{
@@ -566,6 +589,19 @@
 		}
 	})
 
+	b.Run("pipeline bytes no invalid structured metadata", func(b *testing.B) {
+		b.ResetTimer()
+		for n := 0; n < b.N; n++ {
+			resLine, resLbs, resMatches = sp.Process(0, line, labels.Label{Name: "valid_name", Value: "foo"})
+		}
+	})
+	b.Run("pipeline string with invalid structured metadata", func(b *testing.B) {
+		b.ResetTimer()
+		for n := 0; n < b.N; n++ {
+			resLine, resLbs, resMatches = sp.Process(0, line, labels.Label{Name: "invalid-name", Value: "foo"}, labels.Label{Name: "other-invalid-name", Value: "foo"})
+		}
+	})
+
 	extractor, err := NewLineSampleExtractor(CountExtractor, stages, []string{"cluster", "level"}, false, false)
 	require.NoError(b, err)
 	ex := extractor.ForStream(lbs)