diff --git a/cmd/logparser.go b/cmd/logparser.go index 6a2e8f6..7e19941 100644 --- a/cmd/logparser.go +++ b/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') diff --git a/go.mod b/go.mod index db10712..b267d0b 100644 --- a/go.mod +++ b/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 ( diff --git a/go.sum b/go.sum index 45d6731..c956e7a 100644 --- a/go.sum +++ b/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= diff --git a/impact_test.go b/impact_test.go index 2d7565d..51432db 100644 --- a/impact_test.go +++ b/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 { diff --git a/json.go b/json.go new file mode 100644 index 0000000..1f21f5c --- /dev/null +++ b/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 +} diff --git a/json_single_decode_test.go b/json_single_decode_test.go new file mode 100644 index 0000000..7339191 --- /dev/null +++ b/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"]) +} diff --git a/json_test.go b/json_test.go new file mode 100644 index 0000000..b787ae7 --- /dev/null +++ b/json_test.go @@ -0,0 +1,204 @@ +package logparser + +import ( + "fmt" + "strings" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestParseJsonLogNotJson(t *testing.T) { + for _, s := range []string{ + "", + "plain text message", + "{incomplete json", + `{"a":1} trailing garbage`, + `{"a":1}{"b":2}`, + "[1, 2, 3]", + "{}", + "ts=2024-01-01 level=info msg=logfmt", + } { + assert.Nil(t, ParseJsonLog(s), s) + } +} + +func TestParseJsonLogSlog(t *testing.T) { + l := ParseJsonLog(`{"time":"2026-07-24T15:49:24.699496383Z","level":"WARN","msg":"retrying request","upstream":"payment-service","attempt":3,"error":"connection refused"}`) + require.NotNil(t, l) + assert.Equal(t, LevelWarning, l.Level) + assert.Equal(t, "retrying request", l.Message) + assert.Equal(t, map[string]string{ + "upstream": "payment-service", + "attempt": "3", + "error": "connection refused", + }, l.Attributes) +} + +func TestParseJsonLogZap(t *testing.T) { + l := ParseJsonLog(`{"level":"error","ts":1784908164.831705,"caller":"app/main.go:159","msg":"request to upstream failed","upstream":"inventory-service","error":"context deadline exceeded"}`) + require.NotNil(t, l) + assert.Equal(t, LevelError, l.Level) + assert.Equal(t, "request to upstream failed", l.Message) + assert.Equal(t, map[string]string{ + "caller": "app/main.go:159", + "upstream": "inventory-service", + "error": "context deadline exceeded", + }, l.Attributes) +} + +func TestParseJsonLogLogrus(t *testing.T) { + l := ParseJsonLog(`{"amount":181.72,"currency":"USD","level":"info","msg":"payment processed","order_id":"ord-753339","time":"2026-07-24T15:49:24Z"}`) + require.NotNil(t, l) + assert.Equal(t, LevelInfo, l.Level) + assert.Equal(t, "payment processed", l.Message) + assert.Equal(t, map[string]string{ + "amount": "181.72", + "currency": "USD", + "order_id": "ord-753339", + }, l.Attributes) +} + +func TestParseJsonLogDotnet(t *testing.T) { + l := ParseJsonLog(`{"Timestamp":"2026-07-24T15:49:25.1236791Z","EventId":0,"LogLevel":"Information","Category":"Demo.Worker","Message":"request completed","State":{"Message":"request completed","status":200,"{OriginalFormat}":"request completed"}}`) + require.NotNil(t, l) + assert.Equal(t, LevelInfo, l.Level) + assert.Equal(t, "request completed", l.Message) + assert.Equal(t, map[string]string{ + "EventId": "0", + "Category": "Demo.Worker", + "State.Message": "request completed", + "State.status": "200", + "State.{OriginalFormat}": "request completed", + }, l.Attributes) +} + +func TestParseJsonLogSerilogClef(t *testing.T) { + l := ParseJsonLog(`{"@t":"2026-07-24T15:49:25.2750856Z","@mt":"slow query","@l":"Warning","duration_ms":2026.79,"rows":7552}`) + require.NotNil(t, l) + assert.Equal(t, LevelWarning, l.Level) + assert.Equal(t, "slow query", l.Message) + assert.Equal(t, map[string]string{ + "duration_ms": "2026.79", + "rows": "7552", + }, l.Attributes) + + l = ParseJsonLog(`{"@t":"2026-07-24T15:49:25.2634236Z","@mt":"request completed","status":200}`) + require.NotNil(t, l) + assert.Equal(t, LevelUnknown, l.Level) + assert.Equal(t, "request completed", l.Message) +} + +func TestParseJsonLogPino(t *testing.T) { + l := ParseJsonLog(`{"level":30,"time":1784908164831,"pid":1,"hostname":"review-service-775df668b6-abcde","msg":"request completed","responseTime":12}`) + require.NotNil(t, l) + assert.Equal(t, LevelInfo, l.Level) + assert.Equal(t, "request completed", l.Message) + assert.Equal(t, "12", l.Attributes["responseTime"]) + + l = ParseJsonLog(`{"level":50,"time":1784908164831,"msg":"unhandled error"}`) + require.NotNil(t, l) + assert.Equal(t, LevelError, l.Level) +} + +func TestParseJsonLogKeyPriority(t *testing.T) { + l := ParseJsonLog(`{"log":"b","msg":"a","message":"c","level":"warn","severity":"error"}`) + require.NotNil(t, l) + assert.Equal(t, "a", l.Message) + assert.Equal(t, LevelWarning, l.Level) + assert.Equal(t, map[string]string{ + "log": "b", + "message": "c", + "severity": "error", + }, l.Attributes) + + l = ParseJsonLog(`{"msg":42,"message":"m"}`) + require.NotNil(t, l) + assert.Equal(t, "m", l.Message) + assert.Equal(t, "42", l.Attributes["msg"]) + + l = ParseJsonLog(`{"msg":"m","level":"weird","severity":"error"}`) + require.NotNil(t, l) + assert.Equal(t, LevelError, l.Level) + assert.Equal(t, "weird", l.Attributes["level"]) +} + +func TestParseJsonLogNestedAndArrays(t *testing.T) { + l := ParseJsonLog(`{"msg":"m","t":{"$date":"2026-09-17T10:15:22.123+00:00"},"http":{"method":"GET","response":{"status":200}},"tags":["a","b"],"ok":true,"ref":null}`) + require.NotNil(t, l) + assert.Equal(t, map[string]string{ // the mongodb-style "t" timestamp object is dropped + "http.method": "GET", + "http.response.status": "200", + "tags": `["a","b"]`, + "ok": "true", + "ref": "", + }, l.Attributes) +} + +func TestParseJsonLogAttrLimits(t *testing.T) { + sb := strings.Builder{} + sb.WriteString(`{"msg":"m"`) + for i := 0; i < 500; i++ { + fmt.Fprintf(&sb, `,"key%d":"v"`, i) + } + sb.WriteString("}") + l := ParseJsonLog(sb.String()) + require.NotNil(t, l) + assert.LessOrEqual(t, len(l.Attributes), jsonMaxAttrs) + + l = ParseJsonLog(`{"msg":"m","big":"` + strings.Repeat("a", 10000) + `"}`) + require.NotNil(t, l) + assert.Equal(t, jsonMaxValueLen, len(l.Attributes["big"])) +} + +func TestParseJsonLogLevels(t *testing.T) { + assert.Equal(t, LevelDebug, jsonLevelFromString("TRACE")) + assert.Equal(t, LevelDebug, jsonLevelFromString("Verbose")) + assert.Equal(t, LevelInfo, jsonLevelFromString("Information")) + assert.Equal(t, LevelInfo, jsonLevelFromString("notice")) + assert.Equal(t, LevelWarning, jsonLevelFromString("WARN")) + assert.Equal(t, LevelError, jsonLevelFromString("Error")) + assert.Equal(t, LevelCritical, jsonLevelFromString("panic")) + assert.Equal(t, LevelCritical, jsonLevelFromString("FATAL")) + assert.Equal(t, LevelUnknown, jsonLevelFromString("something else")) +} + +var ( + benchSlogLine = `{"time":"2026-07-24T15:49:24.699496383Z","level":"WARN","msg":"retrying request","upstream":"payment-service","attempt":3,"error":"dial tcp 10.96.14.21:8080: connect: connection refused"}` + benchPlainLine = `2026-07-24 15:49:24.699 WARN [main] retrying request to payment-service, attempt 3: connection refused` + benchBracedLine = "{this line starts with a brace but is not json, like a C++ initializer dump or a truncated record" + strings.Repeat(" x", 30) + "}" + benchNestedLine = `{"time":"2026-07-24T15:49:24.699496383Z","level":"info","msg":"request completed","http":{"request":{"method":"POST","path":"/api/checkout","headers":{"user-agent":"Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36","x-request-id":"e2b1a8c0-6f4e-4b6a-9a3e-2c8d1f0b7a55"}},"response":{"status":200,"duration_ms":138.04,"bytes":2048}},"db":{"queries":12,"duration_ms":41.7},"user":{"id":"user-1453","tenant":"acme","roles":["admin","billing"]},"cache":{"hits":9,"misses":1}}` +) + +func BenchmarkParseJsonLogSlog(b *testing.B) { + for i := 0; i < b.N; i++ { + if ParseJsonLog(benchSlogLine) == nil { + b.Fatal("expected json") + } + } +} + +func BenchmarkParseJsonLogNested(b *testing.B) { + for i := 0; i < b.N; i++ { + if ParseJsonLog(benchNestedLine) == nil { + b.Fatal("expected json") + } + } +} + +func BenchmarkParseJsonLogPlainText(b *testing.B) { + for i := 0; i < b.N; i++ { + if ParseJsonLog(benchPlainLine) != nil { + b.Fatal("expected non-json") + } + } +} + +func BenchmarkParseJsonLogBracedNonJson(b *testing.B) { + for i := 0; i < b.N; i++ { + if ParseJsonLog(benchBracedLine) != nil { + b.Fatal("expected non-json") + } + } +} diff --git a/level.go b/level.go index 5f65bdd..b206d24 100644 --- a/level.go +++ b/level.go @@ -103,7 +103,7 @@ func GuessLevel(line string) Level { switch sf[:4] { case "debu": return LevelDebug - case "info", "noti": + case "info", "noti", "note", "syst": return LevelInfo case "warn": return LevelWarning diff --git a/level_test.go b/level_test.go index 305bbc0..4bf9487 100644 --- a/level_test.go +++ b/level_test.go @@ -70,6 +70,19 @@ func TestParseLevelValue(t *testing.T) { assert.Equal(t, LevelError, parseLevelValue(" error ")) } +func TestGuessLevelMySQL(t *testing.T) { + // MySQL 8.x + assert.Equal(t, LevelInfo, GuessLevel(`2026-09-30T18:48:49.795807Z 8 [Note] [MY-010926] [Server] Access denied for user 'root'@'localhost' (using password: YES)`)) + assert.Equal(t, LevelInfo, GuessLevel(`2026-09-30T18:33:14.028354Z 0 [System] [MY-010931] [Server] /usr/sbin/mysqld: ready for connections. Version: '8.0.46' socket: '/var/lib/mysql/mysql.sock' port: 3306 MySQL Community Server - GPL.`)) + assert.Equal(t, LevelWarning, GuessLevel(`2026-09-30T18:31:33.217632Z 0 [Warning] [MY-010068] [Server] CA certificate ca.pem is self signed.`)) + assert.Equal(t, LevelError, GuessLevel(`2026-09-30T18:31:33.217632Z 0 [ERROR] [MY-010119] [Server] Aborting`)) + // MySQL 5.7 + assert.Equal(t, LevelInfo, GuessLevel(`2019-05-14T12:32:17.123456Z 0 [Note] InnoDB: Buffer pool(s) load completed at 190514 12:32:17`)) + // MariaDB + assert.Equal(t, LevelInfo, GuessLevel(`2024-01-01 12:00:00 0 [Note] InnoDB: 10.11.6 started; log sequence number 47776; transaction id 14`)) + assert.Equal(t, LevelWarning, GuessLevel(`2024-01-01 12:00:00 5 [Warning] Access denied for user 'root'@'localhost' (using password: NO)`)) +} + 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`)) diff --git a/multiline.go b/multiline.go index e77a3d9..853e126 100644 --- a/multiline.go +++ b/multiline.go @@ -98,9 +98,9 @@ func (m *MultilineCollector) add(entry LogEntry) { } if len(m.lines) == 0 { m.ts = entry.Timestamp - m.level = GuessLevel(entry.Content) - if m.level == LevelUnknown && entry.Level != LevelUnknown { - m.level = entry.Level + m.level = entry.Level // the level reported by the source is trusted, otherwise it is guessed from the content + if m.level == LevelUnknown { + m.level = GuessLevel(entry.Content) } m.isFirstLineContainsTimestamp = containsTimestamp(entry.Content) } @@ -124,6 +124,10 @@ func (m *MultilineCollector) isNextMessage(l string) bool { return false } + if strings.HasPrefix(l, `{"`) && strings.HasSuffix(strings.TrimRight(l, " \t\r"), "}") { + return true + } + if m.isFirstLineContainsTimestamp { return containsTimestamp(l) } diff --git a/multiline_test.go b/multiline_test.go index 2e8803c..5b47106 100644 --- a/multiline_test.go +++ b/multiline_test.go @@ -47,6 +47,20 @@ Order response: {"timestamp":1648205755430,"status":406,"error":"Not Acceptable" assert.Equal(t, strings.Split(data, "\n")[1], msgs[1].Content) } +func TestMultilineCollectorJsonLines(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + m := NewMultilineCollector(ctx, 10*time.Millisecond, multilineCollectorLimit) + defer cancel() + data := `{"hit":false,"key":"cart:user-1664","level":"debug","msg":"cache lookup","time":"2026-07-27T13:46:33Z"} +{"error":"upstream returned status 503","level":"error","msg":"request to upstream failed","time":"2026-07-27T13:46:34Z","upstream":"product-catalog"} +{"currency":"USD","items":8,"level":"info","msg":"order created","order_id":"ord-003007","time":"2026-07-27T13:46:34Z","total":413.98,"user_id":"user-4735"}` + msgs := writeByLine(m, data, time.Unix(0, 0)) + require.Len(t, msgs, 3) + for i, line := range strings.Split(data, "\n") { + assert.Equal(t, line, msgs[i].Content) + } +} + func TestMultilineCollectorPython(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) m := NewMultilineCollector(ctx, 10*time.Millisecond, multilineCollectorLimit) @@ -408,3 +422,17 @@ func TestMultilineCollectorLimit(t *testing.T) { assert.Equal(t, 97, len(msgs[0].Content)) assert.True(t, utf8.ValidString(msgs[0].Content)) } + +func TestMultilineCollectorReportedLevel(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + m := NewMultilineCollector(ctx, 10*time.Millisecond, multilineCollectorLimit) + defer cancel() + + m.Add(LogEntry{Timestamp: time.Unix(0, 0), Content: "Memory: avail=27.3G, total=31.1G, warnings=1.0G/500M/100M", Level: LevelInfo}) + msg := <-m.Messages + assert.Equal(t, LevelInfo, msg.Level) // the reported level wins over the guess ("warnings") + + m.Add(LogEntry{Timestamp: time.Unix(1, 0), Content: "ERROR: relation does not exist"}) + msg = <-m.Messages + assert.Equal(t, LevelError, msg.Level) // guessed when no level is reported +} diff --git a/parser.go b/parser.go index 1a78503..44b4556 100644 --- a/parser.go +++ b/parser.go @@ -9,6 +9,8 @@ import ( "strings" "sync" "time" + + "golang.org/x/time/rate" ) //go:embed sensitive_patterns.json @@ -17,6 +19,9 @@ var sensitivePatternsJSON []byte var ( unclassifiedPatternLabel = "unclassified pattern (pattern limit reached)" unclassifiedPatternHash = "00000000000000000000000000000000" + + sampledPatternLabel = "event was sampled" + sampledPatternHash = "11111111111111111111111111111111" ) // Shared pattern caches: compiled once, shared across all parsers. @@ -82,6 +87,10 @@ type Parser struct { multilineCollector *MultilineCollector + parseJson bool + + limiter *rate.Limiter + stop func() onMsgCb OnMsgCallbackF @@ -92,14 +101,16 @@ type Parser struct { sensitiveCounter uint64 } -type OnMsgCallbackF func(ts time.Time, level Level, patternHash string, msg string) +type OnMsgCallbackF func(ts time.Time, level Level, patternHash string, msg string, attributes map[string]string) -func NewParser(ch <-chan LogEntry, decoder Decoder, onMsgCallback OnMsgCallbackF, multilineCollectorTimeout time.Duration, patternsPerLevelLimit int, sensitiveCfg SensitiveConfig) *Parser { +func NewParser(ch <-chan LogEntry, decoder Decoder, onMsgCallback OnMsgCallbackF, multilineCollectorTimeout time.Duration, patternsPerLevelLimit int, parseJson bool, limiter *rate.Limiter, sensitiveCfg SensitiveConfig) *Parser { p := &Parser{ decoder: decoder, patterns: map[patternKey]*patternStat{}, patternsPerLevel: map[Level]int{}, patternsPerLevelLimit: patternsPerLevelLimit, + parseJson: parseJson, + limiter: limiter, onMsgCb: onMsgCallback, sensitivePatterns: map[sensitivePatternKey]*sensitivePatternStat{}, sensitiveConfig: sensitiveCfg, @@ -155,30 +166,76 @@ func (p *Parser) inc(msg Message) { // 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) + // unreliable GuessLevel text-scan for structured logs. With parseJson, a + // JSON line is decoded once, for the pattern content and for the emitted + // message and attributes alike. + var normalizedContent string + var structuredLevel Level + var jl *JsonLog + if p.parseJson { + if fields := decodeJsonObject(msg.Content); fields != nil { + normalizedContent, structuredLevel = structuredFromJSON(fields) + jl = jsonLogFromFields(fields) + } + } + if jl == nil { + 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: ""} + // A JSON log is emitted as its message, with the other fields as + // attributes. Patterns still come from normalizedContent. + content := msg.Content + var attributes map[string]string + if jl != nil { + if jl.Level != LevelUnknown { + msg.Level = jl.Level + } + if jl.Message != "" { + content = jl.Message + } + attributes = jl.Attributes + } + level := msg.Level + + if level == LevelUnknown || level == LevelDebug || level == LevelInfo { + key := patternKey{level: level, hash: ""} if stat := p.patterns[key]; stat == nil { p.patterns[key] = &patternStat{} } p.patterns[key].messages++ if p.onMsgCb != nil { - p.onMsgCb(msg.Timestamp, msg.Level, "", msg.Content) + p.onMsgCb(msg.Timestamp, level, "", content, attributes) } pattern := NewPatternFromNormalized(normalizedContent) p.processSensitivePattern(msg, pattern) return } + // Under a log storm, skip pattern extraction and sensitive-data detection + // (the CPU-heavy part), counting the message under a dedicated pattern. + // Unlike upstream, the message is still emitted: the limit caps CPU, it + // must not drop log records. + if p.limiter != nil && !p.limiter.Allow() { + key := patternKey{level: level, hash: sampledPatternHash} + stat := p.patterns[key] + if stat == nil { + stat = &patternStat{sample: sampledPatternLabel} + p.patterns[key] = stat + } + stat.messages++ + if p.onMsgCb != nil { + p.onMsgCb(msg.Timestamp, level, sampledPatternHash, content, attributes) + } + return + } + pattern := NewPatternFromNormalized(normalizedContent) - stat, key := p.getPatternStat(msg.Level, pattern, msg.Content) + stat, key := p.getPatternStat(level, pattern, content) if p.onMsgCb != nil { - p.onMsgCb(msg.Timestamp, msg.Level, key.hash, msg.Content) + p.onMsgCb(msg.Timestamp, level, key.hash, content, attributes) } stat.messages++ p.processSensitivePattern(msg, pattern) diff --git a/parser_test.go b/parser_test.go index 5d5eac6..a5bb6ae 100644 --- a/parser_test.go +++ b/parser_test.go @@ -2,16 +2,18 @@ package logparser import ( "sort" + "strings" "testing" "time" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" + "golang.org/x/time/rate" ) func TestParser(t *testing.T) { ch := make(chan LogEntry) - parser := NewParser(ch, nil, nil, time.Second, 256, SensitiveConfig{ + parser := NewParser(ch, nil, nil, time.Second, 256, false, nil, SensitiveConfig{ Enabled: true, MinConfidence: "high", }) @@ -27,7 +29,7 @@ func TestParser(t *testing.T) { func TestParserSensitiveDisabled(t *testing.T) { ch := make(chan LogEntry) - parser := NewParser(ch, nil, nil, time.Second, 256, SensitiveConfig{ + parser := NewParser(ch, nil, nil, time.Second, 256, false, nil, SensitiveConfig{ Enabled: false, }) @@ -41,7 +43,7 @@ func TestParserSensitiveDisabled(t *testing.T) { func TestParserSensitiveSampling(t *testing.T) { ch := make(chan LogEntry) // Sample 1 in 10 lines - parser := NewParser(ch, nil, nil, time.Second, 256, SensitiveConfig{ + parser := NewParser(ch, nil, nil, time.Second, 256, false, nil, SensitiveConfig{ Enabled: true, SampleRate: 10, MinConfidence: "high", @@ -67,7 +69,7 @@ func TestParserSensitiveSampling(t *testing.T) { func TestParserSensitiveMaxDetections(t *testing.T) { ch := make(chan LogEntry) - parser := NewParser(ch, nil, nil, time.Second, 256, SensitiveConfig{ + parser := NewParser(ch, nil, nil, time.Second, 256, false, nil, SensitiveConfig{ Enabled: true, MinConfidence: "high", MaxDetections: 1, @@ -144,6 +146,40 @@ func TestParserJSONLevelOverride(t *testing.T) { assert.NotEmpty(t, errorCounters[0].Hash, "ERROR logs should have a pattern hash") } +func TestParserJson(t *testing.T) { + line := `{"level":"error","msg":"payment failed","order_id":"ord-1"}` + + var gotLevel Level + var gotMsg string + var gotAttrs map[string]string + cb := func(ts time.Time, level Level, patternHash string, msg string, attributes map[string]string) { + gotLevel, gotMsg, gotAttrs = level, msg, attributes + } + + p := &Parser{ + patterns: map[patternKey]*patternStat{}, + patternsPerLevel: map[Level]int{}, + patternsPerLevelLimit: 10, + parseJson: true, + onMsgCb: cb, + } + p.inc(Message{Timestamp: time.Now(), Content: line, Level: LevelUnknown}) + assert.Equal(t, LevelError, gotLevel) + assert.Equal(t, "payment failed", gotMsg) + assert.Equal(t, map[string]string{"order_id": "ord-1"}, gotAttrs) + + p = &Parser{ + patterns: map[patternKey]*patternStat{}, + patternsPerLevel: map[Level]int{}, + patternsPerLevelLimit: 10, + onMsgCb: cb, + } + p.inc(Message{Timestamp: time.Now(), Content: line, Level: LevelError}) + assert.Equal(t, LevelError, gotLevel) + assert.Equal(t, line, gotMsg) + assert.Nil(t, gotAttrs) +} + func TestParserCardinalityLimit(t *testing.T) { p := &Parser{ patterns: map[patternKey]*patternStat{}, @@ -178,3 +214,58 @@ func TestParserCardinalityLimit(t *testing.T) { assert.Equal(t, unclassifiedPatternLabel, counters[2].Sample) assert.Equal(t, unclassifiedPatternHash, counters[2].Hash) } + +func TestParserRateLimit(t *testing.T) { + calls, sampled := 0, 0 + p := &Parser{ + patterns: map[patternKey]*patternStat{}, + patternsPerLevel: map[Level]int{}, + patternsPerLevelLimit: 256, + limiter: rate.NewLimiter(0, 3), // never refills + onMsgCb: func(ts time.Time, level Level, patternHash string, msg string, attributes map[string]string) { + calls++ + if patternHash == sampledPatternHash { + sampled++ + } + }, + } + + for i := 0; i < 10; i++ { + p.inc(Message{Timestamp: time.Now(), Content: "error" + strings.Repeat(" word", i+1), Level: LevelError}) + } + + assert.Equal(t, 3, p.patternsPerLevel[LevelError]) + // Over the limit, pattern extraction is skipped but every message is + // still emitted, under the sampled pattern hash. + assert.Equal(t, 10, calls) + assert.Equal(t, 7, sampled) + + stat, ok := p.patterns[patternKey{level: LevelError, hash: sampledPatternHash}] + require.True(t, ok) + assert.Equal(t, 7, stat.messages) + assert.Equal(t, sampledPatternLabel, stat.sample) + + total := 0 + for _, c := range p.GetCounters() { + assert.Equal(t, LevelError, c.Level) + total += c.Messages + } + assert.Equal(t, 10, total) +} + +func TestParserRateLimitLevels(t *testing.T) { + p := &Parser{ + patterns: map[patternKey]*patternStat{}, + patternsPerLevel: map[Level]int{}, + patternsPerLevelLimit: 256, + limiter: rate.NewLimiter(0, 0), // always over the limit + } + + p.inc(Message{Timestamp: time.Now(), Content: "some info message", Level: LevelInfo}) + p.inc(Message{Timestamp: time.Now(), Content: "some debug message", Level: LevelDebug}) + + assert.Equal(t, 1, p.patterns[patternKey{level: LevelInfo, hash: ""}].messages) + assert.Equal(t, 1, p.patterns[patternKey{level: LevelDebug, hash: ""}].messages) + _, ok := p.patterns[patternKey{level: LevelError, hash: sampledPatternHash}] + assert.False(t, ok) +} diff --git a/pattern.go b/pattern.go index 49b44d8..0090170 100644 --- a/pattern.go +++ b/pattern.go @@ -7,6 +7,7 @@ import ( "fmt" "regexp" "sort" + "strconv" "strings" "sync" ) @@ -224,16 +225,16 @@ func removeQuotedAndBrackets(s string, buf *bytes.Buffer) string { return buf.String() } -// jsonMessageKeys lists the JSON field names (lowercase) used for pattern extraction. +// patternMessageKeys lists the JSON field names (lowercase) used for pattern extraction. // Following industry standard (Datadog, New Relic, Elastic, Better Stack), pattern // hashing uses only the message/error content, not metadata fields like timestamps, // file paths, line numbers, IDs, or data blobs which produce unstable hashes. -var jsonMessageKeys = []string{"msg", "message", "error", "err", "reason", "log", "text"} +var patternMessageKeys = []string{"msg", "message", "error", "err", "reason", "log", "text"} -// jsonLevelKeys lists the JSON field names (lowercase) checked for log level. +// patternLevelKeys lists the JSON field names (lowercase) checked for log level. // Covers: slog/zerolog/zap (level), GCP/Stackdriver (severity), Bunyan (lvl), // Python logging (levelname), and common variants. -var jsonLevelKeys = []string{"level", "severity", "lvl", "log.level", "loglevel", "log_level", "levelname", "log_type"} +var patternLevelKeys = []string{"level", "severity", "lvl", "log.level", "loglevel", "log_level", "levelname", "log_type"} // maxFallbackFieldLen caps individual field values in the fallback path to prevent // large data blobs (HTML, XML, stack traces) from overwhelming the pattern. @@ -246,7 +247,15 @@ func parseJSONLog(line string) (message string, level Level, ok bool) { if err := json.Unmarshal([]byte(line), &m); err != nil { return line, LevelUnknown, false } + message, level = structuredFromJSON(m) + return message, level, true +} +// structuredFromJSON extracts the pattern content and the level from a +// decoded JSON log line. Numbers may be float64 (json.Unmarshal) or +// json.Number (decoding with UseNumber): both render the same, so pattern +// hashes do not depend on how the line was decoded. +func structuredFromJSON(m map[string]interface{}) (message string, level Level) { // Build a lowercase-key lookup for case-insensitive matching. lowerMap := make(map[string]interface{}, len(m)) for k, v := range m { @@ -255,7 +264,7 @@ func parseJSONLog(line string) (message string, level Level, ok bool) { // Extract level from structured field. level = LevelUnknown - for _, k := range jsonLevelKeys { + for _, k := range patternLevelKeys { if v, found := lowerMap[k]; found { if s, isStr := v.(string); isStr { level = parseLevelValue(s) @@ -266,9 +275,9 @@ func parseJSONLog(line string) (message string, level Level, ok bool) { // Extract only message-relevant fields for stable pattern hashing. var buf strings.Builder - for _, k := range jsonMessageKeys { + for _, k := range patternMessageKeys { if v, found := lowerMap[k]; found { - s := fmt.Sprintf("%v", v) + s := fmt.Sprintf("%v", float64Numbers(v)) if s != "" { buf.WriteString(s) buf.WriteByte(' ') @@ -276,7 +285,7 @@ func parseJSONLog(line string) (message string, level Level, ok bool) { } } if buf.Len() > 0 { - return strings.TrimSpace(buf.String()), level, true + return strings.TrimSpace(buf.String()), level } // Fallback: no known message fields found, use all values sorted by key @@ -287,14 +296,39 @@ func parseJSONLog(line string) (message string, level Level, ok bool) { } sort.Strings(keys) for _, k := range keys { - s := fmt.Sprintf("%v", m[k]) + s := fmt.Sprintf("%v", float64Numbers(m[k])) if len(s) > maxFallbackFieldLen { s = s[:maxFallbackFieldLen] } buf.WriteString(s) buf.WriteByte(' ') } - return strings.TrimSpace(buf.String()), level, true + return strings.TrimSpace(buf.String()), level +} + +// float64Numbers converts json.Number values, including nested ones, to the +// float64 that json.Unmarshal produces, so both decodings render alike. +func float64Numbers(v interface{}) interface{} { + switch t := v.(type) { + case json.Number: + if f, err := strconv.ParseFloat(string(t), 64); err == nil { + return f + } + return string(t) + case map[string]interface{}: + res := make(map[string]interface{}, len(t)) + for k, vv := range t { + res[k] = float64Numbers(vv) + } + return res + case []interface{}: + res := make([]interface{}, len(t)) + for i, vv := range t { + res[i] = float64Numbers(vv) + } + return res + } + return v } // ParseStructuredLog attempts to parse a log line as a structured format