From d66d1856355377cf0e58c5b2e90ab3fecf0714d7 Mon Sep 17 00:00:00 2001 From: Nikolay Sivko Date: Fri, 24 Jul 2026 17:10:06 -0300 Subject: [PATCH 1/7] json parser --- cmd/logparser.go | 2 +- json.go | 201 ++++++++++++++++++++++++++++++++++++++++++++++ json_test.go | 204 +++++++++++++++++++++++++++++++++++++++++++++++ parser.go | 39 +++++++-- parser_test.go | 34 ++++++++ 5 files changed, 471 insertions(+), 9 deletions(-) create mode 100644 json.go create mode 100644 json_test.go diff --git a/cmd/logparser.go b/cmd/logparser.go index d8ef744..2ba4dec 100644 --- a/cmd/logparser.go +++ b/cmd/logparser.go @@ -23,7 +23,7 @@ func main() { reader := bufio.NewReader(os.Stdin) ch := make(chan logparser.LogEntry) - parser := logparser.NewParser(ch, nil, nil, time.Second, 256) + parser := logparser.NewParser(ch, nil, nil, time.Second, 256, true) t := time.Now() for { line, err := reader.ReadString('\n') diff --git a/json.go b/json.go new file mode 100644 index 0000000..c45de6a --- /dev/null +++ b/json.go @@ -0,0 +1,201 @@ +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, "@timestamp": true, "datetime": true, "asctime": true, + } +) + +type JsonLog struct { + Level Level + Message string + Attributes map[string]string +} + +func ParseJsonLog(content string) *JsonLog { + 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 + } + + 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] { + switch v.(type) { + case string, json.Number: + continue + } + } + flattenJsonField(k, v, res.Attributes, jsonMaxDepth) + } + res.Message = strings.TrimSuffix(msgVal, "\n") + return res +} + +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_test.go b/json_test.go new file mode 100644 index 0000000..e673d01 --- /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","http":{"method":"GET","response":{"status":200}},"tags":["a","b"],"ok":true,"ref":null}`) + require.NotNil(t, l) + assert.Equal(t, map[string]string{ + "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/parser.go b/parser.go index 4dad7c5..d7f7418 100644 --- a/parser.go +++ b/parser.go @@ -34,19 +34,22 @@ type Parser struct { multilineCollector *MultilineCollector + parseJson bool + stop func() onMsgCb OnMsgCallbackF } -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) *Parser { +func NewParser(ch <-chan LogEntry, decoder Decoder, onMsgCallback OnMsgCallbackF, multilineCollectorTimeout time.Duration, patternsPerLevelLimit int, parseJson bool) *Parser { p := &Parser{ decoder: decoder, patterns: map[patternKey]*patternStat{}, patternsPerLevel: map[Level]int{}, patternsPerLevelLimit: patternsPerLevelLimit, + parseJson: parseJson, onMsgCb: onMsgCallback, } ctx, stop := context.WithCancel(context.Background()) @@ -92,26 +95,46 @@ func (p *Parser) inc(msg Message) { p.lock.Lock() defer p.lock.Unlock() - if msg.Level == LevelUnknown || msg.Level == LevelDebug || msg.Level == LevelInfo { - key := patternKey{level: msg.Level, hash: ""} + content := msg.Content + level := msg.Level + var attributes map[string]string + if jl := p.parseJsonLog(content); jl != nil { + if jl.Level != LevelUnknown { + level = jl.Level + } + if jl.Message != "" { + content = jl.Message + } + attributes = jl.Attributes + } + + 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) } return } - pattern := NewPattern(msg.Content) - stat, key := p.getPatternStat(msg.Level, pattern, msg.Content) + pattern := NewPattern(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++ } +func (p *Parser) parseJsonLog(content string) *JsonLog { + if !p.parseJson { + return nil + } + return ParseJsonLog(content) +} + func (p *Parser) getPatternStat(level Level, pattern *Pattern, sample string) (*patternStat, patternKey) { key := patternKey{level: level, hash: pattern.Hash()} if stat := p.patterns[key]; stat != nil { diff --git a/parser_test.go b/parser_test.go index 41261c6..6bb129b 100644 --- a/parser_test.go +++ b/parser_test.go @@ -9,6 +9,40 @@ import ( "github.com/stretchr/testify/require" ) +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{}, From 3be0d826161ae8c344a6f2d3b0d5b613af4da20b Mon Sep 17 00:00:00 2001 From: Nikolay Sivko Date: Mon, 27 Jul 2026 10:59:15 -0300 Subject: [PATCH 2/7] fix multiline parser for json logs --- multiline.go | 4 ++++ multiline_test.go | 14 ++++++++++++++ 2 files changed, 18 insertions(+) diff --git a/multiline.go b/multiline.go index e77a3d9..4f3e58f 100644 --- a/multiline.go +++ b/multiline.go @@ -124,6 +124,10 @@ func (m *MultilineCollector) isNextMessage(l string) bool { return false } + if strings.HasPrefix(l, `{"`) && strings.HasSuffix(l, "}") { + return true + } + if m.isFirstLineContainsTimestamp { return containsTimestamp(l) } diff --git a/multiline_test.go b/multiline_test.go index 2e8803c..836a9f1 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) From dff1e4b3736a9435bfda56a3c5435bb19149028f Mon Sep 17 00:00:00 2001 From: Nikolay Sivko Date: Mon, 10 Aug 2026 14:06:17 -0300 Subject: [PATCH 3/7] rate limit pattern extraction to cap CPU usage under log storms --- cmd/logparser.go | 2 +- go.mod | 5 ++++- go.sum | 2 ++ parser.go | 21 +++++++++++++++++++- parser_test.go | 51 ++++++++++++++++++++++++++++++++++++++++++++++++ 5 files changed, 78 insertions(+), 3 deletions(-) diff --git a/cmd/logparser.go b/cmd/logparser.go index 2ba4dec..5875e16 100644 --- a/cmd/logparser.go +++ b/cmd/logparser.go @@ -23,7 +23,7 @@ func main() { reader := bufio.NewReader(os.Stdin) ch := make(chan logparser.LogEntry) - parser := logparser.NewParser(ch, nil, nil, time.Second, 256, true) + parser := logparser.NewParser(ch, nil, nil, time.Second, 256, true, nil) t := time.Now() for { line, err := reader.ReadString('\n') diff --git a/go.mod b/go.mod index 8378898..b27cc7d 100644 --- a/go.mod +++ b/go.mod @@ -2,7 +2,10 @@ module github.com/coroot/logparser go 1.24 -require github.com/stretchr/testify v1.8.4 +require ( + github.com/stretchr/testify v1.8.4 + golang.org/x/time v0.12.0 +) require ( github.com/davecgh/go-spew v1.1.1 // indirect diff --git a/go.sum b/go.sum index fa4b6e6..e3408e1 100644 --- a/go.sum +++ b/go.sum @@ -4,6 +4,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.8.4 h1:CcVxjf3Q8PM0mHUKJCdn+eZZtm5yQwehR5yeSVQQcUk= github.com/stretchr/testify v1.8.4/go.mod h1:sz/lmYIOXD/1dqDmKjjqLyZ2RngseejIcXlSw2iwfAo= +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/parser.go b/parser.go index d7f7418..169f240 100644 --- a/parser.go +++ b/parser.go @@ -4,11 +4,16 @@ import ( "context" "sync" "time" + + "golang.org/x/time/rate" ) var ( unclassifiedPatternLabel = "unclassified pattern (pattern limit reached)" unclassifiedPatternHash = "00000000000000000000000000000000" + + sampledPatternLabel = "event was sampled" + sampledPatternHash = "11111111111111111111111111111111" ) type LogEntry struct { @@ -36,6 +41,8 @@ type Parser struct { parseJson bool + limiter *rate.Limiter + stop func() onMsgCb OnMsgCallbackF @@ -43,13 +50,14 @@ type Parser struct { 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, parseJson bool) *Parser { +func NewParser(ch <-chan LogEntry, decoder Decoder, onMsgCallback OnMsgCallbackF, multilineCollectorTimeout time.Duration, patternsPerLevelLimit int, parseJson bool, limiter *rate.Limiter) *Parser { p := &Parser{ decoder: decoder, patterns: map[patternKey]*patternStat{}, patternsPerLevel: map[Level]int{}, patternsPerLevelLimit: patternsPerLevelLimit, parseJson: parseJson, + limiter: limiter, onMsgCb: onMsgCallback, } ctx, stop := context.WithCancel(context.Background()) @@ -120,6 +128,17 @@ func (p *Parser) inc(msg Message) { return } + 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++ + return + } + pattern := NewPattern(content) stat, key := p.getPatternStat(level, pattern, content) if p.onMsgCb != nil { diff --git a/parser_test.go b/parser_test.go index 6bb129b..65c30e9 100644 --- a/parser_test.go +++ b/parser_test.go @@ -2,11 +2,13 @@ package logparser import ( "sort" + "strings" "testing" "time" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" + "golang.org/x/time/rate" ) func TestParserJson(t *testing.T) { @@ -76,3 +78,52 @@ 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 := 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++ + }, + } + + 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]) + assert.Equal(t, 3, calls) + + 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) +} From 4a17e7476dfa7e8bc87f50411039a502286d36bf Mon Sep 17 00:00:00 2001 From: Nikolay Sivko Date: Thu, 17 Sep 2026 13:47:10 -0300 Subject: [PATCH 4/7] json: treat mongodb {"$date": ...} objects under the "t" key as timestamps --- json.go | 24 ++++++++++++++++++------ json_test.go | 4 ++-- 2 files changed, 20 insertions(+), 8 deletions(-) diff --git a/json.go b/json.go index c45de6a..0897e28 100644 --- a/json.go +++ b/json.go @@ -33,7 +33,7 @@ var ( "levelname": 7, } jsonTimestampKeys = map[string]bool{ - "time": true, "timestamp": true, "ts": true, "@t": true, "@timestamp": true, "datetime": true, "asctime": true, + "time": true, "timestamp": true, "ts": true, "t": true, "@t": true, "@timestamp": true, "datetime": true, "asctime": true, } ) @@ -89,11 +89,8 @@ func ParseJsonLog(content string) *JsonLog { } continue } - } else if jsonTimestampKeys[lk] { - switch v.(type) { - case string, json.Number: - continue - } + } else if jsonTimestampKeys[lk] && jsonIsTimestampValue(v) { + continue } flattenJsonField(k, v, res.Attributes, jsonMaxDepth) } @@ -101,6 +98,21 @@ func ParseJsonLog(content string) *JsonLog { 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: diff --git a/json_test.go b/json_test.go index e673d01..b787ae7 100644 --- a/json_test.go +++ b/json_test.go @@ -125,9 +125,9 @@ func TestParseJsonLogKeyPriority(t *testing.T) { } func TestParseJsonLogNestedAndArrays(t *testing.T) { - l := ParseJsonLog(`{"msg":"m","http":{"method":"GET","response":{"status":200}},"tags":["a","b"],"ok":true,"ref":null}`) + 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{ + assert.Equal(t, map[string]string{ // the mongodb-style "t" timestamp object is dropped "http.method": "GET", "http.response.status": "200", "tags": `["a","b"]`, From cbc5b5bb945d350cb3972c00da89532dcf756ae0 Mon Sep 17 00:00:00 2001 From: Nikolay Sivko Date: Thu, 17 Sep 2026 13:47:27 -0300 Subject: [PATCH 5/7] multiline: trust the level reported with an entry, guess only when it is unknown --- multiline.go | 6 +++--- multiline_test.go | 14 ++++++++++++++ 2 files changed, 17 insertions(+), 3 deletions(-) diff --git a/multiline.go b/multiline.go index 4f3e58f..119e080 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) } diff --git a/multiline_test.go b/multiline_test.go index 836a9f1..5b47106 100644 --- a/multiline_test.go +++ b/multiline_test.go @@ -422,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 +} From 03c7b0c564527aecc3d015badd87d1ce4fc445fc Mon Sep 17 00:00:00 2001 From: Nikolay Sivko Date: Wed, 30 Sep 2026 16:40:53 -0300 Subject: [PATCH 6/7] level: recognize the MySQL/MariaDB error log priorities [Note] and [System] as info --- level.go | 2 +- level_test.go | 13 +++++++++++++ 2 files changed, 14 insertions(+), 1 deletion(-) diff --git a/level.go b/level.go index a64f95d..2baf343 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 3a059d9..fe69846 100644 --- a/level_test.go +++ b/level_test.go @@ -21,6 +21,19 @@ func TestGuessLevelRedis(t *testing.T) { assert.Equal(t, LevelDebug, GuessLevel(`1:S 12 Nov 2019 07:52:11.999 . verbosed`)) } +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`)) From 6da0f06ad2f31bd86396d5197f0e3d29874a0296 Mon Sep 17 00:00:00 2001 From: mayankpande88 Date: Fri, 9 Oct 2026 01:36:12 +0530 Subject: [PATCH 7/7] decode a JSON log line once; accept CRLF/trailing whitespace on JSON lines With parseJson, inc() decoded each JSON line twice: in ParseStructuredLog for the pattern content and level, and in ParseJsonLog for the message and attributes. It now decodes once (decodeJsonObject) and derives both from the same map. structuredFromJSON renders json.Number exactly as the json.Unmarshal float64 path did, so pattern hashes are unchanged; a test checks this across numeric, nested and fallback cases. Per JSON line: 16.0 -> 10.7 us, 103 -> 57 allocations. A JSON line ending in \r or other trailing whitespace was not parsed as JSON and not recognised as the start of a new multiline message; both checks now ignore it. --- json.go | 17 ++++++++++ json_single_decode_test.go | 68 ++++++++++++++++++++++++++++++++++++++ multiline.go | 2 +- parser.go | 30 ++++++++++------- pattern.go | 42 ++++++++++++++++++++--- 5 files changed, 142 insertions(+), 17 deletions(-) create mode 100644 json_single_decode_test.go diff --git a/json.go b/json.go index 0897e28..1f21f5c 100644 --- a/json.go +++ b/json.go @@ -44,6 +44,18 @@ type JsonLog struct { } 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 } @@ -56,7 +68,12 @@ func ParseJsonLog(content string) *JsonLog { 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 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/multiline.go b/multiline.go index 119e080..853e126 100644 --- a/multiline.go +++ b/multiline.go @@ -124,7 +124,7 @@ func (m *MultilineCollector) isNextMessage(l string) bool { return false } - if strings.HasPrefix(l, `{"`) && strings.HasSuffix(l, "}") { + if strings.HasPrefix(l, `{"`) && strings.HasSuffix(strings.TrimRight(l, " \t\r"), "}") { return true } diff --git a/parser.go b/parser.go index cb3e31c..44b4556 100644 --- a/parser.go +++ b/parser.go @@ -166,17 +166,30 @@ 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 } - // With parseJson, a JSON log is emitted as its message, with the other - // fields as attributes. Patterns still come from normalizedContent. + // 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 := p.parseJsonLog(content); jl != nil { + if jl != nil { if jl.Level != LevelUnknown { msg.Level = jl.Level } @@ -264,13 +277,6 @@ func (p *Parser) processSensitivePattern(msg Message, pattern *Pattern) { } } -func (p *Parser) parseJsonLog(content string) *JsonLog { - if !p.parseJson { - return nil - } - return ParseJsonLog(content) -} - func (p *Parser) getPatternStat(level Level, pattern *Pattern, sample string) (*patternStat, patternKey) { key := patternKey{level: level, hash: pattern.Hash()} if stat := p.patterns[key]; stat != nil { diff --git a/pattern.go b/pattern.go index a04f5db..0090170 100644 --- a/pattern.go +++ b/pattern.go @@ -7,6 +7,7 @@ import ( "fmt" "regexp" "sort" + "strconv" "strings" "sync" ) @@ -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 { @@ -268,7 +277,7 @@ func parseJSONLog(line string) (message string, level Level, ok bool) { var buf strings.Builder 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