Merge pull request influxdata#22 from influxdb/nc-merge_langs

Merge both TICKscript and the expression language into the TICKscript language.

alert.go

@@ -10,9 +10,9 @@ import (
 
 	"github.com/influxdb/influxdb/influxql"
 	imodels "github.com/influxdb/influxdb/models"
-	"github.com/influxdb/kapacitor/expr"
 	"github.com/influxdb/kapacitor/models"
 	"github.com/influxdb/kapacitor/pipeline"
+	"github.com/influxdb/kapacitor/tick"
 )
 
 // Number of previous states to remember when computing flapping percentage.
@@ -64,7 +64,7 @@ type AlertNode struct {
 	a        *pipeline.AlertNode
 	endpoint string
 	handlers []AlertHandler
-	levels   []*expr.StatefulExpr
+	levels   []*tick.StatefulExpr
 	history  []AlertLevel
 	hIdx     int
 	flapping bool
@@ -92,27 +92,15 @@ func newAlertNode(et *ExecutingTask, n *pipeline.AlertNode) (an *AlertNode, err
 		an.handlers = append(an.handlers, an.handleExec)
 	}
 	// Parse level expressions
-	an.levels = make([]*expr.StatefulExpr, CritAlert+1)
-	if n.Info != "" {
-		tree, err := expr.ParseForType(n.Info, expr.ReturnBool)
-		if err != nil {
-			return nil, err
-		}
-		an.levels[InfoAlert] = &expr.StatefulExpr{Tree: tree, Funcs: expr.Functions()}
+	an.levels = make([]*tick.StatefulExpr, CritAlert+1)
+	if n.Info != nil {
+		an.levels[InfoAlert] = tick.NewStatefulExpr(n.Info)
 	}
-	if n.Warn != "" {
-		tree, err := expr.ParseForType(n.Warn, expr.ReturnBool)
-		if err != nil {
-			return nil, err
-		}
-		an.levels[WarnAlert] = &expr.StatefulExpr{Tree: tree, Funcs: expr.Functions()}
+	if n.Warn != nil {
+		an.levels[WarnAlert] = tick.NewStatefulExpr(n.Warn)
 	}
-	if n.Crit != "" {
-		tree, err := expr.ParseForType(n.Crit, expr.ReturnBool)
-		if err != nil {
-			return nil, err
-		}
-		an.levels[CritAlert] = &expr.StatefulExpr{Tree: tree, Funcs: expr.Functions()}
+	if n.Crit != nil {
+		an.levels[CritAlert] = tick.NewStatefulExpr(n.Crit)
 	}
 	// Configure flapping
 	if n.UseFlapping {

eval.go

@@ -0,0 +1,73 @@
+package kapacitor
+
+import (
+	"github.com/influxdb/kapacitor/models"
+	"github.com/influxdb/kapacitor/pipeline"
+	"github.com/influxdb/kapacitor/tick"
+)
+
+type EvalNode struct {
+	node
+	e          *pipeline.EvalNode
+	expression *tick.StatefulExpr
+}
+
+// Create a new  ApplyNode which applies a transformation func to each point in a stream and returns a single point.
+func newApplyNode(et *ExecutingTask, n *pipeline.EvalNode) (*EvalNode, error) {
+	en := &EvalNode{
+		node:       node{Node: n, et: et},
+		e:          n,
+		expression: tick.NewStatefulExpr(n.Expression),
+	}
+	en.node.runF = en.runApply
+	return en, nil
+}
+
+func (e *EvalNode) runApply() error {
+	switch e.Provides() {
+	case pipeline.StreamEdge:
+		for p, ok := e.ins[0].NextPoint(); ok; p, ok = e.ins[0].NextPoint() {
+			fields, err := e.eval(p.Fields, p.Tags)
+			if err != nil {
+				return err
+			}
+			p.Fields = fields
+			for _, child := range e.outs {
+				err := child.CollectPoint(p)
+				if err != nil {
+					return err
+				}
+			}
+		}
+	case pipeline.BatchEdge:
+		for b, ok := e.ins[0].NextBatch(); ok; b, ok = e.ins[0].NextBatch() {
+			for i, p := range b.Points {
+				fields, err := e.eval(p.Fields, b.Tags)
+				if err != nil {
+					return err
+				}
+				b.Points[i].Fields = fields
+			}
+			for _, child := range e.outs {
+				err := child.CollectBatch(b)
+				if err != nil {
+					return err
+				}
+			}
+		}
+	}
+	return nil
+}
+
+func (e *EvalNode) eval(fields models.Fields, tags map[string]string) (models.Fields, error) {
+	vars, err := mergeFieldsAndTags(fields, tags)
+	if err != nil {
+		return nil, err
+	}
+	v, err := e.expression.EvalNum(vars)
+	if err != nil {
+		return nil, err
+	}
+	fields[e.e.As] = v
+	return fields, nil
+}

expr.go

@@ -3,63 +3,33 @@ package kapacitor
 import (
 	"fmt"
 
-	"github.com/influxdb/kapacitor/expr"
 	"github.com/influxdb/kapacitor/models"
+	"github.com/influxdb/kapacitor/tick"
 )
 
-func ExprFunc(field, e string) (TransFunc, error) {
-	t, err := expr.Parse(e)
+// Evaluate a given expression as a boolean predicate against a set of fields and tags
+func EvalPredicate(se *tick.StatefulExpr, fields models.Fields, tags map[string]string) (bool, error) {
+	vars, err := mergeFieldsAndTags(fields, tags)
 	if err != nil {
-		return nil, err
-	}
-	if t.RType() != expr.ReturnNum {
-		return nil, fmt.Errorf("expression does not evaluate to a number")
-	}
-	x := &expression{
-		field: field,
-		se:    &expr.StatefulExpr{Tree: t, Funcs: expr.Functions()},
-	}
-	return TransFunc(x.Eval), nil
-}
-
-type expression struct {
-	field string
-	se    *expr.StatefulExpr
-}
-
-func (x *expression) Eval(fields models.Fields) (models.Fields, error) {
-	vars := make(expr.Vars)
-	for k, v := range fields {
-		if f, ok := v.(float64); ok {
-			vars[k] = f
-		} else {
-			return nil, fmt.Errorf("field values must be float64")
-		}
+		return false, err
 	}
-
-	v, err := x.se.EvalNum(vars)
+	b, err := se.EvalBool(vars)
 	if err != nil {
-		return nil, err
+		return false, err
 	}
-	fields[x.field] = v
-	return fields, nil
+	return b, nil
 }
 
-// Evaluate a given expression as a boolean predicate against a set of fields and tags
-func EvalPredicate(se *expr.StatefulExpr, fields models.Fields, tags map[string]string) (bool, error) {
-	vars := make(expr.Vars)
+func mergeFieldsAndTags(fields models.Fields, tags map[string]string) (tick.Vars, error) {
+	vars := make(tick.Vars)
 	for k, v := range fields {
-		if tags[k] != "" {
-			return false, fmt.Errorf("cannot have field and tags with same name %q", k)
+		if _, ok := tags[k]; ok {
+			return nil, fmt.Errorf("cannot have field and tags with same name %q", k)
 		}
 		vars[k] = v
 	}
 	for k, v := range tags {
 		vars[k] = v
 	}
-	b, err := se.EvalBool(vars)
-	if err != nil {
-		return false, err
-	}
-	return b, nil
+	return vars, nil
 }