add onMsgCallback

Author: def

cmd/logparser.go

@@ -22,7 +22,7 @@ func main() {
 
 	reader := bufio.NewReader(os.Stdin)
 	ch := make(chan logparser.LogEntry)
-	parser := logparser.NewParser(ch, nil)
+	parser := logparser.NewParser(ch, nil, nil)
 	t := time.Now()
 	for {
 		line, err := reader.ReadString('\n')
@@ -32,7 +32,7 @@ func main() {
 			}
 			break
 		}
-		ch <- logparser.LogEntry{Content: strings.TrimSuffix(line, "\n"), Level: logparser.LevelUnknown}
+		ch <- logparser.LogEntry{Timestamp: time.Now(), Content: strings.TrimSuffix(line, "\n"), Level: logparser.LevelUnknown}
 	}
 	d := time.Since(t)
 	defer parser.Stop()

multiline.go

@@ -13,12 +13,14 @@ var (
 )
 
 type Message struct {
-	Content string
-	Level   Level
+	Timestamp time.Time
+	Content   string
+	Level     Level
 }
 
 type MultilineCollector struct {
 	Messages                     chan Message
+	ts                           time.Time
 	level                        Level
 	lines                        []string
 	isFirstLineContainsTimestamp bool
@@ -80,26 +82,18 @@ func (m *MultilineCollector) Add(entry LogEntry) {
 
 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.isFirstLineContainsTimestamp = containsTimestamp(entry.Content)
 	}
-	switch m.level {
-	case LevelCritical, LevelError, LevelWarning:
-		if len(m.lines) == 0 {
-			m.appendLine(clean(entry.Content), entry.Content)
-		} else {
-			m.appendLine(entry.Content, entry.Content)
-		}
-	default:
-		m.appendLine("", entry.Content)
-	}
+	m.appendLine(entry.Content)
 	m.lastReceiveTime = time.Now()
 }
 
