mirror of
https://github.com/SigNoz/signoz.git
synced 2026-07-15 19:00:34 +01:00
Compare commits
1 Commits
issue_5601
...
issue_5602
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
d250f190a7 |
76
pkg/querybuilder/expr_inspect.go
Normal file
76
pkg/querybuilder/expr_inspect.go
Normal file
@@ -0,0 +1,76 @@
|
||||
package querybuilder
|
||||
|
||||
import (
|
||||
"strings"
|
||||
|
||||
"github.com/SigNoz/signoz/pkg/errors"
|
||||
grammar "github.com/SigNoz/signoz/pkg/parser/filterquery/grammar"
|
||||
qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
|
||||
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
|
||||
"github.com/antlr4-go/antlr/v4"
|
||||
)
|
||||
|
||||
// ExprKeys returns the field keys referenced in *key positions* of a filter
|
||||
// expression. Unlike QueryStringToKeysSelectors (which scans raw KEY tokens and so
|
||||
// also picks up unquoted values — in `x > $threshold` it reports `$threshold`), this
|
||||
// walks the parse tree and collects only KeyContext nodes.
|
||||
func ExprKeys(query string) []*telemetrytypes.TelemetryFieldKey {
|
||||
var keys []*telemetrytypes.TelemetryFieldKey
|
||||
var walk func(node antlr.Tree)
|
||||
walk = func(node antlr.Tree) {
|
||||
if kc, ok := node.(*grammar.KeyContext); ok {
|
||||
key := telemetrytypes.GetFieldKeyFromKeyText(kc.GetText())
|
||||
keys = append(keys, &key)
|
||||
return
|
||||
}
|
||||
for i := 0; i < node.GetChildCount(); i++ {
|
||||
walk(node.GetChild(i))
|
||||
}
|
||||
}
|
||||
walk(parseFilterQuery(query))
|
||||
return keys
|
||||
}
|
||||
|
||||
// ValidateVariablesInExpr checks the variable references in an expression's value
|
||||
// positions upfront, so a broken reference fails with a targeted error instead of
|
||||
// the where-clause visitor's combined "Found N errors" (whose details ride in the
|
||||
// error's additionals). Lookup mirrors the visitor: verbatim, then with a leading
|
||||
// `$` stripped. A `$`-prefixed token that resolves to nothing is an error — it can
|
||||
// never be a valid literal; a bare token that resolves to nothing is left to mean
|
||||
// itself.
|
||||
func ValidateVariablesInExpr(query string, variables map[string]qbtypes.VariableItem) error {
|
||||
var err error
|
||||
var walk func(node antlr.Tree)
|
||||
walk = func(node antlr.Tree) {
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
if vc, ok := node.(*grammar.ValueContext); ok {
|
||||
// only unquoted textual values can be variable references
|
||||
if vc.KEY() == nil {
|
||||
return
|
||||
}
|
||||
text := vc.GetText()
|
||||
item, ok := variables[text]
|
||||
if !ok {
|
||||
item, ok = variables[strings.TrimPrefix(text, "$")]
|
||||
}
|
||||
if !ok {
|
||||
if strings.HasPrefix(text, "$") {
|
||||
err = errors.NewInvalidInputf(errors.CodeInvalidInput, "unknown variable %q", text)
|
||||
}
|
||||
return
|
||||
}
|
||||
if values, isList := item.Value.([]any); isList && len(values) == 0 {
|
||||
err = errors.NewInvalidInputf(errors.CodeInvalidInput,
|
||||
"variable %q used in expression has an empty list value", strings.TrimPrefix(text, "$"))
|
||||
}
|
||||
return
|
||||
}
|
||||
for i := 0; i < node.GetChildCount(); i++ {
|
||||
walk(node.GetChild(i))
|
||||
}
|
||||
}
|
||||
walk(parseFilterQuery(query))
|
||||
return err
|
||||
}
|
||||
39
pkg/querybuilder/expr_inspect_test.go
Normal file
39
pkg/querybuilder/expr_inspect_test.go
Normal file
@@ -0,0 +1,39 @@
|
||||
package querybuilder
|
||||
|
||||
import (
|
||||
"testing"
|
||||
|
||||
qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
func TestExprKeys(t *testing.T) {
|
||||
names := func(expr string) []string {
|
||||
var out []string
|
||||
for _, k := range ExprKeys(expr) {
|
||||
out = append(out, k.Name)
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// value-position tokens are not keys, unlike QueryStringToKeysSelectors
|
||||
require.Equal(t, []string{"output_tokens"}, names("output_tokens > $threshold"))
|
||||
require.Equal(t, []string{"trace.output_tokens"}, names("trace.output_tokens > 1000"))
|
||||
require.Equal(t, []string{"a", "b"}, names("a > 1 AND b IN ('x', 'y')"))
|
||||
}
|
||||
|
||||
func TestValidateVariablesInExpr(t *testing.T) {
|
||||
vars := map[string]qbtypes.VariableItem{
|
||||
"threshold": {Type: qbtypes.TextBoxVariableType, Value: float64(1000)},
|
||||
"empty": {Type: qbtypes.QueryVariableType, Value: []any{}},
|
||||
"all": {Type: qbtypes.DynamicVariableType, Value: "__all__"},
|
||||
}
|
||||
|
||||
require.NoError(t, ValidateVariablesInExpr("x > $threshold", vars))
|
||||
require.NoError(t, ValidateVariablesInExpr("x > threshold", vars))
|
||||
require.NoError(t, ValidateVariablesInExpr("x IN $all", vars))
|
||||
require.NoError(t, ValidateVariablesInExpr("m = 'cost$usd'", vars)) // quoted literals are not references
|
||||
require.NoError(t, ValidateVariablesInExpr("x > bare_word", vars)) // bare non-variable means itself
|
||||
require.ErrorContains(t, ValidateVariablesInExpr("x > $bogus", vars), `unknown variable "$bogus"`)
|
||||
require.ErrorContains(t, ValidateVariablesInExpr("x IN $empty", vars), "empty list")
|
||||
}
|
||||
@@ -50,6 +50,11 @@ func (genAIColumnProvider) Columns() []scopedtraces.TraceColumn {
|
||||
|
||||
func (genAIColumnProvider) DefaultOrderAlias() string { return "last_activity_time" }
|
||||
|
||||
// Trace-level aggregations only see traces with LLM activity in the window/bucket —
|
||||
// a tool/agent-only slice would contribute NULL tokens but still count as a trace,
|
||||
// making count(trace.trace_id) and avg(trace.output_tokens) disagree.
|
||||
func (genAIColumnProvider) ActivityGateAlias() string { return "llm_call_count" }
|
||||
|
||||
func (p genAIColumnProvider) AggregateAliases() []string {
|
||||
// Derived from Columns() so a new column can't be forgotten; SpanLevel columns
|
||||
// are filtered span-level, so skip them.
|
||||
|
||||
@@ -915,8 +915,9 @@ func TestBuild_TraceList_TracefieldPrefixMatchesTracePrefix(t *testing.T) {
|
||||
require.Contains(t, err.Error(), "cannot be used")
|
||||
}
|
||||
|
||||
// Query variables in a trace-level condition are substituted into the HAVING (the
|
||||
// span path binds them via PrepareWhereClause; the HAVING is a text rewrite).
|
||||
// Query variables in a trace-level condition resolve through the standard filter
|
||||
// pipeline, exactly like span-level filters: bound args, list/IN handling, dynamic
|
||||
// __all__ dropping the condition.
|
||||
func TestBuild_TraceList_VariableInAggregateFilter(t *testing.T) {
|
||||
b := newTestBuilder(t)
|
||||
build := func(expr string, vars map[string]qbtypes.VariableItem) (*qbtypes.Statement, error) {
|
||||
@@ -928,17 +929,19 @@ func TestBuild_TraceList_VariableInAggregateFilter(t *testing.T) {
|
||||
}, vars)
|
||||
}
|
||||
|
||||
// scalar variable -> literal in HAVING
|
||||
// scalar variable -> replaced to a literal (canonical pkg/variables semantics),
|
||||
// then parsed and bound as an arg by the filter pipeline
|
||||
stmt, err := build("trace.output_tokens > $threshold",
|
||||
map[string]qbtypes.VariableItem{"threshold": {Value: 700}})
|
||||
require.NoError(t, err)
|
||||
require.Contains(t, stmt.Query, "HAVING output_tokens > 700")
|
||||
require.Contains(t, stmt.Query, "HAVING output_tokens > ?")
|
||||
require.Contains(t, stmt.Args, float64(700))
|
||||
|
||||
// list variable with IN
|
||||
stmt, err = build("trace.llm_call_count IN $counts",
|
||||
map[string]qbtypes.VariableItem{"counts": {Value: []any{1, 2}}})
|
||||
require.NoError(t, err)
|
||||
require.Contains(t, stmt.Query, "HAVING llm_call_count IN")
|
||||
require.Contains(t, stmt.Query, "HAVING llm_call_count IN (?, ?)")
|
||||
|
||||
// dynamic __all__ -> condition dropped, no HAVING at all
|
||||
stmt, err = build("trace.output_tokens > $threshold",
|
||||
|
||||
398
pkg/telemetryai/trace_aggregation_test.go
Normal file
398
pkg/telemetryai/trace_aggregation_test.go
Normal file
@@ -0,0 +1,398 @@
|
||||
package telemetryai
|
||||
|
||||
import (
|
||||
"context"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
|
||||
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
// Scalar / time-series (trace-level aggregation) Build tests. The
|
||||
// rewriteTraceAggregation unit tests live in pkg/telemetryscopedtraces; these
|
||||
// exercise the full builder through the gen_ai provider pair.
|
||||
|
||||
// Mixing domains across separate aggregations of one query is rejected.
|
||||
func TestBuild_Aggregation_MixedDomainsRejected(t *testing.T) {
|
||||
b := newTestBuilder(t)
|
||||
_, err := b.Build(context.Background(), testStartMs, testEndMs, qbtypes.RequestTypeScalar,
|
||||
qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
|
||||
Signal: telemetrytypes.SignalTraces, Source: telemetrytypes.SourceAI,
|
||||
Aggregations: []qbtypes.TraceAggregation{
|
||||
{Expression: "avg(trace.output_tokens)"},
|
||||
{Expression: "sum(gen_ai.usage.output_tokens)"},
|
||||
},
|
||||
}, nil)
|
||||
require.ErrorContains(t, err, "cannot be mixed")
|
||||
}
|
||||
|
||||
// A trace-level filter over an output-only aggregate is rejected on the
|
||||
// aggregation paths too (it is not computable in the mask-pruned scan).
|
||||
func TestBuild_Aggregation_OutputOnlyFilterRejected(t *testing.T) {
|
||||
b := newTestBuilder(t)
|
||||
for _, rt := range []qbtypes.RequestType{qbtypes.RequestTypeScalar, qbtypes.RequestTypeRaw} {
|
||||
_, err := b.Build(context.Background(), testStartMs, testEndMs, rt,
|
||||
qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
|
||||
Signal: telemetrytypes.SignalTraces, Source: telemetrytypes.SourceAI,
|
||||
Aggregations: []qbtypes.TraceAggregation{{Expression: "count()"}},
|
||||
Filter: &qbtypes.Filter{Expression: "trace.span_count > 3"},
|
||||
}, nil)
|
||||
require.ErrorContains(t, err, `aggregate "span_count" cannot be used`)
|
||||
}
|
||||
}
|
||||
|
||||
// Trace-level per-trace columns are rejected as group-by / order keys with a
|
||||
// targeted error (not the field mapper's generic "field not found").
|
||||
func TestBuild_Aggregation_GroupByOrderValidation(t *testing.T) {
|
||||
b := newTestBuilder(t)
|
||||
ctx := context.Background()
|
||||
|
||||
_, err := b.Build(ctx, testStartMs, testEndMs, qbtypes.RequestTypeScalar,
|
||||
qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
|
||||
Signal: telemetrytypes.SignalTraces, Source: telemetrytypes.SourceAI,
|
||||
Aggregations: []qbtypes.TraceAggregation{{Expression: "avg(trace.output_tokens)"}},
|
||||
GroupBy: []qbtypes.GroupByKey{{TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{Name: "trace.llm_call_count"}}},
|
||||
}, nil)
|
||||
require.ErrorContains(t, err, `grouping by trace-level aggregate "trace.llm_call_count" is not supported`)
|
||||
|
||||
_, err = b.Build(ctx, testStartMs, testEndMs, qbtypes.RequestTypeRaw,
|
||||
qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
|
||||
Signal: telemetrytypes.SignalTraces, Source: telemetrytypes.SourceAI,
|
||||
Order: []qbtypes.OrderBy{{Key: qbtypes.OrderByKey{TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{Name: "trace.output_tokens"}}}},
|
||||
}, nil)
|
||||
require.ErrorContains(t, err, `ordering the span list by trace-level aggregate "trace.output_tokens" is not supported`)
|
||||
|
||||
_, err = b.Build(ctx, testStartMs, testEndMs, qbtypes.RequestTypeScalar,
|
||||
qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
|
||||
Signal: telemetrytypes.SignalTraces, Source: telemetrytypes.SourceAI,
|
||||
Aggregations: []qbtypes.TraceAggregation{{Expression: "avg(trace.output_tokens)"}},
|
||||
Order: []qbtypes.OrderBy{{Key: qbtypes.OrderByKey{TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{Name: "trace.total_tokens"}}}},
|
||||
}, nil)
|
||||
require.ErrorContains(t, err, `ordering by trace-level aggregate "trace.total_tokens" is not supported`)
|
||||
|
||||
// ordering by the aggregation itself (expression or alias) stays valid
|
||||
_, err = b.Build(ctx, testStartMs, testEndMs, qbtypes.RequestTypeScalar,
|
||||
qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
|
||||
Signal: telemetrytypes.SignalTraces, Source: telemetrytypes.SourceAI,
|
||||
Aggregations: []qbtypes.TraceAggregation{{Expression: "avg(trace.output_tokens)", Alias: "avg_out"}},
|
||||
Order: []qbtypes.OrderBy{{Key: qbtypes.OrderByKey{TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{Name: "avg_out"}}, Direction: qbtypes.OrderDirectionAsc}},
|
||||
}, nil)
|
||||
require.NoError(t, err)
|
||||
}
|
||||
|
||||
// Query variables resolve inside trace-level filter conditions on every request type,
|
||||
// as bound args via the standard filter pipeline; unknown $vars fail with a variable
|
||||
// error, not an "unknown aggregate" one.
|
||||
func TestBuild_Aggregation_VariablesInTraceFilter(t *testing.T) {
|
||||
b := newTestBuilder(t)
|
||||
ctx := context.Background()
|
||||
vars := map[string]qbtypes.VariableItem{
|
||||
"threshold": {Type: qbtypes.TextBoxVariableType, Value: float64(1000)},
|
||||
}
|
||||
|
||||
for _, rt := range []qbtypes.RequestType{qbtypes.RequestTypeScalar, qbtypes.RequestTypeRaw, qbtypes.RequestTypeTrace} {
|
||||
q := qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
|
||||
Signal: telemetrytypes.SignalTraces, Source: telemetrytypes.SourceAI,
|
||||
Filter: &qbtypes.Filter{Expression: "trace.output_tokens > $threshold"},
|
||||
}
|
||||
if rt == qbtypes.RequestTypeScalar {
|
||||
q.Aggregations = []qbtypes.TraceAggregation{{Expression: "avg(trace.output_tokens)"}}
|
||||
}
|
||||
stmt, err := b.Build(ctx, testStartMs, testEndMs, rt, q, vars)
|
||||
require.NoError(t, err, rt.StringValue())
|
||||
require.Contains(t, stmt.Query, "HAVING output_tokens > ?", rt.StringValue())
|
||||
require.Contains(t, stmt.Args, float64(1000), rt.StringValue())
|
||||
|
||||
_, err = b.Build(ctx, testStartMs, testEndMs, rt, q, nil)
|
||||
require.ErrorContains(t, err, `unknown variable "$threshold"`, rt.StringValue())
|
||||
}
|
||||
|
||||
// a dynamic variable resolved to __all__ skips the trace-level condition, exactly
|
||||
// like span filters — no qualification CTE is built
|
||||
stmt, err := b.Build(ctx, testStartMs, testEndMs, qbtypes.RequestTypeScalar,
|
||||
qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
|
||||
Signal: telemetrytypes.SignalTraces, Source: telemetrytypes.SourceAI,
|
||||
Aggregations: []qbtypes.TraceAggregation{{Expression: "avg(trace.output_tokens)"}},
|
||||
Filter: &qbtypes.Filter{Expression: "trace.output_tokens IN $models"},
|
||||
}, map[string]qbtypes.VariableItem{
|
||||
"models": {Type: qbtypes.DynamicVariableType, Value: "__all__"},
|
||||
})
|
||||
require.NoError(t, err)
|
||||
require.NotContains(t, stmt.Query, "__qualified")
|
||||
|
||||
// list variables render as IN with bound args
|
||||
stmt, err = b.Build(ctx, testStartMs, testEndMs, qbtypes.RequestTypeScalar,
|
||||
qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
|
||||
Signal: telemetrytypes.SignalTraces, Source: telemetrytypes.SourceAI,
|
||||
Aggregations: []qbtypes.TraceAggregation{{Expression: "count(trace.trace_id)"}},
|
||||
Filter: &qbtypes.Filter{Expression: "trace.llm_call_count IN $counts"},
|
||||
}, map[string]qbtypes.VariableItem{
|
||||
"counts": {Type: qbtypes.QueryVariableType, Value: []any{float64(1), float64(2)}},
|
||||
})
|
||||
require.NoError(t, err)
|
||||
require.Contains(t, stmt.Query, "HAVING llm_call_count IN (?, ?)")
|
||||
}
|
||||
|
||||
// A resource-attribute condition prunes the qualification scan the same way it prunes
|
||||
// the trace list's matched pass: __qualified references the __resource_filter CTE, the
|
||||
// delegated __trace_scope inlines the fingerprint subquery.
|
||||
func TestBuild_Aggregation_QualificationResourcePruned(t *testing.T) {
|
||||
keys := otelKeysMap()
|
||||
keys["service.name"] = []*telemetrytypes.TelemetryFieldKey{{
|
||||
Name: "service.name",
|
||||
Signal: telemetrytypes.SignalTraces,
|
||||
FieldContext: telemetrytypes.FieldContextResource,
|
||||
FieldDataType: telemetrytypes.FieldDataTypeString,
|
||||
}}
|
||||
b := newTestBuilderWithKeys(t, keys)
|
||||
ctx := context.Background()
|
||||
filter := &qbtypes.Filter{Expression: "service.name = 'api' AND trace.output_tokens > 1000"}
|
||||
|
||||
stmt, err := b.Build(ctx, testStartMs, testEndMs, qbtypes.RequestTypeScalar,
|
||||
qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
|
||||
Signal: telemetrytypes.SignalTraces, Source: telemetrytypes.SourceAI,
|
||||
Aggregations: []qbtypes.TraceAggregation{{Expression: "avg(trace.output_tokens)"}},
|
||||
Filter: filter,
|
||||
}, nil)
|
||||
require.NoError(t, err)
|
||||
qualified := stmt.Query[strings.Index(stmt.Query, "__qualified"):strings.Index(stmt.Query, "__ai_traces")]
|
||||
require.Contains(t, qualified, "resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter)")
|
||||
|
||||
stmt, err = b.Build(ctx, testStartMs, testEndMs, qbtypes.RequestTypeRaw,
|
||||
qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
|
||||
Signal: telemetrytypes.SignalTraces, Source: telemetrytypes.SourceAI,
|
||||
Filter: filter,
|
||||
Limit: 10,
|
||||
}, nil)
|
||||
require.NoError(t, err)
|
||||
scope := stmt.Query[strings.Index(stmt.Query, "__trace_scope"):strings.Index(stmt.Query, "SELECT timestamp")]
|
||||
require.Contains(t, scope, "resource_fingerprint GLOBAL IN (SELECT fingerprint FROM (SELECT")
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Full-query goldens — native trace-domain pipeline
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
// Scalar over per-trace values, no filter: one window-clipped per-trace scan, outer
|
||||
// avg across traces.
|
||||
func TestBuild_FullSQL_Scalar_TraceAgg(t *testing.T) {
|
||||
b := newTestBuilder(t)
|
||||
stmt, err := b.Build(context.Background(), testStartMs, testEndMs, qbtypes.RequestTypeScalar,
|
||||
qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
|
||||
Signal: telemetrytypes.SignalTraces, Source: telemetrytypes.SourceAI,
|
||||
Aggregations: []qbtypes.TraceAggregation{{Expression: "avg(trace.output_tokens)"}},
|
||||
}, nil)
|
||||
require.NoError(t, err)
|
||||
|
||||
requireSQLEqual(t, `
|
||||
WITH __ai_traces AS (
|
||||
SELECT trace_id,
|
||||
sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.output_tokens') = true, toFloat64(attributes_number['gen_ai.usage.output_tokens']), NULL)) AS output_tokens
|
||||
FROM signoz_traces.distributed_signoz_index_v3
|
||||
WHERE timestamp >= '1747947419000000000'
|
||||
AND timestamp < '1747983448000000000'
|
||||
AND ts_bucket_start >= 1747945619
|
||||
AND ts_bucket_start <= 1747983448
|
||||
AND (mapContains(attributes_string, 'gen_ai.request.model') = true OR mapContains(attributes_string, 'gen_ai.tool.name') = true OR mapContains(attributes_string, 'gen_ai.agent.name') = true)
|
||||
GROUP BY trace_id
|
||||
HAVING (countIf(mapContains(attributes_string, 'gen_ai.request.model') = true)) > 0
|
||||
)
|
||||
SELECT avg(output_tokens) AS __result_0
|
||||
FROM __ai_traces
|
||||
ORDER BY __result_0 DESC
|
||||
SETTINGS distributed_product_mode='allow', max_memory_usage=10000000000
|
||||
`, stmt)
|
||||
}
|
||||
|
||||
// Scalar with a span-level + trace-level filter and a groupBy: __qualified holds the
|
||||
// whole-window qualification, the per-trace scan is per (trace, group) with the span
|
||||
// filter ANDed, the outer aggregation is per group.
|
||||
func TestBuild_FullSQL_Scalar_TraceAgg_FilterAndGroupBy(t *testing.T) {
|
||||
b := newTestBuilder(t)
|
||||
stmt, err := b.Build(context.Background(), testStartMs, testEndMs, qbtypes.RequestTypeScalar,
|
||||
qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
|
||||
Signal: telemetrytypes.SignalTraces, Source: telemetrytypes.SourceAI,
|
||||
Aggregations: []qbtypes.TraceAggregation{{Expression: "sum(trace.total_tokens)"}},
|
||||
Filter: &qbtypes.Filter{Expression: "gen_ai.request.model = 'gpt-4o-mini' AND trace.output_tokens > 1000"},
|
||||
GroupBy: []qbtypes.GroupByKey{
|
||||
{TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{Name: "gen_ai.request.model"}},
|
||||
},
|
||||
Limit: 5,
|
||||
}, nil)
|
||||
require.NoError(t, err)
|
||||
|
||||
requireSQLEqual(t, `
|
||||
WITH __qualified AS (
|
||||
SELECT trace_id,
|
||||
sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.output_tokens') = true, toFloat64(attributes_number['gen_ai.usage.output_tokens']), NULL)) AS output_tokens
|
||||
FROM signoz_traces.distributed_signoz_index_v3
|
||||
WHERE timestamp >= '1747947419000000000'
|
||||
AND timestamp < '1747983448000000000'
|
||||
AND ts_bucket_start >= 1747945619
|
||||
AND ts_bucket_start <= 1747983448
|
||||
AND (mapContains(attributes_string, 'gen_ai.request.model') = true OR mapContains(attributes_string, 'gen_ai.tool.name') = true OR mapContains(attributes_string, 'gen_ai.agent.name') = true)
|
||||
GROUP BY trace_id
|
||||
HAVING output_tokens > 1000
|
||||
),
|
||||
__ai_traces AS (
|
||||
SELECT trace_id,
|
||||
toString(multiIf(mapContains(attributes_string, 'gen_ai.request.model') = true, attributes_string['gen_ai.request.model'], NULL)) AS gen_ai.request.model,
|
||||
coalesce(sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.input_tokens') = true, toFloat64(attributes_number['gen_ai.usage.input_tokens']), NULL)), 0) + coalesce(sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.output_tokens') = true, toFloat64(attributes_number['gen_ai.usage.output_tokens']), NULL)), 0) AS total_tokens
|
||||
FROM signoz_traces.distributed_signoz_index_v3
|
||||
WHERE timestamp >= '1747947419000000000'
|
||||
AND timestamp < '1747983448000000000'
|
||||
AND ts_bucket_start >= 1747945619
|
||||
AND ts_bucket_start <= 1747983448
|
||||
AND (mapContains(attributes_string, 'gen_ai.request.model') = true OR mapContains(attributes_string, 'gen_ai.tool.name') = true OR mapContains(attributes_string, 'gen_ai.agent.name') = true)
|
||||
AND (attributes_string['gen_ai.request.model'] = 'gpt-4o-mini' AND mapContains(attributes_string, 'gen_ai.request.model') = true)
|
||||
AND trace_id GLOBAL IN (SELECT trace_id FROM __qualified)
|
||||
GROUP BY trace_id, gen_ai.request.model
|
||||
HAVING (countIf(mapContains(attributes_string, 'gen_ai.request.model') = true)) > 0
|
||||
)
|
||||
SELECT gen_ai.request.model, sum(total_tokens) AS __result_0
|
||||
FROM __ai_traces
|
||||
GROUP BY gen_ai.request.model
|
||||
ORDER BY __result_0 DESC
|
||||
LIMIT 5
|
||||
SETTINGS distributed_product_mode='allow', max_memory_usage=10000000000
|
||||
`, stmt)
|
||||
}
|
||||
|
||||
// Time series over per-trace values: the per-trace scan buckets by span time
|
||||
// (per-bucket clipping), the outer aggregation is per bucket.
|
||||
func TestBuild_FullSQL_TimeSeries_TraceAgg(t *testing.T) {
|
||||
b := newTestBuilder(t)
|
||||
stmt, err := b.Build(context.Background(), testStartMs, testEndMs, qbtypes.RequestTypeTimeSeries,
|
||||
qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
|
||||
Signal: telemetrytypes.SignalTraces, Source: telemetrytypes.SourceAI,
|
||||
StepInterval: qbtypes.Step{Duration: 60 * time.Second},
|
||||
Aggregations: []qbtypes.TraceAggregation{{Expression: "avg(trace.output_tokens)"}},
|
||||
}, nil)
|
||||
require.NoError(t, err)
|
||||
|
||||
requireSQLEqual(t, `
|
||||
WITH __ai_traces AS (
|
||||
SELECT trace_id,
|
||||
toStartOfInterval(timestamp, INTERVAL 60 SECOND) AS ts,
|
||||
sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.output_tokens') = true, toFloat64(attributes_number['gen_ai.usage.output_tokens']), NULL)) AS output_tokens
|
||||
FROM signoz_traces.distributed_signoz_index_v3
|
||||
WHERE timestamp >= '1747947419000000000'
|
||||
AND timestamp < '1747983448000000000'
|
||||
AND ts_bucket_start >= 1747945619
|
||||
AND ts_bucket_start <= 1747983448
|
||||
AND (mapContains(attributes_string, 'gen_ai.request.model') = true OR mapContains(attributes_string, 'gen_ai.tool.name') = true OR mapContains(attributes_string, 'gen_ai.agent.name') = true)
|
||||
GROUP BY trace_id, ts
|
||||
HAVING (countIf(mapContains(attributes_string, 'gen_ai.request.model') = true)) > 0
|
||||
)
|
||||
SELECT ts, avg(output_tokens) AS __result_0
|
||||
FROM __ai_traces
|
||||
GROUP BY ts
|
||||
SETTINGS distributed_product_mode='allow', max_memory_usage=10000000000
|
||||
`, stmt)
|
||||
}
|
||||
|
||||
// Grouped, limited time series: groups are ranked on whole-window per-trace values
|
||||
// (__ai_traces_total → __limit_cte) and the main per-bucket query is constrained to
|
||||
// the top-N groups.
|
||||
func TestBuild_FullSQL_TimeSeries_TraceAgg_TopN(t *testing.T) {
|
||||
b := newTestBuilder(t)
|
||||
stmt, err := b.Build(context.Background(), testStartMs, testEndMs, qbtypes.RequestTypeTimeSeries,
|
||||
qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
|
||||
Signal: telemetrytypes.SignalTraces, Source: telemetrytypes.SourceAI,
|
||||
StepInterval: qbtypes.Step{Duration: 60 * time.Second},
|
||||
Aggregations: []qbtypes.TraceAggregation{{Expression: "avg(trace.output_tokens)"}},
|
||||
GroupBy: []qbtypes.GroupByKey{
|
||||
{TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{Name: "gen_ai.request.model"}},
|
||||
},
|
||||
Limit: 3,
|
||||
}, nil)
|
||||
require.NoError(t, err)
|
||||
|
||||
requireSQLEqual(t, `
|
||||
WITH __ai_traces_total AS (
|
||||
SELECT trace_id,
|
||||
toString(multiIf(mapContains(attributes_string, 'gen_ai.request.model') = true, attributes_string['gen_ai.request.model'], NULL)) AS gen_ai.request.model,
|
||||
sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.output_tokens') = true, toFloat64(attributes_number['gen_ai.usage.output_tokens']), NULL)) AS output_tokens
|
||||
FROM signoz_traces.distributed_signoz_index_v3
|
||||
WHERE timestamp >= '1747947419000000000'
|
||||
AND timestamp < '1747983448000000000'
|
||||
AND ts_bucket_start >= 1747945619
|
||||
AND ts_bucket_start <= 1747983448
|
||||
AND (mapContains(attributes_string, 'gen_ai.request.model') = true OR mapContains(attributes_string, 'gen_ai.tool.name') = true OR mapContains(attributes_string, 'gen_ai.agent.name') = true)
|
||||
GROUP BY trace_id, gen_ai.request.model
|
||||
HAVING (countIf(mapContains(attributes_string, 'gen_ai.request.model') = true)) > 0
|
||||
),
|
||||
__limit_cte AS (
|
||||
SELECT gen_ai.request.model, avg(output_tokens) AS __result_0
|
||||
FROM __ai_traces_total
|
||||
GROUP BY gen_ai.request.model
|
||||
ORDER BY __result_0 DESC
|
||||
LIMIT 3
|
||||
),
|
||||
__ai_traces AS (
|
||||
SELECT trace_id,
|
||||
toStartOfInterval(timestamp, INTERVAL 60 SECOND) AS ts,
|
||||
toString(multiIf(mapContains(attributes_string, 'gen_ai.request.model') = true, attributes_string['gen_ai.request.model'], NULL)) AS gen_ai.request.model,
|
||||
sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.output_tokens') = true, toFloat64(attributes_number['gen_ai.usage.output_tokens']), NULL)) AS output_tokens
|
||||
FROM signoz_traces.distributed_signoz_index_v3
|
||||
WHERE timestamp >= '1747947419000000000'
|
||||
AND timestamp < '1747983448000000000'
|
||||
AND ts_bucket_start >= 1747945619
|
||||
AND ts_bucket_start <= 1747983448
|
||||
AND (mapContains(attributes_string, 'gen_ai.request.model') = true OR mapContains(attributes_string, 'gen_ai.tool.name') = true OR mapContains(attributes_string, 'gen_ai.agent.name') = true)
|
||||
GROUP BY trace_id, ts, gen_ai.request.model
|
||||
HAVING (countIf(mapContains(attributes_string, 'gen_ai.request.model') = true)) > 0
|
||||
)
|
||||
SELECT ts, gen_ai.request.model, avg(output_tokens) AS __result_0
|
||||
FROM __ai_traces
|
||||
WHERE (gen_ai.request.model) IN (SELECT gen_ai.request.model FROM __limit_cte)
|
||||
GROUP BY ts, gen_ai.request.model
|
||||
SETTINGS distributed_product_mode='allow', max_memory_usage=10000000000
|
||||
`, stmt)
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Full-query goldens — delegated span-domain with a trace-level filter
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
// Span-level scalar with a trace-level filter: delegated to the trace builder with
|
||||
// the gate ANDed, constrained by the __trace_scope qualification.
|
||||
func TestBuild_FullSQL_Scalar_SpanAgg_TraceScoped(t *testing.T) {
|
||||
b := newTestBuilder(t)
|
||||
stmt, err := b.Build(context.Background(), testStartMs, testEndMs, qbtypes.RequestTypeScalar,
|
||||
qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
|
||||
Signal: telemetrytypes.SignalTraces, Source: telemetrytypes.SourceAI,
|
||||
Aggregations: []qbtypes.TraceAggregation{{Expression: "sum(gen_ai.usage.output_tokens)"}},
|
||||
Filter: &qbtypes.Filter{Expression: "trace.output_tokens > 1000"},
|
||||
}, nil)
|
||||
require.NoError(t, err)
|
||||
|
||||
requireSQLEqual(t, `
|
||||
WITH __trace_scope AS (
|
||||
SELECT trace_id,
|
||||
sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.output_tokens') = true, toFloat64(attributes_number['gen_ai.usage.output_tokens']), NULL)) AS output_tokens
|
||||
FROM signoz_traces.distributed_signoz_index_v3
|
||||
WHERE timestamp >= '1747947419000000000'
|
||||
AND timestamp < '1747983448000000000'
|
||||
AND ts_bucket_start >= 1747945619
|
||||
AND ts_bucket_start <= 1747983448
|
||||
AND (mapContains(attributes_string, 'gen_ai.request.model') = true OR mapContains(attributes_string, 'gen_ai.tool.name') = true OR mapContains(attributes_string, 'gen_ai.agent.name') = true)
|
||||
GROUP BY trace_id
|
||||
HAVING output_tokens > 1000
|
||||
)
|
||||
SELECT sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.output_tokens') = true, toFloat64(attributes_number['gen_ai.usage.output_tokens']), NULL)) AS __result_0
|
||||
FROM signoz_traces.distributed_signoz_index_v3
|
||||
WHERE trace_id GLOBAL IN (SELECT trace_id FROM __trace_scope)
|
||||
AND (mapContains(attributes_string, 'gen_ai.request.model') = true
|
||||
OR mapContains(attributes_string, 'gen_ai.tool.name') = true
|
||||
OR mapContains(attributes_string, 'gen_ai.agent.name') = true)
|
||||
AND timestamp >= '1747947419000000000'
|
||||
AND timestamp < '1747983448000000000'
|
||||
AND ts_bucket_start >= 1747945619
|
||||
AND ts_bucket_start <= 1747983448
|
||||
ORDER BY __result_0 DESC
|
||||
`, stmt)
|
||||
}
|
||||
@@ -110,7 +110,7 @@ func TestBuild_SpanList_TraceFilter_Validation(t *testing.T) {
|
||||
}
|
||||
|
||||
// Variables in a trace-level condition on the span list get the trace list's
|
||||
// treatment: substituted as literals, __all__ drops the condition (no scope CTE),
|
||||
// treatment: resolved and bound as args, __all__ drops the condition (no scope CTE),
|
||||
// tracefield. spelling behaves like trace..
|
||||
func TestBuild_SpanList_TraceFilter_Variables(t *testing.T) {
|
||||
b := newTestBuilder(t)
|
||||
@@ -126,7 +126,8 @@ func TestBuild_SpanList_TraceFilter_Variables(t *testing.T) {
|
||||
stmt, err := build("trace.output_tokens > $threshold",
|
||||
map[string]qbtypes.VariableItem{"threshold": {Value: 700}})
|
||||
require.NoError(t, err)
|
||||
require.Contains(t, stmt.Query, "HAVING output_tokens > 700")
|
||||
require.Contains(t, stmt.Query, "HAVING output_tokens > ?")
|
||||
require.Contains(t, stmt.Args, float64(700))
|
||||
|
||||
stmt, err = build("trace.output_tokens > $threshold",
|
||||
map[string]qbtypes.VariableItem{"threshold": {Type: qbtypes.DynamicVariableType, Value: "__all__"}})
|
||||
|
||||
@@ -165,6 +165,9 @@ type ColumnProvider interface {
|
||||
// AggregateAliases are the computed per-trace column names, used to classify a
|
||||
// filter key as trace-level vs span-level. Excludes SpanLevel columns.
|
||||
AggregateAliases() []string
|
||||
// ActivityGateAlias names the column that must be > 0 for a per-trace row to feed
|
||||
// trace-level (trace.) aggregations; empty disables the gate.
|
||||
ActivityGateAlias() string
|
||||
}
|
||||
|
||||
// CommonTraceColumns are domain-neutral columns any trace list can reuse. All
|
||||
|
||||
@@ -4,7 +4,6 @@ import (
|
||||
"context"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"sort"
|
||||
"strings"
|
||||
|
||||
"github.com/SigNoz/signoz/pkg/errors"
|
||||
@@ -16,7 +15,6 @@ import (
|
||||
"github.com/SigNoz/signoz/pkg/telemetrytraces"
|
||||
qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
|
||||
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
|
||||
qbvariables "github.com/SigNoz/signoz/pkg/variables"
|
||||
"github.com/huandu/go-sqlbuilder"
|
||||
)
|
||||
|
||||
@@ -102,10 +100,12 @@ func (b *scopedTraceStatementBuilder) Build(
|
||||
case qbtypes.RequestTypeTrace:
|
||||
return b.buildTraceListQuery(ctx, querybuilder.ToNanoSecs(start), querybuilder.ToNanoSecs(end), query, variables)
|
||||
case qbtypes.RequestTypeRaw:
|
||||
if err := b.validateRawOrderKeys(query); err != nil {
|
||||
if err := b.validateGroupByAndOrder(requestType, query); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return b.buildDelegated(ctx, start, end, requestType, query, variables)
|
||||
case qbtypes.RequestTypeScalar, qbtypes.RequestTypeTimeSeries:
|
||||
return b.buildAggregation(ctx, start, end, requestType, query, variables)
|
||||
default:
|
||||
return nil, ErrUnsupportedRequestType
|
||||
}
|
||||
@@ -117,38 +117,28 @@ type traceScopedStatementBuilder interface {
|
||||
BuildTraceScoped(ctx context.Context, start, end uint64, requestType qbtypes.RequestType, query qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation], variables map[string]qbtypes.VariableItem, traceScope *qbtypes.Statement) (*qbtypes.Statement, error)
|
||||
}
|
||||
|
||||
// substituteTraceLevelVariables resolves query variables in a trace-level expression.
|
||||
// The span-level parts bind variables via PrepareWhereClause; trace-level parts are
|
||||
// text rewrites, so variables become literals (a dynamic __all__ drops its condition).
|
||||
func substituteTraceLevelVariables(expr string, variables map[string]qbtypes.VariableItem) (string, error) {
|
||||
if strings.TrimSpace(expr) == "" || len(variables) == 0 {
|
||||
return expr, nil
|
||||
// splitUserFilter partitions query.Filter into a span-level expression (re-parsed by
|
||||
// the delegate / span predicate resolution) and the resolved trace-level part (nil
|
||||
// when there is none, or when every trace-level condition was skipped).
|
||||
func (b *scopedTraceStatementBuilder) splitUserFilter(ctx context.Context, query qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation], variables map[string]qbtypes.VariableItem) (string, *traceHaving, error) {
|
||||
if query.Filter == nil || strings.TrimSpace(query.Filter.Expression) == "" {
|
||||
return "", nil, nil
|
||||
}
|
||||
return qbvariables.ReplaceVariablesInExpression(expr, variables)
|
||||
spanExpr, traceExpr, err := querybuilder.SplitFilterForAggregates(query.Filter.Expression, b.aggregateAliasSet())
|
||||
if err != nil {
|
||||
return "", nil, err
|
||||
}
|
||||
having, err := b.resolveTraceHaving(ctx, traceExpr, variables)
|
||||
if err != nil {
|
||||
return "", nil, err
|
||||
}
|
||||
return spanExpr, having, nil
|
||||
}
|
||||
|
||||
// validateRawOrderKeys rejects ordering the span list by an explicitly trace-level
|
||||
// aggregate (trace./tracefield. prefix or trace field context) — the per-trace value
|
||||
// does not exist on span rows. Bare names pass through: they may legitimately be span
|
||||
// columns (duration_nano, timestamp).
|
||||
func (b *scopedTraceStatementBuilder) validateRawOrderKeys(query qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]) error {
|
||||
aliases := b.aggregateAliasSet()
|
||||
for _, o := range query.Order {
|
||||
name := strings.TrimPrefix(strings.TrimPrefix(o.Key.Name, "tracefield."), "trace.")
|
||||
if _, ok := aliases[name]; !ok {
|
||||
continue
|
||||
}
|
||||
if name != o.Key.Name || o.Key.FieldContext == telemetrytypes.FieldContextTrace {
|
||||
return errors.NewInvalidInputf(errors.CodeInvalidInput,
|
||||
"ordering the span list by trace-level aggregate %q is not supported; order by span columns instead (e.g. timestamp, duration_nano)", o.Key.Name)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// buildDelegated ANDs the base gate into the span-level filter part and delegates to
|
||||
// the standard trace builder. A trace-level part (trace.output_tokens > 1000) becomes
|
||||
// a __trace_scope qualification the delegate constrains trace_id by.
|
||||
// buildDelegated splits the user filter, ANDs the base gate into its span-level part,
|
||||
// and delegates to the standard trace builder. A trace-level part (trace.output_tokens
|
||||
// > 1000) becomes a window-clipped qualification the delegate constrains trace_id by.
|
||||
// Serves the span list (raw) and span-level scalar/time-series.
|
||||
func (b *scopedTraceStatementBuilder) buildDelegated(
|
||||
ctx context.Context,
|
||||
start, end uint64,
|
||||
@@ -156,15 +146,7 @@ func (b *scopedTraceStatementBuilder) buildDelegated(
|
||||
query qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation],
|
||||
variables map[string]qbtypes.VariableItem,
|
||||
) (*qbtypes.Statement, error) {
|
||||
spanExpr, traceExpr := "", ""
|
||||
if query.Filter != nil && strings.TrimSpace(query.Filter.Expression) != "" {
|
||||
var err error
|
||||
spanExpr, traceExpr, err = querybuilder.SplitFilterForAggregates(query.Filter.Expression, b.aggregateAliasSet())
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
traceExpr, err := substituteTraceLevelVariables(traceExpr, variables)
|
||||
spanExpr, having, err := b.splitUserFilter(ctx, query, variables)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
@@ -179,7 +161,7 @@ func (b *scopedTraceStatementBuilder) buildDelegated(
|
||||
gated := query
|
||||
gated.Filter = &qbtypes.Filter{Expression: expr}
|
||||
|
||||
if strings.TrimSpace(traceExpr) == "" {
|
||||
if having == nil {
|
||||
return b.traceStmtBuilder.Build(ctx, start, end, requestType, gated, variables)
|
||||
}
|
||||
|
||||
@@ -187,69 +169,13 @@ func (b *scopedTraceStatementBuilder) buildDelegated(
|
||||
if !ok {
|
||||
return nil, errors.NewInternalf(errors.CodeInternal, "trace statement builder does not support trace-scoped queries")
|
||||
}
|
||||
scope, err := b.buildQualifiedStatement(ctx, querybuilder.ToNanoSecs(start), querybuilder.ToNanoSecs(end), traceExpr)
|
||||
scope, err := b.buildQualifiedStatement(ctx, querybuilder.ToNanoSecs(start), querybuilder.ToNanoSecs(end), having, query, variables)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return scoped.BuildTraceScoped(ctx, start, end, requestType, gated, variables, scope)
|
||||
}
|
||||
|
||||
// buildQualifiedStatement builds the __trace_scope statement: trace ids whose
|
||||
// window-clipped per-trace aggregates satisfy traceExpr, from one windowed,
|
||||
// mask-pruned GROUP BY trace_id scan. start/end are ns.
|
||||
func (b *scopedTraceStatementBuilder) buildQualifiedStatement(ctx context.Context, start, end uint64, traceExpr string) (*qbtypes.Statement, error) {
|
||||
keys, err := b.fetchKeys(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
maskExpr, maskArgs, err := b.resolveMask(ctx, start, end, keys)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
resolved, err := b.resolveColumns(ctx, start, end, keys, maskExpr, maskArgs)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
orderableSet := orderableAliasSet(resolved)
|
||||
if err := validateAggregateFilter(traceExpr, orderableSet); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
having, err := b.buildHaving(traceExpr, orderableSet)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
startBucket := start/querybuilder.NsToSeconds - querybuilder.BucketAdjustment
|
||||
endBucket := end / querybuilder.NsToSeconds
|
||||
|
||||
sb := sqlbuilder.NewSelectBuilder()
|
||||
needed := neededMatchedAliases(nil, traceExpr, orderableSet)
|
||||
selects := []string{"trace_id"}
|
||||
for _, rc := range resolved {
|
||||
if _, ok := needed[rc.alias]; !ok {
|
||||
continue
|
||||
}
|
||||
colExpr, err := embedExpr(sb, rc.expr, rc.args)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
selects = append(selects, colExpr+" AS "+quoteAlias(rc.alias))
|
||||
}
|
||||
sb.Select(selects...)
|
||||
sb.From(spanTable())
|
||||
mask, err := embedExpr(sb, maskExpr, maskArgs)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
sb.Where(append(windowWhere(sb, start, end, startBucket, endBucket), mask)...)
|
||||
sb.GroupBy("trace_id")
|
||||
if strings.TrimSpace(having) != "" {
|
||||
sb.Having(having)
|
||||
}
|
||||
sql, args := sb.BuildWithFlavor(sqlbuilder.ClickHouse)
|
||||
return &qbtypes.Statement{Query: sql, Args: args}, nil
|
||||
}
|
||||
|
||||
// buildTraceListQuery wires the CTE pipeline (benchmarked, see ai-qb-handoff.md): one
|
||||
// windowed pass picks the top-N traces, then a bucket-pruned pass enriches only those.
|
||||
// Helpers appear in this file in the order they run. start/end are nanoseconds.
|
||||
@@ -306,7 +232,6 @@ func (b *scopedTraceStatementBuilder) buildTraceListQuery(
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
orderableSet := orderableAliasSet(resolved)
|
||||
|
||||
// If the filter references resource attributes, add a __resource_filter CTE and
|
||||
// narrow the matched scan by resource_fingerprint; skipResourceFilter then drops
|
||||
@@ -317,13 +242,13 @@ func (b *scopedTraceStatementBuilder) buildTraceListQuery(
|
||||
}
|
||||
|
||||
// Split the user filter: span-level predicate + trace-level HAVING expression.
|
||||
fp, err := b.splitFilter(ctx, query, b.aggregateAliasSet(), orderableSet, start, end, skipResourceFilter, variables)
|
||||
fp, err := b.splitFilter(ctx, query, b.aggregateAliasSet(), start, end, skipResourceFilter, variables)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
// matched → ranked → buckets → enrichment
|
||||
matchedFrag, matchedArgs, err := b.buildMatchedCTE(start, end, startBucket, endBucket, resolved, orders, orderableSet, maskExpr, maskArgs, fp, resourcePred, limit, query.Offset)
|
||||
matchedFrag, matchedArgs, err := b.buildMatchedCTE(start, end, startBucket, endBucket, resolved, orders, maskExpr, maskArgs, fp, resourcePred, limit, query.Offset)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
@@ -362,20 +287,37 @@ func (b *scopedTraceStatementBuilder) maybeAttachResourceFilter(
|
||||
start, end uint64,
|
||||
variables map[string]qbtypes.VariableItem,
|
||||
) (cteFrag string, cteArgs []any, fingerprintPred string, skipResourceFilter bool, err error) {
|
||||
stmt, skipResourceFilter, err := b.resolveResourceFilterStmt(ctx, query, start, end, variables)
|
||||
if err != nil || stmt == nil {
|
||||
return "", nil, "", skipResourceFilter, err
|
||||
}
|
||||
return fmt.Sprintf("__resource_filter AS (%s)", stmt.Query), stmt.Args,
|
||||
"resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter)", skipResourceFilter, nil
|
||||
}
|
||||
|
||||
// resolveResourceFilterStmt resolves the fingerprint statement for the resource
|
||||
// attributes in the query's filter; nil when there are none (or the skip decision
|
||||
// applies). skipResourceFilter follows the same contract as maybeAttachResourceFilter.
|
||||
func (b *scopedTraceStatementBuilder) resolveResourceFilterStmt(
|
||||
ctx context.Context,
|
||||
query qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation],
|
||||
start, end uint64,
|
||||
variables map[string]qbtypes.VariableItem,
|
||||
) (*qbtypes.Statement, bool, error) {
|
||||
if b.resourceFilterResolver == nil {
|
||||
return "", nil, "", false, nil
|
||||
return nil, false, nil
|
||||
}
|
||||
|
||||
if b.skipResourceFingerprintEnabled {
|
||||
decision, err := b.resourceFilterResolver.Resolve(ctx, query, start, end, variables)
|
||||
if err != nil {
|
||||
return "", nil, "", true, err
|
||||
return nil, true, err
|
||||
}
|
||||
switch decision {
|
||||
case qbtypes.ResourceFilterResolveKindNoOp:
|
||||
return "", nil, "", true, nil
|
||||
return nil, true, nil
|
||||
case qbtypes.ResourceFilterResolveKindFallback:
|
||||
return "", nil, "", false, nil
|
||||
return nil, false, nil
|
||||
}
|
||||
}
|
||||
|
||||
@@ -383,13 +325,12 @@ func (b *scopedTraceStatementBuilder) maybeAttachResourceFilter(
|
||||
ctx, start, end, qbtypes.RequestTypeRaw, query, variables,
|
||||
)
|
||||
if err != nil {
|
||||
return "", nil, "", true, err
|
||||
return nil, true, err
|
||||
}
|
||||
if stmt == nil {
|
||||
return "", nil, "", true, nil
|
||||
return nil, true, nil
|
||||
}
|
||||
return fmt.Sprintf("__resource_filter AS (%s)", stmt.Query), stmt.Args,
|
||||
"resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter)", true, nil
|
||||
return stmt, true, nil
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
@@ -560,28 +501,28 @@ func (b *scopedTraceStatementBuilder) resolveListOrders(order []qbtypes.OrderBy,
|
||||
}
|
||||
|
||||
// filterParts is the user filter split into a span-level predicate (widens the
|
||||
// matched WHERE prune and becomes a countIf existence check in HAVING) and a
|
||||
// trace-level HAVING expression.
|
||||
// matched WHERE prune and becomes a countIf existence check in HAVING) and the
|
||||
// resolved trace-level HAVING (nil when there is none).
|
||||
type filterParts struct {
|
||||
spanPred string
|
||||
spanArgs []any
|
||||
hasSpanFilter bool
|
||||
havingExpr string
|
||||
having *traceHaving
|
||||
warnings []string
|
||||
warningsURL string
|
||||
}
|
||||
|
||||
// splitFilter splits query.Filter into a span-level predicate and a trace-level
|
||||
// HAVING expression (an explicit query.Having is ANDed onto the latter), then
|
||||
// validates the trace-level part against the matched-pass aggregates.
|
||||
func (b *scopedTraceStatementBuilder) splitFilter(ctx context.Context, query qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation], classifySet, orderableSet map[string]struct{}, start, end uint64, skipResourceFilter bool, variables map[string]qbtypes.VariableItem) (filterParts, error) {
|
||||
// HAVING; an explicit query.Having is ANDed onto the latter before resolution.
|
||||
func (b *scopedTraceStatementBuilder) splitFilter(ctx context.Context, query qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation], classifySet map[string]struct{}, start, end uint64, skipResourceFilter bool, variables map[string]qbtypes.VariableItem) (filterParts, error) {
|
||||
var fp filterParts
|
||||
havingExpr := ""
|
||||
if query.Filter != nil && strings.TrimSpace(query.Filter.Expression) != "" {
|
||||
spanExpr, traceExpr, err := querybuilder.SplitFilterForAggregates(query.Filter.Expression, classifySet)
|
||||
if err != nil {
|
||||
return fp, err
|
||||
}
|
||||
fp.havingExpr = traceExpr
|
||||
havingExpr = traceExpr
|
||||
if strings.TrimSpace(spanExpr) != "" {
|
||||
pred, args, warnings, url, err := b.resolveSpanPredicate(ctx, start, end, spanExpr, skipResourceFilter, variables)
|
||||
if err != nil {
|
||||
@@ -596,20 +537,17 @@ func (b *scopedTraceStatementBuilder) splitFilter(ctx context.Context, query qbt
|
||||
}
|
||||
}
|
||||
if query.Having != nil && strings.TrimSpace(query.Having.Expression) != "" {
|
||||
if fp.havingExpr != "" {
|
||||
fp.havingExpr = fmt.Sprintf("(%s) AND (%s)", fp.havingExpr, query.Having.Expression)
|
||||
if havingExpr != "" {
|
||||
havingExpr = fmt.Sprintf("(%s) AND (%s)", havingExpr, query.Having.Expression)
|
||||
} else {
|
||||
fp.havingExpr = query.Having.Expression
|
||||
havingExpr = query.Having.Expression
|
||||
}
|
||||
}
|
||||
havingExpr, err := substituteTraceLevelVariables(fp.havingExpr, variables)
|
||||
having, err := b.resolveTraceHaving(ctx, havingExpr, variables)
|
||||
if err != nil {
|
||||
return fp, err
|
||||
}
|
||||
fp.havingExpr = havingExpr
|
||||
if err := validateAggregateFilter(fp.havingExpr, orderableSet); err != nil {
|
||||
return fp, err
|
||||
}
|
||||
fp.having = having
|
||||
return fp, nil
|
||||
}
|
||||
|
||||
@@ -652,11 +590,11 @@ func (b *scopedTraceStatementBuilder) resolveSpanPredicate(ctx context.Context,
|
||||
// buildMatchedCTE builds `matched`: the single windowed GROUP BY trace_id scan that
|
||||
// fuses gate + span filter + HAVING + ORDER BY + LIMIT/OFFSET, selecting only the
|
||||
// aliases the ORDER BY / HAVING reference.
|
||||
func (b *scopedTraceStatementBuilder) buildMatchedCTE(start, end, startBucket, endBucket uint64, resolved []resolvedColumn, orders []listOrder, orderableSet map[string]struct{}, maskExpr string, maskArgs []any, fp filterParts, resourcePred string, limit, offset int) (string, []any, error) {
|
||||
func (b *scopedTraceStatementBuilder) buildMatchedCTE(start, end, startBucket, endBucket uint64, resolved []resolvedColumn, orders []listOrder, maskExpr string, maskArgs []any, fp filterParts, resourcePred string, limit, offset int) (string, []any, error) {
|
||||
sb := sqlbuilder.NewSelectBuilder()
|
||||
|
||||
// SELECT trace_id + only the aggregates ORDER BY / HAVING reference (as aliases).
|
||||
needed := neededMatchedAliases(orders, fp.havingExpr, orderableSet)
|
||||
needed := neededMatchedAliases(orders, fp.having)
|
||||
selects := []string{"trace_id"}
|
||||
for _, rc := range resolved {
|
||||
if _, ok := needed[rc.alias]; !ok {
|
||||
@@ -709,14 +647,12 @@ func (b *scopedTraceStatementBuilder) buildMatchedCTE(start, end, startBucket, e
|
||||
having = append(having, "countIf("+havingMask+") > 0")
|
||||
having = append(having, "countIf("+havingPred+") > 0")
|
||||
}
|
||||
if strings.TrimSpace(fp.havingExpr) != "" {
|
||||
hv, err := b.buildHaving(fp.havingExpr, orderableSet)
|
||||
if fp.having != nil {
|
||||
hv, err := embedExpr(sb, fp.having.pred, fp.having.args)
|
||||
if err != nil {
|
||||
return "", nil, err
|
||||
}
|
||||
if hv != "" {
|
||||
having = append(having, hv)
|
||||
}
|
||||
having = append(having, hv)
|
||||
}
|
||||
if len(having) > 0 {
|
||||
sb.Having(strings.Join(having, " AND "))
|
||||
@@ -783,18 +719,6 @@ func (b *scopedTraceStatementBuilder) buildEnrichmentSelect(resolved []resolvedC
|
||||
return sql, append(append([]any{}, selectArgs...), builtArgs...)
|
||||
}
|
||||
|
||||
// buildHaving rewrites a trace-level HAVING expression to the matched-pass column
|
||||
// aliases; bare, trace., and tracefield. forms all map to the same alias.
|
||||
func (b *scopedTraceStatementBuilder) buildHaving(havingExpr string, orderableSet map[string]struct{}) (string, error) {
|
||||
columnMap := make(map[string]string, len(orderableSet)*3)
|
||||
for a := range orderableSet {
|
||||
columnMap[a] = quoteAlias(a)
|
||||
columnMap["trace."+a] = quoteAlias(a)
|
||||
columnMap["tracefield."+a] = quoteAlias(a)
|
||||
}
|
||||
return querybuilder.NewHavingExpressionRewriter().Rewrite(havingExpr, columnMap)
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Small shared SQL-builder utilities
|
||||
// ---------------------------------------------------------------------------
|
||||
@@ -819,51 +743,35 @@ func (b *scopedTraceStatementBuilder) aggregateAliasSet() map[string]struct{} {
|
||||
return set
|
||||
}
|
||||
|
||||
// orderableAliasSet is the subset of aliases computable in the matched pass — the
|
||||
// only ones usable in ORDER BY and the aggregate filter.
|
||||
func orderableAliasSet(resolved []resolvedColumn) map[string]struct{} {
|
||||
set := make(map[string]struct{})
|
||||
for _, rc := range resolved {
|
||||
if rc.orderable {
|
||||
set[rc.alias] = struct{}{}
|
||||
}
|
||||
}
|
||||
return set
|
||||
}
|
||||
|
||||
// neededMatchedAliases is the minimal alias set the matched pass must select: those
|
||||
// in ORDER BY plus those in the aggregate HAVING. Everything else is left to the
|
||||
// enrichment scan.
|
||||
func neededMatchedAliases(orders []listOrder, havingExpr string, orderableSet map[string]struct{}) map[string]struct{} {
|
||||
// in ORDER BY plus those the resolved trace-level HAVING touches. Everything else is
|
||||
// left to the enrichment scan.
|
||||
func neededMatchedAliases(orders []listOrder, having *traceHaving) map[string]struct{} {
|
||||
needed := make(map[string]struct{})
|
||||
for _, o := range orders {
|
||||
needed[o.alias] = struct{}{}
|
||||
}
|
||||
for _, sel := range querybuilder.QueryStringToKeysSelectors(havingExpr) {
|
||||
name := strings.TrimPrefix(strings.TrimPrefix(sel.Name, "trace."), "tracefield.")
|
||||
if _, ok := orderableSet[name]; ok {
|
||||
needed[name] = struct{}{}
|
||||
if having != nil {
|
||||
for a := range having.used {
|
||||
needed[a] = struct{}{}
|
||||
}
|
||||
}
|
||||
return needed
|
||||
}
|
||||
|
||||
// validateAggregateFilter rejects a trace-level filter referencing an aggregate not
|
||||
// computable in the matched pass (e.g. span_count, duration_nano).
|
||||
// computable in the matched pass (e.g. span_count, duration_nano) with a targeted
|
||||
// top-level error — the same check inside the where-clause visitor would surface only
|
||||
// as a detail of a combined error. Key positions only: `x > $threshold` references x.
|
||||
func validateAggregateFilter(havingExpr string, orderableSet map[string]struct{}) error {
|
||||
if strings.TrimSpace(havingExpr) == "" {
|
||||
return nil
|
||||
}
|
||||
allowed := make([]string, 0, len(orderableSet))
|
||||
for a := range orderableSet {
|
||||
allowed = append(allowed, a)
|
||||
}
|
||||
sort.Strings(allowed)
|
||||
for _, sel := range querybuilder.QueryStringToKeysSelectors(havingExpr) {
|
||||
name := strings.TrimPrefix(strings.TrimPrefix(sel.Name, "trace."), "tracefield.")
|
||||
for _, key := range querybuilder.ExprKeys(havingExpr) {
|
||||
name := strings.TrimPrefix(strings.TrimPrefix(key.Name, "trace."), "tracefield.")
|
||||
if _, ok := orderableSet[name]; !ok {
|
||||
return errors.NewInvalidInputf(errors.CodeInvalidInput,
|
||||
"aggregate %q cannot be used in the trace-list filter; filterable aggregates: %s", name, strings.Join(allowed, ", "))
|
||||
"aggregate %q cannot be used in the trace-list filter; filterable aggregates: %s", name, strings.Join(sortedAliases(orderableSet), ", "))
|
||||
}
|
||||
}
|
||||
return nil
|
||||
|
||||
@@ -42,6 +42,7 @@ func (s stubColumns) Columns() []TraceColumn {
|
||||
TraceColumn{Alias: "scoped_span_count", Orderable: true, Expr: CountExists(s.key)})
|
||||
}
|
||||
func (s stubColumns) DefaultOrderAlias() string { return "scoped_span_count" }
|
||||
func (s stubColumns) ActivityGateAlias() string { return "" }
|
||||
func (s stubColumns) AggregateAliases() []string {
|
||||
aliases := make([]string, 0)
|
||||
for _, c := range s.Columns() {
|
||||
|
||||
785
pkg/telemetryscopedtraces/trace_aggregation.go
Normal file
785
pkg/telemetryscopedtraces/trace_aggregation.go
Normal file
@@ -0,0 +1,785 @@
|
||||
package telemetryscopedtraces
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"sort"
|
||||
"strings"
|
||||
|
||||
chparser "github.com/AfterShip/clickhouse-sql-parser/parser"
|
||||
"github.com/SigNoz/signoz/pkg/errors"
|
||||
"github.com/SigNoz/signoz/pkg/querybuilder"
|
||||
qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
|
||||
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
|
||||
"github.com/SigNoz/signoz/pkg/valuer"
|
||||
"github.com/huandu/go-sqlbuilder"
|
||||
)
|
||||
|
||||
// This file implements scalar / time-series for source=ai.
|
||||
//
|
||||
// Aggregations come in two domains, chosen per expression by the `trace.` prefix:
|
||||
// - span-level (bare keys): aggregate over individual gen_ai spans. Delegated to the
|
||||
// standard trace builder with the gate ANDed in; a trace-level filter part becomes
|
||||
// a __trace_scope qualification (see buildDelegated).
|
||||
// - trace-level (`trace.` prefix): aggregate over window-clipped per-trace values
|
||||
// (avg(trace.output_tokens) = average per trace). Runs the native pipeline below.
|
||||
//
|
||||
// Native pipeline (buildTraceAggregationQuery):
|
||||
//
|
||||
// __qualified traces whose window-clipped aggregates satisfy the trace-level
|
||||
// │ filter — whole-window values, so a trace qualifies once. Only
|
||||
// ▼ present when the filter has a trace-level part.
|
||||
// __ai_traces per-trace values: windowed, mask-pruned GROUP BY trace_id
|
||||
// │ (+ time bucket for time series → per-bucket clipping, + group-by
|
||||
// ▼ columns), spans filtered by gate AND span-level filter; rows with
|
||||
// main no LLM activity are dropped (activity gate). Outer aggregation over
|
||||
// the per-trace rows → __result_i.
|
||||
|
||||
// traceAggregation is one aggregation rewritten to run over the per-trace scan.
|
||||
type traceAggregation struct {
|
||||
expr string // rewritten SQL over the per-trace column aliases
|
||||
used map[string]struct{} // per-trace aliases referenced
|
||||
isRate bool
|
||||
}
|
||||
|
||||
// buildAggregation routes scalar/time-series requests by aggregation domain.
|
||||
func (b *scopedTraceStatementBuilder) buildAggregation(
|
||||
ctx context.Context,
|
||||
start, end uint64,
|
||||
requestType qbtypes.RequestType,
|
||||
query qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation],
|
||||
variables map[string]qbtypes.VariableItem,
|
||||
) (*qbtypes.Statement, error) {
|
||||
traceAggs, err := b.classifyAggregations(query.Aggregations)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if err := b.validateGroupByAndOrder(requestType, query); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if len(traceAggs) == 0 {
|
||||
return b.buildDelegated(ctx, start, end, requestType, query, variables)
|
||||
}
|
||||
return b.buildTraceAggregationQuery(ctx, querybuilder.ToNanoSecs(start), querybuilder.ToNanoSecs(end), requestType, query, variables, traceAggs)
|
||||
}
|
||||
|
||||
// classifyAggregations splits the aggregations into span-domain (delegated) vs
|
||||
// trace-domain (over per-trace values). Returns the rewritten trace-domain
|
||||
// aggregations, nil when all are span-domain; mixing the two domains is rejected.
|
||||
func (b *scopedTraceStatementBuilder) classifyAggregations(aggs []qbtypes.TraceAggregation) ([]traceAggregation, error) {
|
||||
traceCols := b.orderableColumnSet()
|
||||
var out []traceAggregation
|
||||
spanCount := 0
|
||||
for _, agg := range aggs {
|
||||
ta, isTrace, err := rewriteTraceAggregation(agg.Expression, traceCols)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if isTrace {
|
||||
out = append(out, *ta)
|
||||
} else {
|
||||
spanCount++
|
||||
}
|
||||
}
|
||||
if len(out) > 0 && spanCount > 0 {
|
||||
return nil, errors.NewInvalidInputf(errors.CodeInvalidInput,
|
||||
"span-level and trace-level (trace.) aggregations cannot be mixed in one query")
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// orderableColumnSet is the gen_ai-scoped per-trace column set (static, from the
|
||||
// provider) usable in trace-level aggregations and filters.
|
||||
func (b *scopedTraceStatementBuilder) orderableColumnSet() map[string]struct{} {
|
||||
set := make(map[string]struct{})
|
||||
for _, c := range b.columnProvider.Columns() {
|
||||
if c.Orderable {
|
||||
set[c.Alias] = struct{}{}
|
||||
}
|
||||
}
|
||||
return set
|
||||
}
|
||||
|
||||
// validateGroupByAndOrder rejects trace-level (trace.) per-trace columns used as a
|
||||
// group-by key or an order key with a targeted error, instead of the generic "field
|
||||
// not found" the field mapper would raise. An order key that names an aggregation
|
||||
// (alias / expression / index) is exempt — that is the way to order by a trace-level
|
||||
// aggregation's result.
|
||||
func (b *scopedTraceStatementBuilder) validateGroupByAndOrder(requestType qbtypes.RequestType, query qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]) error {
|
||||
aliases := b.aggregateAliasSet()
|
||||
for _, gb := range query.GroupBy {
|
||||
if isTraceLevelKey(gb.Name, gb.FieldContext, aliases) {
|
||||
return errors.NewInvalidInputf(errors.CodeInvalidInput,
|
||||
"grouping by trace-level aggregate %q is not supported; group by span attributes instead (e.g. gen_ai.request.model, service.name)", gb.Name)
|
||||
}
|
||||
}
|
||||
for _, o := range query.Order {
|
||||
if _, isAgg := traceAggOrderIndex(o, query); isAgg {
|
||||
continue
|
||||
}
|
||||
if !isTraceLevelKey(o.Key.Name, o.Key.FieldContext, aliases) {
|
||||
continue
|
||||
}
|
||||
if requestType == qbtypes.RequestTypeRaw {
|
||||
return errors.NewInvalidInputf(errors.CodeInvalidInput,
|
||||
"ordering the span list by trace-level aggregate %q is not supported; order by span columns instead (e.g. timestamp, duration_nano)", o.Key.Name)
|
||||
}
|
||||
return errors.NewInvalidInputf(errors.CodeInvalidInput,
|
||||
"ordering by trace-level aggregate %q is not supported; order by the aggregation itself (its alias or expression) or a group-by key", o.Key.Name)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// isTraceLevelKey reports whether a group-by / order key explicitly names a
|
||||
// trace-level per-trace aggregate (trace./tracefield. prefix or trace field context).
|
||||
// Bare names pass through: they may legitimately be span columns that share a name
|
||||
// with an aggregate alias (duration_nano, timestamp).
|
||||
func isTraceLevelKey(name string, fieldContext telemetrytypes.FieldContext, aliases map[string]struct{}) bool {
|
||||
stripped := strings.TrimPrefix(strings.TrimPrefix(name, "tracefield."), "trace.")
|
||||
if _, ok := aliases[stripped]; !ok {
|
||||
return false
|
||||
}
|
||||
return stripped != name || fieldContext == telemetrytypes.FieldContextTrace
|
||||
}
|
||||
|
||||
// rewriteTraceAggregation parses one aggregation expression. When it references
|
||||
// trace.-prefixed per-trace columns it returns the expression rewritten to run over
|
||||
// the per-trace scan (trace.output_tokens → output_tokens, arithmetic between
|
||||
// trace. columns allowed, function names mapped via AggreFuncMap) with isTrace=true;
|
||||
// a pure span-level expression returns isTrace=false and is left for the delegate.
|
||||
func rewriteTraceAggregation(expr string, traceCols map[string]struct{}) (*traceAggregation, bool, error) {
|
||||
p := chparser.NewParser("SELECT " + expr)
|
||||
stmts, err := p.ParseStmts()
|
||||
if err != nil {
|
||||
return nil, false, errors.WrapInvalidInputf(err, errors.CodeInvalidInput, "failed to parse aggregation expression %q", expr)
|
||||
}
|
||||
if len(stmts) == 0 {
|
||||
return nil, false, errors.NewInvalidInputf(errors.CodeInvalidInput, "invalid aggregation expression %q", expr)
|
||||
}
|
||||
sel, ok := stmts[0].(*chparser.SelectQuery)
|
||||
if !ok || len(sel.SelectItems) == 0 {
|
||||
return nil, false, errors.NewInvalidInputf(errors.CodeInvalidInput, "invalid aggregation expression %q", expr)
|
||||
}
|
||||
|
||||
v := &traceAggVisitor{traceCols: traceCols, used: make(map[string]struct{})}
|
||||
if err := sel.SelectItems[0].Accept(v); err != nil {
|
||||
return nil, false, err
|
||||
}
|
||||
if !v.hasTrace {
|
||||
return nil, false, nil
|
||||
}
|
||||
if v.hasSpan {
|
||||
return nil, false, errors.NewInvalidInputf(errors.CodeInvalidInput,
|
||||
"aggregation %q mixes trace-level (trace.) and span-level columns; use one domain per aggregation", expr)
|
||||
}
|
||||
return &traceAggregation{expr: sel.SelectItems[0].String(), used: v.used, isRate: v.isRate}, true, nil
|
||||
}
|
||||
|
||||
// traceAggVisitor walks the aggregation AST, classifying column references and
|
||||
// rewriting trace.-prefixed ones (bare paths, backquoted identifiers, and either
|
||||
// nested in arithmetic) to the per-trace column aliases in place. It keeps an
|
||||
// ancestor stack (Enter/Leave) to tell a column identifier from a path segment,
|
||||
// function name, or alias, and to reject trace. columns inside *If combinators.
|
||||
type traceAggVisitor struct {
|
||||
chparser.DefaultASTVisitor
|
||||
traceCols map[string]struct{}
|
||||
used map[string]struct{}
|
||||
stack []chparser.Expr
|
||||
hasTrace bool
|
||||
hasSpan bool
|
||||
isRate bool
|
||||
}
|
||||
|
||||
func (v *traceAggVisitor) Enter(expr chparser.Expr) { v.stack = append(v.stack, expr) }
|
||||
func (v *traceAggVisitor) Leave(expr chparser.Expr) { v.stack = v.stack[:len(v.stack)-1] }
|
||||
|
||||
// parent is the node enclosing the one currently being visited (the visited node
|
||||
// itself is the stack top).
|
||||
func (v *traceAggVisitor) parent() chparser.Expr {
|
||||
if len(v.stack) < 2 {
|
||||
return nil
|
||||
}
|
||||
return v.stack[len(v.stack)-2]
|
||||
}
|
||||
|
||||
// enclosingCombinator returns the name of a surrounding *If-combinator function, if any.
|
||||
func (v *traceAggVisitor) enclosingCombinator() (string, bool) {
|
||||
for _, e := range v.stack {
|
||||
fn, ok := e.(*chparser.FunctionExpr)
|
||||
if !ok {
|
||||
continue
|
||||
}
|
||||
if agg, known := querybuilder.AggreFuncMap[valuer.NewString(strings.ToLower(fn.Name.Name))]; known && agg.FuncCombinator {
|
||||
return fn.Name.Name, true
|
||||
}
|
||||
}
|
||||
return "", false
|
||||
}
|
||||
|
||||
// VisitPath classifies a dotted reference (trace.output_tokens); trace-level ones are
|
||||
// rewritten in place to the bare per-trace alias.
|
||||
func (v *traceAggVisitor) VisitPath(p *chparser.Path) error {
|
||||
col, isTrace := traceColumnRef(p.String())
|
||||
if !isTrace {
|
||||
v.hasSpan = true
|
||||
return nil
|
||||
}
|
||||
if err := v.acceptTraceColumn(p.String(), col); err != nil {
|
||||
return err
|
||||
}
|
||||
p.Fields = p.Fields[len(p.Fields)-1:]
|
||||
p.Fields[0].Name = col
|
||||
return nil
|
||||
}
|
||||
|
||||
// VisitIdent classifies a plain identifier: a backquoted `trace.output_tokens` is a
|
||||
// trace-level reference (rewritten in place); any other column identifier is
|
||||
// span-level. Path segments, function names, and aliases are structural, not columns.
|
||||
func (v *traceAggVisitor) VisitIdent(i *chparser.Ident) error {
|
||||
switch parent := v.parent().(type) {
|
||||
case *chparser.Path:
|
||||
return nil // segments are classified whole by VisitPath
|
||||
case *chparser.FunctionExpr:
|
||||
if parent.Name == i {
|
||||
return nil
|
||||
}
|
||||
case *chparser.ColumnExpr:
|
||||
if parent.Alias == i {
|
||||
return nil
|
||||
}
|
||||
}
|
||||
col, isTrace := traceColumnRef(i.Name)
|
||||
if !isTrace {
|
||||
v.hasSpan = true
|
||||
return nil
|
||||
}
|
||||
if err := v.acceptTraceColumn(i.Name, col); err != nil {
|
||||
return err
|
||||
}
|
||||
i.Name = col
|
||||
return nil
|
||||
}
|
||||
|
||||
// acceptTraceColumn validates one trace-level column reference and records it.
|
||||
func (v *traceAggVisitor) acceptTraceColumn(ref, col string) error {
|
||||
if name, in := v.enclosingCombinator(); in {
|
||||
return errors.NewInvalidInputf(errors.CodeInvalidInput,
|
||||
"%q over trace-level (trace.) columns is not supported; put the trace-level condition in the filter expression instead", name)
|
||||
}
|
||||
// trace_id is always selected by the per-trace scan (count(trace.trace_id)
|
||||
// counts traces); everything else must be a gen_ai-scoped column.
|
||||
if col != "trace_id" {
|
||||
if _, known := v.traceCols[col]; !known {
|
||||
return errors.NewInvalidInputf(errors.CodeInvalidInput,
|
||||
"unknown trace-level aggregation column %q; usable columns: %s", ref, strings.Join(sortedAliases(v.traceCols), ", "))
|
||||
}
|
||||
v.used[col] = struct{}{}
|
||||
}
|
||||
v.hasTrace = true
|
||||
return nil
|
||||
}
|
||||
|
||||
// VisitFunctionExpr validates and maps the function name. Children were already
|
||||
// visited (post-order), so classification is complete for this subtree.
|
||||
func (v *traceAggVisitor) VisitFunctionExpr(fn *chparser.FunctionExpr) error {
|
||||
name := strings.ToLower(fn.Name.Name)
|
||||
aggFunc, ok := querybuilder.AggreFuncMap[valuer.NewString(name)]
|
||||
if !ok {
|
||||
return errors.NewInvalidInputf(errors.CodeInvalidInput, "unrecognized function: %s", name)
|
||||
}
|
||||
if fn.Params != nil && fn.Params.Items != nil && len(fn.Params.Items.Items) > 0 && aggFunc.FuncCombinator {
|
||||
// combinator predicates over span columns stay span-level (countIf(has_error=true))
|
||||
v.hasSpan = true
|
||||
return nil
|
||||
}
|
||||
fn.Name.Name = aggFunc.FuncName
|
||||
if aggFunc.Rate {
|
||||
v.isRate = true
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// traceColumnRef reports whether text is a pure trace.-prefixed column reference
|
||||
// (trace.output_tokens / tracefield.output_tokens) and returns the bare column name.
|
||||
func traceColumnRef(text string) (string, bool) {
|
||||
text = strings.TrimSpace(text)
|
||||
var rest string
|
||||
if r, ok := strings.CutPrefix(text, "trace."); ok {
|
||||
rest = r
|
||||
} else if r, ok := strings.CutPrefix(text, "tracefield."); ok {
|
||||
rest = r
|
||||
} else {
|
||||
return "", false
|
||||
}
|
||||
if rest == "" || strings.ContainsAny(rest, " ()'\"`,+-*/<>=!") {
|
||||
return "", false
|
||||
}
|
||||
return rest, true
|
||||
}
|
||||
|
||||
func sortedAliases(set map[string]struct{}) []string {
|
||||
out := make([]string, 0, len(set))
|
||||
for a := range set {
|
||||
out = append(out, a)
|
||||
}
|
||||
sort.Strings(out)
|
||||
return out
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Qualification + per-trace scan
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
// buildQualifiedStatement builds the qualification statement — trace ids whose
|
||||
// window-clipped per-trace aggregates satisfy the resolved trace-level filter — used
|
||||
// as the delegate's __trace_scope and the native pipeline's __qualified. When the
|
||||
// query's filter references resource attributes, the scan is pruned to matching
|
||||
// resource fingerprints (inlined, since the caller embeds this statement standalone),
|
||||
// matching the trace list's matched pass. start/end are ns.
|
||||
func (b *scopedTraceStatementBuilder) buildQualifiedStatement(
|
||||
ctx context.Context,
|
||||
start, end uint64,
|
||||
having *traceHaving,
|
||||
query qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation],
|
||||
variables map[string]qbtypes.VariableItem,
|
||||
) (*qbtypes.Statement, error) {
|
||||
keys, err := b.fetchKeys(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
maskExpr, maskArgs, err := b.resolveMask(ctx, start, end, keys)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
resolved, err := b.resolveColumns(ctx, start, end, keys, maskExpr, maskArgs)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
var resourcePred string
|
||||
var resourceArgs []any
|
||||
if stmt, _, err := b.resolveResourceFilterStmt(ctx, query, start, end, variables); err != nil {
|
||||
return nil, err
|
||||
} else if stmt != nil {
|
||||
resourcePred = fmt.Sprintf("resource_fingerprint GLOBAL IN (SELECT fingerprint FROM (%s))", sqlbuilder.Escape(stmt.Query))
|
||||
resourceArgs = stmt.Args
|
||||
}
|
||||
sql, args, err := b.qualifiedScanSQL(start, end, having, resolved, maskExpr, maskArgs, resourcePred, resourceArgs)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &qbtypes.Statement{Query: sql, Args: args}, nil
|
||||
}
|
||||
|
||||
// qualifiedScanSQL renders the qualification scan given already-resolved columns.
|
||||
func (b *scopedTraceStatementBuilder) qualifiedScanSQL(start, end uint64, having *traceHaving, resolved []resolvedColumn, maskExpr string, maskArgs []any, resourcePred string, resourceArgs []any) (string, []any, error) {
|
||||
return b.buildPerTraceScan(start, end, resolved, maskExpr, maskArgs, perTraceScanOpts{
|
||||
needed: having.used,
|
||||
havingExpr: having.pred,
|
||||
havingArgs: having.args,
|
||||
resourcePred: resourcePred,
|
||||
resourcePredArgs: resourceArgs,
|
||||
})
|
||||
}
|
||||
|
||||
// groupColumn is a resolved span-attribute group-by column.
|
||||
type groupColumn struct {
|
||||
name string
|
||||
expr string
|
||||
args []any
|
||||
}
|
||||
|
||||
// perTraceScanOpts parametrize one windowed, mask-pruned GROUP BY trace_id scan.
|
||||
type perTraceScanOpts struct {
|
||||
stepSeconds int64 // >0 → bucket per-trace values by time (ts column)
|
||||
groupCols []groupColumn
|
||||
needed map[string]struct{} // per-trace aliases to select
|
||||
spanPred string // resolved span-level filter, ANDed per span
|
||||
spanPredArgs []any
|
||||
resourcePred string // resource-fingerprint prune (CTE reference or inline subquery)
|
||||
resourcePredArgs []any
|
||||
qualified bool // constrain to __qualified
|
||||
havingExpr string // resolved HAVING predicate over the selected aliases
|
||||
havingArgs []any
|
||||
activityExpr string // aggregate expr that must be > 0 for a row to survive (LLM-activity gate)
|
||||
activityArgs []any
|
||||
}
|
||||
|
||||
// buildPerTraceScan renders the scan: window + gate mask (+ span filter, resource
|
||||
// prune, qualification), grouped by trace_id (+ ts bucket, group-by columns).
|
||||
func (b *scopedTraceStatementBuilder) buildPerTraceScan(start, end uint64, resolved []resolvedColumn, maskExpr string, maskArgs []any, o perTraceScanOpts) (string, []any, error) {
|
||||
startBucket := start/querybuilder.NsToSeconds - querybuilder.BucketAdjustment
|
||||
endBucket := end / querybuilder.NsToSeconds
|
||||
|
||||
sb := sqlbuilder.NewSelectBuilder()
|
||||
selects := []string{"trace_id"}
|
||||
if o.stepSeconds > 0 {
|
||||
selects = append(selects, fmt.Sprintf("toStartOfInterval(timestamp, INTERVAL %d SECOND) AS ts", o.stepSeconds))
|
||||
}
|
||||
for _, gc := range o.groupCols {
|
||||
gcExpr, err := embedExpr(sb, gc.expr, gc.args)
|
||||
if err != nil {
|
||||
return "", nil, err
|
||||
}
|
||||
selects = append(selects, fmt.Sprintf("toString(%s) AS `%s`", gcExpr, gc.name))
|
||||
}
|
||||
for _, rc := range resolved {
|
||||
if _, ok := o.needed[rc.alias]; !ok {
|
||||
continue
|
||||
}
|
||||
colExpr, err := embedExpr(sb, rc.expr, rc.args)
|
||||
if err != nil {
|
||||
return "", nil, err
|
||||
}
|
||||
selects = append(selects, colExpr+" AS "+quoteAlias(rc.alias))
|
||||
}
|
||||
sb.Select(selects...)
|
||||
sb.From(spanTable())
|
||||
|
||||
where := windowWhere(sb, start, end, startBucket, endBucket)
|
||||
mask, err := embedExpr(sb, maskExpr, maskArgs)
|
||||
if err != nil {
|
||||
return "", nil, err
|
||||
}
|
||||
where = append(where, mask)
|
||||
if strings.TrimSpace(o.spanPred) != "" {
|
||||
pred, err := embedExpr(sb, o.spanPred, o.spanPredArgs)
|
||||
if err != nil {
|
||||
return "", nil, err
|
||||
}
|
||||
where = append(where, pred)
|
||||
}
|
||||
if o.resourcePred != "" {
|
||||
pred, err := embedExpr(sb, o.resourcePred, o.resourcePredArgs)
|
||||
if err != nil {
|
||||
return "", nil, err
|
||||
}
|
||||
where = append(where, pred)
|
||||
}
|
||||
if o.qualified {
|
||||
where = append(where, "trace_id GLOBAL IN (SELECT trace_id FROM __qualified)")
|
||||
}
|
||||
sb.Where(where...)
|
||||
|
||||
groupBy := []string{"trace_id"}
|
||||
if o.stepSeconds > 0 {
|
||||
groupBy = append(groupBy, "ts")
|
||||
}
|
||||
for _, gc := range o.groupCols {
|
||||
groupBy = append(groupBy, "`"+gc.name+"`")
|
||||
}
|
||||
sb.GroupBy(groupBy...)
|
||||
var having []string
|
||||
if strings.TrimSpace(o.activityExpr) != "" {
|
||||
activity, err := embedExpr(sb, o.activityExpr, o.activityArgs)
|
||||
if err != nil {
|
||||
return "", nil, err
|
||||
}
|
||||
having = append(having, "("+activity+") > 0")
|
||||
}
|
||||
if strings.TrimSpace(o.havingExpr) != "" {
|
||||
hv, err := embedExpr(sb, o.havingExpr, o.havingArgs)
|
||||
if err != nil {
|
||||
return "", nil, err
|
||||
}
|
||||
having = append(having, hv)
|
||||
}
|
||||
if len(having) > 0 {
|
||||
sb.Having(strings.Join(having, " AND "))
|
||||
}
|
||||
sql, args := sb.BuildWithFlavor(sqlbuilder.ClickHouse)
|
||||
return sql, args, nil
|
||||
}
|
||||
|
||||
// resolveGroupColumns resolves span-attribute group-by keys through the field mapper
|
||||
// (metadata-aware), for selection inside the per-trace scan.
|
||||
func (b *scopedTraceStatementBuilder) resolveGroupColumns(ctx context.Context, start, end uint64, groupBy []qbtypes.GroupByKey) ([]groupColumn, error) {
|
||||
if len(groupBy) == 0 {
|
||||
return nil, nil
|
||||
}
|
||||
selectors := make([]*telemetrytypes.FieldKeySelector, 0, len(groupBy))
|
||||
for i := range groupBy {
|
||||
selectors = append(selectors, &telemetrytypes.FieldKeySelector{
|
||||
Name: groupBy[i].Name,
|
||||
Signal: telemetrytypes.SignalTraces,
|
||||
FieldContext: groupBy[i].FieldContext,
|
||||
FieldDataType: groupBy[i].FieldDataType,
|
||||
SelectorMatchType: telemetrytypes.FieldSelectorMatchTypeExact,
|
||||
})
|
||||
}
|
||||
keys, _, err := b.metadataStore.GetKeysMulti(ctx, selectors)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
out := make([]groupColumn, 0, len(groupBy))
|
||||
for i := range groupBy {
|
||||
expr, args, err := querybuilder.CollisionHandledFinalExpr(ctx, start, end, &groupBy[i].TelemetryFieldKey, b.fm, b.cb, keys, telemetrytypes.FieldDataTypeString, nil, false)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
out = append(out, groupColumn{name: groupBy[i].Name, expr: expr, args: args})
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Native trace-domain aggregation query
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
// buildTraceAggregationQuery builds the native pipeline (see the file comment).
|
||||
// start/end are ns.
|
||||
func (b *scopedTraceStatementBuilder) buildTraceAggregationQuery(
|
||||
ctx context.Context,
|
||||
start, end uint64,
|
||||
requestType qbtypes.RequestType,
|
||||
query qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation],
|
||||
variables map[string]qbtypes.VariableItem,
|
||||
traceAggs []traceAggregation,
|
||||
) (*qbtypes.Statement, error) {
|
||||
keys, err := b.fetchKeys(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
maskExpr, maskArgs, err := b.resolveMask(ctx, start, end, keys)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
resolved, err := b.resolveColumns(ctx, start, end, keys, maskExpr, maskArgs)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
spanExpr, traceHavingPart, err := b.splitUserFilter(ctx, query, variables)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
resourceFrag, resourceArgs, resourcePred, skipResourceFilter, err := b.maybeAttachResourceFilter(ctx, query, start, end, variables)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
var warnings []string
|
||||
var warningsURL string
|
||||
var spanPred string
|
||||
var spanPredArgs []any
|
||||
if strings.TrimSpace(spanExpr) != "" {
|
||||
pred, args, warns, url, err := b.resolveSpanPredicate(ctx, start, end, spanExpr, skipResourceFilter, variables)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
spanPred, spanPredArgs, warnings, warningsURL = pred, args, warns, url
|
||||
}
|
||||
|
||||
var cteFragments []string
|
||||
var cteArgs [][]any
|
||||
if resourceFrag != "" {
|
||||
cteFragments = append(cteFragments, resourceFrag)
|
||||
cteArgs = append(cteArgs, resourceArgs)
|
||||
}
|
||||
|
||||
qualified := traceHavingPart != nil
|
||||
if qualified {
|
||||
qsql, qargs, err := b.qualifiedScanSQL(start, end, traceHavingPart, resolved, maskExpr, maskArgs, resourcePred, nil)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
cteFragments = append(cteFragments, fmt.Sprintf("__qualified AS (%s)", qsql))
|
||||
cteArgs = append(cteArgs, qargs)
|
||||
}
|
||||
|
||||
groupCols, err := b.resolveGroupColumns(ctx, start, end, query.GroupBy)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
groupNames := make([]string, 0, len(groupCols))
|
||||
for _, gc := range groupCols {
|
||||
groupNames = append(groupNames, "`"+gc.name+"`")
|
||||
}
|
||||
|
||||
needed := make(map[string]struct{})
|
||||
for _, ta := range traceAggs {
|
||||
for a := range ta.used {
|
||||
needed[a] = struct{}{}
|
||||
}
|
||||
}
|
||||
|
||||
stepSeconds := int64(0)
|
||||
rateInterval := (end - start) / querybuilder.NsToSeconds
|
||||
if requestType == qbtypes.RequestTypeTimeSeries {
|
||||
stepSeconds = int64(query.StepInterval.Seconds())
|
||||
rateInterval = uint64(stepSeconds)
|
||||
}
|
||||
|
||||
// LLM-activity gate: per-trace rows with no LLM span in their window/bucket slice
|
||||
// are dropped, so e.g. count(trace.trace_id) and avg(trace.output_tokens) agree on
|
||||
// the set of traces they see.
|
||||
activityExpr, activityArgs := activityGate(b.columnProvider, resolved)
|
||||
|
||||
scanOpts := perTraceScanOpts{
|
||||
stepSeconds: stepSeconds,
|
||||
groupCols: groupCols,
|
||||
needed: needed,
|
||||
spanPred: spanPred,
|
||||
spanPredArgs: spanPredArgs,
|
||||
resourcePred: resourcePred,
|
||||
qualified: qualified,
|
||||
activityExpr: activityExpr,
|
||||
activityArgs: activityArgs,
|
||||
}
|
||||
|
||||
// outer aggregation over the per-trace rows
|
||||
sb := sqlbuilder.NewSelectBuilder()
|
||||
selects := []string{}
|
||||
if stepSeconds > 0 {
|
||||
selects = append(selects, "ts")
|
||||
}
|
||||
selects = append(selects, groupNames...)
|
||||
for i, ta := range traceAggs {
|
||||
selects = append(selects, fmt.Sprintf("%s AS __result_%d", ta.rendered(rateInterval), i))
|
||||
}
|
||||
sb.Select(selects...)
|
||||
sb.From("__ai_traces")
|
||||
|
||||
// grouped, limited time series → rank groups on whole-window per-trace values
|
||||
// (exact for non-composable aggregates) and constrain the main query to the top-N.
|
||||
if requestType == qbtypes.RequestTypeTimeSeries && query.Limit > 0 && len(groupCols) > 0 {
|
||||
totalOpts := scanOpts
|
||||
totalOpts.stepSeconds = 0
|
||||
totalSQL, totalArgs, err := b.buildPerTraceScan(start, end, resolved, maskExpr, maskArgs, totalOpts)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
cteFragments = append(cteFragments, fmt.Sprintf("__ai_traces_total AS (%s)", totalSQL))
|
||||
cteArgs = append(cteArgs, totalArgs)
|
||||
|
||||
limitSQL, limitArgs := outerLimitSQL(query, traceAggs, groupNames, (end-start)/querybuilder.NsToSeconds)
|
||||
cteFragments = append(cteFragments, fmt.Sprintf("__limit_cte AS (%s)", limitSQL))
|
||||
cteArgs = append(cteArgs, limitArgs)
|
||||
|
||||
tuple := "(" + strings.Join(groupNames, ", ") + ")"
|
||||
sb.Where(fmt.Sprintf("%s IN (SELECT %s FROM __limit_cte)", tuple, strings.Join(groupNames, ", ")))
|
||||
}
|
||||
|
||||
perTraceSQL, perTraceArgs, err := b.buildPerTraceScan(start, end, resolved, maskExpr, maskArgs, scanOpts)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
cteFragments = append(cteFragments, fmt.Sprintf("__ai_traces AS (%s)", perTraceSQL))
|
||||
cteArgs = append(cteArgs, perTraceArgs)
|
||||
|
||||
groupBys := []string{}
|
||||
if stepSeconds > 0 {
|
||||
groupBys = append(groupBys, "ts")
|
||||
}
|
||||
groupBys = append(groupBys, groupNames...)
|
||||
if len(groupBys) > 0 {
|
||||
sb.GroupBy(groupBys...)
|
||||
}
|
||||
|
||||
if query.Having != nil && strings.TrimSpace(query.Having.Expression) != "" {
|
||||
rewritten, err := querybuilder.NewHavingExpressionRewriter().RewriteForTraces(query.Having.Expression, query.Aggregations)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
sb.Having(rewritten)
|
||||
}
|
||||
|
||||
if requestType == qbtypes.RequestTypeTimeSeries {
|
||||
if len(query.Order) != 0 {
|
||||
for _, orderBy := range query.Order {
|
||||
if _, ok := traceAggOrderIndex(orderBy, query); !ok {
|
||||
sb.OrderBy(fmt.Sprintf("`%s` %s", orderBy.Key.Name, orderBy.Direction.StringValue()))
|
||||
}
|
||||
}
|
||||
sb.OrderBy("ts desc")
|
||||
}
|
||||
} else {
|
||||
for _, orderBy := range query.Order {
|
||||
if idx, ok := traceAggOrderIndex(orderBy, query); ok {
|
||||
sb.OrderBy(fmt.Sprintf("__result_%d %s", idx, orderBy.Direction.StringValue()))
|
||||
} else {
|
||||
sb.OrderBy(fmt.Sprintf("`%s` %s", orderBy.Key.Name, orderBy.Direction.StringValue()))
|
||||
}
|
||||
}
|
||||
if len(query.Order) == 0 {
|
||||
sb.OrderBy("__result_0 DESC")
|
||||
}
|
||||
if query.Limit > 0 {
|
||||
sb.Limit(query.Limit)
|
||||
}
|
||||
}
|
||||
|
||||
mainSQL, mainArgs := sb.BuildWithFlavor(sqlbuilder.ClickHouse)
|
||||
finalSQL := querybuilder.CombineCTEs(cteFragments) + mainSQL + " SETTINGS distributed_product_mode='allow', max_memory_usage=10000000000"
|
||||
finalArgs := querybuilder.PrependArgs(cteArgs, mainArgs)
|
||||
|
||||
return &qbtypes.Statement{
|
||||
Query: finalSQL,
|
||||
Args: finalArgs,
|
||||
Warnings: warnings,
|
||||
WarningsDocURL: warningsURL,
|
||||
}, nil
|
||||
}
|
||||
|
||||
// activityGate resolves the provider's activity-gate column (llm_call_count for
|
||||
// gen_ai) to its aggregate expression; empty when the provider declares none.
|
||||
func activityGate(provider ColumnProvider, resolved []resolvedColumn) (string, []any) {
|
||||
alias := provider.ActivityGateAlias()
|
||||
if alias == "" {
|
||||
return "", nil
|
||||
}
|
||||
for _, rc := range resolved {
|
||||
if rc.alias == alias {
|
||||
return rc.expr, rc.args
|
||||
}
|
||||
}
|
||||
return "", nil
|
||||
}
|
||||
|
||||
// rendered returns the outer aggregation SQL, dividing rate aggregations by the
|
||||
// interval (step for time series, window length for scalar).
|
||||
func (ta traceAggregation) rendered(rateInterval uint64) string {
|
||||
if ta.isRate {
|
||||
return fmt.Sprintf("%s/%d", ta.expr, rateInterval)
|
||||
}
|
||||
return ta.expr
|
||||
}
|
||||
|
||||
// outerLimitSQL renders the top-N group selection for a grouped, limited time
|
||||
// series: the outer aggregations over whole-window per-trace values, ranked and
|
||||
// limited.
|
||||
func outerLimitSQL(query qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation], traceAggs []traceAggregation, groupNames []string, windowSeconds uint64) (string, []any) {
|
||||
sb := sqlbuilder.NewSelectBuilder()
|
||||
selects := append([]string{}, groupNames...)
|
||||
for i, ta := range traceAggs {
|
||||
selects = append(selects, fmt.Sprintf("%s AS __result_%d", ta.rendered(windowSeconds), i))
|
||||
}
|
||||
sb.Select(selects...)
|
||||
sb.From("__ai_traces_total")
|
||||
sb.GroupBy(groupNames...)
|
||||
for _, orderBy := range query.Order {
|
||||
if idx, ok := traceAggOrderIndex(orderBy, query); ok {
|
||||
sb.OrderBy(fmt.Sprintf("__result_%d %s", idx, orderBy.Direction.StringValue()))
|
||||
} else {
|
||||
sb.OrderBy(fmt.Sprintf("`%s` %s", orderBy.Key.Name, orderBy.Direction.StringValue()))
|
||||
}
|
||||
}
|
||||
if len(query.Order) == 0 {
|
||||
sb.OrderBy("__result_0 DESC")
|
||||
}
|
||||
sb.Limit(query.Limit)
|
||||
return sb.BuildWithFlavor(sqlbuilder.ClickHouse)
|
||||
}
|
||||
|
||||
// traceAggOrderIndex reports whether an order key refers to the i-th aggregation
|
||||
// (by alias, expression, or index), mirroring the trace builder.
|
||||
func traceAggOrderIndex(k qbtypes.OrderBy, q qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]) (int, bool) {
|
||||
for i, agg := range q.Aggregations {
|
||||
if k.Key.Name == agg.Alias ||
|
||||
k.Key.Name == agg.Expression ||
|
||||
k.Key.Name == fmt.Sprintf("%d", i) {
|
||||
return i, true
|
||||
}
|
||||
}
|
||||
return 0, false
|
||||
}
|
||||
63
pkg/telemetryscopedtraces/trace_aggregation_test.go
Normal file
63
pkg/telemetryscopedtraces/trace_aggregation_test.go
Normal file
@@ -0,0 +1,63 @@
|
||||
package telemetryscopedtraces
|
||||
|
||||
import (
|
||||
"testing"
|
||||
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// rewriteTraceAggregation unit tests
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
func TestRewriteTraceAggregation(t *testing.T) {
|
||||
cols := map[string]struct{}{
|
||||
"input_tokens": {}, "output_tokens": {}, "total_tokens": {}, "llm_call_count": {}, "max_llm_latency_ns": {},
|
||||
}
|
||||
|
||||
cases := []struct {
|
||||
name string
|
||||
expr string
|
||||
isTrace bool
|
||||
want string // rewritten expr, only checked when isTrace
|
||||
used []string
|
||||
wantErr string
|
||||
}{
|
||||
{name: "avg trace col", expr: "avg(trace.output_tokens)", isTrace: true, want: "avg(output_tokens)", used: []string{"output_tokens"}},
|
||||
{name: "tracefield prefix", expr: "sum(tracefield.total_tokens)", isTrace: true, want: "sum(total_tokens)", used: []string{"total_tokens"}},
|
||||
{name: "count traces", expr: "count(trace.trace_id)", isTrace: true, want: "count(trace_id)"},
|
||||
{name: "p90 trace col", expr: "p90(trace.max_llm_latency_ns)", isTrace: true, want: "quantile(0.90)(max_llm_latency_ns)", used: []string{"max_llm_latency_ns"}},
|
||||
{name: "arithmetic between trace cols", expr: "avg(trace.output_tokens + trace.input_tokens)", isTrace: true, want: "avg(output_tokens + input_tokens)", used: []string{"output_tokens", "input_tokens"}},
|
||||
{name: "arithmetic with constant", expr: "sum(trace.output_tokens * 1.5)", isTrace: true, want: "sum(output_tokens * 1.5)", used: []string{"output_tokens"}},
|
||||
{name: "ratio of two aggregations", expr: "sum(trace.output_tokens)/count(trace.trace_id)", isTrace: true, want: "sum(output_tokens) / count(trace_id)", used: []string{"output_tokens"}},
|
||||
{name: "backquoted trace col", expr: "avg(`trace.output_tokens`)", isTrace: true, want: "avg(`output_tokens`)", used: []string{"output_tokens"}},
|
||||
{name: "bare count is span-level", expr: "count()", isTrace: false},
|
||||
{name: "span attribute is span-level", expr: "sum(gen_ai.usage.output_tokens)", isTrace: false},
|
||||
{name: "countIf span predicate is span-level", expr: "countIf(has_error = true)", isTrace: false},
|
||||
{name: "mixed domains in one expression", expr: "sum(trace.output_tokens) + sum(gen_ai.usage.input_tokens)", wantErr: "mixes trace-level"},
|
||||
{name: "mixed domains in one function", expr: "sum(trace.output_tokens + gen_ai.usage.input_tokens)", wantErr: "mixes trace-level"},
|
||||
{name: "output-only column rejected", expr: "avg(trace.span_count)", wantErr: "unknown trace-level aggregation column"},
|
||||
{name: "unknown column rejected", expr: "avg(trace.bogus)", wantErr: "unknown trace-level aggregation column"},
|
||||
{name: "countIf over trace col rejected", expr: "countIf(trace.output_tokens > 1000)", wantErr: "not supported"},
|
||||
}
|
||||
|
||||
for _, tc := range cases {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
ta, isTrace, err := rewriteTraceAggregation(tc.expr, cols)
|
||||
if tc.wantErr != "" {
|
||||
require.ErrorContains(t, err, tc.wantErr)
|
||||
return
|
||||
}
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, tc.isTrace, isTrace)
|
||||
if !tc.isTrace {
|
||||
return
|
||||
}
|
||||
require.Equal(t, tc.want, ta.expr)
|
||||
for _, u := range tc.used {
|
||||
require.Contains(t, ta.used, u)
|
||||
}
|
||||
require.Len(t, ta.used, len(tc.used))
|
||||
})
|
||||
}
|
||||
}
|
||||
156
pkg/telemetryscopedtraces/trace_having.go
Normal file
156
pkg/telemetryscopedtraces/trace_having.go
Normal file
@@ -0,0 +1,156 @@
|
||||
package telemetryscopedtraces
|
||||
|
||||
import (
|
||||
"context"
|
||||
"strings"
|
||||
|
||||
"github.com/SigNoz/signoz/pkg/errors"
|
||||
"github.com/SigNoz/signoz/pkg/querybuilder"
|
||||
qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
|
||||
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
|
||||
qbvariables "github.com/SigNoz/signoz/pkg/variables"
|
||||
"github.com/huandu/go-sqlbuilder"
|
||||
)
|
||||
|
||||
// traceHaving is the resolved trace-level filter part: a HAVING predicate over the
|
||||
// per-trace column aliases (escaped, `?` placeholders) plus the aliases it references
|
||||
// (so scans select only what the predicate needs).
|
||||
type traceHaving struct {
|
||||
pred string
|
||||
args []any
|
||||
used map[string]struct{}
|
||||
}
|
||||
|
||||
// resolveTraceHaving resolves a trace-level filter expression through the standard
|
||||
// filter pipeline (PrepareWhereClause) against the per-trace column aliases, so
|
||||
// operators and bound args behave exactly as in span-level filters. Query variables
|
||||
// are resolved by the canonical replacement (pkg/variables) first — a dynamic
|
||||
// variable set to __all__ drops its condition for any operator — and the resulting
|
||||
// literals are then parsed and bound as args. Returns nil when the expression is
|
||||
// empty or every condition was dropped.
|
||||
func (b *scopedTraceStatementBuilder) resolveTraceHaving(ctx context.Context, expr string, variables map[string]qbtypes.VariableItem) (*traceHaving, error) {
|
||||
if strings.TrimSpace(expr) == "" {
|
||||
return nil, nil
|
||||
}
|
||||
allowed := b.orderableColumnSet()
|
||||
// upfront targeted errors: the visitor folds condition errors into a combined
|
||||
// "Found N errors" whose details are not part of the error message
|
||||
if err := validateAggregateFilter(expr, allowed); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if err := querybuilder.ValidateVariablesInExpr(expr, variables); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if len(variables) > 0 {
|
||||
replaced, err := qbvariables.ReplaceVariablesInExpression(expr, variables)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
expr = replaced
|
||||
if strings.TrimSpace(expr) == "" {
|
||||
return nil, nil
|
||||
}
|
||||
}
|
||||
|
||||
// every user-facing spelling of an alias resolves to the same synthetic key:
|
||||
// bare + trace.-prefixed here; tracefield. parses to FieldContextTrace, which
|
||||
// matches the bare entry's context
|
||||
fieldKeys := make(map[string][]*telemetrytypes.TelemetryFieldKey, len(allowed)*2)
|
||||
for alias := range allowed {
|
||||
key := &telemetrytypes.TelemetryFieldKey{Name: alias, FieldContext: telemetrytypes.FieldContextTrace}
|
||||
fieldKeys[alias] = []*telemetrytypes.TelemetryFieldKey{key}
|
||||
fieldKeys["trace."+alias] = []*telemetrytypes.TelemetryFieldKey{key}
|
||||
}
|
||||
|
||||
cb := &aliasConditionBuilder{allowed: allowed, used: make(map[string]struct{})}
|
||||
prepared, err := querybuilder.PrepareWhereClause(expr, querybuilder.FilterExprVisitorOpts{
|
||||
Context: ctx,
|
||||
Logger: b.logger,
|
||||
ConditionBuilder: cb,
|
||||
FieldKeys: fieldKeys,
|
||||
Variables: variables,
|
||||
})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if prepared.IsEmpty() {
|
||||
return nil, nil
|
||||
}
|
||||
sb := sqlbuilder.NewSelectBuilder()
|
||||
sb.Select("1")
|
||||
sb.AddWhereClause(prepared.WhereClause)
|
||||
sql, args := sb.BuildWithFlavor(sqlbuilder.ClickHouse)
|
||||
pred := sql[strings.Index(sql, "WHERE ")+len("WHERE "):]
|
||||
return &traceHaving{pred: sqlbuilder.Escape(pred), args: args, used: cb.used}, nil
|
||||
}
|
||||
|
||||
// aliasConditionBuilder renders filter conditions directly against the per-trace
|
||||
// column aliases. It records the aliases it touches; a key that resolves to no alias
|
||||
// is an unknown/unfilterable aggregate.
|
||||
type aliasConditionBuilder struct {
|
||||
allowed map[string]struct{}
|
||||
used map[string]struct{}
|
||||
}
|
||||
|
||||
var _ qbtypes.ConditionBuilder = (*aliasConditionBuilder)(nil)
|
||||
|
||||
func (c *aliasConditionBuilder) ConditionFor(
|
||||
_ context.Context,
|
||||
_, _ uint64,
|
||||
key *telemetrytypes.TelemetryFieldKey,
|
||||
matching []*telemetrytypes.TelemetryFieldKey,
|
||||
op qbtypes.FilterOperator,
|
||||
value any,
|
||||
sb *sqlbuilder.SelectBuilder,
|
||||
) ([]string, []string, error) {
|
||||
if len(matching) == 0 {
|
||||
name := strings.TrimPrefix(strings.TrimPrefix(key.Name, "tracefield."), "trace.")
|
||||
return nil, nil, errors.NewInvalidInputf(errors.CodeInvalidInput,
|
||||
"aggregate %q cannot be used in an AI trace-list filter; filterable aggregates: %s",
|
||||
name, strings.Join(sortedAliases(c.allowed), ", "))
|
||||
}
|
||||
alias := matching[0].Name
|
||||
c.used[alias] = struct{}{}
|
||||
col := quoteAlias(alias)
|
||||
|
||||
var cond string
|
||||
switch op {
|
||||
case qbtypes.FilterOperatorEqual:
|
||||
cond = sb.E(col, value)
|
||||
case qbtypes.FilterOperatorNotEqual:
|
||||
cond = sb.NE(col, value)
|
||||
case qbtypes.FilterOperatorGreaterThan:
|
||||
cond = sb.G(col, value)
|
||||
case qbtypes.FilterOperatorGreaterThanOrEq:
|
||||
cond = sb.GE(col, value)
|
||||
case qbtypes.FilterOperatorLessThan:
|
||||
cond = sb.L(col, value)
|
||||
case qbtypes.FilterOperatorLessThanOrEq:
|
||||
cond = sb.LE(col, value)
|
||||
case qbtypes.FilterOperatorIn, qbtypes.FilterOperatorNotIn:
|
||||
values, ok := value.([]any)
|
||||
if !ok {
|
||||
values = []any{value}
|
||||
}
|
||||
if op == qbtypes.FilterOperatorIn {
|
||||
cond = sb.In(col, values...)
|
||||
} else {
|
||||
cond = sb.NotIn(col, values...)
|
||||
}
|
||||
case qbtypes.FilterOperatorBetween, qbtypes.FilterOperatorNotBetween:
|
||||
values, ok := value.([]any)
|
||||
if !ok || len(values) != 2 {
|
||||
return nil, nil, errors.NewInvalidInputf(errors.CodeInvalidInput,
|
||||
"between on trace-level aggregate %q requires exactly two values", alias)
|
||||
}
|
||||
if op == qbtypes.FilterOperatorBetween {
|
||||
cond = sb.Between(col, values[0], values[1])
|
||||
} else {
|
||||
cond = sb.NotBetween(col, values[0], values[1])
|
||||
}
|
||||
default:
|
||||
return nil, nil, errors.NewInvalidInputf(errors.CodeInvalidInput,
|
||||
"trace-level aggregate %q supports only comparison operators (=, !=, <, <=, >, >=, in, between)", alias)
|
||||
}
|
||||
return []string{cond}, nil, nil
|
||||
}
|
||||
@@ -29,8 +29,9 @@ type traceQueryStatementBuilder struct {
|
||||
resourceFilterResolver *telemetryresourcefilter.ResourceFingerprintResolver[qbtypes.TraceAggregation]
|
||||
aggExprRewriter qbtypes.AggExprRewriter
|
||||
skipResourceFingerprintEnabled bool
|
||||
// traceScope is set only on the per-call copy made by BuildTraceScoped; it
|
||||
// constrains the query to trace ids selected by the __trace_scope CTE.
|
||||
// traceScope, when set (only on the per-call copy made by BuildTraceScoped),
|
||||
// constrains raw/scalar/time-series queries to spans whose trace_id is in the
|
||||
// scope statement, attached as a __trace_scope CTE.
|
||||
traceScope *qbtypes.Statement
|
||||
}
|
||||
|
||||
@@ -73,8 +74,9 @@ func NewTraceQueryStatementBuilder(
|
||||
}
|
||||
}
|
||||
|
||||
// BuildTraceScoped is Build additionally constrained to spans whose trace_id is
|
||||
// selected by traceScope. The receiver is copied so the shared builder stays stateless.
|
||||
// BuildTraceScoped is Build with the query additionally constrained to spans whose
|
||||
// trace_id is selected by traceScope. The receiver is copied so the shared builder
|
||||
// stays stateless.
|
||||
func (b *traceQueryStatementBuilder) BuildTraceScoped(
|
||||
ctx context.Context,
|
||||
start uint64,
|
||||
@@ -525,6 +527,11 @@ func (b *traceQueryStatementBuilder) buildTimeSeriesQuery(
|
||||
cteArgs = append(cteArgs, args)
|
||||
}
|
||||
|
||||
if scopeFrag, scopeArgs := b.attachTraceScope(sb); scopeFrag != "" {
|
||||
cteFragments = append(cteFragments, scopeFrag)
|
||||
cteArgs = append(cteArgs, scopeArgs)
|
||||
}
|
||||
|
||||
sb.SelectMore(fmt.Sprintf(
|
||||
"toStartOfInterval(timestamp, INTERVAL %d SECOND) AS ts",
|
||||
int64(query.StepInterval.Seconds()),
|
||||
@@ -679,6 +686,13 @@ func (b *traceQueryStatementBuilder) buildScalarQuery(
|
||||
cteArgs = append(cteArgs, args)
|
||||
}
|
||||
|
||||
// skipResourceCTE means this scalar is embedded as a CTE of a time-series query,
|
||||
// which has already emitted the __trace_scope fragment — add only the condition.
|
||||
if scopeFrag, scopeArgs := b.attachTraceScope(sb); scopeFrag != "" && !skipResourceCTE {
|
||||
cteFragments = append(cteFragments, scopeFrag)
|
||||
cteArgs = append(cteArgs, scopeArgs)
|
||||
}
|
||||
|
||||
allAggChArgs := []any{}
|
||||
|
||||
var allGroupByArgs []any
|
||||
|
||||
598
tests/integration/tests/querierai/02_ai_aggregations.py
Normal file
598
tests/integration/tests/querierai/02_ai_aggregations.py
Normal file
@@ -0,0 +1,598 @@
|
||||
"""
|
||||
Integration tests for source="ai" scalar / time-series aggregations.
|
||||
|
||||
Aggregations come in two domains, chosen per expression by the `trace.` prefix:
|
||||
- span-level (bare keys): over individual gen_ai spans (count(), sum(gen_ai.*))
|
||||
- trace-level (trace.*): over window-clipped per-trace values (avg(trace.output_tokens))
|
||||
A trace-level condition in the filter (trace.output_tokens > N) qualifies traces by
|
||||
their window-clipped per-trace values in every request type — the span-list variant
|
||||
of this is covered in 01_ai_traces.py (test_ai_span_list_trace_level_filter).
|
||||
|
||||
Each test tags its spans with a unique service.name and filters on it, so tests do
|
||||
not interfere with each other's data.
|
||||
"""
|
||||
|
||||
from collections.abc import Callable
|
||||
from datetime import UTC, datetime, timedelta
|
||||
from http import HTTPStatus
|
||||
|
||||
import pytest
|
||||
|
||||
from fixtures import types
|
||||
from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
|
||||
from fixtures.querier import (
|
||||
Aggregation,
|
||||
BuilderQuery,
|
||||
OrderBy,
|
||||
RequestType,
|
||||
TelemetryFieldKey,
|
||||
get_scalar_table_data,
|
||||
make_query_request,
|
||||
)
|
||||
from fixtures.traces import TraceIdGenerator, Traces, TracesKind, TracesStatusCode
|
||||
|
||||
|
||||
def _ai_trace(
|
||||
*,
|
||||
now: datetime,
|
||||
service: str,
|
||||
in_tokens: int,
|
||||
out_tokens: int,
|
||||
model: str = "gpt-4o-mini",
|
||||
) -> list[Traces]:
|
||||
"""A minimal AI trace: root span + one LLM span with gen_ai attributes."""
|
||||
trace_id = TraceIdGenerator.trace_id()
|
||||
root_id = TraceIdGenerator.span_id()
|
||||
resources = {"service.name": service}
|
||||
|
||||
root = Traces(
|
||||
timestamp=now - timedelta(seconds=5),
|
||||
duration=timedelta(seconds=2),
|
||||
trace_id=trace_id,
|
||||
span_id=root_id,
|
||||
parent_span_id="",
|
||||
name="POST /api/chat",
|
||||
kind=TracesKind.SPAN_KIND_SERVER,
|
||||
status_code=TracesStatusCode.STATUS_CODE_OK,
|
||||
resources=resources,
|
||||
attributes={"http.request.method": "POST"},
|
||||
)
|
||||
llm = Traces(
|
||||
timestamp=now - timedelta(seconds=4),
|
||||
duration=timedelta(seconds=1),
|
||||
trace_id=trace_id,
|
||||
span_id=TraceIdGenerator.span_id(),
|
||||
parent_span_id=root_id,
|
||||
name="chat",
|
||||
kind=TracesKind.SPAN_KIND_CLIENT,
|
||||
status_code=TracesStatusCode.STATUS_CODE_OK,
|
||||
resources=resources,
|
||||
attributes={
|
||||
"gen_ai.request.model": model,
|
||||
"gen_ai.usage.input_tokens": in_tokens,
|
||||
"gen_ai.usage.output_tokens": out_tokens,
|
||||
},
|
||||
)
|
||||
return [root, llm]
|
||||
|
||||
|
||||
def _tool_only_trace(*, now: datetime, service: str) -> list[Traces]:
|
||||
"""Root + one tool span: passes the gen_ai gate but has NO LLM span."""
|
||||
trace_id = TraceIdGenerator.trace_id()
|
||||
root_id = TraceIdGenerator.span_id()
|
||||
resources = {"service.name": service}
|
||||
return [
|
||||
Traces(
|
||||
timestamp=now - timedelta(seconds=5),
|
||||
duration=timedelta(seconds=2),
|
||||
trace_id=trace_id,
|
||||
span_id=root_id,
|
||||
parent_span_id="",
|
||||
name="POST /api/tool",
|
||||
kind=TracesKind.SPAN_KIND_SERVER,
|
||||
status_code=TracesStatusCode.STATUS_CODE_OK,
|
||||
resources=resources,
|
||||
attributes={"http.request.method": "POST"},
|
||||
),
|
||||
Traces(
|
||||
timestamp=now - timedelta(seconds=4),
|
||||
duration=timedelta(seconds=0.5),
|
||||
trace_id=trace_id,
|
||||
span_id=TraceIdGenerator.span_id(),
|
||||
parent_span_id=root_id,
|
||||
name="execute_tool",
|
||||
kind=TracesKind.SPAN_KIND_INTERNAL,
|
||||
status_code=TracesStatusCode.STATUS_CODE_OK,
|
||||
resources=resources,
|
||||
attributes={"gen_ai.tool.name": "get_weather", "gen_ai.tool.type": "function"},
|
||||
),
|
||||
]
|
||||
|
||||
|
||||
def _window_ms(now: datetime) -> tuple[int, int]:
|
||||
start_ms = int((now - timedelta(minutes=10)).timestamp() * 1000)
|
||||
end_ms = int((now + timedelta(minutes=1)).timestamp() * 1000)
|
||||
return start_ms, end_ms
|
||||
|
||||
|
||||
def _scalar_query(
|
||||
service: str,
|
||||
expression: str,
|
||||
*,
|
||||
filter_extra: str = "",
|
||||
group_by: list[TelemetryFieldKey] | None = None,
|
||||
alias: str | None = None,
|
||||
having: str | None = None,
|
||||
limit: int | None = None,
|
||||
) -> dict:
|
||||
filter_expression = f"service.name = '{service}'"
|
||||
if filter_extra:
|
||||
filter_expression += f" AND {filter_extra}"
|
||||
return BuilderQuery(
|
||||
signal="traces",
|
||||
source="ai",
|
||||
name="A",
|
||||
filter_expression=filter_expression,
|
||||
aggregations=[Aggregation(expression=expression, alias=alias)],
|
||||
group_by=group_by,
|
||||
having_expression=having,
|
||||
limit=limit,
|
||||
).to_dict()
|
||||
|
||||
|
||||
def _scalar_value(signoz: types.SigNoz, token: str, start_ms: int, end_ms: int, service: str, expression: str) -> float:
|
||||
"""Run one single-aggregation scalar query and return its value."""
|
||||
resp = make_query_request(
|
||||
signoz,
|
||||
token,
|
||||
start_ms,
|
||||
end_ms,
|
||||
[_scalar_query(service, expression)],
|
||||
request_type=RequestType.SCALAR,
|
||||
)
|
||||
assert resp.status_code == HTTPStatus.OK, f"{expression}: {resp.text}"
|
||||
data = get_scalar_table_data(resp.json())
|
||||
assert len(data) == 1, f"{expression}: expected one row, got {data}"
|
||||
return float(data[0][-1])
|
||||
|
||||
|
||||
def _series_values(response_json: dict) -> list[list[float]]:
|
||||
"""Per-series lists of bucket values (bucket order as returned)."""
|
||||
series = response_json["data"]["data"]["results"][0]["aggregations"][0]["series"]
|
||||
return [[v["value"] for v in ser["values"]] for ser in series]
|
||||
|
||||
|
||||
def test_ai_scalar_trace_level_aggregations(
|
||||
signoz: types.SigNoz,
|
||||
create_user_admin: None, # pylint: disable=unused-argument
|
||||
get_token: Callable[[str, str], str],
|
||||
insert_traces: Callable[[list[Traces]], None],
|
||||
) -> None:
|
||||
"""
|
||||
Trace-level (trace.) scalar aggregations over per-trace values: two traces with
|
||||
out-tokens 100 / 300 give avg(trace.output_tokens)=200 and count(trace.trace_id)=2,
|
||||
while the span-level count() sees the two LLM spans (root spans are gated out).
|
||||
"""
|
||||
now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
|
||||
service = "ai-it-agg-scalar"
|
||||
insert_traces(_ai_trace(now=now, service=service, in_tokens=10, out_tokens=100) + _ai_trace(now=now, service=service, in_tokens=30, out_tokens=300))
|
||||
|
||||
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
|
||||
start_ms, end_ms = _window_ms(now)
|
||||
|
||||
def scalar_value(expression: str) -> float:
|
||||
return _scalar_value(signoz, token, start_ms, end_ms, service, expression)
|
||||
|
||||
assert scalar_value("avg(trace.output_tokens)") == pytest.approx(200)
|
||||
assert scalar_value("count(trace.trace_id)") == 2
|
||||
assert scalar_value("max(trace.total_tokens)") == pytest.approx(330)
|
||||
assert scalar_value("p50(trace.output_tokens)") == pytest.approx(200) # AggreFuncMap -> quantile(0.50)
|
||||
# arithmetic inside one function and between functions
|
||||
assert scalar_value("avg(trace.output_tokens + trace.input_tokens)") == pytest.approx(220)
|
||||
assert scalar_value("sum(trace.output_tokens)/count(trace.trace_id)") == pytest.approx(200)
|
||||
# span-level domain still works through the same request type
|
||||
assert scalar_value("count()") == 2 # the two LLM spans; roots are not gen_ai
|
||||
assert scalar_value("sum(gen_ai.usage.output_tokens)") == pytest.approx(400)
|
||||
|
||||
# multiple trace-level aggregations in one query -> one column per aggregation
|
||||
multi = BuilderQuery(
|
||||
signal="traces",
|
||||
source="ai",
|
||||
name="A",
|
||||
filter_expression=f"service.name = '{service}'",
|
||||
aggregations=[Aggregation(expression="avg(trace.output_tokens)"), Aggregation(expression="count(trace.trace_id)")],
|
||||
)
|
||||
resp = make_query_request(signoz, token, start_ms, end_ms, [multi.to_dict()], request_type=RequestType.SCALAR)
|
||||
assert resp.status_code == HTTPStatus.OK, resp.text
|
||||
data = get_scalar_table_data(resp.json())
|
||||
assert len(data) == 1 and [float(v) for v in data[0]] == [pytest.approx(200), 2], data
|
||||
|
||||
|
||||
def test_ai_scalar_trace_level_filter_qualifies_traces(
|
||||
signoz: types.SigNoz,
|
||||
create_user_admin: None, # pylint: disable=unused-argument
|
||||
get_token: Callable[[str, str], str],
|
||||
insert_traces: Callable[[list[Traces]], None],
|
||||
) -> None:
|
||||
"""
|
||||
A trace-level condition in the filter qualifies whole traces before aggregation:
|
||||
with out-tokens 100 / 300, `trace.output_tokens > 100` keeps only the 300 trace
|
||||
for both trace-level and span-level aggregations.
|
||||
"""
|
||||
now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
|
||||
service = "ai-it-agg-qualify"
|
||||
insert_traces(_ai_trace(now=now, service=service, in_tokens=10, out_tokens=100) + _ai_trace(now=now, service=service, in_tokens=30, out_tokens=300))
|
||||
|
||||
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
|
||||
start_ms, end_ms = _window_ms(now)
|
||||
|
||||
for expression, expected in (
|
||||
("sum(trace.output_tokens)", 300), # native trace-domain path
|
||||
("sum(gen_ai.usage.output_tokens)", 300), # delegated span-domain path (__trace_scope)
|
||||
):
|
||||
resp = make_query_request(
|
||||
signoz,
|
||||
token,
|
||||
start_ms,
|
||||
end_ms,
|
||||
[_scalar_query(service, expression, filter_extra="trace.output_tokens > 100")],
|
||||
request_type=RequestType.SCALAR,
|
||||
)
|
||||
assert resp.status_code == HTTPStatus.OK, resp.text
|
||||
data = get_scalar_table_data(resp.json())
|
||||
assert len(data) == 1 and float(data[0][-1]) == pytest.approx(expected), f"{expression}: {data}"
|
||||
|
||||
# the qualification also constrains delegated (span-domain) time series
|
||||
ts = BuilderQuery(
|
||||
signal="traces",
|
||||
source="ai",
|
||||
name="A",
|
||||
filter_expression=f"service.name = '{service}' AND trace.output_tokens > 100",
|
||||
aggregations=[Aggregation(expression="sum(gen_ai.usage.output_tokens)")],
|
||||
step_interval=60,
|
||||
)
|
||||
resp = make_query_request(signoz, token, start_ms, end_ms, [ts.to_dict()], request_type=RequestType.TIME_SERIES)
|
||||
assert resp.status_code == HTTPStatus.OK, resp.text
|
||||
values = _series_values(resp.json())
|
||||
assert values == [[pytest.approx(300)]], values
|
||||
|
||||
|
||||
def test_ai_scalar_group_by_model(
|
||||
signoz: types.SigNoz,
|
||||
create_user_admin: None, # pylint: disable=unused-argument
|
||||
get_token: Callable[[str, str], str],
|
||||
insert_traces: Callable[[list[Traces]], None],
|
||||
) -> None:
|
||||
"""Trace-level aggregation grouped by a span attribute: per-model avg of per-trace tokens."""
|
||||
now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
|
||||
service = "ai-it-agg-groupby"
|
||||
insert_traces(_ai_trace(now=now, service=service, in_tokens=10, out_tokens=100, model="gpt-4o") + _ai_trace(now=now, service=service, in_tokens=10, out_tokens=300, model="gpt-4o") + _ai_trace(now=now, service=service, in_tokens=10, out_tokens=50, model="gpt-4o-mini"))
|
||||
|
||||
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
|
||||
start_ms, end_ms = _window_ms(now)
|
||||
|
||||
resp = make_query_request(
|
||||
signoz,
|
||||
token,
|
||||
start_ms,
|
||||
end_ms,
|
||||
[_scalar_query(service, "avg(trace.output_tokens)", group_by=[TelemetryFieldKey(name="gen_ai.request.model")])],
|
||||
request_type=RequestType.SCALAR,
|
||||
)
|
||||
assert resp.status_code == HTTPStatus.OK, resp.text
|
||||
data = get_scalar_table_data(resp.json())
|
||||
by_model = {row[0]: float(row[-1]) for row in data}
|
||||
assert by_model == {"gpt-4o": pytest.approx(200), "gpt-4o-mini": pytest.approx(50)}, data
|
||||
|
||||
|
||||
def test_ai_timeseries_trace_level_aggregation(
|
||||
signoz: types.SigNoz,
|
||||
create_user_admin: None, # pylint: disable=unused-argument
|
||||
get_token: Callable[[str, str], str],
|
||||
insert_traces: Callable[[list[Traces]], None],
|
||||
) -> None:
|
||||
"""Time-series over per-trace values: all spans fall in one step bucket, avg=200."""
|
||||
now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
|
||||
service = "ai-it-agg-ts"
|
||||
insert_traces(_ai_trace(now=now, service=service, in_tokens=10, out_tokens=100) + _ai_trace(now=now, service=service, in_tokens=30, out_tokens=300))
|
||||
|
||||
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
|
||||
start_ms, end_ms = _window_ms(now)
|
||||
|
||||
query = BuilderQuery(
|
||||
signal="traces",
|
||||
source="ai",
|
||||
name="A",
|
||||
filter_expression=f"service.name = '{service}'",
|
||||
aggregations=[Aggregation(expression="avg(trace.output_tokens)")],
|
||||
step_interval=60,
|
||||
)
|
||||
resp = make_query_request(
|
||||
signoz,
|
||||
token,
|
||||
start_ms,
|
||||
end_ms,
|
||||
[query.to_dict()],
|
||||
request_type=RequestType.TIME_SERIES,
|
||||
)
|
||||
assert resp.status_code == HTTPStatus.OK, resp.text
|
||||
|
||||
values = _series_values(resp.json())
|
||||
assert values == [[pytest.approx(200)]], values
|
||||
|
||||
|
||||
def test_ai_timeseries_top_n_groups(
|
||||
signoz: types.SigNoz,
|
||||
create_user_admin: None, # pylint: disable=unused-argument
|
||||
get_token: Callable[[str, str], str],
|
||||
insert_traces: Callable[[list[Traces]], None],
|
||||
) -> None:
|
||||
"""
|
||||
Grouped, limited time series ranks groups on whole-window per-trace values
|
||||
(__ai_traces_total -> __limit_cte) and returns only the top-N: gpt-4o sums to
|
||||
400 across two traces vs gpt-4o-mini's 50, so limit=1 keeps only gpt-4o.
|
||||
"""
|
||||
now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
|
||||
service = "ai-it-agg-topn"
|
||||
insert_traces(
|
||||
_ai_trace(now=now, service=service, in_tokens=10, out_tokens=300, model="gpt-4o")
|
||||
+ _ai_trace(now=now, service=service, in_tokens=10, out_tokens=100, model="gpt-4o")
|
||||
+ _ai_trace(now=now, service=service, in_tokens=10, out_tokens=50, model="gpt-4o-mini")
|
||||
)
|
||||
|
||||
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
|
||||
start_ms, end_ms = _window_ms(now)
|
||||
|
||||
query = BuilderQuery(
|
||||
signal="traces",
|
||||
source="ai",
|
||||
name="A",
|
||||
filter_expression=f"service.name = '{service}'",
|
||||
aggregations=[Aggregation(expression="sum(trace.output_tokens)")],
|
||||
group_by=[TelemetryFieldKey(name="gen_ai.request.model")],
|
||||
step_interval=60,
|
||||
limit=1,
|
||||
)
|
||||
resp = make_query_request(signoz, token, start_ms, end_ms, [query.to_dict()], request_type=RequestType.TIME_SERIES)
|
||||
assert resp.status_code == HTTPStatus.OK, resp.text
|
||||
|
||||
series = resp.json()["data"]["data"]["results"][0]["aggregations"][0]["series"]
|
||||
assert len(series) == 1, f"limit=1 must keep only the top group, got {len(series)} series"
|
||||
assert series[0]["labels"][0]["value"] == "gpt-4o", series[0]["labels"]
|
||||
assert [v["value"] for v in series[0]["values"]] == [pytest.approx(400)]
|
||||
|
||||
|
||||
def test_ai_timeseries_span_time_bucketing(
|
||||
signoz: types.SigNoz,
|
||||
create_user_admin: None, # pylint: disable=unused-argument
|
||||
get_token: Callable[[str, str], str],
|
||||
insert_traces: Callable[[list[Traces]], None],
|
||||
) -> None:
|
||||
"""
|
||||
Per-trace values are clipped per (bucket, trace): one trace with two LLM calls
|
||||
two minutes apart contributes each call's tokens to its own bucket, not the
|
||||
whole-trace total to both.
|
||||
"""
|
||||
now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
|
||||
service = "ai-it-agg-buckets"
|
||||
|
||||
trace_id = TraceIdGenerator.trace_id()
|
||||
root_id = TraceIdGenerator.span_id()
|
||||
resources = {"service.name": service}
|
||||
|
||||
def _llm(offset_s: float, out_tokens: int) -> Traces:
|
||||
return Traces(
|
||||
timestamp=now - timedelta(seconds=offset_s),
|
||||
duration=timedelta(seconds=1),
|
||||
trace_id=trace_id,
|
||||
span_id=TraceIdGenerator.span_id(),
|
||||
parent_span_id=root_id,
|
||||
name="chat",
|
||||
kind=TracesKind.SPAN_KIND_CLIENT,
|
||||
status_code=TracesStatusCode.STATUS_CODE_OK,
|
||||
resources=resources,
|
||||
attributes={"gen_ai.request.model": "gpt-4o-mini", "gen_ai.usage.output_tokens": out_tokens},
|
||||
)
|
||||
|
||||
root = Traces(
|
||||
timestamp=now - timedelta(seconds=130),
|
||||
duration=timedelta(seconds=130),
|
||||
trace_id=trace_id,
|
||||
span_id=root_id,
|
||||
parent_span_id="",
|
||||
name="POST /api/chat",
|
||||
kind=TracesKind.SPAN_KIND_SERVER,
|
||||
status_code=TracesStatusCode.STATUS_CODE_OK,
|
||||
resources=resources,
|
||||
attributes={"http.request.method": "POST"},
|
||||
)
|
||||
insert_traces([root, _llm(124, 100), _llm(4, 300)])
|
||||
|
||||
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
|
||||
start_ms, end_ms = _window_ms(now)
|
||||
|
||||
query = BuilderQuery(
|
||||
signal="traces",
|
||||
source="ai",
|
||||
name="A",
|
||||
filter_expression=f"service.name = '{service}'",
|
||||
aggregations=[Aggregation(expression="avg(trace.output_tokens)")],
|
||||
step_interval=60,
|
||||
)
|
||||
resp = make_query_request(signoz, token, start_ms, end_ms, [query.to_dict()], request_type=RequestType.TIME_SERIES)
|
||||
assert resp.status_code == HTTPStatus.OK, resp.text
|
||||
|
||||
values = _series_values(resp.json())
|
||||
assert len(values) == 1, values
|
||||
assert sorted(values[0]) == [pytest.approx(100), pytest.approx(300)], f"each call's tokens in its own bucket: {values}"
|
||||
|
||||
|
||||
def test_ai_scalar_variables_in_trace_level_filter(
|
||||
signoz: types.SigNoz,
|
||||
create_user_admin: None, # pylint: disable=unused-argument
|
||||
get_token: Callable[[str, str], str],
|
||||
insert_traces: Callable[[list[Traces]], None],
|
||||
) -> None:
|
||||
"""
|
||||
Query variables resolve inside trace-level conditions with span-filter semantics;
|
||||
an unresolvable $var is a 400, not a silent literal comparison.
|
||||
"""
|
||||
now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
|
||||
service = "ai-it-agg-vars"
|
||||
insert_traces(_ai_trace(now=now, service=service, in_tokens=10, out_tokens=100) + _ai_trace(now=now, service=service, in_tokens=30, out_tokens=300))
|
||||
|
||||
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
|
||||
start_ms, end_ms = _window_ms(now)
|
||||
query = _scalar_query(service, "sum(trace.output_tokens)", filter_extra="trace.output_tokens > $threshold")
|
||||
|
||||
resp = make_query_request(
|
||||
signoz,
|
||||
token,
|
||||
start_ms,
|
||||
end_ms,
|
||||
[query],
|
||||
request_type=RequestType.SCALAR,
|
||||
variables={"threshold": {"type": "text", "value": 100}},
|
||||
)
|
||||
assert resp.status_code == HTTPStatus.OK, resp.text
|
||||
data = get_scalar_table_data(resp.json())
|
||||
assert len(data) == 1 and float(data[0][-1]) == pytest.approx(300), data
|
||||
|
||||
resp = make_query_request(
|
||||
signoz,
|
||||
token,
|
||||
start_ms,
|
||||
end_ms,
|
||||
[query],
|
||||
request_type=RequestType.SCALAR,
|
||||
)
|
||||
assert resp.status_code == HTTPStatus.BAD_REQUEST, resp.text
|
||||
assert "unknown variable" in resp.text
|
||||
|
||||
# a dynamic variable resolved to __all__ drops the condition (both traces count)
|
||||
resp = make_query_request(
|
||||
signoz,
|
||||
token,
|
||||
start_ms,
|
||||
end_ms,
|
||||
[query],
|
||||
request_type=RequestType.SCALAR,
|
||||
variables={"threshold": {"type": "dynamic", "value": "__all__"}},
|
||||
)
|
||||
assert resp.status_code == HTTPStatus.OK, resp.text
|
||||
data = get_scalar_table_data(resp.json())
|
||||
assert len(data) == 1 and float(data[0][-1]) == pytest.approx(400), data
|
||||
|
||||
|
||||
def test_ai_scalar_activity_gate_excludes_tool_only_traces(
|
||||
signoz: types.SigNoz,
|
||||
create_user_admin: None, # pylint: disable=unused-argument
|
||||
get_token: Callable[[str, str], str],
|
||||
insert_traces: Callable[[list[Traces]], None],
|
||||
) -> None:
|
||||
"""
|
||||
A tool-only trace (in the gen_ai gate, no LLM span) must not feed trace-level
|
||||
aggregations: count(trace.trace_id) sees only the LLM trace, while the span-level
|
||||
count() still sees both gen_ai spans (LLM + tool).
|
||||
"""
|
||||
now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
|
||||
service = "ai-it-agg-gate"
|
||||
insert_traces(_ai_trace(now=now, service=service, in_tokens=10, out_tokens=100) + _tool_only_trace(now=now, service=service))
|
||||
|
||||
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
|
||||
start_ms, end_ms = _window_ms(now)
|
||||
|
||||
def scalar_value(expression: str) -> float:
|
||||
return _scalar_value(signoz, token, start_ms, end_ms, service, expression)
|
||||
|
||||
# count and avg agree on the trace set — the gate's purpose
|
||||
assert scalar_value("count(trace.trace_id)") == 1, "tool-only trace must be dropped by the LLM-activity gate"
|
||||
assert scalar_value("avg(trace.output_tokens)") == pytest.approx(100), "avg over the same gated trace set"
|
||||
assert scalar_value("count()") == 2, "span-level count still sees the tool span"
|
||||
|
||||
|
||||
def test_ai_scalar_having_on_aggregation(
|
||||
signoz: types.SigNoz,
|
||||
create_user_admin: None, # pylint: disable=unused-argument
|
||||
get_token: Callable[[str, str], str],
|
||||
insert_traces: Callable[[list[Traces]], None],
|
||||
) -> None:
|
||||
"""The outer having filters aggregation results per group (by alias)."""
|
||||
now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
|
||||
service = "ai-it-agg-having"
|
||||
insert_traces(_ai_trace(now=now, service=service, in_tokens=10, out_tokens=300, model="gpt-4o") + _ai_trace(now=now, service=service, in_tokens=10, out_tokens=50, model="gpt-4o-mini"))
|
||||
|
||||
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
|
||||
start_ms, end_ms = _window_ms(now)
|
||||
|
||||
resp = make_query_request(
|
||||
signoz,
|
||||
token,
|
||||
start_ms,
|
||||
end_ms,
|
||||
[
|
||||
_scalar_query(
|
||||
service,
|
||||
"avg(trace.output_tokens)",
|
||||
group_by=[TelemetryFieldKey(name="gen_ai.request.model")],
|
||||
alias="avg_out",
|
||||
having="avg_out > 100",
|
||||
)
|
||||
],
|
||||
request_type=RequestType.SCALAR,
|
||||
)
|
||||
assert resp.status_code == HTTPStatus.OK, resp.text
|
||||
data = get_scalar_table_data(resp.json())
|
||||
assert len(data) == 1 and data[0][0] == "gpt-4o", data
|
||||
|
||||
|
||||
def test_ai_aggregation_rejections(
|
||||
signoz: types.SigNoz,
|
||||
create_user_admin: None, # pylint: disable=unused-argument
|
||||
get_token: Callable[[str, str], str],
|
||||
insert_traces: Callable[[list[Traces]], None],
|
||||
) -> None:
|
||||
"""Targeted 400s: mixed domains, group-by on a trace column, raw order by a trace column."""
|
||||
now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
|
||||
service = "ai-it-agg-reject"
|
||||
insert_traces(_ai_trace(now=now, service=service, in_tokens=10, out_tokens=100))
|
||||
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
|
||||
start_ms, end_ms = _window_ms(now)
|
||||
|
||||
# span-level and trace-level aggregations cannot be mixed in one query
|
||||
mixed = BuilderQuery(
|
||||
signal="traces",
|
||||
source="ai",
|
||||
name="A",
|
||||
filter_expression=f"service.name = '{service}'",
|
||||
aggregations=[Aggregation(expression="avg(trace.output_tokens)"), Aggregation(expression="count()")],
|
||||
)
|
||||
resp = make_query_request(signoz, token, start_ms, end_ms, [mixed.to_dict()], request_type=RequestType.SCALAR)
|
||||
assert resp.status_code == HTTPStatus.BAD_REQUEST, resp.text
|
||||
assert "cannot be mixed" in resp.text
|
||||
|
||||
# grouping by a trace-level per-trace column is rejected with a targeted error
|
||||
bad_group = BuilderQuery(
|
||||
signal="traces",
|
||||
source="ai",
|
||||
name="A",
|
||||
filter_expression=f"service.name = '{service}'",
|
||||
aggregations=[Aggregation(expression="avg(trace.output_tokens)")],
|
||||
group_by=[TelemetryFieldKey(name="trace.llm_call_count")],
|
||||
)
|
||||
resp = make_query_request(signoz, token, start_ms, end_ms, [bad_group.to_dict()], request_type=RequestType.SCALAR)
|
||||
assert resp.status_code == HTTPStatus.BAD_REQUEST, resp.text
|
||||
assert "grouping by trace-level aggregate" in resp.text
|
||||
|
||||
# ordering the span list by a trace-level column is rejected with a targeted error
|
||||
bad_order = BuilderQuery(
|
||||
signal="traces",
|
||||
source="ai",
|
||||
name="A",
|
||||
filter_expression=f"service.name = '{service}'",
|
||||
order=[OrderBy(key=TelemetryFieldKey(name="trace.output_tokens"), direction="desc")],
|
||||
limit=10,
|
||||
)
|
||||
resp = make_query_request(signoz, token, start_ms, end_ms, [bad_order.to_dict()], request_type=RequestType.RAW)
|
||||
assert resp.status_code == HTTPStatus.BAD_REQUEST, resp.text
|
||||
assert "ordering the span list by trace-level aggregate" in resp.text
|
||||
Reference in New Issue
Block a user