Merge branch 'main' into helm-query-frontend-lb-optional

Author: GarnerBot

pkg/distributor/field_detection.go

@@ -79,17 +79,25 @@ 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
+	// Check if detected_level is already present in entry.StructuredMetadata and normalize it
+	for i, sm := range entry.StructuredMetadata {
+		if sm.Name == constants.LevelLabel {
+			normalizedLevel := normalizeLogLevel(sm.Value)
+			if sm.Value != normalizedLevel {
+				// Update the value in-place with the normalized version
+				entry.StructuredMetadata[i].Value = normalizedLevel
+			}
+			// Level already exists and has been normalized if needed
+			return logproto.LabelAdapter{}, false
+		}
 	}
 
 	levelFromLabel, hasLevelLabel := labelsContainAny(labels, l.allowedLevelLabels)
 	var logLevel string
 	if hasLevelLabel {
-		logLevel = levelFromLabel
+		logLevel = normalizeLogLevel(levelFromLabel)
 	} else if levelFromMetadata, ok := labelsContainAny(structuredMetadata, l.allowedLevelLabels); ok {
-		logLevel = levelFromMetadata
+		logLevel = normalizeLogLevel(levelFromMetadata)
 	} else {
 		logLevel = l.detectLogLevelFromLogEntry(entry, structuredMetadata)
 	}
@@ -129,6 +137,30 @@ func labelsContainAny(labels labels.Labels, names []string) (string, bool) {
 	return "", false
 }
 
+// normalizeLogLevel normalizes log level strings to lowercase standard values
+func normalizeLogLevel(level string) string {
+	levelBytes := unsafe.Slice(unsafe.StringData(level), len(level)) // #nosec G103 -- we know the string is not mutated -- nosemgrep: use-of-unsafe-block
+	switch {
+	case bytes.EqualFold(levelBytes, traceBytes), bytes.EqualFold(levelBytes, traceAbbrv):
+		return constants.LogLevelTrace
+	case bytes.EqualFold(levelBytes, debug), bytes.EqualFold(levelBytes, debugAbbrv):
+		return constants.LogLevelDebug
+	case bytes.EqualFold(levelBytes, info), bytes.EqualFold(levelBytes, infoAbbrv), bytes.EqualFold(levelBytes, infoFull):
+		return constants.LogLevelInfo
+	case bytes.EqualFold(levelBytes, warn), bytes.EqualFold(levelBytes, warnAbbrv), bytes.EqualFold(levelBytes, warning):
+		return constants.LogLevelWarn
+	case bytes.EqualFold(levelBytes, errorStr), bytes.EqualFold(levelBytes, errorAbbrv):
+		return constants.LogLevelError
+	case bytes.EqualFold(levelBytes, critical):
+		return constants.LogLevelCritical
+	case bytes.EqualFold(levelBytes, fatal):
+		return constants.LogLevelFatal
+	default:
+		// Return the original value if it doesn't match any known level
+		return level
+	}
+}
+
 func (l *FieldDetector) detectLogLevelFromLogEntry(entry logproto.Entry, structuredMetadata labels.Labels) string {
 	// otlp logs have a severity number, using which we are defining the log levels.
 	// Significance of severity number is explained in otel docs here https://opentelemetry.io/docs/specs/otel/logs/data-model/#field-severitynumber

pkg/distributor/field_detection_test.go

@@ -168,6 +168,81 @@ func Test_DetectLogLevels(t *testing.T) {
 		})
 	})
 
+	t.Run("detected_level with mixed case value gets normalized to lowercase", func(t *testing.T) {
+		// Test various mixed case values
+		testCases := []struct {
+			input    string
+			expected string
+		}{
+			{"WaRn", constants.LogLevelWarn},
+			{"InFo", constants.LogLevelInfo},
+			{"CRITICAL", constants.LogLevelCritical},
+			{"Debug", constants.LogLevelDebug},
+			{"FaTaL", constants.LogLevelFatal},
+			{"tRaCe", constants.LogLevelTrace},
+			{"ERROR", constants.LogLevelError},
+		}
+
+		for _, tc := range testCases {
+			// Create a fresh setup for each test case
+			limits, ingester := setup(true)
+			distributors, _ := prepare(t, 1, 5, limits, func(_ string) (ring_client.PoolClient, error) { return ingester, nil })
+
+			writeReq := makeWriteRequestWithLabels(1, 10, []string{`{foo="bar"}`}, false, false, false)
+			writeReq.Streams[0].Entries[0].Line = `test log`
+			writeReq.Streams[0].Entries[0].StructuredMetadata = push.LabelsAdapter{
+				{
+					Name:  constants.LevelLabel,
+					Value: tc.input,
+				},
+			}
+
+			_, err := distributors[0].Push(ctx, writeReq)
+			require.NoError(t, err)
+			topVal := ingester.Peek()
+
+			sm := topVal.Streams[0].Entries[0].StructuredMetadata
+			require.Len(t, sm, 1)
+			require.Equal(t, constants.LevelLabel, sm[0].Name)
+			require.Equal(t, tc.expected, sm[0].Value, "Input %q should normalize to %q", tc.input, tc.expected)
+		}
+	})
+
+	t.Run("level from stream labels gets normalized to lowercase in detected_level", func(t *testing.T) {
+		// Test various mixed case values in stream labels
+		testCases := []struct {
+			streamLabel string
+			expected    string
+		}{
+			{`{foo="bar", level="ERROR"}`, constants.LogLevelError},
+			{`{foo="bar", level="WaRn"}`, constants.LogLevelWarn},
+			{`{foo="bar", level="InFo"}`, constants.LogLevelInfo},
+			{`{foo="bar", level="CRITICAL"}`, constants.LogLevelCritical},
+			{`{foo="bar", level="Debug"}`, constants.LogLevelDebug},
+			{`{foo="bar", level="FaTaL"}`, constants.LogLevelFatal},
+			{`{foo="bar", level="tRaCe"}`, constants.LogLevelTrace},
+		}
+
+		for _, tc := range testCases {
+			// Create a fresh setup for each test case
+			limits, ingester := setup(true)
+			distributors, _ := prepare(t, 1, 5, limits, func(_ string) (ring_client.PoolClient, error) { return ingester, nil })
+
+			writeReq := makeWriteRequestWithLabels(1, 10, []string{tc.streamLabel}, false, false, false)
+			writeReq.Streams[0].Entries[0].Line = `log message without level`
+
+			_, err := distributors[0].Push(ctx, writeReq)
+			require.NoError(t, err)
+			topVal := ingester.Peek()
+
+			// Verify that detected_level is normalized to lowercase
+			sm := topVal.Streams[0].Entries[0].StructuredMetadata
+			require.Len(t, sm, 1, "Expected detected_level in structured metadata for stream label %s", tc.streamLabel)
+			require.Equal(t, constants.LevelLabel, sm[0].Name)
+			require.Equal(t, tc.expected, sm[0].Value, "Stream label %q should normalize to %q in detected_level", tc.streamLabel, tc.expected)
+		}
+	})
+
 	t.Run("indexed OTEL severity takes precedence over structured metadata or log line", func(t *testing.T) {
 		limits, ingester := setup(true)
 		distributors, _ := prepare(t, 1, 5, limits, func(_ string) (ring_client.PoolClient, error) { return ingester, nil })