-func (m *MultilineCollector) appendLine(value, rawValue string) {
+func (m *MultilineCollector) appendLine(value string) {
 	m.lines = append(m.lines, value)
 }
 
@@ -145,14 +139,16 @@ func (m *MultilineCollector) flushMessage() {
 	}
 	content := strings.TrimSpace(strings.Join(m.lines, "\n"))
 	msg := Message{
-		Content: content,
-		Level:   m.level,
+		Timestamp: m.ts,
+		Content:   content,
+		Level:     m.level,
 	}
 	m.reset()
 	m.Messages <- msg
 }
 
 func (m *MultilineCollector) reset() {
+	m.ts = time.Time{}
 	m.lines = m.lines[:0]
 	m.isFirstLineContainsTimestamp = false
 	m.level = LevelUnknown

multiline_test.go

@@ -8,9 +8,9 @@ import (
 	"time"
 )
 
-func writeByLine(m *MultilineCollector, data string) {
-	for _, line := range strings.Split(data, "\n") {
-		m.Add(LogEntry{Content: line, Level: LevelUnknown})
+func writeByLine(m *MultilineCollector, data string, ts time.Time) {
+	for i, line := range strings.Split(data, "\n") {
+		m.Add(LogEntry{Timestamp: ts.Add(time.Millisecond * time.Duration(i)), Content: line, Level: LevelUnknown})
 	}
 }
 
@@ -19,8 +19,7 @@ func TestMultilineCollector(t *testing.T) {
 	m := NewMultilineCollector(ctx, 10*time.Millisecond)
 	defer cancel()
 
-	dateStr := "2020-03-20 08:48:57,067 "
-	tracebackStr := `ERROR [django.request:222] log 46 140452532862280 Internal Server Error: /article
+	tracebackStr := `2020-03-20 08:48:57,067 ERROR [django.request:222] log 46 140452532862280 Internal Server Error: /article
 Traceback (most recent call last):
   File "/usr/local/lib/python3.8/site-packages/django/db/backends/base/base.py", line 220, in ensure_connection
     self.connect()
@@ -79,28 +78,28 @@ Traceback (most recent call last):
     super(Connection, self).__init__(*args, **kwargs2)
 django.db.utils.OperationalError: (1040, 'Too many connections')`
 
-	writeByLine(m, dateStr+tracebackStr)
+	writeByLine(m, tracebackStr, time.Unix(100500, 0))
 	msg := <-m.Messages
 	assert.Equal(t, tracebackStr, msg.Content)
+	assert.Equal(t, int64(100500), msg.Timestamp.Unix())
 
-	tracebackStr = `ERROR:__main__:Traceback (most recent call last):
+	tracebackStr = `2020-03-20 08:48:57,067 ERROR:__main__:Traceback (most recent call last):
   File "<stdin>", line 2, in <module>
   File "<stdin>", line 2, in do_something_that_might_error
   File "<stdin>", line 2, in raise_error
 RuntimeError: something bad happened!`
-	writeByLine(m, dateStr+tracebackStr)
+	writeByLine(m, tracebackStr, time.Unix(0, 0))
 	msg = <-m.Messages
 	assert.Equal(t, tracebackStr, msg.Content)
 
 	m.Add(LogEntry{Content: "E0504 07:38:36.184861       1 replica_set.go:450] starting worker #224", Level: LevelUnknown})
 	m.Add(LogEntry{Content: "E0504 07:38:36.184861       1 replica_set.go:450] starting worker #225", Level: LevelUnknown})
 	msg = <-m.Messages
-	assert.Equal(t, "E0504 1 replica_set.go:450] starting worker #224", msg.Content)
+	assert.Equal(t, "E0504 07:38:36.184861       1 replica_set.go:450] starting worker #224", msg.Content)
 	msg = <-m.Messages
-	assert.Equal(t, "E0504 1 replica_set.go:450] starting worker #225", msg.Content)
+	assert.Equal(t, "E0504 07:38:36.184861       1 replica_set.go:450] starting worker #225", msg.Content)
 
-	dateStr = "2020-03-31 11:35:06.158 "
-	javaStackTraceStr := `[ERROR] javax.servlet.ServletException: Something bad happened
+	javaStackTraceStr := `2020-03-31 11:35:06.158 [ERROR] javax.servlet.ServletException: Something bad happened
 	at com.example.myproject.OpenSessionInViewFilter.doFilter(OpenSessionInViewFilter.java:60)
 	at org.mortbay.jetty.servlet.ServletHandler$CachedChain.doFilter(ServletHandler.java:1157)
 	at com.example.myproject.ExceptionHandlerFilter.doFilter(ExceptionHandlerFilter.java:28)
@@ -163,16 +162,14 @@ Caused by: java.sql.SQLException: Violation of unique constraint MY_ENTITY_UK_1:
 	at org.hibernate.id.insert.AbstractSelectingDelegate.performInsert(AbstractSelectingDelegate.java:57)
 	... 54 more`
 
-	writeByLine(m, dateStr+javaStackTraceStr)
+	writeByLine(m, javaStackTraceStr, time.Unix(0, 0))
 	msg = <-m.Messages
 	assert.Equal(t, javaStackTraceStr, msg.Content)
 
 	data := `Order response: {"statusCode":406,"body":{"timestamp":1648205755430,"status":406,"error":"Not Acceptable","exception":"works.weave.socks.orders.controllers.OrdersController$PaymentDeclinedException","message":"Payment declined: amount exceeds 100.00","path":"/orders"},"headers":{"x-application-context":"orders:80","content-type":"application/json;charset=UTF-8","transfer-encoding":"chunked","date":"Fri, 25 Mar 2022 10:55:55 GMT","connection":"close"},"request":{"uri":{"protocol":"http:","slashes":true,"auth":null,"host":"orders","port":80,"hostname":"orders","hash":null,"search":null,"query":null,"pathname":"/orders","path":"/orders","href":"http://orders/orders"},"method":"POST","headers":{"accept":"application/json","content-type":"application/json","content-length":232}}}
 Order response: {"timestamp":1648205755430,"status":406,"error":"Not Acceptable","exception":"works.weave.socks.orders.controllers.OrdersController$PaymentDeclinedException","message":"Payment declined: amount exceeds 100.00","path":"/orders"}
 `
-	writeByLine(m, data)
+	writeByLine(m, data, time.Unix(0, 0))
 	msg = <-m.Messages
-	assert.Equal(t,
-		clean(strings.Split(data, "\n")[0]),
-		msg.Content)
+	assert.Equal(t, strings.Split(data, "\n")[0], msg.Content)
 }

parser.go

@@ -3,11 +3,13 @@ package logparser
 import (
 	"context"
 	"sync"
+	"time"
 )
 
 type LogEntry struct {
-	Content string
-	Level   Level
+	Timestamp time.Time
+	Content   string
+	Level     Level
 }
 
 type LogCounter struct {
@@ -26,10 +28,18 @@ type Parser struct {
 	multilineCollector *MultilineCollector
 
 	stop func()
+
+	onMsgCb OnMsgCallbackF
 }
 
-func NewParser(ch <-chan LogEntry, decoder Decoder) *Parser {
-	p := &Parser{decoder: decoder, patterns: map[patternKey]*patternStat{}}
+type OnMsgCallbackF func(ts time.Time, level Level, patternHash string, msg string)
+
+func NewParser(ch <-chan LogEntry, decoder Decoder, onMsgCallback OnMsgCallbackF) *Parser {
+	p := &Parser{
+		decoder:  decoder,
+		patterns: map[patternKey]*patternStat{},
+		onMsgCb:  onMsgCallback,
+	}
 	ctx, stop := context.WithCancel(context.Background())
 	p.stop = stop
 	p.multilineCollector = NewMultilineCollector(ctx, multilineCollectorTimeout)
@@ -79,6 +89,9 @@ func (p *Parser) inc(msg Message) {
 			p.patterns[key] = &patternStat{}
 		}
 		p.patterns[key].messages++
+		if p.onMsgCb != nil {
+			p.onMsgCb(msg.Timestamp, msg.Level, "", msg.Content)
+		}
 		return
 	}
 
@@ -97,6 +110,9 @@ func (p *Parser) inc(msg Message) {
 			p.patterns[key] = stat
 		}
 	}
+	if p.onMsgCb != nil {
+		p.onMsgCb(msg.Timestamp, msg.Level, key.hash, msg.Content)
+	}
 	stat.messages++
 }