cmd/bosun: series operations

cmd/bosun/expr/expr.go

@@ -189,6 +189,10 @@ func (s Series) MarshalJSON() ([]byte, error) {
 	return json.Marshal(r)
 }
 
+func (a Series) Equal(b Series) bool {
+	return reflect.DeepEqual(a, b)
+}
+
 type ESQuery struct {
 	Query elastic.Query
 }
@@ -282,9 +286,19 @@ func (a *Results) Equal(b *Results) (bool, error) {
 		if !result.Group.Equal(sortedB[i].Group) {
 			return false, fmt.Errorf("mismatched groups a: %v, b: %v", result.Group, sortedB[i].Group)
 		}
-		if result.Value != sortedB[i].Value {
-			return false, fmt.Errorf("values do not match a: %v, b: %v", result.Value, sortedB[i].Value)
+		switch t := result.Value.(type) {
+		case Number, Scalar, String:
+			if result.Value != sortedB[i].Value {
+				return false, fmt.Errorf("values do not match a: %v, b: %v", result.Value, sortedB[i].Value)
+			}
+		case Series:
+			if !t.Equal(sortedB[i].Value.(Series)) {
+				return false, fmt.Errorf("mismatched series in result a: %v, b: %v", t, sortedB[i].Value.(Series))
+			}
+		default:
+			panic(fmt.Sprintf("can't compare results with type %T", t))
 		}
+
 	}
 	return true, nil
 }
@@ -516,6 +530,14 @@ func (e *State) walkBinary(node *parse.BinaryNode, T miniprofiler.Timer) *Result
 						s[k] = operate(node.OpStr, float64(v), bv)
 					}
 					value = s
+				case Series:
+					s := make(Series)
+					for k, av := range at {
+						if bv, ok := bt[k]; ok {
+							s[k] = operate(node.OpStr, av, bv)
+						}
+					}
+					value = s
 				default:
 					panic(ErrUnknownOp)
 				}

cmd/bosun/expr/expr_test.go

@@ -248,3 +248,61 @@ func TestQueryExpr(t *testing.T) {
 		}
 	}
 }
+
+func TestSeriesOperations(t *testing.T) {
+	seriesA := `series("key=a", 0, 1, 1, 2, 2, 1, 3, 4)`
+	seriesB := `series("key=a", 0, 1,       2, 0, 3, 4)`
+	seriesC := `series("key=a", 4, 1,       6, 0, 7, 4)`
+	template := "%v %v %v"
+	tests := []exprInOut{
+		{
+			fmt.Sprintf(template, seriesA, "+", seriesB),
+			Results{
+				Results: ResultSlice{
+					&Result{
+						Value: Series{
+							time.Unix(0, 0): 2,
+							time.Unix(2, 0): 1,
+							time.Unix(3, 0): 8,
+						},
+						Group: opentsdb.TagSet{"key": "a"},
+					},
+				},
+			},
+		},
+		{
+			fmt.Sprintf(template, seriesA, "+", seriesC),
+			Results{
+				Results: ResultSlice{
+					&Result{
+						Value: Series{
+						// Should be empty
+						},
+						Group: opentsdb.TagSet{"key": "a"},
+					},
+				},
+			},
+		},
+		{
+			fmt.Sprintf(template, seriesA, "/", seriesB),
+			Results{
+				Results: ResultSlice{
+					&Result{
+						Value: Series{
+							time.Unix(0, 0): 1,
+							time.Unix(2, 0): math.Inf(1),
+							time.Unix(3, 0): 1,
+						},
+						Group: opentsdb.TagSet{"key": "a"},
+					},
+				},
+			},
+		},
+	}
+	for _, test := range tests {
+		err := testExpression(test)
+		if err != nil {
+			t.Error(err)
+		}
+	}
+}

cmd/bosun/expr/funcs.go

@@ -313,6 +313,12 @@ var builtins = map[string]parse.Func{
 		Tags:   tagFirst,
 		F:      DropNA,
 	},
+	"dropbool": {
+		Args:   []models.FuncType{models.TypeSeriesSet, models.TypeSeriesSet},
+		Return: models.TypeSeriesSet,
+		Tags:   tagFirst,
+		F:      DropBool,
+	},
 	"epoch": {
 		Args:   []models.FuncType{},
 		Return: models.TypeScalar,
@@ -388,6 +394,27 @@ func SeriesFunc(e *State, T miniprofiler.Timer, tags string, pairs ...float64) (
 	}, nil
 }
 
