From 799b9b69ba08b898e13026b7ecab9f9f58580a82 Mon Sep 17 00:00:00 2001 From: Paul Buetow Date: Tue, 8 Dec 2020 14:49:41 +0000 Subject: merge develop --- internal/mapr/server/aggregate.go | 81 +++++++++++++++++++++------------------ 1 file changed, 44 insertions(+), 37 deletions(-) (limited to 'internal/mapr') diff --git a/internal/mapr/server/aggregate.go b/internal/mapr/server/aggregate.go index 1028943..28bb074 100644 --- a/internal/mapr/server/aggregate.go +++ b/internal/mapr/server/aggregate.go @@ -6,6 +6,7 @@ import ( "strings" "time" + "github.com/mimecast/dtail/internal" "github.com/mimecast/dtail/internal/config" "github.com/mimecast/dtail/internal/io/line" "github.com/mimecast/dtail/internal/io/logger" @@ -15,6 +16,7 @@ import ( // Aggregate is for aggregating mapreduce data on the DTail server side. type Aggregate struct { + done *internal.Done // Log lines to process (parsing MAPREDUCE lines). Lines chan line.Line // Hostname of the current server (used to populate $hostname field). @@ -23,12 +25,12 @@ type Aggregate struct { serialize chan struct{} // Signals to flush data. flush chan struct{} + // Signals that data has been flushed + flushed chan struct{} // The mapr query query *mapr.Query // The mapr log format parser parser *logformat.Parser - cancel context.CancelFunc - ctx context.Context } // NewAggregate return a new server side aggregator. @@ -64,56 +66,64 @@ func NewAggregate(queryStr string) (*Aggregate, error) { } } - ctx, cancel := context.WithCancel(context.Background()) - a := Aggregate{ + done: internal.NewDone(), Lines: make(chan line.Line, 100), serialize: make(chan struct{}), flush: make(chan struct{}), + flushed: make(chan struct{}), hostname: s[0], query: query, parser: logParser, - ctx: ctx, - cancel: cancel, } return &a, nil } +// Shutdown the aggregation engine. +func (a *Aggregate) Shutdown() { + a.Flush() + a.done.Shutdown() +} + // Start an aggregation. func (a *Aggregate) Start(ctx context.Context, maprLines chan<- string) { - defer a.cancel() - fieldsCh := a.linesToFields(ctx) + myCtx, cancel := context.WithCancel(ctx) + defer cancel() + + go func() { + select { + case <-myCtx.Done(): + a.done.Shutdown() + case <-a.done.Done(): + cancel() + } + }() + + fieldsCh := a.makeFields(myCtx) // Add fields (e.g. via 'set' clause) if len(a.query.Set) > 0 { - fieldsCh = a.addMoreFields(ctx, fieldsCh) + fieldsCh = a.addFields(myCtx, fieldsCh) } - go a.fieldsToMaprLines(ctx, fieldsCh, maprLines) - a.periodicAggregateTimer(ctx) + go a.aggregateTimer(myCtx) + a.makeMaprLines(myCtx, fieldsCh, maprLines) } -// Cancel the aggregation. -func (a *Aggregate) Cancel() { - a.cancel() -} - -func (a *Aggregate) periodicAggregateTimer(ctx context.Context) { +func (a *Aggregate) aggregateTimer(ctx context.Context) { for { select { case <-time.After(a.query.Interval): a.Serialize(ctx) case <-ctx.Done(): return - case <-a.ctx.Done(): - return } } } -func (a *Aggregate) linesToFields(ctx context.Context) <-chan map[string]string { +func (a *Aggregate) makeFields(ctx context.Context) <-chan map[string]string { ch := make(chan map[string]string) go func() { @@ -144,8 +154,6 @@ func (a *Aggregate) linesToFields(ctx context.Context) <-chan map[string]string } case <-ctx.Done(): return - case <-a.ctx.Done(): - return } } }() @@ -153,14 +161,14 @@ func (a *Aggregate) linesToFields(ctx context.Context) <-chan map[string]string return ch } -func (a *Aggregate) addMoreFields(ctx context.Context, fieldsCh <-chan map[string]string) <-chan map[string]string { +func (a *Aggregate) addFields(ctx context.Context, fieldsCh <-chan map[string]string) <-chan map[string]string { ch := make(chan map[string]string) go func() { defer close(ch) for { - // fieldsCh will be closed via 'linesToFields' if ctx is done + // fieldsCh will be closed via 'makeFields' if ctx is done fields, ok := <-fieldsCh if !ok { return @@ -179,7 +187,7 @@ func (a *Aggregate) addMoreFields(ctx context.Context, fieldsCh <-chan map[strin return ch } -func (a *Aggregate) fieldsToMaprLines(ctx context.Context, fieldsCh <-chan map[string]string, maprLines chan<- string) { +func (a *Aggregate) makeMaprLines(ctx context.Context, fieldsCh <-chan map[string]string, maprLines chan<- string) { group := mapr.NewGroupSet() serialize := func() { @@ -200,18 +208,10 @@ func (a *Aggregate) fieldsToMaprLines(ctx context.Context, fieldsCh <-chan map[s case <-a.serialize: serialize() case <-a.flush: - logger.Info("Flushing mapreduce result") serialize() - a.flush <- struct{}{} - logger.Info("Done flushing mapreduce result") + a.flushed <- struct{}{} case <-ctx.Done(): return - case <-a.ctx.Done(): - logger.Info("Flushing mapreduce result") - serialize() - a.flush <- struct{}{} - logger.Info("Done flushing mapreduce result") - return } } } @@ -254,6 +254,8 @@ func (a *Aggregate) aggregate(group *mapr.GroupSet, fields map[string]string) { func (a *Aggregate) Serialize(ctx context.Context) { select { case a.serialize <- struct{}{}: + case <-time.After(time.Minute): + logger.Warn("Starting to serialize mapredice data takes over a minute") case <-ctx.Done(): } } @@ -261,15 +263,20 @@ func (a *Aggregate) Serialize(ctx context.Context) { // Flush all data. func (a *Aggregate) Flush() { select { - case <-a.ctx.Done(): - return case a.flush <- struct{}{}: + logger.Info("Flushing mapreduce data") case <-time.After(time.Minute): + logger.Warn("Starting to flush mapreduce data takes over a minute") + return + case <-a.done.Done(): return } select { - case <-a.flush: + case <-a.flushed: + logger.Info("Done flushing") case <-time.After(time.Minute): + logger.Warn("Waiting for data to be flushed takes over a minute") + case <-a.done.Done(): } } -- cgit v1.2.3 From 44041db5db3d630413067606ae2402d10a61774e Mon Sep 17 00:00:00 2001 From: Paul Buetow Date: Thu, 10 Dec 2020 14:37:19 +0000 Subject: add hasprefix, nhasprefix, hassuffix, nhassuffix operation support to where clause --- internal/mapr/wherecondition.go | 22 ++++++++++++++++++++++ 1 file changed, 22 insertions(+) (limited to 'internal/mapr') diff --git a/internal/mapr/wherecondition.go b/internal/mapr/wherecondition.go index ff1b489..70e9c32 100644 --- a/internal/mapr/wherecondition.go +++ b/internal/mapr/wherecondition.go @@ -19,6 +19,10 @@ const ( StringNe QueryOperation = iota StringContains QueryOperation = iota StringNotContains QueryOperation = iota + StringHasPrefix QueryOperation = iota + StringNotHasPrefix QueryOperation = iota + StringHasSuffix QueryOperation = iota + StringNotHasSuffix QueryOperation = iota FloatOperation QueryOperation = iota FloatEq QueryOperation = iota FloatNe QueryOperation = iota @@ -78,7 +82,17 @@ func makeWhereConditions(tokens []token) (where []whereCondition, err error) { case "contains": wc.Operation = StringContains case "lacks": + fallthrough + case "ncontains": wc.Operation = StringNotContains + case "hasprefix": + wc.Operation = StringHasPrefix + case "nhasprefix": + wc.Operation = StringNotHasPrefix + case "hassuffix": + wc.Operation = StringHasSuffix + case "nhassuffix": + wc.Operation = StringNotHasSuffix default: return wc, nil, errors.New(invalidQuery + "Unknown operation in 'where' clause: " + whereOp) } @@ -169,6 +183,14 @@ func (wc *whereCondition) stringClause(lValue string, rValue string) bool { return strings.Contains(lValue, rValue) case StringNotContains: return !strings.Contains(lValue, rValue) + case StringHasPrefix: + return strings.HasPrefix(lValue, rValue) + case StringNotHasPrefix: + return !strings.HasPrefix(lValue, rValue) + case StringHasSuffix: + return strings.HasSuffix(lValue, rValue) + case StringNotHasSuffix: + return !strings.HasSuffix(lValue, rValue) default: logger.Error("Unknown string operation", lValue, wc.Operation, rValue) } -- cgit v1.2.3 From ab676c2b484225ed22765b23d8f0545088ecd610 Mon Sep 17 00:00:00 2001 From: Paul Buetow Date: Sat, 26 Dec 2020 10:48:51 +0000 Subject: code cleanup and minor refactorings --- internal/mapr/funcs/function.go | 1 + internal/mapr/query.go | 6 ------ internal/mapr/selectcondition.go | 1 + internal/mapr/token.go | 3 +++ internal/mapr/wherecondition.go | 2 ++ 5 files changed, 7 insertions(+), 6 deletions(-) (limited to 'internal/mapr') diff --git a/internal/mapr/funcs/function.go b/internal/mapr/funcs/function.go index 52aaa98..1a89c3a 100644 --- a/internal/mapr/funcs/function.go +++ b/internal/mapr/funcs/function.go @@ -12,6 +12,7 @@ type CallbackFunc func(text string) string type Function struct { // Name of the callback function Name string + // The Go-callback function to call for this DTail function. call CallbackFunc } diff --git a/internal/mapr/query.go b/internal/mapr/query.go index 7f6b63c..01852da 100644 --- a/internal/mapr/query.go +++ b/internal/mapr/query.go @@ -177,12 +177,6 @@ func (q *Query) parse(tokens []token) error { } } - // Comment out for empty table support, which is "all" log lines. - /* - if q.Table == "" { - return errors.New(invalidQuery + "Empty table specified in 'from' clause") - } - */ if len(q.Select) < 1 { return errors.New(invalidQuery + "Expected at least one field in 'select' clause but got none") } diff --git a/internal/mapr/selectcondition.go b/internal/mapr/selectcondition.go index 1882b7e..d6aa0d4 100644 --- a/internal/mapr/selectcondition.go +++ b/internal/mapr/selectcondition.go @@ -92,5 +92,6 @@ func makeSelectConditions(tokens []token) ([]selectCondition, error) { } sel = append(sel, sc) } + return sel, nil } diff --git a/internal/mapr/token.go b/internal/mapr/token.go index d337bd2..8972188 100644 --- a/internal/mapr/token.go +++ b/internal/mapr/token.go @@ -22,6 +22,7 @@ func (t token) isKeyword() bool { return true } } + return false } @@ -94,6 +95,7 @@ func tokensConsumeStr(tokens []token) ([]token, []string) { for _, token := range found { strings = append(strings, token.str) } + return tokens, strings } @@ -104,5 +106,6 @@ func tokensConsumeOptional(tokens []token, optional string) []token { if strings.ToLower(tokens[0].str) == strings.ToLower(optional) { return tokens[1:] } + return tokens } diff --git a/internal/mapr/wherecondition.go b/internal/mapr/wherecondition.go index 70e9c32..7a60dba 100644 --- a/internal/mapr/wherecondition.go +++ b/internal/mapr/wherecondition.go @@ -170,6 +170,7 @@ func (wc *whereCondition) floatClause(lValue float64, rValue float64) bool { default: logger.Error("Unknown float operation", lValue, wc.Operation, rValue) } + return false } @@ -194,5 +195,6 @@ func (wc *whereCondition) stringClause(lValue string, rValue string) bool { default: logger.Error("Unknown string operation", lValue, wc.Operation, rValue) } + return false } -- cgit v1.2.3