fix: short circuit level detection when already detected (#17028)

Author: trevorwhitney

pkg/distributor/field_detection.go

@@ -94,6 +94,11 @@ func (l *FieldDetector) shouldDiscoverGenericFields() bool {
 }
 
 func (l *FieldDetector) extractLogLevel(labels labels.Labels, structuredMetadata labels.Labels, entry logproto.Entry) (logproto.LabelAdapter, bool) {
+	// If the level is already set in the structured metadata, we don't need to do anything.
+	if structuredMetadata.Has(constants.LevelLabel) {
+		return logproto.LabelAdapter{}, false
+	}
+
 	levelFromLabel, hasLevelLabel := labelsContainAny(labels, l.allowedLevelLabels)
 	var logLevel string
 	if hasLevelLabel {

pkg/distributor/field_detection_test.go

@@ -118,6 +118,55 @@ func Test_DetectLogLevels(t *testing.T) {
 			},
 		}, sm)
 	})
+
+	t.Run("detected_level structured metadata takes precedence over other level detection methods", func(t *testing.T) {
+		limits, ingester := setup(true)
+		distributors, _ := prepare(t, 1, 5, limits, func(_ string) (ring_client.PoolClient, error) { return ingester, nil })
+
+		// Create a write request with multiple potential level sources:
+		// 1. A JSON log line with level=error
+		// 2. A log level in stream labels (level=debug)
+		// 3. OTLP severity number in structured metadata
+		// 4. A severity field in structured metadata
+		// 5. The detected_level field in structured metadata (should take precedence)
+		writeReq := makeWriteRequestWithLabels(1, 10, []string{`{foo="bar", level="debug"}`}, false, false, false)
+		writeReq.Streams[0].Entries[0].Line = `{"msg":"this is a test message", "level":"error"}`
+		writeReq.Streams[0].Entries[0].StructuredMetadata = push.LabelsAdapter{
+			{
+				Name:  loghttp_push.OTLPSeverityNumber,
+				Value: fmt.Sprintf("%d", plog.SeverityNumberWarn),
+			},
+			{
+				Name:  "severity",
+				Value: constants.LogLevelCritical,
+			},
+			{
+				Name:  constants.LevelLabel, // detected_level
+				Value: constants.LogLevelTrace,
+			},
+		}
+
+		_, err := distributors[0].Push(ctx, writeReq)
+		require.NoError(t, err)
+		topVal := ingester.Peek()
+		require.Equal(t, `{foo="bar", level="debug"}`, topVal.Streams[0].Labels)
+
+		// Verify that detected_level from structured metadata is preserved and used
+		sm := topVal.Streams[0].Entries[0].StructuredMetadata
+
+		detectedLevelLbls := make([]logproto.LabelAdapter, 0, len(sm))
+		for _, sm := range sm {
+			if sm.Name == constants.LevelLabel {
+				detectedLevelLbls = append(detectedLevelLbls, sm)
+			}
+		}
+
+		require.Len(t, detectedLevelLbls, 1)
+		require.Contains(t, detectedLevelLbls, logproto.LabelAdapter{
+			Name:  constants.LevelLabel,
+			Value: constants.LogLevelTrace,
+		})
+	})
 }
 
 func Test_detectLogLevelFromLogEntry(t *testing.T) {