refactor: single-pass structured log parsing for level detection

level.go

@@ -184,17 +184,23 @@ func redisCharToLevel(level string) Level {
 	return LevelUnknown
 }
 
-// todo
-//
-// python
-//
-//   loglevels: DEBUG, INFO, WARNING, ERROR,CRITICAL
-// pylogging:
-//  default "%(levelname)s:%(name)s:%(message)s" https://github.com/python/cpython/blob/master/Lib/logging/__init__.py#L502
-// django:
-//   {levelname} {message}
-//   {asctime} {module} [{levelname}] okserver
-//   [%(asctime)s] %(levelname)s
-//   %(levelname)s %(asctime)s %(module)s: %(message)s
-// asctime: %(asctime)s Human-readable time when the LogRecord was created. By default this is of the form ‘2003-07-08 16:49:45,896’ (the numbers after the comma are millisecond portion of the time).
-//
+// parseLevelValue maps a structured log level field value (from JSON "level",
+// logfmt level=, etc.) to a Level. Case-insensitive.
+func parseLevelValue(val string) Level {
+	val = strings.ToLower(strings.TrimSpace(val))
+	switch val {
+	case "debug", "dbg":
+		return LevelDebug
+	case "trace", "trc":
+		return LevelDebug
+	case "info", "inf", "notice", "informational", "log":
+		return LevelInfo
+	case "warn", "warning", "wrn":
+		return LevelWarning
+	case "error", "err":
+		return LevelError
+	case "fatal", "critical", "crit", "panic", "dpanic", "alert", "emerg", "emergency":
+		return LevelCritical
+	}
+	return LevelUnknown
+}

level_test.go

@@ -21,6 +21,55 @@ func TestGuessLevelRedis(t *testing.T) {
 	assert.Equal(t, LevelDebug, GuessLevel(`1:S 12 Nov 2019 07:52:11.999 . verbosed`))
 }
 
