feat(logparser): add logfmt

Update: #243

go.mod

@@ -14,6 +14,7 @@ require (
 	github.com/go-logfmt/logfmt v0.6.0
 	github.com/google/uuid v1.4.0
 	github.com/grafana/pyroscope-go v1.0.4
+	github.com/kr/logfmt v0.0.0-20140226030751-b84e30acd515
 	github.com/ogen-go/ogen v0.79.1
 	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/pdatautil v0.90.1
 	github.com/prometheus/common v0.45.0

go.sum

@@ -378,6 +378,7 @@ github.com/kolo/xmlrpc v0.0.0-20220921171641-a4b6fa1dd06b h1:udzkj9S/zlT5X367kqJ
 github.com/kolo/xmlrpc v0.0.0-20220921171641-a4b6fa1dd06b/go.mod h1:pcaDhQK0/NJZEvtCO0qQPPropqV0sJOJ6YW7X+9kRwM=
 github.com/konsorten/go-windows-terminal-sequences v1.0.1/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ=
 github.com/konsorten/go-windows-terminal-sequences v1.0.3/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ=
+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/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo=
 github.com/kr/pretty v0.2.0/go.mod h1:ipq/a2n7PKx3OHsz4KJII5eveXtPO4qwEXGdVfWzfnI=

internal/logparser/_golden/logfmt_logrus_01.json

@@ -0,0 +1,13 @@
+{
+  "severity": {
+    "str": "Warn",
+    "number": 13,
+    "text": "warning"
+  },
+  "body": "The group's number increased tremendously!",
+  "timestamp": "2015-03-26T05:27:38Z",
+  "attrs": {
+    "number": 122,
+    "omg": true
+  }
+}

internal/logparser/_golden/logfmt_logrus_02.json

@@ -0,0 +1,14 @@
+{
+  "severity": {
+    "str": "Fatal",
+    "number": 21,
+    "text": "fatal"
+  },
+  "body": "The ice breaks!",
+  "timestamp": "2015-03-26T05:27:38Z",
+  "attrs": {
+    "err": "&{0x2082280c0 map[animal:orca size:9009] 2015-03-26 01:27:38.441574009 -0400 EDT panic It's over 9000!}",
+    "number": 100,
+    "omg": true
+  }
+}

internal/logparser/_golden/logfmt_test_01.json

@@ -0,0 +1,10 @@
+{
+  "attrs": {
+    "foo": "bar",
+    "a": 14,
+    "baz": "hello kitty",
+    "cool%story": "bro",
+    "f": true,
+    "%^asdf": true
+  }
+}

internal/logparser/_testdata/logfmt/logrus.txt

@@ -0,0 +1,2 @@
+time="2015-03-26T01:27:38-04:00" level=warning msg="The group's number increased tremendously!" number=122 omg=true
+time="2015-03-26T01:27:38-04:00" level=fatal msg="The ice breaks!" err="&{0x2082280c0 map[animal:orca size:9009] 2015-03-26 01:27:38.441574009 -0400 EDT panic It's over 9000!}" number=100 omg=true

internal/logparser/_testdata/logfmt/test.txt

@@ -0,0 +1 @@
+foo=bar a=14 baz="hello kitty" cool%story=bro f %^asdf

internal/logparser/logfmt.go

@@ -0,0 +1,116 @@
+package logparser
+
+import (
+	"encoding/hex"
+	"strings"
+	"time"
+	"unicode"
+
+	"github.com/go-faster/jx"
+	"github.com/google/uuid"
+	"github.com/kr/logfmt"
+	"go.opentelemetry.io/collector/pdata/pcommon"
+
+	"github.com/go-faster/oteldb/internal/otelstorage"
+)
+
+// LogFmtParser parses logfmt lines.
+type LogFmtParser struct{}
+
+// Parse line.
+func (LogFmtParser) Parse(data []byte) (*Line, error) {
+	line := &Line{}
+	attrs := pcommon.NewMap()
+	hf := logfmt.HandlerFunc(func(key, val []byte) error {
+		k := string(key)
+		v := string(val)
+		switch k {
+		case "msg":
+			line.Body = v
+		case "level", "lvl", "levelStr", "severity_text", "severity", "levelname":
+			line.SeverityText = v
+			line.SeverityNumber = _severityMap[unicode.ToLower(rune(v[0]))]
+		case "span_id", "spanid", "spanID", "spanId":
+			raw, _ := hex.DecodeString(v)
+			if len(raw) != 8 {
+				attrs.PutStr(k, v)
+				return nil
+			}
+			var spanID otelstorage.SpanID
+			copy(spanID[:], raw)
+			line.SpanID = spanID
+		case "trace_id", "traceid", "traceID", "traceId":
+			traceID, err := otelstorage.ParseTraceID(strings.ToLower(v))
+			if err != nil {
+				// Trying to parse as UUID.
+				id, err := uuid.Parse(v)
+				if err != nil {
+					attrs.PutStr(k, v)
+					return nil
+				}
+				traceID = otelstorage.TraceID(id)
+			}
+			line.TraceID = traceID
+		case "ts", "time", "@timestamp", "timestamp":
+			for _, layout := range []string{
+				time.RFC3339Nano,
+				time.RFC3339,
+				ISO8601Millis,
+			} {
+				ts, err := time.Parse(layout, v)
+				if err != nil {
+					continue
+				}
+				line.Timestamp = otelstorage.Timestamp(ts.UnixNano())
+			}
+			if line.Timestamp == 0 {
+				attrs.PutStr(k, v)
+			}
+		default:
+			// Try deduct a type.
+			if v == "" {
+				attrs.PutBool(k, true)
+				return nil
+			}
+			dec := jx.DecodeBytes(val)
+			switch dec.Next() {
+			case jx.Number:
+				n, err := dec.Num()
+				if err == nil && n.IsInt() {
+					i, err := n.Int64()
+					if err == nil {
+						attrs.PutInt(k, i)
+						return nil
+					}
+				} else if err == nil {
+					f, err := n.Float64()
+					if err == nil {
+						attrs.PutDouble(k, f)
+						return nil
+					}
+				}
+			case jx.Bool:
+				v, err := dec.Bool()
+				if err == nil {
+					attrs.PutBool(k, v)
+					return nil
+				}
+			}
+			// Fallback.
+			attrs.PutStr(k, v)
+		}
+		return nil
+	})
+	if err := logfmt.Unmarshal(data, hf); err != nil {
+		return nil, err
+	}
+	if attrs.Len() > 0 {
+		line.Attrs = otelstorage.Attrs(attrs)
+	}
+	return line, nil
+}
+
+// Detect if line is parsable by this parser.
+func (LogFmtParser) Detect(line string) bool {
+	return jx.DecodeStr(line).Next() == jx.Object
+}

internal/logparser/logfmt_test.go

@@ -0,0 +1,49 @@
+package logparser
+
+import (
+	"bufio"
+	"bytes"
+	"fmt"
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+
+	"github.com/go-faster/sdk/gold"
+	"github.com/stretchr/testify/require"
+)
+
+func TestLogFmtParser_Parse(t *testing.T) {
+	const name = "logfmt"
+	files, err := os.ReadDir(filepath.Join("_testdata", name))
+	require.NoError(t, err, "read testdata")
+
+	for _, file := range files {
+		t.Run(file.Name(), func(t *testing.T) {
+			data, err := os.ReadFile(filepath.Join("_testdata", name, file.Name()))
+			require.NoError(t, err, "read testdata")
+
+			var parser LogFmtParser
+
+			scanner := bufio.NewScanner(bytes.NewReader(data))
+
+			var i int
+			for scanner.Scan() {
+				s := strings.TrimSpace(scanner.Text())
+				if s == "" {
+					continue
+				}
+				i++
+				t.Run(fmt.Sprintf("Line%02d", i), func(t *testing.T) {
+					t.Logf("%s", s)
+					line, err := parser.Parse([]byte(s))
+					require.NoError(t, err, "parse")
+					fileName := fmt.Sprintf("%s_%s_%02d.json",
+						name, strings.TrimSuffix(file.Name(), filepath.Ext(file.Name())), i,
+					)
+					gold.Str(t, line.String(), fileName)
+				})
+			}
+		})
+	}
+}