Produce MAPREDUCE lines, can aggregate these via default log format

2 files changed, 35 insertions, 19 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]
 	}
diff --git a/internal/mapr/logformat/default_test.go b/internal/mapr/logformat/default_test.go
index d7a4da4..6284008 100644
--- a/internal/mapr/logformat/default_test.go
+++ b/internal/mapr/logformat/default_test.go
@@ -10,26 +10,33 @@ func TestDefaultLogFormat(t *testing.T) {
 		t.Errorf("Unable to create parser: %s", err.Error())
 	}
 
-	fields, err := parser.MakeFields("foo=bar|baz=bay")
-
-	if err != nil {
-		t.Errorf("Unable to parse: %s", err.Error())
+	inputs := []string{
+		"foo=bar|baz=bay",
+		"INFO|20210907-065632|SERVER|MAPREDUCE:TEST|foo=bar|baz=bay",
 	}
 
-	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 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)
+		}
+
+		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 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)
 	}
 
-	fields, err = parser.MakeFields("foo=bar|bazbay")
-	if err == nil {
-		t.Errorf("Expected error but didn't: %s", err.Error())
+	if _, err := parser.MakeFields("foo=bar|bazbay"); err == nil {
+		t.Errorf("Expected error but didn't")
 	}
 }