fix: Fix log level detection (#12651)

docs/sources/shared/configuration.md

@@ -2925,8 +2925,10 @@ The `limits_config` block configures global and per-tenant limits in Loki. The v
 [discover_service_name: <list of strings> | default = [service app application name app_kubernetes_io_name container container_name component workload job]]
 
 # Discover and add log levels during ingestion, if not present already. Levels
-# would be added to Structured Metadata with name 'level' and one of the values
-# from 'debug', 'info', 'warn', 'error', 'critical', 'fatal'.
+# would be added to Structured Metadata with name
+# level/LEVEL/Level/Severity/severity/SEVERITY/lvl/LVL/Lvl (case-sensitive) and
+# one of the values from 'trace', 'debug', 'info', 'warn', 'error', 'critical',
+# 'fatal' (case insensitive).
 # CLI flag: -validation.discover-log-levels
 [discover_log_levels: <boolean> | default = true]
 

go.mod

@@ -119,6 +119,7 @@ require (
 	github.com/IBM/go-sdk-core/v5 v5.13.1
 	github.com/IBM/ibm-cos-sdk-go v1.10.0
 	github.com/axiomhq/hyperloglog v0.0.0-20240124082744-24bca3a5b39b
+	github.com/buger/jsonparser v1.1.1
 	github.com/d4l3k/messagediff v1.2.1
 	github.com/dolthub/swiss v0.2.1
 	github.com/efficientgo/core v1.0.0-rc.2

go.sum

@@ -401,6 +401,8 @@ github.com/bmatcuk/doublestar v1.3.4/go.mod h1:wiQtGV+rzVYxB7WIlirSN++5HPtPlXEo9
 github.com/bmizerany/assert v0.0.0-20160611221934-b7ed37b82869 h1:DDGfHa7BWjL4YnC6+E63dPcxHo2sUxDIu8g3QgEJdRY=
 github.com/bmizerany/assert v0.0.0-20160611221934-b7ed37b82869/go.mod h1:Ekp36dRnpXw/yCqJaO+ZrUyxD+3VXMFFr56k5XYrpB4=
 github.com/boltdb/bolt v1.3.1/go.mod h1:clJnj/oiGkjum5o1McbSZDSLxVThjynRyGBgiAx27Ps=
+github.com/buger/jsonparser v1.1.1 h1:2PnMjfWD7wBILjqQbt530v576A/cAbQvEW9gGIpYMUs=
+github.com/buger/jsonparser v1.1.1/go.mod h1:6RYKKt7H4d4+iWqouImQ9R2FZql3VbhNgx27UK13J/0=
 github.com/c2h5oh/datasize v0.0.0-20220606134207-859f65c6625b h1:6+ZFm0flnudZzdSE0JxlhR2hKnGPcNB35BjQf4RYQDY=
 github.com/c2h5oh/datasize v0.0.0-20220606134207-859f65c6625b/go.mod h1:S/7n9copUssQ56c7aAgHqftWO4LTf4xY6CGWt8Bc+3M=
 github.com/caddyserver/caddy v1.0.4/go.mod h1:uruyfVsyMcDb3IOzSKsi1x0wOjy1my/PxOSTcD+24jM=

pkg/distributor/distributor.go

@@ -12,8 +12,10 @@ import (
 	"time"
 	"unicode"
 
+	"github.com/buger/jsonparser"
 	"github.com/go-kit/log"
 	"github.com/go-kit/log/level"
+	"github.com/go-logfmt/logfmt"
 	"github.com/gogo/status"
 	"github.com/prometheus/prometheus/model/labels"
 	"go.opentelemetry.io/collector/pdata/plog"
@@ -59,20 +61,28 @@ const (
 
 	labelServiceName = "service_name"
 	serviceUnknown   = "unknown_service"
-	labelLevel       = "level"
+	levelLabel       = "detected_level"
 	logLevelDebug    = "debug"
 	logLevelInfo     = "info"
 	logLevelWarn     = "warn"
 	logLevelError    = "error"
 	logLevelFatal    = "fatal"
 	logLevelCritical = "critical"
+	logLevelTrace    = "trace"
+	logLevelUnknown  = "unknown"
 )
 
 var (
 	maxLabelCacheSize = 100000
 	rfStats           = analytics.NewInt("distributor_replication_factor")
 )
 
+var allowedLabelsForLevel = map[string]struct{}{
+	"level": {}, "LEVEL": {}, "Level": {},
+	"severity": {}, "SEVERITY": {}, "Severity": {},
+	"lvl": {}, "LVL": {}, "Lvl": {},
+}
+
 // Config for a Distributor.
 type Config struct {
 	// Distributors ring
@@ -376,7 +386,9 @@ func (d *Distributor) Push(ctx context.Context, req *logproto.PushRequest) (*log
 			n := 0
 			pushSize := 0
 			prevTs := stream.Entries[0].Timestamp
-			addLogLevel := validationContext.allowStructuredMetadata && validationContext.discoverLogLevels && !lbs.Has(labelLevel)
+
+			shouldDiscoverLevels := validationContext.allowStructuredMetadata && validationContext.discoverLogLevels
+			levelFromLabel, hasLevelLabel := hasAnyLevelLabels(lbs)
 			for _, entry := range stream.Entries {
 				if err := d.validator.ValidateEntry(ctx, validationContext, lbs, entry); err != nil {
 					d.writeFailuresManager.Log(tenantID, err)
@@ -385,12 +397,21 @@ func (d *Distributor) Push(ctx context.Context, req *logproto.PushRequest) (*log
 				}
 
 				structuredMetadata := logproto.FromLabelAdaptersToLabels(entry.StructuredMetadata)
-				if addLogLevel && !structuredMetadata.Has(labelLevel) {
-					logLevel := detectLogLevelFromLogEntry(entry, structuredMetadata)
-					entry.StructuredMetadata = append(entry.StructuredMetadata, logproto.LabelAdapter{
-						Name:  labelLevel,
-						Value: logLevel,
-					})
+				if shouldDiscoverLevels {
+					var logLevel string
+					if hasLevelLabel {
+						logLevel = levelFromLabel
+					} else if levelFromMetadata, ok := hasAnyLevelLabels(structuredMetadata); ok {
+						logLevel = levelFromMetadata
+					} else {
+						logLevel = detectLogLevelFromLogEntry(entry, structuredMetadata)
+					}
+					if logLevel != logLevelUnknown && logLevel != "" {
+						entry.StructuredMetadata = append(entry.StructuredMetadata, logproto.LabelAdapter{
+							Name:  levelLabel,
+							Value: logLevel,
+						})
+					}
 				}
 				stream.Entries[n] = entry
 
@@ -537,6 +558,15 @@ func (d *Distributor) Push(ctx context.Context, req *logproto.PushRequest) (*log
 	}
 }
 
+func hasAnyLevelLabels(l labels.Labels) (string, bool) {
+	for lbl := range allowedLabelsForLevel {
+		if l.Has(lbl) {
+			return l.Get(lbl), true
+		}
+	}
+	return "", false
+}
+
 // shardStream shards (divides) the given stream into N smaller streams, where
 // N is the sharding size for the given stream. shardSteam returns the smaller
 // streams and their associated keys for hashing to ingesters.
@@ -865,7 +895,11 @@ func detectLogLevelFromLogEntry(entry logproto.Entry, structuredMetadata labels.
 		if err != nil {
 			return logLevelInfo
 		}
-		if otlpSeverityNumber <= int(plog.SeverityNumberDebug4) {
+		if otlpSeverityNumber == int(plog.SeverityNumberUnspecified) {
+			return logLevelUnknown
+		} else if otlpSeverityNumber <= int(plog.SeverityNumberTrace4) {
+			return logLevelTrace
+		} else if otlpSeverityNumber <= int(plog.SeverityNumberDebug4) {
 			return logLevelDebug
 		} else if otlpSeverityNumber <= int(plog.SeverityNumberInfo4) {
 			return logLevelInfo
@@ -876,74 +910,87 @@ func detectLogLevelFromLogEntry(entry logproto.Entry, structuredMetadata labels.
 		} else if otlpSeverityNumber <= int(plog.SeverityNumberFatal4) {
 			return logLevelFatal
 		}
-		return logLevelInfo
+		return logLevelUnknown
 	}
 
 	return extractLogLevelFromLogLine(entry.Line)
 }
 
 func extractLogLevelFromLogLine(log string) string {
-	// check for log levels in known log formats to avoid any false detection
+	var v string
+	if isJSON(log) {
+		v = getValueUsingJSONParser(log)
+	} else {
+		v = getValueUsingLogfmtParser(log)
+	}
 
-	// json logs:
+	switch strings.ToLower(v) {
+	case "trace", "trc":
+		return logLevelTrace
+	case "debug", "dbg":
+		return logLevelDebug
+	case "info", "inf":
+		return logLevelInfo
+	case "warn", "wrn":
+		return logLevelWarn
+	case "error", "err":
+		return logLevelError
+	case "critical":
+		return logLevelCritical
+	case "fatal":
+		return logLevelFatal
+	default:
+		return detectLevelFromLogLine(log)
+	}
+}
+
+func getValueUsingLogfmtParser(line string) string {
+	equalIndex := strings.Index(line, "=")
+	if len(line) == 0 || equalIndex == -1 {
+		return logLevelUnknown
+	}
+	d := logfmt.NewDecoder(strings.NewReader(line))
+	d.ScanRecord()
+	for d.ScanKeyval() {
+		if _, ok := allowedLabelsForLevel[string(d.Key())]; ok {
+			return string(d.Value())
+		}
+	}
+	return logLevelUnknown
+}
+
+func getValueUsingJSONParser(log string) string {
+	for allowedLabel := range allowedLabelsForLevel {
+		l, err := jsonparser.GetString([]byte(log), allowedLabel)
+		if err == nil {
+			return l
+		}
+	}
+	return logLevelUnknown
+}
+
+func isJSON(line string) bool {
 	var firstNonSpaceChar rune
-	for _, char := range log {
+	for _, char := range line {
 		if !unicode.IsSpace(char) {
 			firstNonSpaceChar = char
 			break
 		}
 	}
 
 	var lastNonSpaceChar rune
-	for i := len(log) - 1; i >= 0; i-- {
-		char := rune(log[i])
+	for i := len(line) - 1; i >= 0; i-- {
+		char := rune(line[i])
 		if !unicode.IsSpace(char) {
 			lastNonSpaceChar = char
 			break
 		}
 	}
 
-	if firstNonSpaceChar == '{' && lastNonSpaceChar == '}' {
-		if strings.Contains(log, `:"err"`) || strings.Contains(log, `:"ERR"`) ||
-			strings.Contains(log, `:"error"`) || strings.Contains(log, `:"ERROR"`) {
-			return logLevelError
-		}
-		if strings.Contains(log, `:"warn"`) || strings.Contains(log, `:"WARN"`) ||
-			strings.Contains(log, `:"warning"`) || strings.Contains(log, `:"WARNING"`) {
-			return logLevelWarn
-		}
-		if strings.Contains(log, `:"critical"`) || strings.Contains(log, `:"CRITICAL"`) {
-			return logLevelCritical
-		}
-		if strings.Contains(log, `:"debug"`) || strings.Contains(log, `:"DEBUG"`) {
-			return logLevelDebug
-		}
-		if strings.Contains(log, `:"info"`) || strings.Contains(log, `:"INFO"`) {
-			return logLevelInfo
-		}
-	}
-
-	// logfmt logs:
-	if strings.Contains(log, "=") {
-		if strings.Contains(log, "=err") || strings.Contains(log, "=ERR") ||
-			strings.Contains(log, "=error") || strings.Contains(log, "=ERROR") {
-			return logLevelError
-		}
-		if strings.Contains(log, "=warn") || strings.Contains(log, "=WARN") ||
-			strings.Contains(log, "=warning") || strings.Contains(log, "=WARNING") {
-			return logLevelWarn
-		}
-		if strings.Contains(log, "=critical") || strings.Contains(log, "=CRITICAL") {
-			return logLevelCritical
-		}
-		if strings.Contains(log, "=debug") || strings.Contains(log, "=DEBUG") {
-			return logLevelDebug
-		}
-		if strings.Contains(log, "=info") || strings.Contains(log, "=INFO") {
-			return logLevelInfo
-		}
-	}
+	return firstNonSpaceChar == '{' && lastNonSpaceChar == '}'
+}
 
+func detectLevelFromLogLine(log string) string {
 	if strings.Contains(log, "err:") || strings.Contains(log, "ERR:") ||
 		strings.Contains(log, "error") || strings.Contains(log, "ERROR") {
 		return logLevelError
@@ -958,7 +1005,5 @@ func extractLogLevelFromLogLine(log string) string {
 	if strings.Contains(log, "debug:") || strings.Contains(log, "DEBUG:") {
 		return logLevelDebug
 	}
-
-	// Default to info if no specific level is found
-	return logLevelInfo
+	return logLevelUnknown
 }