4 files changed, 142 insertions, 33 deletions

diff --git a/internal/mapr/logformat/default.go b/internal/mapr/logformat/default.go
index 44bf558..9b6c855 100644
--- a/internal/mapr/logformat/default.go
+++ b/internal/mapr/logformat/default.go
@@ -1,14 +1,23 @@
 package logformat
 
 import (
-	"errors"
+	"fmt"
 	"strings"
+
+	"github.com/mimecast/dtail/internal/protocol"
 )
 
-// MakeFieldsDEFAULT is the default log file mapreduce parser.
+// MakeFieldsDEFAULT is the default DTail log file key-value parser.
 func (p *Parser) MakeFieldsDEFAULT(maprLine string) (map[string]string, error) {
-	fields := make(map[string]string, 20)
-	splitted := strings.Split(maprLine, "|")
+	splitted := strings.Split(maprLine, protocol.FieldDelimiter)
+
+	if len(splitted) < 11 || !strings.HasPrefix(splitted[9], "MAPREDUCE:") ||
+		!strings.HasPrefix(splitted[0], "INFO") {
+		// Not a DTail mapreduce log line.
+		return nil, ErrIgnoreFields
+	}
+
+	fields := make(map[string]string, len(splitted)+8)
 
 	fields["*"] = "*"
 	fields["$line"] = maprLine
@@ -17,10 +26,30 @@ func (p *Parser) MakeFieldsDEFAULT(maprLine string) (map[string]string, error) {
 	fields["$timezone"] = p.timeZoneName
 	fields["$timeoffset"] = p.timeZoneOffset
 
-	for _, kv := range splitted {
+	fields["$severity"] = splitted[0]
+	fields["$loglevel"] = splitted[0]
+
+	time := splitted[1]
+	fields["$time"] = time
+	if len(time) == 15 {
+		// Example: 20211002-071209
+		fields["$date"] = time[0:8]
+		fields["$hour"] = time[9:11]
+		fields["$minute"] = time[11:13]
+		fields["$second"] = time[13:]
+	}
+	fields["$pid"] = splitted[2]
+	fields["$caller"] = splitted[3]
+	fields["$cpus"] = splitted[4]
+	fields["$goroutines"] = splitted[5]
+	fields["$cgocalls"] = splitted[6]
+	fields["$loadavg"] = splitted[7]
+	fields["$uptime"] = splitted[8]
+
+	for _, kv := range splitted[10:] {
 		keyAndValue := strings.SplitN(kv, "=", 2)
 		if len(keyAndValue) != 2 {
-			return fields, errors.New("Error parsing mapr token: " + kv)
+			return fields, fmt.Errorf("Unable to parse key-value token '%s'", kv)
 		}
 		fields[strings.ToLower(keyAndValue[0])] = keyAndValue[1]
 	}
diff --git a/internal/mapr/logformat/default_test.go b/internal/mapr/logformat/default_test.go
index 10ec8b7..28e1acc 100644
--- a/internal/mapr/logformat/default_test.go
+++ b/internal/mapr/logformat/default_test.go
@@ -1,6 +1,7 @@
 package logformat
 
 import (
+	"fmt"
 	"testing"
 )
 
@@ -10,26 +11,83 @@ func TestDefaultLogFormat(t *testing.T) {
 		t.Errorf("Unable to create parser: %s", err.Error())
 	}
 
-	fields, err := parser.MakeFields("foo=bar|baz=bay")
+	date := "20211002"
+	hour := "07"
+	minute := "23"
+	second := "42"
+	time := fmt.Sprintf("%s-%s%s%s", date, hour, minute, second)
 
-	if err != nil {
-		t.Errorf("Unable to parse: %s", err.Error())
+	inputs := []string{
+		fmt.Sprintf("INFO|%s|1|default_test.go:0|8|14|7|0.21|471h0m21s|MAPREDUCE:STATS|foo=bar|bar=foo", time),
+		fmt.Sprintf("INFO|%s|1|default_test.go:0|8|14|7|0.21|471h0m21s|MAPREDUCE:STATS|bar=foo|foo=bar", time),
 	}
 
-	if bar, ok := fields["foo"]; !ok {
-		t.Errorf("Expected field 'foo', but no such field there\n")
-	} else if bar != "bar" {
-		t.Errorf("Expected 'bar' stored in field 'foo', but got '%s'\n", bar)
-	}
+	for _, input := range inputs {
+		fields, err := parser.MakeFields(input)
+
+		if err != nil {
+			t.Errorf("Parser unable to make fields: %s", err.Error())
+		}
+
+		if val, ok := fields["$severity"]; !ok {
+			t.Errorf("Expected field '$severity', but no such field there in '%s'\n", input)
+		} else if val != "INFO" {
+			t.Errorf("Expected 'Info' stored in field '$severity', but got '%s' in '%s'\n",
+				val, input)
+		}
+
+		if val, ok := fields["$time"]; !ok {
+			t.Errorf("Expected field '$time', but no such field there in '%s'\n", input)
+		} else if val != time {
+			t.Errorf("Expected '%s' stored in field '$time', but got '%s' in '%s'\n",
+				time, val, input)
+		}
+
+		if val, ok := fields["$date"]; !ok {
+			t.Errorf("Expected field '$date', but no such field there in '%s'\n", input)
+		} else if val != date {
+			t.Errorf("Expected '%s' stored in field '$date', but got '%s' in '%s'\n",
+				date, val, input)
+		}
+
+		if val, ok := fields["$hour"]; !ok {
+			t.Errorf("Expected field '$hour', but no such field there in '%s'\n", input)
+		} else if val != hour {
+			t.Errorf("Expected '%s' stored in field '$hour', but got '%s' in '%s'\n",
+				hour, val, input)
+		}
+
+		if val, ok := fields["$minute"]; !ok {
+			t.Errorf("Expected field '$minute', but no such field there in '%s'\n", input)
+		} else if val != minute {
+			t.Errorf("Expected '%s' stored in field '$minute', but got '%s' in '%s'\n",
+				minute, val, input)
+		}
+
+		if val, ok := fields["$second"]; !ok {
+			t.Errorf("Expected field '$second', but no such field there in '%s'\n", input)
+		} else if val != second {
+			t.Errorf("Expected '%s' stored in field '$second', but got '%s' in '%s'\n",
+				second, val, input)
+		}
+
+		if val, ok := fields["foo"]; !ok {
+			t.Errorf("Expected field 'foo', but no such field there in '%s'\n", input)
+		} else if val != "bar" {
+			t.Errorf("Expected 'bar' stored in field 'foo', but got '%s' in '%s'\n",
+				val, input)
+		}
 
-	if bay, ok := fields["baz"]; !ok {
-		t.Errorf("Expected field 'baz', but no such field there\n")
-	} else if bay != "bay" {
-		t.Errorf("Expected 'bay' stored in field 'baz', but got '%s'\n", bay)
+		if val, ok := fields["bar"]; !ok {
+			t.Errorf("Expected field 'bar', but no such field there in '%s'\n", input)
+		} else if val != "foo" {
+			t.Errorf("Expected 'foo' stored in field 'bar', but got '%s' in '%s'\n",
+				val, input)
+		}
 	}
 
-	_, err = parser.MakeFields("foo=bar|bazbay")
-	if err == nil {
-		t.Errorf("Expected error but didn't: %s", err.Error())
+	fields, err := parser.MakeFields("foozoo=bar|bazbay")
+	if _, ok := fields["foo"]; ok {
+		t.Errorf("Expected fiending field 'foo', but found it\n")
 	}
 }
diff --git a/internal/mapr/logformat/generickv.go b/internal/mapr/logformat/generickv.go
new file mode 100644
index 0000000..433eb5f
--- /dev/null
+++ b/internal/mapr/logformat/generickv.go
@@ -0,0 +1,31 @@
+package logformat
+
+import (
+	"strings"
+
+	"github.com/mimecast/dtail/internal/protocol"
+)
+
+// MakeFieldsGENERIGKV is the generic key-value logfile parser.
+func (p *Parser) MakeFieldsGENERIGKV(maprLine string) (map[string]string, error) {
+	splitted := strings.Split(maprLine, protocol.FieldDelimiter)
+	fields := make(map[string]string, len(splitted))
+
+	fields["*"] = "*"
+	fields["$line"] = maprLine
+	fields["$empty"] = ""
+	fields["$hostname"] = p.hostname
+	fields["$timezone"] = p.timeZoneName
+	fields["$timeoffset"] = p.timeZoneOffset
+
+	for _, kv := range splitted[0:] {
+		keyAndValue := strings.SplitN(kv, "=", 2)
+		if len(keyAndValue) != 2 {
+			//dlog.Common.Debug("Unable to parse key-value token, ignoring it", kv)
+			continue
+		}
+		fields[strings.ToLower(keyAndValue[0])] = keyAndValue[1]
+	}
+
+	return fields, nil
+}
diff --git a/internal/mapr/logformat/parser.go b/internal/mapr/logformat/parser.go
index c53729a..129081d 100644
--- a/internal/mapr/logformat/parser.go
+++ b/internal/mapr/logformat/parser.go
@@ -8,10 +8,12 @@ import (
 	"strings"
 	"time"
 
-	"github.com/mimecast/dtail/internal/io/logger"
 	"github.com/mimecast/dtail/internal/mapr"
 )
 
+// ErrIgnoreFields indicates that the fields should be ignored.
+var ErrIgnoreFields error = errors.New("Ignore this field set")
+
 // Parser is used to parse the mapreduce information from the server log files.
 type Parser struct {
 	hostname           string
@@ -25,11 +27,9 @@ type Parser struct {
 // NewParser returns a new log parser.
 func NewParser(logFormatName string, query *mapr.Query) (*Parser, error) {
 	hostname, err := os.Hostname()
-
 	if err != nil {
 		return nil, err
 	}
-
 	now := time.Now()
 	zone, offset := now.Zone()
 
@@ -43,7 +43,6 @@ func NewParser(logFormatName string, query *mapr.Query) (*Parser, error) {
 	if err != nil {
 		return nil, err
 	}
-
 	return &p, nil
 }
 
@@ -52,7 +51,6 @@ func NewParser(logFormatName string, query *mapr.Query) (*Parser, error) {
 // Parser. Whereas MODULENAME must be a upeprcase string.
 func (p *Parser) reflectLogFormat(logFormatName string) error {
 	methodName := fmt.Sprintf("MakeFields%s", strings.ToUpper(logFormatName))
-
 	rt := reflect.TypeOf(p)
 	method, ok := rt.MethodByName(methodName)
 	if !ok {
@@ -61,7 +59,6 @@ func (p *Parser) reflectLogFormat(logFormatName string) error {
 
 	p.makeFieldsFunc = method.Func
 	p.makeFieldsReceiver = reflect.ValueOf(p)
-
 	return nil
 }
 
@@ -69,17 +66,11 @@ func (p *Parser) reflectLogFormat(logFormatName string) error {
 func (p *Parser) MakeFields(maprLine string) (fields map[string]string, err error) {
 	inputValues := []reflect.Value{p.makeFieldsReceiver, reflect.ValueOf(maprLine)}
 	returnValues := p.makeFieldsFunc.Call(inputValues)
-
 	errInterface := returnValues[1].Interface()
-
 	if errInterface == nil {
 		fields, err = returnValues[0].Interface().(map[string]string), nil
-		logger.Trace("parser.MakeFields", fields, err)
 		return
 	}
-
 	fields, err = returnValues[0].Interface().(map[string]string), errInterface.(error)
-	logger.Trace("parser.MakeFields", fields, err)
-
 	return
 }