Merge branch 'master' of github.com:influxdata/kapacitor

CHANGELOG.md

@@ -54,6 +54,7 @@ For example, let's say we want to store all data that triggered an alert in Infl
 - [#388](https://github.com/influxdata/kapacitor/issues/388): The duration of an alert is now tracked and exposed as part of the alert data as well as can be set as a field via `.durationField('duration')`.
 - [#486](https://github.com/influxdata/kapacitor/pull/486): Default config file location.
 - [#461](https://github.com/influxdata/kapacitor/pull/461): Make Alerta `event` property configurable.
+- [#491](https://github.com/influxdata/kapacitor/pull/491): BREAKING: Rewriting stateful expression in order to improve performance, the only breaking change is: short circuit evaluation for booleans - for example: ``lambda: "bool_value" && (count() > 100)`` if "bool_value" is false, we won't evaluate "count".
 
 ### Bugfixes
 

alert.go

@@ -19,7 +19,7 @@ import (
 	imodels "github.com/influxdata/influxdb/models"
 	"github.com/influxdata/kapacitor/models"
 	"github.com/influxdata/kapacitor/pipeline"
-	"github.com/influxdata/kapacitor/tick"
+	"github.com/influxdata/kapacitor/tick/stateful"
 )
 
 const (
@@ -97,7 +97,7 @@ type AlertNode struct {
 	a           *pipeline.AlertNode
 	endpoint    string
 	handlers    []AlertHandler
-	levels      []*tick.StatefulExpr
+	levels      []stateful.Expression
 	states      map[models.GroupID]*alertState
 	idTmpl      *text.Template
 	messageTmpl *text.Template
@@ -273,16 +273,32 @@ func newAlertNode(et *ExecutingTask, n *pipeline.AlertNode, l *log.Logger) (an *
 	}
 
 	// Parse level expressions
-	an.levels = make([]*tick.StatefulExpr, CritAlert+1)
+	an.levels = make([]stateful.Expression, CritAlert+1)
+
 	if n.Info != nil {
-		an.levels[InfoAlert] = tick.NewStatefulExpr(n.Info)
+		statefulExpression, expressionCompileError := stateful.NewExpression(n.Info)
+		if expressionCompileError != nil {
+			return nil, fmt.Errorf("Failed to compile stateful expression for info: %s", expressionCompileError)
+		}
+		an.levels[InfoAlert] = statefulExpression
 	}
+
 	if n.Warn != nil {
-		an.levels[WarnAlert] = tick.NewStatefulExpr(n.Warn)
+		statefulExpression, expressionCompileError := stateful.NewExpression(n.Warn)
+		if expressionCompileError != nil {
+			return nil, fmt.Errorf("Failed to compile stateful expression for warn: %s", expressionCompileError)
+		}
+		an.levels[WarnAlert] = statefulExpression
 	}
+
 	if n.Crit != nil {
-		an.levels[CritAlert] = tick.NewStatefulExpr(n.Crit)
+		statefulExpression, expressionCompileError := stateful.NewExpression(n.Crit)
+		if expressionCompileError != nil {
+			return nil, fmt.Errorf("Failed to compile stateful expression for crit: %s", expressionCompileError)
+		}
+		an.levels[CritAlert] = statefulExpression
 	}
+
 	// Setup states
 	if n.History < 2 {
 		n.History = 2

eval.go

@@ -2,13 +2,14 @@ package kapacitor
 
 import (
 	"errors"
+	"fmt"
 	"log"
 	"time"
 
 	"github.com/influxdata/kapacitor/expvar"
 	"github.com/influxdata/kapacitor/models"
 	"github.com/influxdata/kapacitor/pipeline"
-	"github.com/influxdata/kapacitor/tick"
+	"github.com/influxdata/kapacitor/tick/stateful"
 )
 
 const (
@@ -18,7 +19,7 @@ const (
 type EvalNode struct {
 	node
 	e           *pipeline.EvalNode
-	expressions []*tick.StatefulExpr
+	expressions []stateful.Expression
 	evalErrors  *expvar.Int
 }
 
@@ -32,9 +33,14 @@ func newEvalNode(et *ExecutingTask, n *pipeline.EvalNode, l *log.Logger) (*EvalN
 		e:    n,
 	}
 	// Create stateful expressions
-	en.expressions = make([]*tick.StatefulExpr, len(n.Expressions))
+	en.expressions = make([]stateful.Expression, len(n.Expressions))
 	for i, expr := range n.Expressions {
-		en.expressions[i] = tick.NewStatefulExpr(expr)
+		statefulExpr, err := stateful.NewExpression(expr)
+		if err != nil {
+			return nil, fmt.Errorf("Failed to compile %v expression: %v", i, err)
+		}
+
+		en.expressions[i] = statefulExpr
 	}
 
 	en.node.runF = en.runEval

expr.go

@@ -6,10 +6,11 @@ import (
 
 	"github.com/influxdata/kapacitor/models"
 	"github.com/influxdata/kapacitor/tick"
+	"github.com/influxdata/kapacitor/tick/stateful"
 )
 
-// Evaluate a given expression as a boolean predicate against a set of fields and tags
-func EvalPredicate(se *tick.StatefulExpr, now time.Time, fields models.Fields, tags models.Tags) (bool, error) {
+// EvalPredicate - Evaluate a given expression as a boolean predicate against a set of fields and tags
+func EvalPredicate(se stateful.Expression, now time.Time, fields models.Fields, tags models.Tags) (bool, error) {
 	vars, err := mergeFieldsAndTags(now, fields, tags)
 	if err != nil {
 		return false, err

stream.go

@@ -1,11 +1,12 @@
 package kapacitor
 
 import (
+	"fmt"
 	"log"
 
 	"github.com/influxdata/kapacitor/models"
 	"github.com/influxdata/kapacitor/pipeline"
-	"github.com/influxdata/kapacitor/tick"
+	"github.com/influxdata/kapacitor/tick/stateful"
 )
 
 type StreamNode struct {
@@ -38,7 +39,7 @@ func (s *StreamNode) runSourceStream([]byte) error {
 type FromNode struct {
 	node
 	s             *pipeline.FromNode
-	expression    *tick.StatefulExpr
+	expression    stateful.Expression
 	dimensions    []string
 	allDimensions bool
 	db            string
@@ -59,7 +60,12 @@ func newFromNode(et *ExecutingTask, n *pipeline.FromNode, l *log.Logger) (*FromN
 	sn.allDimensions, sn.dimensions = determineDimensions(n.Dimensions)
 
 	if n.Expression != nil {
-		sn.expression = tick.NewStatefulExpr(n.Expression)
+		expr, err := stateful.NewExpression(n.Expression)
+		if err != nil {
+			return nil, fmt.Errorf("Failed to compile from expression: %v", err)
+		}
+
+		sn.expression = expr
 	}
 
 	return sn, nil

tick/lex.go

@@ -161,24 +161,24 @@ func (t TokenType) String() string {
 		return "TRUE"
 	case t == TokenFalse:
 		return "FALSE"
-	case isExprOperator(t):
+	case IsExprOperator(t):
 		return operatorStr[t]
 	}
 	return fmt.Sprintf("%d", t)
 }
 
 // True if token type is an operator used in mathematical or boolean expressions.
-func isExprOperator(typ TokenType) bool {
+func IsExprOperator(typ TokenType) bool {
 	return typ > begin_tok_operator && typ < end_tok_operator
 }
 
 // True if token type is an operator used in mathematical expressions.
-func isMathOperator(typ TokenType) bool {
+func IsMathOperator(typ TokenType) bool {
 	return typ > begin_tok_operator_math && typ < end_tok_operator_math
 }
 
 // True if token type is an operator used in comparisons.
-func isCompOperator(typ TokenType) bool {
+func IsCompOperator(typ TokenType) bool {
 	return typ > begin_tok_operator_comp && typ < end_tok_operator_comp
 }
 

tick/parser.go

@@ -342,12 +342,12 @@ var precedence = [...]int{
 // https://en.wikipedia.org/wiki/Operator-precedence_parser#Pseudo-code
 func (p *parser) precedence(lhs Node, minP int) Node {
 	look := p.peek()
-	for isExprOperator(look.typ) && precedence[look.typ] >= minP {
+	for IsExprOperator(look.typ) && precedence[look.typ] >= minP {
 		op := p.next()
 		rhs := p.primary(nil)
 		look = p.peek()
 		// left-associative
-		for isExprOperator(look.typ) && precedence[look.typ] > precedence[op.typ] {
+		for IsExprOperator(look.typ) && precedence[look.typ] > precedence[op.typ] {
 			rhs = p.precedence(rhs, precedence[look.typ])
 			look = p.peek()
 		}
@@ -389,7 +389,7 @@ func (p *parser) lparameters() (args []Node) {
 
 func (p *parser) lparameter() (n Node) {
 	n = p.primary(nil)
-	if isExprOperator(p.peek().typ) {
+	if IsExprOperator(p.peek().typ) {
 		n = p.precedence(n, 0)
 	}
 	return