first_over_time and last_over_time (#3050)

* Wip Signed-off-by: Cyril Tovena <cyril.tovena@gmail.com> * [POC] last_over_time and first_over_time. This is an attempt to allow users to select the first or last discrete sample values from a range vector. This way we can create gauge, the only drawback being the log need to exists at least in every range. May be we could transfer the last or first value to the next range, but it will still be a problem for small ranges. Still wondering how useful this is compare to min/max_over_time. Signed-off-by: Cyril Tovena <cyril.tovena@gmail.com>
grafana · Apr 1, 2021 · 4d1da2e · 4d1da2e
1 parent ecca5d3
commit 4d1da2e
Show file tree

Hide file tree

Showing 8 changed files with 283 additions and 198 deletions.
diff --git a/docs/sources/logql/_index.md b/docs/sources/logql/_index.md
@@ -534,6 +534,8 @@ Supported function for operating over unwrapped ranges are:
 - `avg_over_time(unwrapped-range)`: the average value of all points in the specified interval.
 - `max_over_time(unwrapped-range)`: the maximum value of all points in the specified interval.
 - `min_over_time(unwrapped-range)`: the minimum value of all points in the specified interval
+- `first_over_time(unwrapped-range)`: the first value of all points in the specified interval
+- `last_over_time(unwrapped-range)`: the last value of all points in the specified interval
 - `stdvar_over_time(unwrapped-range)`: the population standard variance of the values in the specified interval.
 - `stddev_over_time(unwrapped-range)`: the population standard deviation of the values in the specified interval.
 - `quantile_over_time(scalar,unwrapped-range)`: the φ-quantile (0 ≤ φ ≤ 1) of the values in the specified interval.

diff --git a/pkg/logql/ast.go b/pkg/logql/ast.go
@@ -574,6 +574,8 @@ const (
 	OpRangeTypeStdvar    = "stdvar_over_time"
 	OpRangeTypeStddev    = "stddev_over_time"
 	OpRangeTypeQuantile  = "quantile_over_time"
+	OpRangeTypeFirst     = "first_over_time"
+	OpRangeTypeLast      = "last_over_time"
 	OpRangeTypeAbsent    = "absent_over_time"
 
 	// binops - logical/set
@@ -691,14 +693,14 @@ func (e *rangeAggregationExpr) Selector() LogSelectorExpr {
 func (e rangeAggregationExpr) validate() error {
 	if e.grouping != nil {
 		switch e.operation {
-		case OpRangeTypeAvg, OpRangeTypeStddev, OpRangeTypeStdvar, OpRangeTypeQuantile, OpRangeTypeMax, OpRangeTypeMin:
+		case OpRangeTypeAvg, OpRangeTypeStddev, OpRangeTypeStdvar, OpRangeTypeQuantile, OpRangeTypeMax, OpRangeTypeMin, OpRangeTypeFirst, OpRangeTypeLast:
 		default:
 			return fmt.Errorf("grouping not allowed for %s aggregation", e.operation)
 		}
 	}
 	if e.left.unwrap != nil {
 		switch e.operation {
-		case OpRangeTypeRate, OpRangeTypeAvg, OpRangeTypeSum, OpRangeTypeMax, OpRangeTypeMin, OpRangeTypeStddev, OpRangeTypeStdvar, OpRangeTypeQuantile, OpRangeTypeAbsent:
+		case OpRangeTypeAvg, OpRangeTypeSum, OpRangeTypeMax, OpRangeTypeMin, OpRangeTypeStddev, OpRangeTypeStdvar, OpRangeTypeQuantile, OpRangeTypeRate, OpRangeTypeAbsent, OpRangeTypeFirst, OpRangeTypeLast:
 			return nil
 		default:
 			return fmt.Errorf("invalid aggregation %s with unwrap", e.operation)

diff --git a/pkg/logql/ast_test.go b/pkg/logql/ast_test.go
@@ -115,6 +115,8 @@ func Test_SampleExpr_String(t *testing.T) {
 			count_over_time({namespace="tns"} | logfmt | label_format foo=bar[5m])
 		)`,
 		`sum_over_time({namespace="tns"} |= "level=error" | json |foo>=5,bar<25ms | unwrap latency | __error__!~".*" | foo >5[5m])`,
+		`last_over_time({namespace="tns"} |= "level=error" | json |foo>=5,bar<25ms | unwrap latency | __error__!~".*" | foo >5[5m])`,
+		`first_over_time({namespace="tns"} |= "level=error" | json |foo>=5,bar<25ms | unwrap latency | __error__!~".*" | foo >5[5m])`,
 		`absent_over_time({namespace="tns"} |= "level=error" | json |foo>=5,bar<25ms | unwrap latency | __error__!~".*" | foo >5[5m])`,
 		`sum by (job) (
 			sum_over_time(

diff --git a/pkg/logql/engine_test.go b/pkg/logql/engine_test.go
@@ -106,6 +106,16 @@ func TestEngine_LogsInstantQuery(t *testing.T) {
 			},
 			promql.Vector{promql.Sample{Point: promql.Point{T: 60 * 1000, V: 6}, Metric: labels.Labels{labels.Label{Name: "app", Value: "foo"}}}},
 		},
+		{
+			`first_over_time({app="foo"} |~".+bar" | unwrap foo [1m])`, time.Unix(60, 0), logproto.BACKWARD, 10,
+			[][]logproto.Series{
+				{newSeries(testSize, factor(10, identity), `{app="foo"}`)}, // 10 , 20 , 30 .. 60 = 6 total
+			},
+			[]SelectSampleParams{
+				{&logproto.SampleQueryRequest{Start: time.Unix(0, 0), End: time.Unix(60, 0), Selector: `first_over_time({app="foo"}|~".+bar"| unwrap foo [1m])`}},
+			},
+			promql.Vector{promql.Sample{Point: promql.Point{T: 60 * 1000, V: 1}, Metric: labels.Labels{labels.Label{Name: "app", Value: "foo"}}}},
+		},
 		{
 			`count_over_time({app="foo"} |~".+bar" [1m] offset 30s)`, time.Unix(90, 0), logproto.BACKWARD, 10,
 			[][]logproto.Series{
@@ -754,6 +764,33 @@ func TestEngine_RangeQuery(t *testing.T) {
 				},
 			},
 		},
+		{
+			`last_over_time(({app="foo"} |~".+bar" | unwrap foo)[5m])`, time.Unix(5*60, 0), time.Unix(5*120, 0), 30 * time.Second, 0, logproto.BACKWARD, 10,
+			[][]logproto.Series{
+				{newSeries(testSize, factor(10, identity), `{app="foo"}`)}, // 10 , 20 , 30 .. 300 = 30 total
+			},
+			[]SelectSampleParams{
+				{&logproto.SampleQueryRequest{Start: time.Unix(0, 0), End: time.Unix(5*120, 0), Selector: `last_over_time({app="foo"}|~".+bar"| unwrap foo[5m])`}},
+			},
+			promql.Matrix{
+				promql.Series{
+					Metric: labels.Labels{{Name: "app", Value: "foo"}},
+					Points: []promql.Point{
+						{T: 300 * 1000, V: 1},
+						{T: 330 * 1000, V: 1},
+						{T: 360 * 1000, V: 1},
+						{T: 390 * 1000, V: 1},
+						{T: 420 * 1000, V: 1},
+						{T: 450 * 1000, V: 1},
+						{T: 480 * 1000, V: 1},
+						{T: 510 * 1000, V: 1},
+						{T: 540 * 1000, V: 1},
+						{T: 570 * 1000, V: 1},
+						{T: 600 * 1000, V: 1},
+					},
+				},
+			},
+		},
 		{
 			`avg(count_over_time({app=~"foo|bar"} |~".+bar" [1m]))`, time.Unix(60, 0), time.Unix(180, 0), 30 * time.Second, 0, logproto.FORWARD, 100,
 			[][]logproto.Series{

diff --git a/pkg/logql/expr.y b/pkg/logql/expr.y
@@ -100,7 +100,7 @@ import (
                   OPEN_PARENTHESIS CLOSE_PARENTHESIS BY WITHOUT COUNT_OVER_TIME RATE SUM AVG MAX MIN COUNT STDDEV STDVAR BOTTOMK TOPK
                   BYTES_OVER_TIME BYTES_RATE BOOL JSON REGEXP LOGFMT PIPE LINE_FMT LABEL_FMT UNWRAP AVG_OVER_TIME SUM_OVER_TIME MIN_OVER_TIME
                   MAX_OVER_TIME STDVAR_OVER_TIME STDDEV_OVER_TIME QUANTILE_OVER_TIME BYTES_CONV DURATION_CONV DURATION_SECONDS_CONV
-                  ABSENT_OVER_TIME LABEL_REPLACE UNPACK OFFSET
+                  FIRST_OVER_TIME LAST_OVER_TIME ABSENT_OVER_TIME LABEL_REPLACE UNPACK OFFSET
 
 // Operators are listed with increasing precedence.
 %left <binOp> OR
@@ -374,6 +374,8 @@ rangeOp:
     | STDVAR_OVER_TIME   { $$ = OpRangeTypeStdvar }
     | STDDEV_OVER_TIME   { $$ = OpRangeTypeStddev }
     | QUANTILE_OVER_TIME { $$ = OpRangeTypeQuantile }
+    | FIRST_OVER_TIME    { $$ = OpRangeTypeFirst }
+    | LAST_OVER_TIME     { $$ = OpRangeTypeLast }
     | ABSENT_OVER_TIME   { $$ = OpRangeTypeAbsent }
     ;