+func DropBool(e *State, T miniprofiler.Timer, target *Results, filter *Results) (*Results, error) {
+	res := Results{}
+	unions := e.union(target, filter, "dropbool union")
+	for _, union := range unions {
+		aSeries := union.A.Value().(Series)
+		bSeries := union.B.Value().(Series)
+		newSeries := make(Series)
+		for k, v := range aSeries {
+			if bv, ok := bSeries[k]; ok {
+				if bv != float64(0) {
+					newSeries[k] = v
+				}
+			}
+		}
+		if len(newSeries) > 0 {
+			res.Results = append(res.Results, &Result{Group: union.Group, Value: newSeries})
+		}
+	}
+	return &res, nil
+}
+
 func Epoch(e *State, T miniprofiler.Timer) (*Results, error) {
 	return &Results{
 		Results: []*Result{

cmd/bosun/expr/parse/node.go

@@ -270,15 +270,11 @@ func (b *BinaryNode) StringAST() string {
 func (b *BinaryNode) Check(t *Tree) error {
 	t1 := b.Args[0].Return()
 	t2 := b.Args[1].Return()
-	if t1 == models.TypeSeriesSet && t2 == models.TypeSeriesSet {
-		return fmt.Errorf("parse: type error in %s: at least one side must be a number", b)
+	if !(t1 == models.TypeSeriesSet || t1 == models.TypeNumberSet || t1 == models.TypeScalar) {
+		return fmt.Errorf("expected NumberSet, SeriesSet, or Scalar, got %v", string(t1))
 	}
-	check := t1
-	if t1 == models.TypeSeriesSet {
-		check = t2
-	}
-	if check != models.TypeNumberSet && check != models.TypeScalar {
-		return fmt.Errorf("parse: type error in %s: expected a number", b)
+	if !(t2 == models.TypeSeriesSet || t2 == models.TypeNumberSet || t2 == models.TypeScalar) {
+		return fmt.Errorf("expected NumberSet, SeriesSet, or Scalar, got %v", string(t1))
 	}
 	if err := b.Args[0].Check(t); err != nil {
 		return err

docs/expressions.md

@@ -40,13 +40,17 @@ Various metrics can be combined by operators as long as one group is a subset of
 
 ## Operators
 
-The standard arithmetic (`+`, binary and unary `-`, `*`, `/`, `%`), relational (`<`, `>`, `==`, `!=`, `>=`, `<=`), and logical (`&&`, `||`, and unary `!`) operators are supported. The binary operators require the value on at least one side to be a scalar or NumberSet. Arrays will have the operator applied to each element. Examples:
+The standard arithmetic (`+`, binary and unary `-`, `*`, `/`, `%`), relational (`<`, `>`, `==`, `!=`, `>=`, `<=`), and logical (`&&`, `||`, and unary `!`) operators are supported. Examples:
 
 * `q("q") + 1`, which adds one to every element of the result of the query `"q"`
 * `-q("q")`, the negation of the results of the query
 * `5 > q("q")`, a series of numbers indicating whether each data point is more than five
 * `6 / 8`, the scalar value three-quarters
 
+### Series Operations
+
+If you combine two seriesSets with an operator (i.e. `q(..)` + `q(..)`), then operations are applied for each point in the series if there is a corresponding datapoint on the right hand side (RH). A corresponding datapoint is one which has the same timestamp (and normal group subset rules apply). If there is no corresponding datapoint on the left side, then the datapoint is dropped. This is a new feature as of 0.5.0.
+
 ### Precedence
 
 From highest to lowest:
@@ -482,6 +486,18 @@ Remove any values lower than or equal to number from a series. Will error if thi
 
 Remove any NaN or Inf values from a series. Will error if this operation results in an empty series.
 
+## dropbool(seriesSet, seriesSet) seriesSet
+Drop datapoints where the corresponding value in the second series set is non-zero. (See Series Operations for what corresponding means). The following example drops tr_avg (avg response time per bucket) datapoints if the count in that bucket was + or - 100 from the average count over the time period. 
+
+Example:
+
+```
+$count = q("sum:traffic.haproxy.route_tr_count{host=literal_or(ny-logsql01),route=Questions/Show}", "30m", "")
+$avg = q("sum:traffic.haproxy.route_tr_avg{host=literal_or(ny-logsql01),route=Questions/Show}", "30m", "")
+$avgCount = avg($count)
+dropbool($avg, !($count < $avgCount-100 || $count > $avgCount+100))
+```
+
 ## epoch() scalar
 
 Returns the Unix epoch in seconds of the expression start time (scalar).