Produce MAPREDUCE lines, can aggregate these via default log format

11 files changed, 141 insertions, 50 deletions

diff --git a/Makefile b/Makefile
index b05e9aa..103e2c4 100644
--- a/Makefile
+++ b/Makefile
@@ -34,6 +34,7 @@ vet:
 	  echo ${GO} vet $$dir; \
 	  ${GO} vet $$dir; \
 	done
+	grep -R TODO .
 lint:
 	${GO} get golang.org/x/lint/golint
 	find . -type d | while read dir; do \
diff --git a/TODO.md b/TODO.md
index 0a8ad16..80329c6 100644
--- a/TODO.md
+++ b/TODO.md
@@ -15,3 +15,5 @@ This is a loose list of what to do. Maybe for the next releae or maybe for a lat
 [x] Client 4.x should print a warning when trying to connect to a 3.x server.
 [ ] Update docs for color configuration
 [ ] Update animated gifs
+[ ] Canary/RC deployment
+[ ] Fix auto-reconnect
diff --git a/docker/Makefile b/docker/Makefile
index a4ffa19..910b23c 100644
--- a/docker/Makefile
+++ b/docker/Makefile
@@ -1,5 +1,6 @@
 all: build
 testrun: build spinup dcat spindown
+serverfarm: spindown build spinup
 build:
 	cp ../dserver .
 	docker build . -t dserver:develop
