wip

lib/logstorage/logfmt_parser.go

@@ -15,6 +15,10 @@ func (p *logfmtParser) reset() {
 }
 
 func (p *logfmtParser) addField(name, value string) {
+	name = strings.TrimSpace(name)
+	if name == "" && value == "" {
+		return
+	}
 	p.fields = append(p.fields, Field{
 		Name:  name,
 		Value: value,
@@ -24,14 +28,21 @@ func (p *logfmtParser) addField(name, value string) {
 func (p *logfmtParser) parse(s string) {
 	for {
 		// Search for field name
-		n := strings.IndexByte(s, '=')
+		n := strings.IndexAny(s, "= ")
 		if n < 0 {
-			// field name couldn't be read
+			// empty value
+			p.addField(s, "")
 			return
 		}
 
-		name := strings.TrimSpace(s[:n])
+		name := s[:n]
+		ch := s[n]
 		s = s[n+1:]
+		if ch == ' ' {
+			// empty value
+			p.addField(name, "")
+			continue
+		}
 		if len(s) == 0 {
 			p.addField(name, "")
 			return

lib/logstorage/logfmt_parser_test.go

@@ -22,9 +22,9 @@ func TestLogfmtParser(t *testing.T) {
 	f(`foo=bar`, `{"foo":"bar"}`)
 	f(`foo="bar=baz x=y"`, `{"foo":"bar=baz x=y"}`)
 	f(`foo=`, `{"foo":""}`)
+	f(`foo`, `{"foo":""}`)
+	f(`foo bar`, `{"foo":"","bar":""}`)
+	f(`foo bar=baz`, `{"foo":"","bar":"baz"}`)
 	f(`foo=bar baz="x y" a=b`, `{"foo":"bar","baz":"x y","a":"b"}`)
-
-	// errors
-	f(`foo`, `{}`)
-	f(`foo=bar  baz=x z qwe`, `{"foo":"bar","baz":"x"}`)
+	f(`  foo=bar  baz=x =z qwe`, `{"foo":"bar","baz":"x","":"z","qwe":""}`)
 }

lib/logstorage/pipe_unpack_logfmt_test.go

@@ -191,17 +191,6 @@ func TestPipeUnpackLogfmt(t *testing.T) {
 		},
 	})
 
-	// single row, unpack from non-json field
-	f("unpack_logfmt from x", [][]Field{
-		{
-			{"x", `foobar`},
-		},
-	}, [][]Field{
-		{
-			{"x", `foobar`},
-		},
-	})
-
 	// single row, unpack from non-logfmt
 	f("unpack_logfmt from x", [][]Field{
 		{
@@ -210,6 +199,7 @@ func TestPipeUnpackLogfmt(t *testing.T) {
 	}, [][]Field{
 		{
 			{"x", `foobar`},
+			{"foobar", ""},
 		},
 	})