[TraceQL] min/max/sum aggregates

CHANGELOG.md

@@ -4,6 +4,7 @@
 * [FEATURE] Add flag to check configuration [#2131](https:/grafana/tempo/issues/2131) (@robertscherbarth @agrib-01)
 * [FEATURE] Add flag to optionally enable all available Go runtime metrics [#2005](https:/grafana/tempo/pull/2005) (@andreasgerstmayr)
 * [FEATURE] Add support for span `kind` to TraceQL [#2217](https:/grafana/tempo/pull/2217) (@joe-elliott)
+* [FEATURE] Add support for min/max/avg aggregates to TraceQL[#2255](https:/grafana/tempo/pull/2255) (@joe-elliott)
 * [CHANGE] Add support for s3 session token in static config [#2093](https:/grafana/tempo/pull/2093) (@farodin91)
 * [CHANGE] **Breaking Change** Remove support for search on v2 blocks. [#2159](https:/grafana/tempo/pull/2062) (@joe-elliott)
  Removed config options:

docs/sources/tempo/traceql/_index.md

@@ -182,7 +182,10 @@ The second expression returns no traces because it's impossible for a single spa
 So far, all of the example queries expressions have been about individual spans. You can use aggregate functions to ask questions about a set of spans. These currently consist of:
 
 - `count` - The count of spans in the spanset.
-- `avg` - The average of a given attribute or intrinsic for a spanset.
+- `avg` - The average of a given numeric attribute or intrinsic for a spanset.
+- `max` - The max value of a given numeric attribute or intrinsic for a spanset.
+- `min` - The min value of a given numeric attribute or intrinsic for a spanset.
+- `sum` - The sum value of a given numeric attribute or intrinsic for a spanset.
 
 Aggregate functions allow you to carry out operations on matching results to further refine the traces returned. For more information on planned future work, refer to [How TraceQL works]({{< relref "architecture" >}}).
 

pkg/traceql/ast_execute.go

@@ -106,6 +106,49 @@ func (a Aggregate) evaluate(input []*Spanset) (output []*Spanset, err error) {
  copy.Scalar = NewStaticFloat(sum / float64(count))
  output = append(output, copy)
 
+ case aggregateMax:
+ max := math.Inf(-1)
+ for _, s := range ss.Spans {
+ val, err := a.e.execute(s)
+ if err != nil {
+ return nil, err
+ }
+ if val.asFloat() > max {
+ max = val.asFloat()
+ }
+ }
+ copy := ss.clone()
+ copy.Scalar = NewStaticFloat(max)
+ output = append(output, copy)
+
+ case aggregateMin:
+ min := math.Inf(1)
+ for _, s := range ss.Spans {
+ val, err := a.e.execute(s)
+ if err != nil {
+ return nil, err
+ }
+ if val.asFloat() < min {
+ min = val.asFloat()
+ }
+ }
+ copy := ss.clone()
+ copy.Scalar = NewStaticFloat(min)
+ output = append(output, copy)
+
+ case aggregateSum:
+ sum := 0.0
+ for _, s := range ss.Spans {
+ val, err := a.e.execute(s)
+ if err != nil {
+ return nil, err
+ }
+ sum += val.asFloat()
+ }
+ copy := ss.clone()
+ copy.Scalar = NewStaticFloat(sum)
+ output = append(output, copy)
+
  default:
  return nil, fmt.Errorf("aggregate operation (%v) not supported", a.op)
  }

pkg/traceql/ast_execute_test.go

@@ -313,6 +313,135 @@ func TestScalarFilterEvaluate(t *testing.T) {
  },
  },
  },
+ // max
+ {
+ "{ .foo = `a` } | max(duration) >= 10ms",
+ []*Spanset{
+ {Spans: []Span{
+ // max duration = 8ms
+ &mockSpan{attributes: map[Attribute]Static{
+ NewAttribute("foo"): NewStaticString("a"),
+ NewIntrinsic(IntrinsicDuration): NewStaticDuration(2 * time.Millisecond)},
+ },
+ &mockSpan{attributes: map[Attribute]Static{
+ NewAttribute("foo"): NewStaticString("a"),
+ NewIntrinsic(IntrinsicDuration): NewStaticDuration(8 * time.Millisecond)},
+ },
+ }},
+ {Spans: []Span{
+ // max duration = 15ms
+ &mockSpan{attributes: map[Attribute]Static{
+ NewAttribute("foo"): NewStaticString("a"),
+ NewIntrinsic(IntrinsicDuration): NewStaticDuration(5 * time.Millisecond)},
+ },
+ &mockSpan{attributes: map[Attribute]Static{
+ NewAttribute("foo"): NewStaticString("a"),
+ NewIntrinsic(IntrinsicDuration): NewStaticDuration(15 * time.Millisecond)},
+ },
+ }},
+ },
+ []*Spanset{
+ {
+ Scalar: NewStaticFloat(15.0 * float64(time.Millisecond)),
+ Spans: []Span{
+ &mockSpan{attributes: map[Attribute]Static{
+ NewAttribute("foo"): NewStaticString("a"),
+ NewIntrinsic(IntrinsicDuration): NewStaticDuration(5 * time.Millisecond)},
+ },
+ &mockSpan{attributes: map[Attribute]Static{
+ NewAttribute("foo"): NewStaticString("a"),
+ NewIntrinsic(IntrinsicDuration): NewStaticDuration(15 * time.Millisecond)},
+ },
+ },
+ },
+ },
+ },
+ // min
+ {
+ "{ .foo = `a` } | min(duration) <= 10ms",
+ []*Spanset{
+ {Spans: []Span{
+ // min duration = 2ms
+ &mockSpan{attributes: map[Attribute]Static{
+ NewAttribute("foo"): NewStaticString("a"),
+ NewIntrinsic(IntrinsicDuration): NewStaticDuration(2 * time.Millisecond)},
+ },
+ &mockSpan{attributes: map[Attribute]Static{
+ NewAttribute("foo"): NewStaticString("a"),
+ NewIntrinsic(IntrinsicDuration): NewStaticDuration(8 * time.Millisecond)},
+ },
+ }},
+ {Spans: []Span{
+ // min duration = 5ms
+ &mockSpan{attributes: map[Attribute]Static{
+ NewAttribute("foo"): NewStaticString("a"),
+ NewIntrinsic(IntrinsicDuration): NewStaticDuration(12 * time.Millisecond)},
+ },
+ &mockSpan{attributes: map[Attribute]Static{
+ NewAttribute("foo"): NewStaticString("a"),
+ NewIntrinsic(IntrinsicDuration): NewStaticDuration(15 * time.Millisecond)},
+ },
+ }},
+ },
+ []*Spanset{
+ {
+ Scalar: NewStaticFloat(2.0 * float64(time.Millisecond)),
+ Spans: []Span{
+ &mockSpan{attributes: map[Attribute]Static{
+ NewAttribute("foo"): NewStaticString("a"),
+ NewIntrinsic(IntrinsicDuration): NewStaticDuration(2 * time.Millisecond)},
+ },
+ &mockSpan{attributes: map[Attribute]Static{
+ NewAttribute("foo"): NewStaticString("a"),
+ NewIntrinsic(IntrinsicDuration): NewStaticDuration(8 * time.Millisecond)},
+ },
+ },
+ },
+ },
+ },
+ // sum
+ {
+ "{ .foo = `a` } | sum(duration) = 10ms",
+ []*Spanset{
+ {Spans: []Span{
+ // sum duration = 10ms
+ &mockSpan{attributes: map[Attribute]Static{
+ NewAttribute("foo"): NewStaticString("a"),
+ NewIntrinsic(IntrinsicDuration): NewStaticDuration(2 * time.Millisecond)},
+ },
+ &mockSpan{attributes: map[Attribute]Static{
+ NewAttribute("foo"): NewStaticString("a"),
+ NewIntrinsic(IntrinsicDuration): NewStaticDuration(8 * time.Millisecond)},
+ },
+ }},
+ {Spans: []Span{
+ // sum duration = 27ms
+ &mockSpan{attributes: map[Attribute]Static{
+ NewAttribute("foo"): NewStaticString("a"),
+ NewIntrinsic(IntrinsicDuration): NewStaticDuration(12 * time.Millisecond)},
+ },
+ &mockSpan{attributes: map[Attribute]Static{
+ NewAttribute("foo"): NewStaticString("a"),
+ NewIntrinsic(IntrinsicDuration): NewStaticDuration(15 * time.Millisecond)},
+ },
+ }},
+ },
+ []*Spanset{
+ {
+ Scalar: NewStaticFloat(10 * float64(time.Millisecond)),
+ Spans: []Span{
+ &mockSpan{attributes: map[Attribute]Static{
+ NewAttribute("foo"): NewStaticString("a"),
+ NewIntrinsic(IntrinsicDuration): NewStaticDuration(2 * time.Millisecond)},
+ },
+ &mockSpan{attributes: map[Attribute]Static{
+ NewAttribute("foo"): NewStaticString("a"),
+ NewIntrinsic(IntrinsicDuration): NewStaticDuration(8 * time.Millisecond)},
+ },
+ },
+ },
+ },
+ },
  }
 
  for _, tc := range testCases {

pkg/traceql/ast_validate.go

@@ -85,7 +85,7 @@ func (a Aggregate) validate() error {
  }
 
  switch a.op {
- case aggregateCount, aggregateAvg:
+ case aggregateCount, aggregateAvg, aggregateMin, aggregateMax, aggregateSum:
  default:
  return newUnsupportedError(fmt.Sprintf("aggregate operation (%v)", a.op))
  }

pkg/traceql/test_examples.yaml

@@ -61,6 +61,17 @@ valid:
  - '{ true } || { true }'
  # scalar filters
  - 'avg(.field) > 1'
+ - 'max(duration) >= 1s'
+ - 'max(duration) > 1' # same note as above for int, float and duration
+ - '{ true } | max(duration) = 1h'
+ - '{ true } | min(duration) = 1h'
+ - '{ true } | sum(duration) = 1h'
+ - '{ true } | max(.a) = 1'
+ - '{ true } | max(span.a) = 1'
+ - '{ true } | max(resource.a) = 1'
+ - '{ true } | max(1 + .a) = 1'
+ - '{ true } | max((1 + .a) * 2) = 1'
+ - 'max(duration) > 3s | { status = error || .http.status = 500 }'
  # pipelines
  - '{ true } | { .a }'
  - '{ true } | count() = 1'
@@ -161,6 +172,7 @@ validate_fails:
  - 'max(status) = ok'
  - 'max(kind) = consumer'
  - 'min(1 = 3) = 1'
+ - 'max(duration) < ok'
  # scalar expressions must reference the span
  - 'sum(3) = 2'
  - 'sum(3) = min(14)'
@@ -186,30 +198,18 @@ unsupported:
  # by - will *not* be valid when supported - group expressions must reference the span
  - '{ true } | by(1)'
  - '{ true } | by("foo")'
- # aggregates - will be valid when supported
- - 'min(childCount) < 2'
- - 'max(duration) >= 1s'
+ # complex scalar filters - will be valid when supported
  - 'min(.field) < max(duration)'
  - 'sum(.field) = min(.field)'
- - 'max(duration) > 1' # same note as above for int, float and duration
  - 'min(.field) + max(.field) > 1'
  - 'min(.field) + max(childCount) > max(duration) - min(.field)'
  - 'min(childCount) < 2 / 6'
  - 'max(1 - (2 + .field)) < avg(3 * duration ^ 2)'
- - '{ true } | max(duration) = 1h'
- - '{ true } | min(duration) = 1h'
- - '{ true } | sum(duration) = 1h'
- - '{ true } | max(.a) = 1'
+ # aggregates - will be valid when supported
+ - 'min(childCount) < 2'
  - '{ true } | max(parent.a) = 1'
- - '{ true } | max(span.a) = 1'
- - '{ true } | max(resource.a) = 1'
- - '{ true } | max(1 + .a) = 1'
- - '{ true } | max((1 + .a) * 2) = 1'
  - '{ true } | by(3 * .field - 2) | max(duration) < 1s'
- - 'max(duration) > 3s | { status = error || .http.status = 500 }'
  - '{ .http.status = 200 } | max(.field) - min(.field) > 3'
- # aggregates - will *not* be valid when supported - comparing number type to status type
- - 'max(duration) < ok'
  # parent - will be valid when supported
  - '{ parent.a != 3 }'
  - '{ parent.resource.a && true }'

tempodb/tempodb_search_test.go

@@ -19,6 +19,7 @@ import (
  v1 "github.com/grafana/tempo/pkg/tempopb/trace/v1"
  "github.com/grafana/tempo/pkg/traceql"
  "github.com/grafana/tempo/pkg/util"
+ "github.com/grafana/tempo/pkg/util/math"
  "github.com/grafana/tempo/pkg/util/test"
  "github.com/grafana/tempo/tempodb/backend"
  "github.com/grafana/tempo/tempodb/backend/local"
@@ -120,6 +121,7 @@ func testAdvancedTraceQLCompleteBlock(t *testing.T, blockVersion string) {
 
  // collect some info about wantTr to use below
  trueConditionsBySpan := [][]string{}
+ durationBySpan := []uint64{}
  falseConditions := []string{
  fmt.Sprintf("name=`%v`", test.RandomString()),
  fmt.Sprintf("duration>%dh", rand.Intn(10)+1),
@@ -141,10 +143,11 @@ func testAdvancedTraceQLCompleteBlock(t *testing.T, blockVersion string) {
  trueC = append(trueC, fmt.Sprintf("duration=%dns", s.EndTimeUnixNano-s.StartTimeUnixNano))
  trueC = append(trueC, fmt.Sprintf("status=%s", status))
  trueC = append(trueC, fmt.Sprintf("kind=%s", kind))
+ trueC = append(trueC, trueResourceC...)
 
  trueConditionsBySpan = append(trueConditionsBySpan, trueC)
- trueConditionsBySpan = append(trueConditionsBySpan, trueResourceC)
  falseConditions = append(falseConditions, falseC...)
+ durationBySpan = append(durationBySpan, s.EndTimeUnixNano-s.StartTimeUnixNano)
  }
  }
  }
@@ -173,12 +176,26 @@ func testAdvancedTraceQLCompleteBlock(t *testing.T, blockVersion string) {
  // counts
  {Query: fmt.Sprintf("{} | count() = %d", totalSpans)},
  {Query: fmt.Sprintf("{} | count() != %d", totalSpans+1)},
- {Query: fmt.Sprintf("{} | count() <= %d", totalSpans)},
- {Query: fmt.Sprintf("{} | count() >= %d", totalSpans)},
  {Query: fmt.Sprintf("{ true } && { true } | count() = %d", totalSpans)},
  {Query: fmt.Sprintf("{ true } || { true } | count() = %d", totalSpans)},
- // avgs
- {Query: "{ } | avg(duration) > 0"}, // todo: make this better
+ {Query: fmt.Sprintf("{ %s && %s } | count() = 1", rando(trueConditionsBySpan[0]), rando(trueConditionsBySpan[0]))},
+ // avgs/min/max/sum
+ {Query: fmt.Sprintf("{ %s && %s } && { %s && %s } | avg(duration) = %dns",
+ rando(trueConditionsBySpan[0]), rando(trueConditionsBySpan[0]),
+ rando(trueConditionsBySpan[1]), rando(trueConditionsBySpan[1]),
+ (durationBySpan[0]+durationBySpan[1])/2)},
+ {Query: fmt.Sprintf("{ %s && %s } && { %s && %s } | min(duration) = %dns",
+ rando(trueConditionsBySpan[0]), rando(trueConditionsBySpan[0]),
+ rando(trueConditionsBySpan[1]), rando(trueConditionsBySpan[1]),
+ math.Min64(int64(durationBySpan[0]), int64(durationBySpan[1])))},
+ {Query: fmt.Sprintf("{ %s && %s } && { %s && %s } | max(duration) = %dns",
+ rando(trueConditionsBySpan[0]), rando(trueConditionsBySpan[0]),
+ rando(trueConditionsBySpan[1]), rando(trueConditionsBySpan[1]),
+ math.Max64(int64(durationBySpan[0]), int64(durationBySpan[1])))},
+ {Query: fmt.Sprintf("{ %s && %s } && { %s && %s } | sum(duration) = %dns",
+ rando(trueConditionsBySpan[0]), rando(trueConditionsBySpan[0]),
+ rando(trueConditionsBySpan[1]), rando(trueConditionsBySpan[1]),
+ durationBySpan[0]+durationBySpan[1])},
  }
  searchesThatDontMatch := []*tempopb.SearchRequest{
  // conditions
@@ -204,6 +221,9 @@ func testAdvancedTraceQLCompleteBlock(t *testing.T, blockVersion string) {
  // avgs
  {Query: "{ } | avg(.dne) != 0"},
  {Query: "{ } | avg(duration) < 0"},
+ {Query: "{ } | min(duration) < 0"},
+ {Query: "{ } | max(duration) < 0"},
+ {Query: "{ } | sum(duration) < 0"},
  }
 
  for _, req := range searchesThatMatch {
@@ -439,7 +459,6 @@ func searchTestSuite() (
  searchesThatMatch []*tempopb.SearchRequest,
  searchesThatDontMatch []*tempopb.SearchRequest,
 ) {
-
  id = test.ValidTraceID(nil)
 
  start = 1000
@@ -499,7 +518,7 @@ func searchTestSuite() (
  TraceId: id,
  Name: "RootSpan",
  StartTimeUnixNano: uint64(1000 * time.Second),
- EndTimeUnixNano: uint64(1001 * time.Second),
+ EndTimeUnixNano: uint64(1002 * time.Second),
  Status: &v1.Status{},
  },
  },
@@ -512,7 +531,7 @@ func searchTestSuite() (
  expected = &tempopb.TraceSearchMetadata{
  TraceID: util.TraceIDToHexString(id),
  StartTimeUnixNano: uint64(1000 * time.Second),
- DurationMs: 1000,
+ DurationMs: 2000,
  RootServiceName: "RootService",
  RootTraceName: "RootSpan",
  }
@@ -524,7 +543,7 @@ func searchTestSuite() (
  },
  {
  MinDurationMs: 999,
- MaxDurationMs: 1001,
+ MaxDurationMs: 2001,
  },
  {
  Start: 1000,
@@ -579,7 +598,7 @@ func searchTestSuite() (
  // Excludes
  searchesThatDontMatch = []*tempopb.SearchRequest{
  {
- MinDurationMs: 1001,
+ MinDurationMs: 2001,
  },
  {
  MaxDurationMs: 999,