From c5a0ba7d29da7effa0ae18bffa10fc0be359b8e7 Mon Sep 17 00:00:00 2001 From: Paul Buetow Date: Fri, 3 Jul 2020 14:13:13 +0100 Subject: bump up version to 3.0.0. can run continuous background mapreduce queries, useful for log file monitorig for example. breaking protocol change which allows to mapreduce aggreate messages containing the default field separator |. add of more unit tests. add logformat mapreduce query keyword. add set mapreduce clause support and support to evaluate built-in functions such as md5sum() and maskdigits(). --- internal/clients/handlers/maprhandler.go | 7 +------ internal/clients/maprclient.go | 10 ++++++++-- 2 files changed, 9 insertions(+), 8 deletions(-) (limited to 'internal/clients') diff --git a/internal/clients/handlers/maprhandler.go b/internal/clients/handlers/maprhandler.go index 44daf7d..b908f3b 100644 --- a/internal/clients/handlers/maprhandler.go +++ b/internal/clients/handlers/maprhandler.go @@ -59,16 +59,11 @@ func (h *MaprHandler) Write(p []byte) (n int, err error) { // related data. func (h *MaprHandler) handleAggregateMessage(message string) { h.count++ - parts := strings.Split(message, "|") + parts := strings.Split(message, "➔") // Index 0 contains 'AGGREGATE', 1 contains server host. // Aggregation data begins from index 2. logger.Debug("Received aggregate data", h.server, h.count, parts) - /* - for k, v := range parts { - logger.Debug(k, v) - } - */ h.aggregate.Aggregate(parts[2:]) logger.Debug("Aggregated aggregate data", h.server, h.count) } diff --git a/internal/clients/maprclient.go b/internal/clients/maprclient.go index e28dadb..c6c341b 100644 --- a/internal/clients/maprclient.go +++ b/internal/clients/maprclient.go @@ -14,11 +14,15 @@ import ( "github.com/mimecast/dtail/internal/omode" ) +// MaprClientMode determines whether to use cumulative mode or not. type MaprClientMode int const ( - DefaultMode MaprClientMode = iota - CumulativeMode MaprClientMode = iota + // DefaultMode behaviour + DefaultMode MaprClientMode = iota + // CumulativeMode means results are added to prev interval + CumulativeMode MaprClientMode = iota + // NonCumulativeMode means results are from 0 for each interval NonCumulativeMode MaprClientMode = iota ) @@ -60,6 +64,8 @@ func NewMaprClient(args Args, queryStr string, maprClientMode MaprClientMode) (* cumulative = args.Mode == omode.MapClient || query.HasOutfile() } + logger.Debug("Cumulative mapreduce mode?", cumulative) + c := MaprClient{ baseClient: baseClient{ Args: args, -- cgit v1.2.3