+func TestParseLevelValue(t *testing.T) {
+	// Error variants
+	assert.Equal(t, LevelError, parseLevelValue("error"))
+	assert.Equal(t, LevelError, parseLevelValue("ERROR"))
+	assert.Equal(t, LevelError, parseLevelValue("Error"))
+	assert.Equal(t, LevelError, parseLevelValue("err"))
+	assert.Equal(t, LevelError, parseLevelValue("ERR"))
+
+	// Info variants
+	assert.Equal(t, LevelInfo, parseLevelValue("info"))
+	assert.Equal(t, LevelInfo, parseLevelValue("INFO"))
+	assert.Equal(t, LevelInfo, parseLevelValue("inf"))
+	assert.Equal(t, LevelInfo, parseLevelValue("notice"))
+	assert.Equal(t, LevelInfo, parseLevelValue("informational"))
+
+	// Warning variants
+	assert.Equal(t, LevelWarning, parseLevelValue("warn"))
+	assert.Equal(t, LevelWarning, parseLevelValue("WARN"))
+	assert.Equal(t, LevelWarning, parseLevelValue("warning"))
+	assert.Equal(t, LevelWarning, parseLevelValue("WARNING"))
+	assert.Equal(t, LevelWarning, parseLevelValue("wrn"))
+
+	// Debug variants
+	assert.Equal(t, LevelDebug, parseLevelValue("debug"))
+	assert.Equal(t, LevelDebug, parseLevelValue("DEBUG"))
+	assert.Equal(t, LevelDebug, parseLevelValue("dbg"))
+	assert.Equal(t, LevelDebug, parseLevelValue("trace"))
+	assert.Equal(t, LevelDebug, parseLevelValue("trc"))
+
+	// Critical/Fatal variants
+	assert.Equal(t, LevelCritical, parseLevelValue("fatal"))
+	assert.Equal(t, LevelCritical, parseLevelValue("FATAL"))
+	assert.Equal(t, LevelCritical, parseLevelValue("critical"))
+	assert.Equal(t, LevelCritical, parseLevelValue("crit"))
+	assert.Equal(t, LevelCritical, parseLevelValue("panic"))
+	assert.Equal(t, LevelCritical, parseLevelValue("dpanic"))
+	assert.Equal(t, LevelCritical, parseLevelValue("alert"))
+	assert.Equal(t, LevelCritical, parseLevelValue("emerg"))
+	assert.Equal(t, LevelCritical, parseLevelValue("emergency"))
+
+	// Unknown
+	assert.Equal(t, LevelUnknown, parseLevelValue("custom"))
+	assert.Equal(t, LevelUnknown, parseLevelValue(""))
+	assert.Equal(t, LevelUnknown, parseLevelValue("verbose"))
+
+	// Whitespace handling
+	assert.Equal(t, LevelError, parseLevelValue("  error  "))
+}
+
 func TestGuessLevel(t *testing.T) {
 	assert.Equal(t, LevelError, GuessLevel(`[Sat Dec 04 04:51:18 2020] [error] mod_jk child workerEnv in error state 6`))
 	assert.Equal(t, LevelInfo, GuessLevel(`[info:2016-02-16T16:04:05.930-08:00] Some log text here`))

logfmt.go

@@ -0,0 +1,57 @@
+package logparser
+
+import "strings"
+
+// logfmtLevelKeys are the key names checked for level in logfmt-style logs.
+var logfmtLevelKeys = []string{"level=", "lvl=", "severity="}
+
+// parseLogfmtLevel extracts the log level from a logfmt-style line.
+// It scans for level=/lvl=/severity= and parses the value.
+// Returns (level, true) if a recognized level field is found.
+func parseLogfmtLevel(line string) (Level, bool) {
+	lower := strings.ToLower(line)
+	for _, key := range logfmtLevelKeys {
+		idx := strings.Index(lower, key)
+		if idx < 0 {
+			continue
+		}
+		// Ensure it's at a word boundary (start of line or preceded by space/tab).
+		if idx > 0 {
+			prev := line[idx-1]
+			if prev != ' ' && prev != '\t' {
+				continue
+			}
+		}
+		valStart := idx + len(key)
+		if valStart >= len(line) {
+			continue
+		}
+		val := extractLogfmtValue(line[valStart:])
+		if lvl := parseLevelValue(val); lvl != LevelUnknown {
+			return lvl, true
+		}
+	}
+	return LevelUnknown, false
+}
+
+// extractLogfmtValue extracts a logfmt value starting at the given position.
+// Handles both quoted ("error") and unquoted (error) values.
+func extractLogfmtValue(s string) string {
+	if len(s) == 0 {
+		return ""
+	}
+	if s[0] == '"' {
+		// Quoted value: find closing quote.
+		end := strings.IndexByte(s[1:], '"')
+		if end < 0 {
+			return ""
+		}
+		return s[1 : 1+end]
+	}
+	// Unquoted value: ends at space or end of string.
+	end := strings.IndexAny(s, " \t")
+	if end < 0 {
+		return s
+	}
+	return s[:end]
+}

logfmt_test.go

@@ -0,0 +1,65 @@
+package logparser
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestParseLogfmtLevel(t *testing.T) {
+	// Basic unquoted values
+	lvl, ok := parseLogfmtLevel(`ts=2024-01-15 level=error msg="failed"`)
+	assert.True(t, ok)
+	assert.Equal(t, LevelError, lvl)
+
+	lvl, ok = parseLogfmtLevel(`level=info msg="started"`)
+	assert.True(t, ok)
+	assert.Equal(t, LevelInfo, lvl)
+
+	lvl, ok = parseLogfmtLevel(`ts=2024-01-15 level=warn msg="slow"`)
+	assert.True(t, ok)
+	assert.Equal(t, LevelWarning, lvl)
+
+	// Quoted values
+	lvl, ok = parseLogfmtLevel(`ts=2024 level="error" msg="timeout"`)
+	assert.True(t, ok)
+	assert.Equal(t, LevelError, lvl)
+
+	lvl, ok = parseLogfmtLevel(`level="WARNING" msg="deprecated"`)
+	assert.True(t, ok)
+	assert.Equal(t, LevelWarning, lvl)
+
+	// Alternative key names
+	lvl, ok = parseLogfmtLevel(`ts=2024 lvl=error msg="fail"`)
+	assert.True(t, ok)
+	assert.Equal(t, LevelError, lvl)
+
+	lvl, ok = parseLogfmtLevel(`severity=fatal msg="crash"`)
+	assert.True(t, ok)
+	assert.Equal(t, LevelCritical, lvl)
+
+	// No level field
+	_, ok = parseLogfmtLevel(`ts=2024 msg="no level here"`)
+	assert.False(t, ok)
+
+	// Not logfmt at all
+	_, ok = parseLogfmtLevel(`just a plain text log line`)
+	assert.False(t, ok)
+
+	// level= not at word boundary (should not match)
+	_, ok = parseLogfmtLevel(`mylevel=error msg="fail"`)
+	assert.False(t, ok)
+
+	// level at start of line (valid)
+	lvl, ok = parseLogfmtLevel(`level=debug msg="trace"`)
+	assert.True(t, ok)
+	assert.Equal(t, LevelDebug, lvl)
+}
+
+func TestExtractLogfmtValue(t *testing.T) {
+	assert.Equal(t, "error", extractLogfmtValue(`error msg="test"`))
+	assert.Equal(t, "error", extractLogfmtValue(`"error" msg="test"`))
+	assert.Equal(t, "warn", extractLogfmtValue("warn"))
+	assert.Equal(t, "", extractLogfmtValue(""))
+	assert.Equal(t, "info", extractLogfmtValue(`"info"`))
+}

parser.go

@@ -153,6 +153,14 @@ func (p *Parser) inc(msg Message) {
 	p.lock.Lock()
 	defer p.lock.Unlock()
 
+	// Single-pass structured log parsing: extract both normalized message
+	// and authoritative level from JSON/logfmt fields. This replaces the
+	// unreliable GuessLevel text-scan for structured logs.
+	normalizedContent, structuredLevel := ParseStructuredLog(msg.Content)
+	if structuredLevel != LevelUnknown {
+		msg.Level = structuredLevel
+	}
+
 	if msg.Level == LevelUnknown || msg.Level == LevelDebug || msg.Level == LevelInfo {
 		key := patternKey{level: msg.Level, hash: ""}
 		if stat := p.patterns[key]; stat == nil {
@@ -162,12 +170,12 @@ func (p *Parser) inc(msg Message) {
 		if p.onMsgCb != nil {
 			p.onMsgCb(msg.Timestamp, msg.Level, "", msg.Content)
 		}
-		pattern := NewPattern(msg.Content)
+		pattern := NewPatternFromNormalized(normalizedContent)
 		p.processSensitivePattern(msg, pattern)
 		return
 	}
 
-	pattern := NewPattern(msg.Content)
+	pattern := NewPatternFromNormalized(normalizedContent)
 	stat, key := p.getPatternStat(msg.Level, pattern, msg.Content)
 	if p.onMsgCb != nil {
 		p.onMsgCb(msg.Timestamp, msg.Level, key.hash, msg.Content)

parser_test.go

@@ -99,6 +99,51 @@ func TestParserMinConfidence(t *testing.T) {
 	}
 }
 
+func TestParserJSONLevelOverride(t *testing.T) {
+	// Verify that JSON structured level overrides GuessLevel in the full pipeline.
+	// An INFO log with "error" in the message must NOT be pattern-grouped as error.
+	p := &Parser{
+		patterns:              map[patternKey]*patternStat{},
+		patternsPerLevel:      map[Level]int{},
+		patternsPerLevelLimit: 256,
+		sensitivePatterns:     map[sensitivePatternKey]*sensitivePatternStat{},
+	}
+
+	// This JSON log has level=INFO but "error" in the msg and "ErrorRate" in type.
+	// Old GuessLevel would misclassify based on text scanning.
+	p.inc(Message{
+		Timestamp: time.Now(),
+		Content:   `{"time":"2026-04-05T06:02:13Z","level":"INFO","msg":"anomaly: processing anomaly","type":"ErrorRate"}`,
+		Level:     LevelUnknown, // as set by multiline collector's GuessLevel
+	})
+
+	// Should be counted as INFO with empty hash (no pattern grouping)
+	counters := p.GetCounters()
+	require.Equal(t, 1, len(counters))
+	assert.Equal(t, LevelInfo, counters[0].Level)
+	assert.Equal(t, "", counters[0].Hash, "INFO logs should have empty pattern hash")
+
+	// Verify no error-level patterns were created
+	assert.Equal(t, 0, p.patternsPerLevel[LevelError])
+
+	// Now send an actual ERROR JSON log
+	p.inc(Message{
+		Timestamp: time.Now(),
+		Content:   `{"time":"2026-04-05T06:02:14Z","level":"ERROR","msg":"connection timeout","err":"dial tcp: timeout"}`,
+		Level:     LevelUnknown,
+	})
+
+	counters = p.GetCounters()
+	var errorCounters []LogCounter
+	for _, c := range counters {
+		if c.Level == LevelError {
+			errorCounters = append(errorCounters, c)
+		}
+	}
+	require.Equal(t, 1, len(errorCounters))
+	assert.NotEmpty(t, errorCounters[0].Hash, "ERROR logs should have a pattern hash")
+}
+
 func TestParserCardinalityLimit(t *testing.T) {
 	p := &Parser{
 		patterns:              map[patternKey]*patternStat{},