Produce MAPREDUCE lines, can aggregate these via default log format

1 files changed, 12 insertions, 3 deletions

diff --git a/internal/mapr/logformat/default.go b/internal/mapr/logformat/default.go
index 32a34bd..2881047 100644
--- a/internal/mapr/logformat/default.go
+++ b/internal/mapr/logformat/default.go
@@ -9,8 +9,8 @@ import (
 
 // MakeFieldsDEFAULT is the default log file mapreduce parser.
 func (p *Parser) MakeFieldsDEFAULT(maprLine string) (map[string]string, error) {
-	fields := make(map[string]string, 20)
 	splitted := strings.Split(maprLine, protocol.FieldDelimiter)
+	fields := make(map[string]string, len(splitted))
 
 	fields["*"] = "*"
 	fields["$line"] = maprLine
@@ -19,10 +19,19 @@ func (p *Parser) MakeFieldsDEFAULT(maprLine string) (map[string]string, error) {
 	fields["$timezone"] = p.timeZoneName
 	fields["$timeoffset"] = p.timeZoneOffset
 
-	for _, kv := range splitted {
+	kvStart := 0
+	// DTail mapreduce format
+	if len(splitted) > 3 && strings.HasPrefix(splitted[3], "MAPREDUCE:") {
+		fields["$severity"] = splitted[0]
+		// TODO: Parse time like we do at Mimecast
+		fields["$time"] = splitted[1]
+		kvStart = 4
+	}
+
+	for _, kv := range splitted[kvStart:] {
 		keyAndValue := strings.SplitN(kv, "=", 2)
 		if len(keyAndValue) != 2 {
-			return fields, errors.New("Error parsing mapr token: " + kv)
+			return fields, errors.New("Error parsing mapreduce token: " + kv)
 		}
 		fields[strings.ToLower(keyAndValue[0])] = keyAndValue[1]
 	}