Binary operators in LogQL (#1662)

* binops in ast

* bin op associativity & precedence

* binOpEvaluator work

* defers close only if constructed without error

* tests binary ops

* more binops

* updates docs

* changelog

* better logql parsing errors for binops

Signed-off-by: Owen Diehl <ow.diehl@gmail.com>

* adds ^ operator

CHANGELOG.md

@@ -2,6 +2,7 @@
 
 ### Features
 
+* [1662](https://github.com/grafana/loki/pull/1662) **owen-d**: Introduces binary operators in LogQL
 * [1572](https://github.com/grafana/loki/pull/1572) **owen-d**: Introduces the `querier.query-ingesters-within` flag and associated yaml config. When enabled, queries for a time range that do not overlap this lookback interval will not be sent to the ingesters.
 * [1558](https://github.com/grafana/loki/pull/1558) **owen-d**: Introduces `ingester.max-chunk-age` which specifies the maximum chunk age before it's cut.
 * [1565](https://github.com/grafana/loki/pull/1565) **owen-d**: The query frontend's `split_queries_by_interval` can now be specified as an override

docs/logql.md

@@ -151,3 +151,50 @@ by level:
 Get the rate of HTTP GET requests from NGINX logs:
 
 > `avg(rate(({job="nginx"} |= "GET")[10s])) by (region)`
+
+### Binary Operators
+
+#### Arithmetic Binary Operators
+
+Arithmetic binary operators
+The following binary arithmetic operators exist in Loki:
+
+- `+` (addition)
+- `-` (subtraction)
+- `*` (multiplication)
+- `/` (division)
+- `%` (modulo)
+- `^` (power/exponentiation)
+
+Binary arithmetic operators are defined only between two vectors.
+
+Between two instant vectors, a binary arithmetic operator is applied to each entry in the left-hand side vector and its matching element in the right-hand vector. The result is propagated into the result vector with the grouping labels becoming the output label set. Entries for which no matching entry in the right-hand vector can be found are not part of the result.
+
+##### Examples
+
+Get proportion of warning logs to error logs for the `foo` app
+
+> `sum(rate({app="foo", level="warn"}[1m])) / sum(rate({app="foo", level="error"}[1m]))`
+
+Operators on the same precedence level are left-associative (queries substituted with numbers here for simplicity). For example, 2 * 3 % 2 is equivalent to (2 * 3) % 2. However, some operators have different priorities: 1 + 2 / 3 will still be 1 + ( 2 / 3 ). These function identically to mathematical conventions.
+
+
+#### Logical/set binary operators
+
+These logical/set binary operators are only defined between two vectors:
+
+- `and` (intersection)
+- `or` (union)
+- `unless` (complement)
+
+`vector1 and vector2` results in a vector consisting of the elements of vector1 for which there are elements in vector2 with exactly matching label sets. Other elements are dropped.
+
+`vector1 or vector2` results in a vector that contains all original elements (label sets + values) of vector1 and additionally all elements of vector2 which do not have matching label sets in vector1.
+
+`vector1 unless vector2` results in a vector consisting of the elements of vector1 for which there are no elements in vector2 with exactly matching label sets. All matching elements in both vectors are dropped.
+
+##### Examples
+
+This contrived query will return the intersection of these queries, effectively `rate({app="bar"})`
+
+> `rate({app=~"foo|bar"}[1m]) and rate({app="bar"}[1m])`

pkg/logql/ast.go

@@ -227,6 +227,17 @@ const (
 	OpTypeTopK          = "topk"
 	OpTypeCountOverTime = "count_over_time"
 	OpTypeRate          = "rate"
+
+	// binops
+	OpTypeOr     = "or"
+	OpTypeAnd    = "and"
+	OpTypeUnless = "unless"
+	OpTypeAdd    = "+"
+	OpTypeSub    = "-"
+	OpTypeMul    = "*"
+	OpTypeDiv    = "/"
+	OpTypeMod    = "%"
+	OpTypePow    = "^"
 )
 
 // SampleExpr is a LogQL expression filtering logs and returning metric samples.
@@ -370,6 +381,41 @@ func (e *vectorAggregationExpr) String() string {
 	return formatOperation(e.operation, e.grouping, params...)
 }
 
+type binOpExpr struct {
+	SampleExpr
+	RHS SampleExpr
+	op  string
+}
+
+func (e *binOpExpr) String() string {
+	return fmt.Sprintf("%s %s %s", e.SampleExpr.String(), e.op, e.RHS.String())
+}
+
+func mustNewBinOpExpr(op string, lhs, rhs Expr) SampleExpr {
+	left, ok := lhs.(SampleExpr)
+	if !ok {
+		panic(newParseError(fmt.Sprintf(
+			"unexpected type for left leg of binary operation (%s): %T",
+			op,
+			lhs,
+		), 0, 0))
+	}
+
+	right, ok := rhs.(SampleExpr)
+	if !ok {
+		panic(newParseError(fmt.Sprintf(
+			"unexpected type for right leg of binary operation (%s): %T",
+			op,
+			rhs,
+		), 0, 0))
+	}
+	return &binOpExpr{
+		SampleExpr: left,
+		RHS:        right,
+		op:         op,
+	}
+}
+
 // helper used to impl Stringer for vector and range aggregations
 // nolint:interfacer
 func formatOperation(op string, grouping *grouping, params ...string) string {

pkg/logql/ast_test.go

@@ -43,6 +43,14 @@ func Test_SampleExpr_String(t *testing.T) {
 		`sum(count_over_time({job="mysql"}[5m]))`,
 		`topk(10,sum(rate({region="us-east1"}[5m])) by (name))`,
 		`avg( rate( ( {job="nginx"} |= "GET" ) [10s] ) ) by (region)`,
+		`sum by (cluster) (count_over_time({job="mysql"}[5m]))`,
+		`sum by (cluster) (count_over_time({job="mysql"}[5m])) / sum by (cluster) (count_over_time({job="postgres"}[5m])) `,
+		`
+		sum by (cluster) (count_over_time({job="postgres"}[5m])) /
+		sum by (cluster) (count_over_time({job="postgres"}[5m])) /
+		sum by (cluster) (count_over_time({job="postgres"}[5m]))
+		`,
+		`sum by (cluster) (count_over_time({job="mysql"}[5m])) / min(count_over_time({job="mysql"}[5m])) `,
 	} {
 		t.Run(tc, func(t *testing.T) {
 			expr, err := ParseExpr(tc)

pkg/logql/engine.go

@@ -213,10 +213,11 @@ func (ng *engine) exec(ctx context.Context, q *query) (promql.Value, error) {
 func (ng *engine) evalSample(ctx context.Context, expr SampleExpr, q *query) (promql.Value, error) {
 
 	stepEvaluator, err := ng.evaluator.Evaluator(ctx, expr, q)
-	defer helpers.LogError("closing SampleExpr", stepEvaluator.Close)
 	if err != nil {
 		return nil, err
 	}
+	defer helpers.LogError("closing SampleExpr", stepEvaluator.Close)
+
 	seriesIndex := map[uint64]*promql.Series{}
 
 	next, ts, vec := stepEvaluator.Next()