diff options
| -rw-r--r-- | Makefile | 2 | ||||
| -rw-r--r-- | TODO.md | 6 | ||||
| -rw-r--r-- | docker/Makefile | 6 | ||||
| -rw-r--r-- | internal/clients/baseclient.go | 1 | ||||
| -rw-r--r-- | internal/clients/handlers/basehandler.go | 29 | ||||
| -rw-r--r-- | internal/clients/handlers/healthhandler.go | 11 | ||||
| -rw-r--r-- | internal/clients/handlers/maprhandler.go | 42 | ||||
| -rw-r--r-- | internal/clients/maprclient.go | 31 | ||||
| -rw-r--r-- | internal/color/paint.go | 13 | ||||
| -rw-r--r-- | internal/config/client.go | 48 | ||||
| -rw-r--r-- | internal/io/fs/readfile.go | 5 | ||||
| -rw-r--r-- | internal/mapr/aggregateset.go | 17 | ||||
| -rw-r--r-- | internal/mapr/client/aggregate.go | 18 | ||||
| -rw-r--r-- | internal/mapr/globalgroupset.go | 1 | ||||
| -rw-r--r-- | internal/mapr/groupset.go | 258 | ||||
| -rw-r--r-- | internal/mapr/logformat/default.go | 28 | ||||
| -rw-r--r-- | internal/mapr/logformat/default_test.go | 15 | ||||
| -rw-r--r-- | internal/mapr/logformat/generickv.go | 31 | ||||
| -rw-r--r-- | internal/mapr/logformat/parser.go | 2 | ||||
| -rw-r--r-- | internal/mapr/server/aggregate.go | 17 | ||||
| -rw-r--r-- | internal/protocol/protocol.go | 11 | ||||
| -rw-r--r-- | internal/server/handlers/controlhandler.go | 3 | ||||
| -rw-r--r-- | internal/server/handlers/serverhandler.go | 118 |
23 files changed, 478 insertions, 235 deletions
@@ -1,5 +1,5 @@ GO ?= go -all: clean test build +all: test build build: dserver dcat dgrep dmap dtail dserver: ifndef USE_ACL @@ -9,11 +9,11 @@ This is a loose list of what to do. Maybe for the next releae or maybe for a lat [x] Have different color conf sections (by REMOTE, CLIENT, SERVER) [x] Paint ^CLIENT messages (e.g. use yellow backgrounds here) [x] Paint ^SERVER messages (e.g. use cyan backgrounds here) -[ ] Adjust dmap with color schemas +[x] Adjust dmap with color schemas [ ] Fix JSONSchema for the colors [ ] Implement Benchmark cat-ing a file and compare to prev version. -[x] Client 4.x should print a warning when trying to connect to a 3.x server. +[?] 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 +[ ] Fix dmap so that it always reads to the end of file diff --git a/docker/Makefile b/docker/Makefile index 910b23c..28d458a 100644 --- a/docker/Makefile +++ b/docker/Makefile @@ -10,8 +10,12 @@ spinup: spindown: ./spindown.sh 10 dtail: - ../dtail --servers serverlist.txt --files '/var/log/dserver/*,/does/not/exist' --trustAllHosts --debug + ../dtail --servers serverlist.txt --files '/var/log/dserver/*' --trustAllHosts --debug +dgrep: + ../dgrep --servers serverlist.txt --files '/var/log/dserver/*' --regex MAPREDUCE --trustAllHosts dcat: ../dcat --servers serverlist.txt --files '/etc/passwd' --trustAllHosts --debug +dmap: + ../dmap --servers serverlist.txt --files '/var/log/dserver/*' --trustAllHosts --query 'from stats select avg(goroutines),max(goroutines),min(goroutines),last(goroutines),count($$hostname),$$hostname group by $$hostname order by avg(goroutines)' spinup1: docker run -p 2222:2222 dserver:develop diff --git a/internal/clients/baseclient.go b/internal/clients/baseclient.go index f20156f..de0c101 100644 --- a/internal/clients/baseclient.go +++ b/internal/clients/baseclient.go @@ -71,6 +71,7 @@ func (c *baseClient) Start(ctx context.Context, statsCh <-chan string) (status i go c.hostKeyCallback.PromptAddHosts(ctx) // Print client stats every time something on statsCh is recieved. go c.stats.Start(ctx, c.throttleCh, statsCh, c.Args.Quiet) + // Keep count of active connections active := make(chan struct{}, len(c.connections)) diff --git a/internal/clients/handlers/basehandler.go b/internal/clients/handlers/basehandler.go index 63ceaac..51f33c1 100644 --- a/internal/clients/handlers/basehandler.go +++ b/internal/clients/handlers/basehandler.go @@ -1,6 +1,7 @@ package handlers import ( + "bytes" "encoding/base64" "fmt" "io" @@ -17,7 +18,7 @@ type baseHandler struct { server string shellStarted bool commands chan string - receiveBuf []byte + receiveBuf bytes.Buffer status int } @@ -56,14 +57,23 @@ func (h *baseHandler) SendMessage(command string) error { // Read data from the dtail server via Writer interface. func (h *baseHandler) Write(p []byte) (n int, err error) { for _, b := range p { - if b == protocol.MessageDelimiter || b == '\n' { - if len(h.receiveBuf) == 0 { + switch b { + /* + // TODO: Next DTail version make it so that '\n' gets ignored. For now + // leave it for compatibility with older DTail server + ability to display + // the protocol mismatch warn message. + case '\n' { + continue + */ + case '\n', protocol.MessageDelimiter: + message := h.receiveBuf.String() + if len(message) == 0 { continue } - message := string(h.receiveBuf) h.handleMessageType(message) - } else { - h.receiveBuf = append(h.receiveBuf, b) + h.receiveBuf.Reset() + default: + h.receiveBuf.WriteByte(b) } } @@ -78,25 +88,22 @@ func (h *baseHandler) Read(p []byte) (n int, err error) { case <-h.Done(): return 0, io.EOF } - return } // Handle various message types. func (h *baseHandler) handleMessageType(message string) { - if len(h.receiveBuf) == 0 { + if len(message) == 0 { return } // Hidden server commands starti with a dot "." - if h.receiveBuf[0] == '.' { + if message[0] == '.' { h.handleHiddenMessage(message) - h.receiveBuf = h.receiveBuf[:0] return } logger.Raw(message) - h.receiveBuf = h.receiveBuf[:0] } // Handle messages received from server which are not meant to be displayed diff --git a/internal/clients/handlers/healthhandler.go b/internal/clients/handlers/healthhandler.go index 213748c..eca0348 100644 --- a/internal/clients/handlers/healthhandler.go +++ b/internal/clients/handlers/healthhandler.go @@ -1,6 +1,7 @@ package handlers import ( + "bytes" "errors" "fmt" "time" @@ -13,7 +14,7 @@ import ( type HealthHandler struct { done *internal.Done // Buffer of incoming data from server. - receiveBuf []byte + receiveBuf bytes.Buffer // To send commands to the server. commands chan string // To receive messages from the server. @@ -72,10 +73,10 @@ func (h *HealthHandler) SendMessage(command string) error { // Server writes byte stream to client. func (h *HealthHandler) Write(p []byte) (n int, err error) { for _, b := range p { - h.receiveBuf = append(h.receiveBuf, b) - if b == protocol.MessageDelimiter { // '\n' { - h.receive <- string(h.receiveBuf) - h.receiveBuf = h.receiveBuf[:0] + h.receiveBuf.WriteByte(b) + if b == protocol.MessageDelimiter { + h.receive <- h.receiveBuf.String() + h.receiveBuf.Reset() } } diff --git a/internal/clients/handlers/maprhandler.go b/internal/clients/handlers/maprhandler.go index afad507..65b1454 100644 --- a/internal/clients/handlers/maprhandler.go +++ b/internal/clients/handlers/maprhandler.go @@ -15,7 +15,6 @@ type MaprHandler struct { baseHandler aggregate *client.Aggregate query *mapr.Query - count uint64 } // NewMaprHandler returns a new mapreduce client handler. @@ -36,19 +35,20 @@ func NewMaprHandler(server string, query *mapr.Query, globalGroup *mapr.GlobalGr // Read data from the dtail server via Writer interface. func (h *MaprHandler) Write(p []byte) (n int, err error) { for _, b := range p { - h.baseHandler.receiveBuf = append(h.baseHandler.receiveBuf, b) - if b == protocol.MessageDelimiter { // '\n' { - if len(h.baseHandler.receiveBuf) == 0 { - continue + switch b { + case '\n': + continue + case protocol.MessageDelimiter: + message := h.baseHandler.receiveBuf.String() + logger.Debug(message) + if message[0] == 'A' { + h.handleAggregateMessage(message) + } else { + h.baseHandler.handleMessageType(message) } - message := string(h.baseHandler.receiveBuf) - - if h.baseHandler.receiveBuf[0] == 'A' { - h.handleAggregateMessage(strings.TrimSpace(message)) - h.baseHandler.receiveBuf = h.baseHandler.receiveBuf[:0] - continue - } - h.baseHandler.handleMessageType(message) + h.baseHandler.receiveBuf.Reset() + default: + h.baseHandler.receiveBuf.WriteByte(b) } } @@ -58,12 +58,12 @@ func (h *MaprHandler) Write(p []byte) (n int, err error) { // Handle a message received from server including mapr aggregation // related data. func (h *MaprHandler) handleAggregateMessage(message string) { - h.count++ - parts := strings.Split(message, protocol.AggregateDelimiter) - - // Index 0 contains 'AGGREGATE', 1 contains server host. - // Aggregation data begins from index 2. - logger.Debug("Received aggregate data", h.server, h.count, parts) - h.aggregate.Aggregate(parts[2:]) - logger.Debug("Aggregated aggregate data", h.server, h.count) + parts := strings.SplitN(message, protocol.FieldDelimiter, 3) + if len(parts) != 3 { + logger.Error("Unable to aggregate data", h.server, message, parts, len(parts), "expected 3 parts") + return + } + if err := h.aggregate.Aggregate(parts[2]); err != nil { + logger.Error("Unable to aggregate data", h.server, message, err) + } } diff --git a/internal/clients/maprclient.go b/internal/clients/maprclient.go index 77b674b..cab9a6c 100644 --- a/internal/clients/maprclient.go +++ b/internal/clients/maprclient.go @@ -9,6 +9,8 @@ import ( "time" "github.com/mimecast/dtail/internal/clients/handlers" + "github.com/mimecast/dtail/internal/color" + "github.com/mimecast/dtail/internal/config" "github.com/mimecast/dtail/internal/io/logger" "github.com/mimecast/dtail/internal/mapr" "github.com/mimecast/dtail/internal/omode" @@ -37,6 +39,8 @@ type MaprClient struct { query *mapr.Query // Additative result or new result every interval run? cumulative bool + // The last result string received + lastResult string } // NewMaprClient returns a new mapreduce client. @@ -154,24 +158,37 @@ func (c *MaprClient) reportResults() { func (c *MaprClient) printResults() { var result string var err error - var numLines int + var numRows int if c.cumulative { - result, numLines, err = c.globalGroup.Result(c.query) + result, numRows, err = c.globalGroup.Result(c.query) } else { - result, numLines, err = c.globalGroup.SwapOut().Result(c.query) + result, numRows, err = c.globalGroup.SwapOut().Result(c.query) } + if err != nil { logger.FatalExit(err) } - if numLines == 0 { - logger.Warn("Empty result set this time...") + if result == c.lastResult { + logger.Debug("Result hasn't changed compared to last time...") return } + c.lastResult = result - //logger.Raw(fmt.Sprintf("%s\n", c.query.RawQuery)) - logger.Raw(c.query.RawQuery) + if numRows == 0 { + logger.Debug("Empty result set this time...") + return + } + + rawQuery := c.query.RawQuery + if config.Client.TermColorsEnable { + rawQuery = color.PaintStrWithAttr(rawQuery, + config.Client.TermColors.MaprTable.RawQueryFg, + config.Client.TermColors.MaprTable.RawQueryBg, + config.Client.TermColors.MaprTable.RawQueryAttr) + } + logger.Raw(rawQuery) logger.Raw(result) } diff --git a/internal/color/paint.go b/internal/color/paint.go index 5430acd..53c9abb 100644 --- a/internal/color/paint.go +++ b/internal/color/paint.go @@ -63,6 +63,19 @@ func PaintWithAttr(sb *strings.Builder, text string, fg FgColor, bg BgColor, att sb.WriteString(string(FgDefault)) } +// PaintWithAttrs is similar to PaintWithAttr, but it takes multiple text attributes. +func PaintWithAttrs(sb *strings.Builder, text string, fg FgColor, bg BgColor, attrs []Attribute) { + sb.WriteString(string(fg)) + sb.WriteString(string(bg)) + for _, attr := range attrs { + sb.WriteString(string(attr)) + } + sb.WriteString(text) + sb.WriteString(string(AttrReset)) + sb.WriteString(string(BgDefault)) + sb.WriteString(string(FgDefault)) +} + // ResetWithAttr resets background, foreground and attributes. func ResetWithAttr(sb *strings.Builder) { sb.WriteString(string(AttrReset)) diff --git a/internal/config/client.go b/internal/config/client.go index 3c2f7de..795c4a4 100644 --- a/internal/config/client.go +++ b/internal/config/client.go @@ -71,11 +71,32 @@ type commonTermColors struct { SeverityWarnFg color.FgColor } +type maprTableTermColors struct { + DataAttr color.Attribute + DataBg color.BgColor + DataFg color.FgColor + DelimiterAttr color.Attribute + DelimiterBg color.BgColor + DelimiterFg color.FgColor + HeaderAttr color.Attribute + HeaderBg color.BgColor + HeaderDelimiterAttr color.Attribute + HeaderDelimiterBg color.BgColor + HeaderDelimiterFg color.FgColor + HeaderFg color.FgColor + HeaderGroupKeyAttr color.Attribute + HeaderSortKeyAttr color.Attribute + RawQueryAttr color.Attribute + RawQueryBg color.BgColor + RawQueryFg color.FgColor +} + type termColors struct { - Remote remoteTermColors - Client clientTermColors - Server serverTermColors - Common commonTermColors + Remote remoteTermColors + Client clientTermColors + Server serverTermColors + Common commonTermColors + MaprTable maprTableTermColors } // ClientConfig represents a DTail client configuration (empty as of now as there @@ -155,6 +176,25 @@ func newDefaultClientConfig() *ClientConfig { SeverityWarnBg: color.BgBlack, SeverityWarnFg: color.FgWhite, }, + MaprTable: maprTableTermColors{ + DataAttr: color.AttrNone, + DataBg: color.BgBlue, + DataFg: color.FgWhite, + DelimiterAttr: color.AttrDim, + DelimiterBg: color.BgBlue, + DelimiterFg: color.FgWhite, + HeaderAttr: color.AttrBold, + HeaderBg: color.BgBlue, + HeaderFg: color.FgWhite, + HeaderDelimiterAttr: color.AttrDim, + HeaderDelimiterBg: color.BgBlue, + HeaderDelimiterFg: color.FgWhite, + HeaderSortKeyAttr: color.AttrUnderline, + HeaderGroupKeyAttr: color.AttrReverse, + RawQueryAttr: color.AttrDim, + RawQueryBg: color.BgBlack, + RawQueryFg: color.FgCyan, + }, }, } } diff --git a/internal/io/fs/readfile.go b/internal/io/fs/readfile.go index f2f672a..c0d44dd 100644 --- a/internal/io/fs/readfile.go +++ b/internal/io/fs/readfile.go @@ -16,7 +16,6 @@ import ( "github.com/mimecast/dtail/internal/io/line" "github.com/mimecast/dtail/internal/io/logger" "github.com/mimecast/dtail/internal/io/pool" - "github.com/mimecast/dtail/internal/protocol" "github.com/mimecast/dtail/internal/regex" "github.com/DataDog/zstd" @@ -187,7 +186,7 @@ func (f readFile) read(ctx context.Context, fd *os.File, rawLines chan *bytes.Bu time.Sleep(time.Millisecond * 100) continue } - message.WriteByte(protocol.MessageDelimiter) + message.WriteString("\n") select { case rawLines <- message: message = pool.BytesBuffer.Get().(*bytes.Buffer) @@ -202,7 +201,7 @@ func (f readFile) read(ctx context.Context, fd *os.File, rawLines chan *bytes.Bu f.serverMessages <- logger.Warn(f.filePath, "Long log line, splitting into multiple lines") warnedAboutLongLine = true } - message.WriteByte(protocol.MessageDelimiter) + message.WriteString("\n") select { case rawLines <- message: message = pool.BytesBuffer.Get().(*bytes.Buffer) diff --git a/internal/mapr/aggregateset.go b/internal/mapr/aggregateset.go index 029d87b..47f4925 100644 --- a/internal/mapr/aggregateset.go +++ b/internal/mapr/aggregateset.go @@ -6,6 +6,8 @@ import ( "strconv" "strings" + "github.com/mimecast/dtail/internal/io/logger" + "github.com/mimecast/dtail/internal/io/pool" "github.com/mimecast/dtail/internal/protocol" ) @@ -68,22 +70,25 @@ func (s *AggregateSet) Merge(query *Query, set *AggregateSet) error { // Serialize the aggregate set so it can be sent over the wire. func (s *AggregateSet) Serialize(ctx context.Context, groupKey string, ch chan<- string) { - //logger.Trace("Serialising mapr.AggregateSet", s) - var sb strings.Builder + logger.Trace("Serialising mapr.AggregateSet", s) + sb := pool.BuilderBuffer.Get().(*strings.Builder) + defer pool.RecycleBuilderBuffer(sb) sb.WriteString(groupKey) sb.WriteString(protocol.AggregateDelimiter) - sb.WriteString(fmt.Sprintf("%d%s", s.Samples, protocol.AggregateDelimiter)) + sb.WriteString(fmt.Sprintf("%d", s.Samples)) + sb.WriteString(protocol.AggregateDelimiter) for k, v := range s.FValues { sb.WriteString(k) - sb.WriteString("=") - sb.WriteString(fmt.Sprintf("%v%s", v, protocol.AggregateDelimiter)) + sb.WriteString(protocol.AggregateKVDelimiter) + sb.WriteString(fmt.Sprintf("%v", v)) + sb.WriteString(protocol.AggregateDelimiter) } for k, v := range s.SValues { sb.WriteString(k) - sb.WriteString("=") + sb.WriteString(protocol.AggregateKVDelimiter) sb.WriteString(v) sb.WriteString(protocol.AggregateDelimiter) } diff --git a/internal/mapr/client/aggregate.go b/internal/mapr/client/aggregate.go index 10b34d4..5cc09a1 100644 --- a/internal/mapr/client/aggregate.go +++ b/internal/mapr/client/aggregate.go @@ -1,11 +1,13 @@ package client import ( + "fmt" "strconv" "strings" "github.com/mimecast/dtail/internal/io/logger" "github.com/mimecast/dtail/internal/mapr" + "github.com/mimecast/dtail/internal/protocol" ) // Aggregate mapreduce data on the DTail client side. @@ -31,12 +33,18 @@ func NewAggregate(server string, query *mapr.Query, globalGroup *mapr.GlobalGrou } // Aggregate data from mapr log line into local (and global) group sets. -func (a *Aggregate) Aggregate(parts []string) { +func (a *Aggregate) Aggregate(message string) error { + parts := strings.Split(message, protocol.AggregateDelimiter) + if len(parts) < 4 { + return fmt.Errorf("aggregate message without any real data") + } + groupKey := parts[0] samples, err := strconv.Atoi(parts[1]) if err != nil { - logger.FatalExit("Unable to parse sample count", parts[1], err, parts) + return fmt.Errorf("unable to parse sample count '%s': %v", parts[1], err) } + fields := a.makeFields(parts[2:]) set := a.group.GetSet(groupKey) @@ -63,6 +71,8 @@ func (a *Aggregate) Aggregate(parts []string) { // Re-init local group (make it empty again). a.group.InitSet() } + + return nil } // Create a map of key-value pairs from a part list such as ["foo=bar", "bar=baz"]. @@ -70,8 +80,8 @@ func (a *Aggregate) makeFields(parts []string) map[string]string { fields := make(map[string]string, len(parts)) for _, part := range parts { - kv := strings.SplitN(part, "=", 2) - if len(kv) < 2 { + kv := strings.SplitN(part, protocol.AggregateKVDelimiter, 2) + if len(kv) != 2 { continue } fields[kv[0]] = kv[1] diff --git a/internal/mapr/globalgroupset.go b/internal/mapr/globalgroupset.go index cfab506..21bf990 100644 --- a/internal/mapr/globalgroupset.go +++ b/internal/mapr/globalgroupset.go @@ -48,6 +48,7 @@ func (g *GlobalGroupSet) MergeNoblock(query *Query, group *GroupSet) (bool, erro // Merge a group set into the global group set. func (g *GlobalGroupSet) merge(query *Query, group *GroupSet) error { + for groupKey, set := range group.sets { s := g.GetSet(groupKey) if err := s.Merge(query, set); err != nil { diff --git a/internal/mapr/groupset.go b/internal/mapr/groupset.go index 6ee2811..d29559a 100644 --- a/internal/mapr/groupset.go +++ b/internal/mapr/groupset.go @@ -4,13 +4,16 @@ import ( "context" "errors" "fmt" - "io/ioutil" "os" "sort" "strconv" "strings" + "github.com/mimecast/dtail/internal/color" + "github.com/mimecast/dtail/internal/config" "github.com/mimecast/dtail/internal/io/logger" + "github.com/mimecast/dtail/internal/io/pool" + "github.com/mimecast/dtail/internal/protocol" ) // GroupSet represents a map of aggregate sets. The group sets @@ -22,6 +25,14 @@ type GroupSet struct { sets map[string]*AggregateSet } +// Internal helper type +type result struct { + groupKey string + values []string + widths []int + orderBy float64 +} + // NewGroupSet returns a new empty group set. func NewGroupSet() *GroupSet { g := GroupSet{} @@ -58,17 +69,118 @@ func (g *GroupSet) Serialize(ctx context.Context, ch chan<- string) { // Result returns a nicely formated result of the query from the group set. func (g *GroupSet) Result(query *Query) (string, int, error) { - return g.limitedResult(query, query.Limit, "\t", " ", false) + rows, widths, err := g.result(query, true) + if err != nil { + return "", 0, err + } + + sb := pool.BuilderBuffer.Get().(*strings.Builder) + defer pool.RecycleBuilderBuffer(sb) + + // Generate header now + lastIndex := len(query.Select) - 1 + for i, sc := range query.Select { + format := fmt.Sprintf(" %%%ds ", widths[i]) + str := fmt.Sprintf(format, sc.FieldStorage) + if config.Client.TermColorsEnable { + attrs := []color.Attribute{config.Client.TermColors.MaprTable.HeaderAttr} + if sc.FieldStorage == query.OrderBy { + attrs = append(attrs, config.Client.TermColors.MaprTable.HeaderSortKeyAttr) + } + for _, groupBy := range query.GroupBy { + if sc.FieldStorage == groupBy { + attrs = append(attrs, config.Client.TermColors.MaprTable.HeaderGroupKeyAttr) + break + } + } + color.PaintWithAttrs(sb, str, + config.Client.TermColors.MaprTable.HeaderFg, + config.Client.TermColors.MaprTable.HeaderBg, + attrs) + } else { + sb.WriteString(str) + } + + if i == lastIndex { + continue + } + if config.Client.TermColorsEnable { + color.PaintWithAttr(sb, protocol.FieldDelimiter, + config.Client.TermColors.MaprTable.HeaderDelimiterFg, + config.Client.TermColors.MaprTable.HeaderDelimiterBg, + config.Client.TermColors.MaprTable.HeaderDelimiterAttr) + } else { + sb.WriteString(protocol.FieldDelimiter) + } + } + sb.WriteString("\n") + + for i := 0; i < len(query.Select); i++ { + str := fmt.Sprintf("-%s-", strings.Repeat("-", widths[i])) + if config.Client.TermColorsEnable { + color.PaintWithAttr(sb, str, + config.Client.TermColors.MaprTable.HeaderDelimiterFg, + config.Client.TermColors.MaprTable.HeaderDelimiterBg, + config.Client.TermColors.MaprTable.HeaderDelimiterAttr) + } else { + sb.WriteString(str) + } + if i == lastIndex { + continue + } + if config.Client.TermColorsEnable { + color.PaintWithAttr(sb, protocol.FieldDelimiter, + config.Client.TermColors.MaprTable.HeaderDelimiterFg, + config.Client.TermColors.MaprTable.HeaderDelimiterBg, + config.Client.TermColors.MaprTable.HeaderDelimiterAttr) + } else { + sb.WriteString(protocol.FieldDelimiter) + } + } + sb.WriteString("\n") + + // And now write the data + for i, r := range rows { + if i == query.Limit { + break + } + for j, value := range r.values { + format := fmt.Sprintf(" %%%ds ", widths[j]) + str := fmt.Sprintf(format, value) + if config.Client.TermColorsEnable { + color.PaintWithAttr(sb, str, + config.Client.TermColors.MaprTable.DataFg, + config.Client.TermColors.MaprTable.DataBg, + config.Client.TermColors.MaprTable.DataAttr) + } else { + sb.WriteString(str) + } + + if j == lastIndex { + continue + } + if config.Client.TermColorsEnable { + color.PaintWithAttr(sb, protocol.FieldDelimiter, + config.Client.TermColors.MaprTable.DelimiterFg, + config.Client.TermColors.MaprTable.DelimiterBg, + config.Client.TermColors.MaprTable.DelimiterAttr) + } else { + sb.WriteString(protocol.FieldDelimiter) + } + } + sb.WriteString("\n") + } + + return sb.String(), len(rows), nil } -// WriteResult writes the result to an outfile. +// WriteResult writes the result to an CSV outfile. func (g *GroupSet) WriteResult(query *Query) error { if !query.HasOutfile() { return errors.New("No outfile specified") } - // -1: Don't limit the result, include all data sets - result, _, err := g.limitedResult(query, -1, "", ",", true) + rows, _, err := g.result(query, false) if err != nil { return err } @@ -76,9 +188,37 @@ func (g *GroupSet) WriteResult(query *Query) error { logger.Info("Writing outfile", query.Outfile) tmpOutfile := fmt.Sprintf("%s.tmp", query.Outfile) - if err := ioutil.WriteFile(tmpOutfile, []byte(result), 0644); err != nil { + file, err := os.Create(tmpOutfile) + if err != nil { return err } + defer file.Close() + + // Generate header now + lastIndex := len(query.Select) - 1 + for i, sc := range query.Select { + file.WriteString(sc.FieldStorage) + if i == lastIndex { + continue + } + file.WriteString(protocol.CSVDelimiter) + } + file.WriteString("\n") + + // And now write the data + for i, r := range rows { + if i == query.Limit { + break + } + for j, value := range r.values { + file.WriteString(value) + if j == lastIndex { + continue + } + file.WriteString(protocol.CSVDelimiter) + } + file.WriteString("\n") + } if err := os.Rename(tmpOutfile, query.Outfile); err != nil { os.Remove(tmpOutfile) @@ -88,32 +228,22 @@ func (g *GroupSet) WriteResult(query *Query) error { return nil } -// Return a nicely formated result of the query from the group set. -func (g *GroupSet) limitedResult(query *Query, limit int, lineStarter, fieldSeparator string, addHeader bool) (string, int, error) { - type result struct { - groupKey string - resultStr string - orderBy float64 - } +// Return a sorted result slice of the query from the group set. +func (g *GroupSet) result(query *Query, gatherWidths bool) ([]result, []int, error) { + var rows []result + widths := make([]int, len(query.Select)) - var resultSlice []result + var valueStr string + var value float64 for groupKey, set := range g.sets { - var sb strings.Builder r := result{groupKey: groupKey} - lastIndex := len(query.Select) - 1 for i, sc := range query.Select { - storage := sc.FieldStorage - orderByThis := storage == query.OrderBy - switch sc.Operation { case Count: - value := set.FValues[storage] - sb.WriteString(fmt.Sprintf("%d", int(value))) - if orderByThis { - r.orderBy = value - } + value = set.FValues[sc.FieldStorage] + valueStr = fmt.Sprintf("%d", int(value)) case Len: fallthrough case Sum: @@ -121,74 +251,48 @@ func (g *GroupSet) limitedResult(query *Query, limit int, lineStarter, fieldSepa case Min: fallthrough case Max: - value := set.FValues[storage] - sb.WriteString(fmt.Sprintf("%f", value)) - if orderByThis { - r.orderBy = value - } + value = set.FValues[sc.FieldStorage] + valueStr = fmt.Sprintf("%f", value) case Last: - value := set.SValues[storage] - if orderByThis { - f, err := strconv.ParseFloat(value, 64) - if err == nil { - r.orderBy = f - } - } - sb.WriteString(value) + valueStr = set.SValues[sc.FieldStorage] + value, _ = strconv.ParseFloat(valueStr, 64) case Avg: - value := set.FValues[storage] / float64(set.Samples) - sb.WriteString(fmt.Sprintf("%f", value)) - if orderByThis { - r.orderBy = value - } + value = set.FValues[sc.FieldStorage] / float64(set.Samples) + valueStr = fmt.Sprintf("%f", value) default: - return "", 0, fmt.Errorf("Unknown aggregation method '%v'", sc.Operation) + return rows, widths, fmt.Errorf("Unknown aggregation method '%v'", sc.Operation) } - if i != lastIndex { - sb.WriteString(fieldSeparator) + + if sc.FieldStorage == query.OrderBy { + r.orderBy = value + } + r.values = append(r.values, valueStr) + + if !gatherWidths { + continue + } + if widths[i] < len(sc.FieldStorage) { + widths[i] = len(sc.FieldStorage) + } + if widths[i] < len(valueStr) { + widths[i] = len(valueStr) } } - r.resultStr = sb.String() - resultSlice = append(resultSlice, r) + rows = append(rows, r) } if query.OrderBy != "" { if query.ReverseOrder { - sort.SliceStable(resultSlice, func(i, j int) bool { - return resultSlice[i].orderBy < resultSlice[j].orderBy + sort.SliceStable(rows, func(i, j int) bool { + return rows[i].orderBy < rows[j].orderBy }) } else { - sort.SliceStable(resultSlice, func(i, j int) bool { - return resultSlice[i].orderBy > resultSlice[j].orderBy + sort.SliceStable(rows, func(i, j int) bool { + return rows[i].orderBy > rows[j].orderBy }) } } - var sb strings.Builder - - // Write header first - if addHeader { - lastIndex := len(query.Select) - 1 - sb.WriteString(lineStarter) - for i, sc := range query.Select { - sb.WriteString(sc.FieldStorage) - if i != lastIndex { - sb.WriteString(fieldSeparator) - } - } - sb.WriteString("\n") - } - - // And now write the data - for i, r := range resultSlice { - if i == limit { - break - } - sb.WriteString(lineStarter) - sb.WriteString(r.resultStr) - sb.WriteString("\n") - } - - return sb.String(), len(resultSlice), nil + return rows, widths, nil } diff --git a/internal/mapr/logformat/default.go b/internal/mapr/logformat/default.go index da976bd..c67137e 100644 --- a/internal/mapr/logformat/default.go +++ b/internal/mapr/logformat/default.go @@ -1,16 +1,22 @@ package logformat import ( + "fmt" "strings" - "github.com/mimecast/dtail/internal/io/logger" "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) { splitted := strings.Split(maprLine, protocol.FieldDelimiter) - fields := make(map[string]string, len(splitted)) + + if len(splitted) < 3 || !strings.HasPrefix(splitted[3], "MAPREDUCE:") || !strings.HasPrefix(splitted[0], "INFO") { + // Not a DTail mapreduce log line. + return nil, IgnoreFieldsErr + } + + fields := make(map[string]string, len(splitted)+8) fields["*"] = "*" fields["$line"] = maprLine @@ -19,20 +25,14 @@ func (p *Parser) MakeFieldsDEFAULT(maprLine string) (map[string]string, error) { fields["$timezone"] = p.timeZoneName fields["$timeoffset"] = p.timeZoneOffset - kvStart := 0 - // DTail mapreduce format - if len(splitted) > 3 && strings.HasPrefix(splitted[3], "MAPREDUCE:") { - fie |