merge coroot/logparser v1.4.2: JSON parser, pattern rate limit, level fixes

cmd/logparser.go

@@ -30,7 +30,7 @@ func main() {
 
 	reader := bufio.NewReader(os.Stdin)
 	ch := make(chan logparser.LogEntry)
-	parser := logparser.NewParser(ch, nil, nil, time.Second, 256, logparser.SensitiveConfig{Enabled: true, MinConfidence: "medium"})
+	parser := logparser.NewParser(ch, nil, nil, time.Second, 256, true, nil, logparser.SensitiveConfig{Enabled: true, MinConfidence: "medium"})
 	t := time.Now()
 	for {
 		line, err := reader.ReadString('\n')

go.mod

@@ -5,6 +5,7 @@ go 1.24
 require (
 	github.com/jaeyo/go-drain3 v0.1.2
 	github.com/stretchr/testify v1.11.1
+	golang.org/x/time v0.12.0
 )
 
 require (

go.sum

@@ -8,6 +8,8 @@ github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZb
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
 github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U=
 github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U=
+golang.org/x/time v0.12.0 h1:ScB/8o8olJvc+CQPWrK3fPZNfh7qgwCrY0zJmoEQLSE=
+golang.org/x/time v0.12.0/go.mod h1:CDIdPxbZBQxdj6cxyCIdrNogrJKMJ7pr37NYpMcMDSg=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=

impact_test.go

@@ -54,7 +54,7 @@ func TestResourceImpact(t *testing.T) {
 	parsers := make([]*Parser, 200)
 	for i := range parsers {
 		ch := make(chan LogEntry, 1)
-		parsers[i] = NewParser(ch, nil, nil, time.Second, 256, SensitiveConfig{
+		parsers[i] = NewParser(ch, nil, nil, time.Second, 256, false, nil, SensitiveConfig{
 			Enabled:       true,
 			MinConfidence: "medium",
 			MaxDetections: 100,
@@ -93,7 +93,7 @@ func TestResourceImpact(t *testing.T) {
 	disabledParsers := make([]*Parser, 200)
 	for i := range disabledParsers {
 		ch := make(chan LogEntry, 1)
-		disabledParsers[i] = NewParser(ch, nil, nil, time.Second, 256, SensitiveConfig{
+		disabledParsers[i] = NewParser(ch, nil, nil, time.Second, 256, false, nil, SensitiveConfig{
 			Enabled: false,
 		})
 	}
@@ -144,7 +144,7 @@ func TestResourceImpact(t *testing.T) {
 	for _, cfg := range configs {
 		// Create a parser with this config
 		ch := make(chan LogEntry, 1000)
-		p := NewParser(ch, nil, nil, time.Second, 256, cfg.cfg)
+		p := NewParser(ch, nil, nil, time.Second, 256, false, nil, cfg.cfg)
 
 		// Warm up
 		for _, line := range sampleLines {

json.go

@@ -0,0 +1,230 @@
+package logparser
+
+import (
+	"encoding/json"
+	"io"
+	"strings"
+	"unicode/utf8"
+)
+
+const (
+	jsonMaxAttrs    = 100
+	jsonMaxDepth    = 8
+	jsonMaxValueLen = 4096
+)
+
+var (
+	jsonMessageKeys = map[string]int{
+		"msg":     0,
+		"message": 1,
+		"@m":      2,
+		"@mt":     3,
+		"body":    4,
+		"log":     5,
+	}
+	jsonLevelKeys = map[string]int{
+		"level":         0,
+		"severity":      1,
+		"loglevel":      2,
+		"log.level":     3,
+		"lvl":           4,
+		"@l":            5,
+		"severity_text": 6,
+		"levelname":     7,
+	}
+	jsonTimestampKeys = map[string]bool{
+		"time": true, "timestamp": true, "ts": true, "t": true, "@t": true, "@timestamp": true, "datetime": true, "asctime": true,
+	}
+)
+
+type JsonLog struct {
+	Level      Level
+	Message    string
+	Attributes map[string]string
+}
+
+func ParseJsonLog(content string) *JsonLog {
+	fields := decodeJsonObject(content)
+	if fields == nil {
+		return nil
+	}
+	return jsonLogFromFields(fields)
+}
+
+// decodeJsonObject decodes a log line that holds a single, non-empty JSON
+// object, or returns nil. Surrounding whitespace, such as the \r of a CRLF
+// line ending, is ignored. Numbers are kept as json.Number.
+func decodeJsonObject(content string) map[string]any {
+	content = strings.TrimSpace(content)
+	if len(content) < 2 || content[0] != '{' || content[len(content)-1] != '}' {
+		return nil
+	}
+	d := json.NewDecoder(strings.NewReader(content))
+	d.UseNumber()
+	var fields map[string]any
+	if d.Decode(&fields) != nil || len(fields) == 0 {
+		return nil
+	}
+	if _, err := d.Token(); err != io.EOF {
+		return nil
+	}
+	return fields
+}
+
+// jsonLogFromFields extracts the message, level and attributes from a decoded
+// JSON log line.
+func jsonLogFromFields(fields map[string]any) *JsonLog {
+	res := &JsonLog{Attributes: map[string]string{}}
+	var msgKey, lvlKey string
+	var msgPrio, lvlPrio int
+	var msgVal string
+	var lvlRaw any
+	for k, v := range fields {
+		lk := strings.ToLower(k)
+		if p, ok := jsonMessageKeys[lk]; ok {
+			if s, ok := v.(string); ok {
+				if msgKey == "" || p < msgPrio || (p == msgPrio && k < msgKey) {
+					if msgKey != "" {
+						flattenJsonField(msgKey, msgVal, res.Attributes, jsonMaxDepth)
+					}
+					msgKey, msgPrio, msgVal = k, p, s
+				} else {
+					flattenJsonField(k, v, res.Attributes, jsonMaxDepth)
+				}
+				continue
+			}
+		} else if p, ok := jsonLevelKeys[lk]; ok {
+			if lvl := jsonLevelFromValue(v); lvl != LevelUnknown {
+				if lvlKey == "" || p < lvlPrio || (p == lvlPrio && k < lvlKey) {
+					if lvlKey != "" {
+						flattenJsonField(lvlKey, lvlRaw, res.Attributes, jsonMaxDepth)
+					}
+					lvlKey, lvlPrio, lvlRaw = k, p, v
+					res.Level = lvl
+				} else {
+					flattenJsonField(k, v, res.Attributes, jsonMaxDepth)
+				}
+				continue
+			}
+		} else if jsonTimestampKeys[lk] && jsonIsTimestampValue(v) {
+			continue
+		}
+		flattenJsonField(k, v, res.Attributes, jsonMaxDepth)
+	}
+	res.Message = strings.TrimSuffix(msgVal, "\n")
+	return res
+}
+
+func jsonIsTimestampValue(v any) bool {
+	switch value := v.(type) {
+	case string, json.Number:
+		return true
+	case map[string]any:
+		if len(value) != 1 {
+			return false
+		}
+		if _, ok := value["$date"]; ok { // mongodb
+			return true
+		}
+	}
+	return false
+}
+
+func jsonLevelFromValue(v any) Level {
+	switch value := v.(type) {
+	case string:
+		return jsonLevelFromString(value)
+	case json.Number:
+		return jsonLevelFromNumber(value)
+	}
+	return LevelUnknown
+}
+
+func flattenJsonField(key string, value any, attrs map[string]string, depth int) {
+	if len(attrs) >= jsonMaxAttrs {
+		return
+	}
+	switch v := value.(type) {
+	case nil:
+		attrs[key] = ""
+	case string:
+		attrs[key] = truncateUtf8(v, jsonMaxValueLen)
+	case bool:
+		if v {
+			attrs[key] = "true"
+		} else {
+			attrs[key] = "false"
+		}
+	case json.Number:
+		attrs[key] = v.String()
+	case map[string]any:
+		if depth <= 0 {
+			attrs[key] = jsonEncode(v)
+			return
+		}
+		for k, vv := range v {
+			flattenJsonField(key+"."+k, vv, attrs, depth-1)
+		}
+	default:
+		attrs[key] = jsonEncode(v)
+	}
+}
+
+func jsonEncode(v any) string {
+	data, err := json.Marshal(v)
+	if err != nil {
+		return ""
+	}
+	return truncateUtf8(string(data), jsonMaxValueLen)
+}
+
+func truncateUtf8(s string, limit int) string {
+	if len(s) <= limit {
+		return s
+	}
+	for limit > 0 && !utf8.RuneStart(s[limit]) {
+		limit--
+	}
+	return s[:limit]
+}
+
+func jsonLevelFromString(s string) Level {
+	if len(s) > 16 {
+		return LevelUnknown
+	}
+	switch strings.ToLower(s) {
+	case "trace", "trc", "debug", "dbg", "verbose":
+		return LevelDebug
+	case "info", "inf", "information", "informational", "notice":
+		return LevelInfo
+	case "warn", "wrn", "warning":
+		return LevelWarning
+	case "error", "err":
+		return LevelError
+	case "fatal", "ftl", "critical", "crit", "panic", "dpanic", "alert", "emerg", "emergency":
+		return LevelCritical
+	}
+	return LevelUnknown
+}
+
+func jsonLevelFromNumber(n json.Number) Level {
+	v, err := n.Int64()
+	if err != nil {
+		return LevelUnknown
+	}
+	switch {
+	case v < 10:
+		return LevelUnknown
+	case v < 30:
+		return LevelDebug
+	case v < 40:
+		return LevelInfo
+	case v < 50:
+		return LevelWarning
+	case v < 60:
+		return LevelError
+	case v < 100:
+		return LevelCritical
+	}
+	return LevelUnknown
+}

json_single_decode_test.go

@@ -0,0 +1,68 @@
+package logparser
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// Decoding a JSON line once (UseNumber) must give the same pattern content and
+// level as the json.Unmarshal path, or pattern hashes would change.
+func TestStructuredFromJSONMatchesUnmarshal(t *testing.T) {
+	lines := []string{
+		`{"level":"error","msg":"payment failed","order_id":"ord-1"}`,
+		`{"severity":"WARNING","message":"retry 3 of 5","attempt":3}`,
+		`{"lvl":"info","error":500,"reason":"upstream returned 1234567890123"}`,
+		`{"level":"debug","err":12.5e3}`,
+		`{"ts":1700000000.123,"code":1234567,"nested":{"a":1,"b":[1,2.5,"x"],"c":{"d":1e21}},"ok":true,"none":null}`,
+		`{"log_type":"error","data":[{"id":9007199254740993},{"id":-0.000001}]}`,
+		`  {"msg":"padded line","n":42}  `,
+	}
+	for _, line := range lines {
+		m1, l1, ok := parseJSONLog(line)
+		require.True(t, ok, line)
+		fields := decodeJsonObject(line)
+		require.NotNil(t, fields, line)
+		m2, l2 := structuredFromJSON(fields)
+		assert.Equal(t, m1, m2, line)
+		assert.Equal(t, l1, l2, line)
+	}
+}
+
+func TestJsonLogCRLF(t *testing.T) {
+	jl := ParseJsonLog("{\"level\":\"error\",\"msg\":\"payment failed\"}\r")
+	require.NotNil(t, jl)
+	assert.Equal(t, "payment failed", jl.Message)
+	assert.Equal(t, LevelError, jl.Level)
+
+	m := &MultilineCollector{}
+	assert.True(t, m.isNextMessage("{\"msg\":\"a\"}\r"))
+	assert.True(t, m.isNextMessage("{\"msg\":\"a\"} "))
+}
+
+// The same JSON line gets the same pattern hash whether or not JSON parsing
+// (message and attributes) is enabled.
+func TestPatternHashIndependentOfParseJson(t *testing.T) {
+	line := `{"level":"error","msg":"payment failed for order 42","order_id":"ord-1","duration_ms":138.5}`
+	hash := func(parseJson bool) (h string, attrs map[string]string) {
+		p := &Parser{
+			patterns:              map[patternKey]*patternStat{},
+			patternsPerLevel:      map[Level]int{},
+			patternsPerLevelLimit: 256,
+			parseJson:             parseJson,
+			onMsgCb: func(ts time.Time, level Level, patternHash string, msg string, a map[string]string) {
+				h, attrs = patternHash, a
+			},
+		}
+		p.inc(Message{Timestamp: time.Now(), Content: line, Level: LevelUnknown})
+		return
+	}
+	h1, a1 := hash(false)
+	h2, a2 := hash(true)
+	assert.NotEmpty(t, h1)
+	assert.Equal(t, h1, h2)
+	assert.Nil(t, a1)
+	assert.Equal(t, "ord-1", a2["order_id"])
+}