Add count_values() aggregator.

This is useful for counting how many instances
of a job are running a particular version/build.

Fixes #622

promql/engine.go

@@ -1076,6 +1076,13 @@ func (ev *evaluator) aggregation(op itemType, grouping model.LabelNames, without
 			return vector{}
 		}
 	}
+	var valueLabel model.LabelName
+	if op == itemCountValues {
+		valueLabel = model.LabelName(ev.evalString(param).Value)
+		if !without {
+			grouping = append(grouping, valueLabel)
+		}
+	}
 
 	for _, s := range vec {
 		withoutMetric := s.Metric
@@ -1084,6 +1091,13 @@ func (ev *evaluator) aggregation(op itemType, grouping model.LabelNames, without
 				withoutMetric.Del(l)
 			}
 			withoutMetric.Del(model.MetricNameLabel)
+			if op == itemCountValues {
+				withoutMetric.Set(valueLabel, model.LabelValue(s.Value.String()))
+			}
+		} else {
+			if op == itemCountValues {
+				s.Metric.Set(valueLabel, model.LabelValue(s.Value.String()))
+			}
 		}
 
 		var groupingKey uint64
@@ -1147,7 +1161,7 @@ func (ev *evaluator) aggregation(op itemType, grouping model.LabelNames, without
 			if groupedResult.value > s.Value || math.IsNaN(float64(groupedResult.value)) {
 				groupedResult.value = s.Value
 			}
-		case itemCount:
+		case itemCount, itemCountValues:
 			groupedResult.groupCount++
 		case itemStdvar, itemStddev:
 			groupedResult.value += s.Value
@@ -1179,7 +1193,7 @@ func (ev *evaluator) aggregation(op itemType, grouping model.LabelNames, without
 		switch op {
 		case itemAvg:
 			aggr.value = aggr.value / model.SampleValue(aggr.groupCount)
-		case itemCount:
+		case itemCount, itemCountValues:
 			aggr.value = model.SampleValue(aggr.groupCount)
 		case itemStdvar:
 			avg := float64(aggr.value) / float64(aggr.groupCount)

promql/lex.go

@@ -60,7 +60,9 @@ func (i itemType) isAggregator() bool { return i > aggregatorsStart && i < aggre
 
 // isAggregator returns true if the item is an aggregator that takes a parameter.
 // Returns false otherwise
-func (i itemType) isAggregatorWithParam() bool { return i == itemTopK || i == itemBottomK }
+func (i itemType) isAggregatorWithParam() bool {
+	return i == itemTopK || i == itemBottomK || i == itemCountValues
+}
 
 // isKeyword returns true if the item corresponds to a keyword.
 // Returns false otherwise.
@@ -176,6 +178,7 @@ const (
 	itemStdvar
 	itemTopK
 	itemBottomK
+	itemCountValues
 	aggregatorsEnd
 
 	keywordsStart
@@ -209,15 +212,16 @@ var key = map[string]itemType{
 	"unless": itemLUnless,
 
 	// Aggregators.
-	"sum":     itemSum,
-	"avg":     itemAvg,
-	"count":   itemCount,
-	"min":     itemMin,
-	"max":     itemMax,
-	"stddev":  itemStddev,
-	"stdvar":  itemStdvar,
-	"topk":    itemTopK,
-	"bottomk": itemBottomK,
+	"sum":          itemSum,
+	"avg":          itemAvg,
+	"count":        itemCount,
+	"min":          itemMin,
+	"max":          itemMax,
+	"stddev":       itemStddev,
+	"stdvar":       itemStdvar,
+	"topk":         itemTopK,
+	"bottomk":      itemBottomK,
+	"count_values": itemCountValues,
 
 	// Keywords.
 	"alert":       itemAlert,

promql/parse.go

@@ -1052,6 +1052,9 @@ func (p *parser) checkType(node Node) (typ model.ValueType) {
 		if n.Op == itemTopK || n.Op == itemBottomK {
 			p.expectType(n.Param, model.ValScalar, "aggregation parameter")
 		}
+		if n.Op == itemCountValues {
+			p.expectType(n.Param, model.ValString, "aggregation parameter")
+		}
 
 	case *BinaryExpr:
 		lt := p.checkType(n.LHS)

promql/parse_test.go

@@ -1213,6 +1213,18 @@ var testExpr = []struct {
 			},
 			Param: &NumberLiteral{5},
 		},
+	}, {
+		input: "count_values(\"value\", some_metric)",
+		expected: &AggregateExpr{
+			Op: itemCountValues,
+			Expr: &VectorSelector{
+				Name: "some_metric",
+				LabelMatchers: metric.LabelMatchers{
+					{Type: metric.Equal, Name: model.MetricNameLabel, Value: "some_metric"},
+				},
+			},
+			Param: &StringLiteral{"value"},
+		},
 	}, {
 		input:  `sum some_metric by (test)`,
 		fail:   true,
@@ -1257,6 +1269,10 @@ var testExpr = []struct {
 		input:  `topk(some_metric, other_metric)`,
 		fail:   true,
 		errMsg: "parse error at char 32: expected type scalar in aggregation parameter, got vector",
+	}, {
+		input:  `count_values(5, other_metric)`,
+		fail:   true,
+		errMsg: "parse error at char 30: expected type string in aggregation parameter, got scalar",
 	},
 	// Test function calls.
 	{

promql/printer_test.go

@@ -74,6 +74,9 @@ func TestExprString(t *testing.T) {
 		{
 			in: `topk(5, task:errors:rate10s{job="s"})`,
 		},
+		{
+			in: `count_values("value", task:errors:rate10s{job="s"})`,
+		},
 		{
 			in: `a - ON(b) c`,
 		},

promql/testdata/aggregators.test

@@ -183,3 +183,40 @@ eval_ordered instant at 50m bottomk(3, http_requests{job="api-server",group="pro
 	http_requests{job="api-server", instance="0", group="production"}	100
 	http_requests{job="api-server", instance="1", group="production"}	200
 	http_requests{job="api-server", instance="2", group="production"}	NaN
+
+clear
+
+# Tests for count_values.
+load 5m
+	version{job="api-server", instance="0", group="production"}	6
+	version{job="api-server", instance="1", group="production"}	6
+	version{job="api-server", instance="2", group="production"}	6
+	version{job="api-server", instance="0", group="canary"}		8
+	version{job="api-server", instance="1", group="canary"}		8
+	version{job="app-server", instance="0", group="production"}	6
+	version{job="app-server", instance="1", group="production"}	6
+	version{job="app-server", instance="0", group="canary"}		7
+	version{job="app-server", instance="1", group="canary"}		7
+
+eval instant at 5m count_values("version", version)
+	{version="6"} 5
+	{version="7"} 2
+	{version="8"} 2
+
+eval instant at 5m count_values without (instance)("version", version)
+	{job="api-server", group="production", version="6"} 3
+	{job="api-server", group="canary", version="8"} 2
+	{job="app-server", group="production", version="6"} 2
+	{job="app-server", group="canary", version="7"} 2
+
+# Overwrite label with output. Don't do this.
+eval instant at 5m count_values without (instance)("job", version)
+	{job="6", group="production"} 5
+	{job="8", group="canary"} 2
+	{job="7", group="canary"} 2
+
+# Overwrite label with output. Don't do this.
+eval instant at 5m count_values by (job, group)("job", version)
+	{job="6", group="production"} 5
+	{job="8", group="canary"} 2
+	{job="7", group="canary"} 2