diff --git a/internal/clients/stats.go b/internal/clients/stats.go
index faeb9fb..6da443c 100644
--- a/internal/clients/stats.go
+++ b/internal/clients/stats.go
@@ -11,12 +11,13 @@ import (
 	"github.com/mimecast/dtail/internal/color"
 	"github.com/mimecast/dtail/internal/config"
 	"github.com/mimecast/dtail/internal/io/logger"
+	"github.com/mimecast/dtail/internal/protocol"
 )
 
 // Used to collect and display various client stats.
 type stats struct {
 	// Total amount servers to connect to.
-	connectionsTotal int
+	servers int
 	// To keep track of what connected and disconnected
 	connectionsEstCh chan struct{}
 	// Amount of servers connections are established.
@@ -25,10 +26,10 @@ type stats struct {
 	mutex sync.Mutex
 }
 
-func newTailStats(connectionsTotal int) *stats {
+func newTailStats(servers int) *stats {
 	return &stats{
-		connectionsTotal: connectionsTotal,
-		connectionsEstCh: make(chan struct{}, connectionsTotal),
+		servers:          servers,
+		connectionsEstCh: make(chan struct{}, servers),
 		connected:        0,
 	}
 }
@@ -59,13 +60,14 @@ func (s *stats) Start(ctx context.Context, throttleCh <-chan struct{}, statsCh <
 			continue
 		}
 
-		stats := s.statsLine(connected, newConnections, throttle)
 		switch force {
 		case true:
+			stats := s.statsLine(connected, newConnections, throttle)
 			messages = append(messages, fmt.Sprintf("Connection stats: %s", stats))
 			s.printStatsDueInterrupt(messages)
 		default:
-			logger.Info(stats)
+			data := s.statsData(connected, newConnections, throttle)
+			logger.Mapreduce("STATS", data)
 		}
 
 		connectedLast = connected
@@ -92,16 +94,37 @@ func (s *stats) printStatsDueInterrupt(messages []string) {
 	logger.Resume()
 }
 
+func (s *stats) statsData(connected, newConnections int, throttle int) map[string]interface{} {
+	percConnected := percentOf(float64(s.servers), float64(connected))
+
+	data := make(map[string]interface{})
+	data["connected"] = connected
+	data["servers"] = s.servers
+	data["connected%"] = int(percConnected)
+	data["new"] = newConnections
+	data["throttle"] = throttle
+	data["goroutines"] = runtime.NumGoroutine()
+	data["cgocalls"] = runtime.NumCgoCall()
+	data["cpu"] = runtime.NumCPU()
+
+	return data
+}
+
 func (s *stats) statsLine(connected, newConnections int, throttle int) string {
-	percConnected := percentOf(float64(s.connectionsTotal), float64(connected))
+	sb := strings.Builder{}
 
-	var stats []string
-	stats = append(stats, fmt.Sprintf("connected=%d/%d(%d%%)", connected, s.connectionsTotal, int(percConnected)))
-	stats = append(stats, fmt.Sprintf("new=%d", newConnections))
-	stats = append(stats, fmt.Sprintf("throttle=%d", throttle))
-	stats = append(stats, fmt.Sprintf("cpus/goroutines=%d/%d", runtime.NumCPU(), runtime.NumGoroutine()))
+	i := 0
+	for k, v := range s.statsData(connected, newConnections, throttle) {
+		if i > 0 {
+			sb.WriteString(protocol.FieldDelimiter)
+		}
+		sb.WriteString(k)
+		sb.WriteByte('=')
+		sb.WriteString(fmt.Sprintf("%v", v))
+		i++
+	}
 
-	return strings.Join(stats, "|")
+	return sb.String()
 }
 
 func (s *stats) numConnected() int {
diff --git a/internal/color/brush/brush.go b/internal/color/brush/brush.go
index 524829b..82fa410 100644
--- a/internal/color/brush/brush.go
+++ b/internal/color/brush/brush.go
@@ -5,6 +5,7 @@ import (
 
 	"github.com/mimecast/dtail/internal/color"
 	"github.com/mimecast/dtail/internal/config"
+	"github.com/mimecast/dtail/internal/io/pool"
 	"github.com/mimecast/dtail/internal/protocol"
 )
 
@@ -171,20 +172,21 @@ func paintServer(sb *strings.Builder, line string) {
 
 // Colorfy a given line based on the line's content.
 func Colorfy(line string) string {
-	sb := strings.Builder{}
+	sb := pool.BuilderBuffer.Get().(*strings.Builder)
+	defer pool.RecycleBuilderBuffer(sb)
 
 	switch {
 	case strings.HasPrefix(line, "REMOTE"):
-		paintRemote(&sb, line)
+		paintRemote(sb, line)
 
 	case strings.HasPrefix(line, "CLIENT"):
-		paintClient(&sb, line)
+		paintClient(sb, line)
 
 	case strings.HasPrefix(line, "SERVER"):
-		paintServer(&sb, line)
+		paintServer(sb, line)
 
 	default:
-		color.PaintWithAttr(&sb, line,
+		color.PaintWithAttr(sb, line,
 			color.FgDefault,
 			color.BgDefault,
 			color.AttrNone)
diff --git a/internal/io/logger/logger.go b/internal/io/logger/logger.go
index 3a3935d..6890201 100644
--- a/internal/io/logger/logger.go
+++ b/internal/io/logger/logger.go
@@ -14,6 +14,8 @@ import (
 
 	"github.com/mimecast/dtail/internal/color/brush"
 	"github.com/mimecast/dtail/internal/config"
+	"github.com/mimecast/dtail/internal/io/pool"
+	"github.com/mimecast/dtail/internal/protocol"
 )
 
 const (
@@ -132,6 +134,24 @@ func Info(args ...interface{}) string {
 	return log(clientStr, infoStr, args)
 }
 
+// Mapreduce message logging.
+func Mapreduce(table string, data map[string]interface{}) string {
+	args := make([]interface{}, len(data)+1)
+
+	args[0] = fmt.Sprintf("MAPREDUCE:%s", strings.ToUpper(table))
+	i := 1
+	for k, v := range data {
+		args[i] = fmt.Sprintf("%s=%v", k, v)
+		i++
+	}
+
+	if Mode.Server {
+		return log(serverStr, infoStr, args)
+	}
+
+	return log(clientStr, infoStr, args)
+}
+
 // Warn message logging.
 func Warn(args ...interface{}) string {
 	if !Mode.Quiet {
@@ -230,24 +250,28 @@ func log(what string, severity string, args []interface{}) string {
 		return ""
 	}
 
-	messages := []string{}
+	sb := pool.BuilderBuffer.Get().(*strings.Builder)
+
+	for i, arg := range args {
+		if i > 0 {
+			sb.WriteString(protocol.FieldDelimiter)
+		}
 
-	for _, arg := range args {
 		switch v := arg.(type) {
 		case string:
-			messages = append(messages, v)
+			sb.WriteString(v)
 		case int:
-			messages = append(messages, fmt.Sprintf("%d", v))
+			sb.WriteString(fmt.Sprintf("%d", v))
 		case error:
-			messages = append(messages, v.Error())
+			sb.WriteString(v.Error())
 		default:
-			messages = append(messages, fmt.Sprintf("%v", v))
+			sb.WriteString(fmt.Sprintf("%v", v))
 		}
 	}
 
-	message := strings.Join(messages, "|")
+	message := sb.String()
+	pool.RecycleBuilderBuffer(sb)
 	write(what, severity, message)
-
 	return fmt.Sprintf("%s|%s", severity, message)
 }
 
diff --git a/internal/io/pool/builder.go b/internal/io/pool/builder.go
new file mode 100644
index 0000000..c9dc221
--- /dev/null
+++ b/internal/io/pool/builder.go
@@ -0,0 +1,18 @@
+package pool
+
+import (
+	"strings"
+	"sync"
+)
+
+var BuilderBuffer = sync.Pool{
+	New: func() interface{} {
+		sb := strings.Builder{}
+		return &sb
+	},
+}
+
+func RecycleBuilderBuffer(sb *strings.Builder) {
+	sb.Reset()
+	BuilderBuffer.Put(sb)
+}
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")
 	}
 }
diff --git a/internal/server/stats.go b/internal/server/stats.go
index ac579ad..3e8c71d 100644
--- a/internal/server/stats.go
+++ b/internal/server/stats.go
@@ -50,10 +50,14 @@ func (s *stats) logServerStats() {
 	s.mutex.Lock()
 	defer s.mutex.Unlock()
 
-	currentConnections := fmt.Sprintf("currentConnections=%d", s.currentConnections)
-	lifetimeConnections := fmt.Sprintf("lifetimeConnections=%d", s.lifetimeConnections)
-	goroutines := fmt.Sprintf("goroutines=%d", runtime.NumGoroutine())
-	logger.Info("stats", currentConnections, lifetimeConnections, goroutines)
+	data := make(map[string]interface{})
+	data["currentConnections"] = s.currentConnections
+	data["lifetimeConnections"] = s.lifetimeConnections
+	data["goroutines"] = runtime.NumGoroutine()
+	data["cgocalls"] = runtime.NumCgoCall()
+	data["cpu"] = runtime.NumCPU()
+
+	logger.Mapreduce("STATS", data)
 }
 
 func (s *stats) serverLimitExceeded() error {
diff --git a/internal/version/version.go b/internal/version/version.go
index 693192f..a54b560 100644
--- a/internal/version/version.go
+++ b/internal/version/version.go
@@ -13,7 +13,7 @@ const (
 	// Name of DTail.
 	Name string = "DTail"
 	// Version of DTail.
-	Version string = "3.2.0"
+	Version string = "4.0.0-RC1"
 	// Additional information for DTail
 	Additional string = "Have a lot of fun!"
 )