Merge pull request #28 from aybabtme/use-go-logfmt

use go logfmt, be more lenient in parsing lines

.github/workflows/go.yml

@@ -0,0 +1,19 @@
+name: Go
+on: [push]
+jobs:
+  test:
+    name: Test
+    runs-on: ubuntu-latest
+    steps:
+
+    - name: Set up Go 1.12
+      uses: actions/setup-go@v1
+      with:
+        go-version: 1.12
+      id: go
+
+    - name: Check out code into the Go module directory
+      uses: actions/checkout@v1
+
+    - name: Test
+      run: go test -mod=vendor -short ./...

go.mod

@@ -5,6 +5,7 @@ go 1.13
 require (
 	github.com/aybabtme/rgbterm v0.0.0-20170906152045-cc83f3b3ce59
 	github.com/fatih/color v1.7.1-0.20180516100307-2d684516a886
+	github.com/go-logfmt/logfmt v0.4.0
 	github.com/kr/logfmt v0.0.0-20140226030751-b84e30acd515
 	github.com/mattn/go-colorable v0.1.0
 	github.com/mattn/go-isatty v0.0.4 // indirect

go.sum

@@ -2,6 +2,8 @@ github.com/aybabtme/rgbterm v0.0.0-20170906152045-cc83f3b3ce59 h1:WWB576BN5zNSZc
 github.com/aybabtme/rgbterm v0.0.0-20170906152045-cc83f3b3ce59/go.mod h1:q/89r3U2H7sSsE2t6Kca0lfwTK8JdoNGS/yzM/4iH5I=
 github.com/fatih/color v1.7.1-0.20180516100307-2d684516a886 h1:NAFoy+QgUpERgK3y1xiVh5HcOvSeZHpXTTo5qnvnuK4=
 github.com/fatih/color v1.7.1-0.20180516100307-2d684516a886/go.mod h1:Zm6kSWBoL9eyXnKyktHP6abPY2pDugNf5KwzbycvMj4=
+github.com/go-logfmt/logfmt v0.4.0 h1:MP4Eh7ZCb31lleYCFuwm0oe4/YGak+5l1vA2NOE80nA=
+github.com/go-logfmt/logfmt v0.4.0/go.mod h1:3RMwSq7FuexP4Kalkev3ejPJsZTpXXBr9+V4qmtdjCk=
 github.com/kr/logfmt v0.0.0-20140226030751-b84e30acd515 h1:T+h1c/A9Gawja4Y9mFVWj2vyii2bbUNDw3kt9VxK2EY=
 github.com/kr/logfmt v0.0.0-20140226030751-b84e30acd515/go.mod h1:+0opPa2QZZtGFBFZlji/RkVcI2GknAs/DXo4wKdlNEc=
 github.com/mattn/go-colorable v0.1.0 h1:v2XXALHHh6zHfYTJ+cSkwtyffnaOyR1MXaA91mTrb8o=

json_handler.go

@@ -29,9 +29,24 @@ type JSONHandler struct {
 	last map[string]string
 }
 
+func checkEachUntilFound(fieldList []string, found func(string) bool) bool {
+	for _, field := range fieldList {
+		if found(field) {
+			return true
+		}
+	}
+	return false
+}
+
 // supportedTimeFields enumerates supported timestamp field names
 var supportedTimeFields = []string{"time", "ts", "@timestamp"}
 
+// supportedMessageFields enumarates supported Message field names
+var supportedMessageFields = []string{"message", "msg"}
+
+// supportedLevelFields enumarates supported level field names
+var supportedLevelFields = []string{"level", "lvl"}
+
 func (h *JSONHandler) clear() {
 	h.Level = ""
 	h.Time = time.Time{}
@@ -45,74 +60,54 @@ func (h *JSONHandler) clear() {
 
 // TryHandle tells if this line was handled by this handler.
 func (h *JSONHandler) TryHandle(d []byte) bool {
-	var ok bool
-
-	for _, field := range supportedTimeFields {
-		ok = bytes.Contains(d, []byte(`"`+field+`":`))
-		if ok {
-			break
-		}
-	}
-
-	if !ok {
-		return false
-	}
-
-	err := h.UnmarshalJSON(d)
-	if err != nil {
+	if !h.UnmarshalJSON(d) {
 		h.clear()
 		return false
 	}
 	return true
 }
 
 // UnmarshalJSON sets the fields of the handler.
-func (h *JSONHandler) UnmarshalJSON(data []byte) error {
+func (h *JSONHandler) UnmarshalJSON(data []byte) bool {
 	raw := make(map[string]interface{})
 	err := json.Unmarshal(data, &raw)
 	if err != nil {
-		return err
+		return false
 	}
 
-	var time interface{}
-	var ok bool
-
-	for _, field := range supportedTimeFields {
-		time, ok = raw[field]
+	checkEachUntilFound(supportedLevelFields, func(field string) bool {
+		time, ok := tryParseTime(raw[field])
 		if ok {
+			h.Time = time
 			delete(raw, field)
-			break
 		}
-	}
+		return ok
+	})
 
-	if ok {
-		h.Time, ok = tryParseTime(time)
-		if !ok {
-			return fmt.Errorf("field time is not a known timestamp: %v", time)
+	checkEachUntilFound(supportedMessageFields, func(field string) bool {
+		msg, ok := raw[field].(string)
+		if ok {
+			h.Message = msg
+			delete(raw, field)
 		}
-	}
+		return ok
+	})
 
-	if h.Message, ok = raw["msg"].(string); ok {
-		delete(raw, "msg")
-	} else if h.Message, ok = raw["message"].(string); ok {
-		delete(raw, "message")
-	}
-
-	h.Level, ok = raw["level"].(string)
-	if !ok {
-		h.Level, ok = raw["lvl"].(string)
-		delete(raw, "lvl")
+	checkEachUntilFound(supportedLevelFields, func(field string) bool {
+		lvl, ok := raw[field]
 		if !ok {
-			// bunyan uses numerical log levels
-			level, ok := raw["level"].(float64)
-			if ok {
-				h.Level = convertBunyanLogLevel(level)
-				delete(raw, "level")
-			} else {
-				h.Level = "???"
-			}
+			return false
 		}
-	}
+		if strLvl, ok := lvl.(string); ok {
+			h.Level = strLvl
+		} else if flLvl, ok := lvl.(float64); ok {
+			h.Level = convertBunyanLogLevel(flLvl)
+		} else {
+			h.Level = "???"
+		}
+		delete(raw, field)
+		return true
+	})
 
 	if h.Fields == nil {
 		h.Fields = make(map[string]string)
@@ -134,7 +129,7 @@ func (h *JSONHandler) UnmarshalJSON(data []byte) error {
 		}
 	}
 
-	return nil
+	return true
 }
 
 // Prettify the output in a logrus like fashion.

logrus_handler.go → logfmt_handler.go

@@ -10,10 +10,11 @@ import (
 	"time"
 
 	"github.com/fatih/color"
+	"github.com/go-logfmt/logfmt"
 )
 
-// LogrusHandler can handle logs emmited by logrus.TextFormatter loggers.
-type LogrusHandler struct {
+// LogfmtHandler can handle logs emmited by logrus.TextFormatter loggers.
+type LogfmtHandler struct {
 	buf     *bytes.Buffer
 	out     *tabwriter.Writer
 	truncKV int
@@ -28,52 +29,81 @@ type LogrusHandler struct {
 	last map[string]string
 }
 
-func (h *LogrusHandler) clear() {
+func (h *LogfmtHandler) clear() {
 	h.Level = ""
 	h.Time = time.Time{}
 	h.Message = ""
 	h.last = h.Fields
 	h.Fields = make(map[string]string)
-	h.buf.Reset()
+	if h.buf != nil {
+		h.buf.Reset()
+	}
 }
 
 // CanHandle tells if this line can be handled by this handler.
-func (h *LogrusHandler) CanHandle(d []byte) bool {
-	if !(bytes.Contains(d, []byte(`level=`)) || bytes.Contains(d, []byte(`lvl=`))) {
-		return false
-	}
-	if !(bytes.Contains(d, []byte(`time=`)) || bytes.Contains(d, []byte(`ts=`))) {
-		return false
-	}
-	if !(bytes.Contains(d, []byte(`message=`)) || bytes.Contains(d, []byte(`msg=`))) {
+func (h *LogfmtHandler) TryHandle(d []byte) bool {
+	if !h.UnmarshalLogfmt(d) {
+		h.clear()
 		return false
 	}
 	return true
 }
 
 // HandleLogfmt sets the fields of the handler.
-func (h *LogrusHandler) visit(key, val []byte) bool {
-	switch {
-	case bytes.Equal(key, []byte("level")):
-		h.setLevel(val)
-	case bytes.Equal(key, []byte("lvl")):
-		h.setLevel(val)
-	case bytes.Equal(key, []byte("msg")):
-		h.setMessage(val)
-	case bytes.Equal(key, []byte("message")):
-		h.setMessage(val)
-	case bytes.Equal(key, []byte("time")):
-		h.setTime(val)
-	case bytes.Equal(key, []byte("ts")):
-		h.setTime(val)
-	default:
-		h.setField(key, val)
+func (h *LogfmtHandler) UnmarshalLogfmt(data []byte) bool {
+	dec := logfmt.NewDecoder(bytes.NewReader(data))
+	for dec.ScanRecord() {
+	next_kv:
+		for dec.ScanKeyval() {
+			key := dec.Key()
+			val := dec.Value()
+			if h.Time.IsZero() {
+				foundTime := checkEachUntilFound(supportedLevelFields, func(field string) bool {
+					time, ok := tryParseTime(string(val))
+					if ok {
+						h.Time = time
+					}
+					return ok
+				})
+				if foundTime {
+					continue next_kv
+				}
+			}
+
+			if len(h.Message) == 0 {
+				foundMessage := checkEachUntilFound(supportedMessageFields, func(field string) bool {
+					if !bytes.Equal(key, []byte(field)) {
+						return false
+					}
+					h.Message = string(val)
+					return true
+				})
+				if foundMessage {
+					continue next_kv
+				}
+			}
+
+			if len(h.Level) == 0 {
+				foundLevel := checkEachUntilFound(supportedLevelFields, func(field string) bool {
+					if !bytes.Equal(key, []byte(field)) {
+						return false
+					}
+					h.Level = string(val)
+					return true
+				})
+				if foundLevel {
+					continue next_kv
+				}
+			}
+
+			h.setField(key, val)
+		}
 	}
-	return true
+	return dec.Err() == nil
 }
 
 // Prettify the output in a logrus like fashion.
-func (h *LogrusHandler) Prettify(skipUnchanged bool) []byte {
+func (h *LogfmtHandler) Prettify(skipUnchanged bool) []byte {
 	defer h.clear()
 	if h.out == nil {
 		if h.Opts == nil {
@@ -137,9 +167,9 @@ func (h *LogrusHandler) Prettify(skipUnchanged bool) []byte {
 	return h.buf.Bytes()
 }
 
-func (h *LogrusHandler) setLevel(val []byte)   { h.Level = string(val) }
-func (h *LogrusHandler) setMessage(val []byte) { h.Message = string(val) }
-func (h *LogrusHandler) setTime(val []byte) (parsed bool) {
+func (h *LogfmtHandler) setLevel(val []byte)   { h.Level = string(val) }
+func (h *LogfmtHandler) setMessage(val []byte) { h.Message = string(val) }
+func (h *LogfmtHandler) setTime(val []byte) (parsed bool) {
 	valStr := string(val)
 	if valFloat, err := strconv.ParseFloat(valStr, 64); err == nil {
 		h.Time, parsed = tryParseTime(valFloat)
@@ -149,14 +179,14 @@ func (h *LogrusHandler) setTime(val []byte) (parsed bool) {
 	return
 }
 
-func (h *LogrusHandler) setField(key, val []byte) {
+func (h *LogfmtHandler) setField(key, val []byte) {
 	if h.Fields == nil {
 		h.Fields = make(map[string]string)
 	}
 	h.Fields[string(key)] = string(val)
 }
 
-func (h *LogrusHandler) joinKVs(skipUnchanged bool, sep string) []string {
+func (h *LogfmtHandler) joinKVs(skipUnchanged bool, sep string) []string {
 
 	kv := make([]string, 0, len(h.Fields))
 	for k, v := range h.Fields {