mirror of
https://github.com/SigNoz/signoz.git
synced 2026-08-05 12:40:46 +01:00
Compare commits
1 Commits
issue_5601
...
issue_5601
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
8eb66acdc1 |
BIN
cmd/enterprise/db-shm
Normal file
BIN
cmd/enterprise/db-shm
Normal file
Binary file not shown.
BIN
cmd/enterprise/db-wal
Normal file
BIN
cmd/enterprise/db-wal
Normal file
Binary file not shown.
@@ -11020,10 +11020,6 @@ paths:
|
||||
name: source
|
||||
schema:
|
||||
$ref: '#/components/schemas/TelemetrytypesSource'
|
||||
- in: query
|
||||
name: type
|
||||
schema:
|
||||
type: string
|
||||
- in: query
|
||||
name: limit
|
||||
schema:
|
||||
@@ -11113,10 +11109,6 @@ paths:
|
||||
name: source
|
||||
schema:
|
||||
$ref: '#/components/schemas/TelemetrytypesSource'
|
||||
- in: query
|
||||
name: type
|
||||
schema:
|
||||
type: string
|
||||
- in: query
|
||||
name: limit
|
||||
schema:
|
||||
@@ -21319,10 +21311,6 @@ paths:
|
||||
name: source
|
||||
schema:
|
||||
$ref: '#/components/schemas/TelemetrytypesSource'
|
||||
- in: query
|
||||
name: type
|
||||
schema:
|
||||
type: string
|
||||
- in: query
|
||||
name: limit
|
||||
schema:
|
||||
@@ -21424,10 +21412,6 @@ paths:
|
||||
name: source
|
||||
schema:
|
||||
$ref: '#/components/schemas/TelemetrytypesSource'
|
||||
- in: query
|
||||
name: type
|
||||
schema:
|
||||
type: string
|
||||
- in: query
|
||||
name: limit
|
||||
schema:
|
||||
|
||||
@@ -10426,11 +10426,6 @@ export type GetFieldsKeysParams = {
|
||||
* @description undefined
|
||||
*/
|
||||
source?: TelemetrytypesSourceDTO;
|
||||
/**
|
||||
* @type string
|
||||
* @description undefined
|
||||
*/
|
||||
type?: string;
|
||||
/**
|
||||
* @type integer
|
||||
* @description undefined
|
||||
@@ -10490,11 +10485,6 @@ export type GetFieldsValuesParams = {
|
||||
* @description undefined
|
||||
*/
|
||||
source?: TelemetrytypesSourceDTO;
|
||||
/**
|
||||
* @type string
|
||||
* @description undefined
|
||||
*/
|
||||
type?: string;
|
||||
/**
|
||||
* @type integer
|
||||
* @description undefined
|
||||
@@ -11802,11 +11792,6 @@ export type GetRuleHistoryFilterKeysParams = {
|
||||
* @description undefined
|
||||
*/
|
||||
source?: TelemetrytypesSourceDTO;
|
||||
/**
|
||||
* @type string
|
||||
* @description undefined
|
||||
*/
|
||||
type?: string;
|
||||
/**
|
||||
* @type integer
|
||||
* @description undefined
|
||||
@@ -11869,11 +11854,6 @@ export type GetRuleHistoryFilterValuesParams = {
|
||||
* @description undefined
|
||||
*/
|
||||
source?: TelemetrytypesSourceDTO;
|
||||
/**
|
||||
* @type string
|
||||
* @description undefined
|
||||
*/
|
||||
type?: string;
|
||||
/**
|
||||
* @type integer
|
||||
* @description undefined
|
||||
|
||||
@@ -234,6 +234,7 @@ func NewSQLMigrationProviderFactories(
|
||||
sqlmigration.NewUpdateRoleTransactionGroupsFactory(),
|
||||
sqlmigration.NewFillDashboardSpecCollectionsFactory(sqlstore, dashboardStore),
|
||||
sqlmigration.NewScrubEmailChannelTransportFactory(sqlstore),
|
||||
sqlmigration.NewAddAiObservabilityQuickFiltersFactory(sqlstore),
|
||||
)
|
||||
}
|
||||
|
||||
|
||||
142
pkg/sqlmigration/108_add_ai_observability_quickfilters.go
Normal file
142
pkg/sqlmigration/108_add_ai_observability_quickfilters.go
Normal file
@@ -0,0 +1,142 @@
|
||||
package sqlmigration
|
||||
|
||||
import (
|
||||
"context"
|
||||
"database/sql"
|
||||
"encoding/json"
|
||||
"time"
|
||||
|
||||
"github.com/SigNoz/signoz/pkg/errors"
|
||||
"github.com/SigNoz/signoz/pkg/factory"
|
||||
"github.com/SigNoz/signoz/pkg/sqlstore"
|
||||
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
|
||||
"github.com/SigNoz/signoz/pkg/valuer"
|
||||
"github.com/uptrace/bun"
|
||||
"github.com/uptrace/bun/migrate"
|
||||
)
|
||||
|
||||
type addAiObservabilityQuickFilters struct {
|
||||
sqlstore sqlstore.SQLStore
|
||||
}
|
||||
|
||||
func NewAddAiObservabilityQuickFiltersFactory(sqlstore sqlstore.SQLStore) factory.ProviderFactory[SQLMigration, Config] {
|
||||
return factory.NewProviderFactory(factory.MustNewName("add_ai_observability_filters"), func(ctx context.Context, providerSettings factory.ProviderSettings, config Config) (SQLMigration, error) {
|
||||
return &addAiObservabilityQuickFilters{sqlstore: sqlstore}, nil
|
||||
})
|
||||
}
|
||||
|
||||
func (migration *addAiObservabilityQuickFilters) Register(migrations *migrate.Migrations) error {
|
||||
if err := migrations.Register(migration.Up, migration.Down); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (migration *addAiObservabilityQuickFilters) Up(ctx context.Context, db *bun.DB) error {
|
||||
// keep in sync with the ai_observability defaults in quickfiltertypes.NewDefaultQuickFilter
|
||||
aiObservabilityFilters := []map[string]interface{}{
|
||||
{"key": "hasError", "dataType": "bool", "type": "tag"},
|
||||
{"key": "deployment.environment", "dataType": "string", "type": "resource"},
|
||||
{"key": "service.name", "dataType": "string", "type": "resource"},
|
||||
{"key": telemetrytypes.GenAIOperationName, "dataType": "string", "type": "tag"},
|
||||
{"key": telemetrytypes.GenAIProviderName, "dataType": "string", "type": "tag"},
|
||||
{"key": telemetrytypes.GenAIRequestModel, "dataType": "string", "type": "tag"},
|
||||
{"key": telemetrytypes.GenAIToolName, "dataType": "string", "type": "tag"},
|
||||
{"key": telemetrytypes.GenAIAgentName, "dataType": "string", "type": "tag"},
|
||||
{"key": "estimated_total_cost", "dataType": "float64", "type": "trace"},
|
||||
{"key": "input_tokens", "dataType": "float64", "type": "trace"},
|
||||
{"key": "output_tokens", "dataType": "float64", "type": "trace"},
|
||||
{"key": "total_tokens", "dataType": "float64", "type": "trace"},
|
||||
{"key": "llm_call_count", "dataType": "float64", "type": "trace"},
|
||||
{"key": "tool_call_count", "dataType": "float64", "type": "trace"},
|
||||
{"key": "distinct_tool_count", "dataType": "float64", "type": "trace"},
|
||||
}
|
||||
|
||||
aiObservabilityJSON, err := json.Marshal(aiObservabilityFilters)
|
||||
if err != nil {
|
||||
return errors.Wrapf(err, errors.TypeInternal, errors.CodeInternal, "failed to marshal ai monitoring filters")
|
||||
}
|
||||
|
||||
type signal struct {
|
||||
valuer.String
|
||||
}
|
||||
|
||||
type identifiable struct {
|
||||
ID valuer.UUID `json:"id" bun:"id,pk,type:text"`
|
||||
}
|
||||
|
||||
type timeAuditable struct {
|
||||
CreatedAt time.Time `bun:"created_at" json:"createdAt"`
|
||||
UpdatedAt time.Time `bun:"updated_at" json:"updatedAt"`
|
||||
}
|
||||
|
||||
type quickFilterType struct {
|
||||
bun.BaseModel `bun:"table:quick_filter"`
|
||||
identifiable
|
||||
OrgID valuer.UUID `bun:"org_id,type:text,notnull"`
|
||||
Filter string `bun:"filter,type:text,notnull"`
|
||||
Signal signal `bun:"signal,type:text,notnull"`
|
||||
timeAuditable
|
||||
}
|
||||
|
||||
tx, err := db.BeginTx(ctx, nil)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
defer func() {
|
||||
_ = tx.Rollback()
|
||||
}()
|
||||
|
||||
var orgIDs []string
|
||||
err = tx.NewSelect().
|
||||
Table("organizations").
|
||||
Column("id").
|
||||
Scan(ctx, &orgIDs)
|
||||
if err != nil && err != sql.ErrNoRows {
|
||||
return err
|
||||
}
|
||||
|
||||
var filtersToInsert []quickFilterType
|
||||
for _, orgIDStr := range orgIDs {
|
||||
orgID, err := valuer.NewUUID(orgIDStr)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
filtersToInsert = append(filtersToInsert, quickFilterType{
|
||||
identifiable: identifiable{
|
||||
ID: valuer.GenerateUUID(),
|
||||
},
|
||||
OrgID: orgID,
|
||||
Filter: string(aiObservabilityJSON),
|
||||
Signal: signal{valuer.NewString("ai_observability")},
|
||||
timeAuditable: timeAuditable{
|
||||
CreatedAt: time.Now(),
|
||||
UpdatedAt: time.Now(),
|
||||
},
|
||||
})
|
||||
}
|
||||
|
||||
if len(filtersToInsert) > 0 {
|
||||
_, err = tx.NewInsert().
|
||||
Model(&filtersToInsert).
|
||||
On("CONFLICT (org_id, signal) DO UPDATE").
|
||||
Set("filter = EXCLUDED.filter, updated_at = EXCLUDED.updated_at").
|
||||
Exec(ctx)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
if err := tx.Commit(); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (migration *addAiObservabilityQuickFilters) Down(ctx context.Context, db *bun.DB) error {
|
||||
return nil
|
||||
}
|
||||
@@ -49,21 +49,20 @@ func Scope() scopedtraces.TraceScope {
|
||||
str := telemetrytypes.FieldDataTypeString
|
||||
columns := append(scopedtraces.CommonTraceColumns(),
|
||||
// LLM calls only (request model present), not the full gate.
|
||||
scopedtraces.TraceColumn{Alias: "llm_call_count", Orderable: true, Filterable: true, Expr: scopedtraces.CountExists(&reqModel)},
|
||||
scopedtraces.TraceColumn{Alias: "tool_call_count", Orderable: true, Filterable: true, Expr: scopedtraces.CountExists(&toolName)},
|
||||
scopedtraces.TraceColumn{Alias: "distinct_tool_count", Orderable: true, Filterable: true, Expr: scopedtraces.UniqCount(&toolName, str)},
|
||||
scopedtraces.TraceColumn{Alias: "llm_call_count", Orderable: true, Expr: scopedtraces.CountExists(&reqModel)},
|
||||
scopedtraces.TraceColumn{Alias: "tool_call_count", Orderable: true, Expr: scopedtraces.CountExists(&toolName)},
|
||||
scopedtraces.TraceColumn{Alias: "distinct_tool_count", Orderable: true, Expr: scopedtraces.UniqCount(&toolName, str)},
|
||||
// tokens live only on LLM spans, so a plain sum needs no gate scoping.
|
||||
scopedtraces.TraceColumn{Alias: "input_tokens", Orderable: true, Filterable: true, Expr: scopedtraces.Reduce(scopedtraces.AggSum, &inTok)},
|
||||
scopedtraces.TraceColumn{Alias: "output_tokens", Orderable: true, Filterable: true, Expr: scopedtraces.Reduce(scopedtraces.AggSum, &outTok)},
|
||||
scopedtraces.TraceColumn{Alias: "total_tokens", Orderable: true, Filterable: true, Expr: scopedtraces.SumOfKeys(telemetrytypes.FieldDataTypeFloat64, &inTok, &outTok)},
|
||||
scopedtraces.TraceColumn{Alias: "input_tokens", Orderable: true, Expr: scopedtraces.Reduce(scopedtraces.AggSum, &inTok)},
|
||||
scopedtraces.TraceColumn{Alias: "output_tokens", Orderable: true, Expr: scopedtraces.Reduce(scopedtraces.AggSum, &outTok)},
|
||||
scopedtraces.TraceColumn{Alias: "total_tokens", Orderable: true, Expr: scopedtraces.SumOfKeys(telemetrytypes.FieldDataTypeFloat64, &inTok, &outTok)},
|
||||
// per-span cost attached by the SigNoz LLM pricing processor.
|
||||
scopedtraces.TraceColumn{Alias: "estimated_total_cost", Orderable: true, Filterable: true, Expr: scopedtraces.Reduce(scopedtraces.AggSum, &cost)},
|
||||
scopedtraces.TraceColumn{Alias: "estimated_total_cost", Orderable: true, Expr: scopedtraces.Reduce(scopedtraces.AggSum, &cost)},
|
||||
// slowest single LLM call in the trace.
|
||||
scopedtraces.TraceColumn{Alias: "max_llm_duration_nano", Orderable: true, Filterable: true, Expr: scopedtraces.ScopedToKeyColumn(scopedtraces.AggMax, scopedtraces.IntrinsicSpanKey("duration_nano"), &reqModel)},
|
||||
scopedtraces.TraceColumn{Alias: "max_llm_duration_nano", Orderable: true, Expr: scopedtraces.ScopedToKeyColumn(scopedtraces.AggMax, scopedtraces.IntrinsicSpanKey("duration_nano"), &reqModel)},
|
||||
// errors across the whole trace (any span), so display-only.
|
||||
scopedtraces.TraceColumn{Alias: "error_count", Expr: scopedtraces.CondCount(scopedtraces.IntrinsicSpanKey("has_error"), qbtypes.FilterOperatorEqual, true)},
|
||||
// timestamp of the last gen_ai span (LLM/tool/agent), hence gate-scoped;
|
||||
// order-only: a raw-nanos threshold makes no sense in the filter bar.
|
||||
// timestamp of the last gen_ai span (LLM/tool/agent), hence gate-scoped.
|
||||
scopedtraces.TraceColumn{Alias: "last_activity_time", Orderable: true, Expr: scopedtraces.ScopedReduce(scopedtraces.AggMax, scopedtraces.IntrinsicSpanKey("timestamp"))},
|
||||
// previews: first call's input (the prompt), last call's output (the answer).
|
||||
scopedtraces.TraceColumn{Alias: "input", SpanLevel: true, Expr: scopedtraces.PickBy(&inMsg, str, scopedtraces.IntrinsicSpanKey("timestamp"), scopedtraces.PickEarliest)},
|
||||
@@ -77,22 +76,3 @@ func Scope() scopedtraces.TraceScope {
|
||||
DefaultOrderAlias: "last_activity_time",
|
||||
}
|
||||
}
|
||||
|
||||
// MetadataFieldKeys returns the aggregates the metadata store surfaces as trace-context
|
||||
// keys for builder_ai_query suggestions; only filterable columns qualify.
|
||||
func MetadataFieldKeys() []*telemetrytypes.TelemetryFieldKey {
|
||||
cols := Scope().Columns
|
||||
keys := make([]*telemetrytypes.TelemetryFieldKey, 0, len(cols))
|
||||
for _, c := range cols {
|
||||
if !c.Orderable || !c.Filterable {
|
||||
continue
|
||||
}
|
||||
keys = append(keys, &telemetrytypes.TelemetryFieldKey{
|
||||
Name: c.Alias,
|
||||
Signal: telemetrytypes.SignalTraces,
|
||||
FieldContext: telemetrytypes.FieldContextTrace,
|
||||
FieldDataType: telemetrytypes.FieldDataTypeFloat64,
|
||||
})
|
||||
}
|
||||
return keys
|
||||
}
|
||||
|
||||
@@ -23,9 +23,10 @@ type TraceColumn struct {
|
||||
// Alias must not reuse a physical span-index column name (e.g. duration_nano):
|
||||
// ClickHouse resolves bare identifiers to same-SELECT aliases first, so any
|
||||
// expression referencing that column would silently bind to the alias.
|
||||
Alias string
|
||||
Orderable bool
|
||||
Filterable bool
|
||||
Alias string
|
||||
// Orderable columns can be used in ORDER BY and the aggregate filter; all-span
|
||||
// aggregates are display-only and set false.
|
||||
Orderable bool
|
||||
// SpanLevel columns surface a real span/resource attribute; a filter on them is
|
||||
// applied span-level, so they are excluded from the trace-level aliases.
|
||||
SpanLevel bool
|
||||
|
||||
@@ -185,19 +185,18 @@ func (b *scopedTraceStatementBuilder) buildTraceListQuery(
|
||||
return nil, err
|
||||
}
|
||||
orderableSet := orderableAliasSet(resolved)
|
||||
filterableSet := filterableAliasSet(resolved)
|
||||
|
||||
resourceFrag, resourceArgs, resourcePred, err := b.maybeAttachResourceFilter(ctx, orgID, query, start, end, variables)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
fp, err := b.splitFilter(ctx, orgID, query, b.aggregateAliasSet(), filterableSet, start, end, variables, matchedSB)
|
||||
fp, err := b.splitFilter(ctx, orgID, query, b.aggregateAliasSet(), orderableSet, start, end, variables, matchedSB)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
matchedFrag, matchedArgs, err := b.buildMatchedCTE(matchedSB, start, end, startBucket, endBucket, resolved, orders, orderableSet, filterableSet, maskExpr, fp, resourcePred, limit, query.Offset)
|
||||
matchedFrag, matchedArgs, err := b.buildMatchedCTE(matchedSB, start, end, startBucket, endBucket, resolved, orders, orderableSet, maskExpr, fp, resourcePred, limit, query.Offset)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
@@ -327,10 +326,9 @@ func (b *scopedTraceStatementBuilder) resolveMask(ctx context.Context, orgID val
|
||||
}
|
||||
|
||||
type resolvedColumn struct {
|
||||
alias string
|
||||
expr string
|
||||
orderable bool
|
||||
filterable bool
|
||||
alias string
|
||||
expr string
|
||||
orderable bool
|
||||
}
|
||||
|
||||
func (b *scopedTraceStatementBuilder) resolveColumns(ctx context.Context, orgID valuer.UUID, start, end uint64, cols *columnResolver, preds *predicateResolver) ([]resolvedColumn, error) {
|
||||
@@ -340,7 +338,7 @@ func (b *scopedTraceStatementBuilder) resolveColumns(ctx context.Context, orgID
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
out = append(out, resolvedColumn{alias: c.Alias, expr: expr, orderable: c.Orderable, filterable: c.Filterable})
|
||||
out = append(out, resolvedColumn{alias: c.Alias, expr: expr, orderable: c.Orderable})
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
@@ -395,7 +393,7 @@ type filterParts struct {
|
||||
// splitFilter splits query.Filter into a span-level predicate (args bound into sb)
|
||||
// and a trace-level HAVING (explicit query.Having ANDed on), then validates the
|
||||
// trace-level part against the matched-pass aggregates.
|
||||
func (b *scopedTraceStatementBuilder) splitFilter(ctx context.Context, orgID valuer.UUID, query qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation], classifySet, filterableSet map[string]struct{}, start, end uint64, variables map[string]qbtypes.VariableItem, sb *sqlbuilder.SelectBuilder) (filterParts, error) {
|
||||
func (b *scopedTraceStatementBuilder) splitFilter(ctx context.Context, orgID valuer.UUID, query qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation], classifySet, orderableSet map[string]struct{}, start, end uint64, variables map[string]qbtypes.VariableItem, sb *sqlbuilder.SelectBuilder) (filterParts, error) {
|
||||
var fp filterParts
|
||||
if query.Filter != nil && strings.TrimSpace(query.Filter.Expression) != "" {
|
||||
spanExpr, traceExpr, err := querybuilder.SplitFilterForAggregates(query.Filter.Expression, classifySet)
|
||||
@@ -431,7 +429,7 @@ func (b *scopedTraceStatementBuilder) splitFilter(ctx context.Context, orgID val
|
||||
}
|
||||
fp.havingExpr = replaced
|
||||
}
|
||||
if err := validateAggregateFilter(fp.havingExpr, filterableSet); err != nil {
|
||||
if err := validateAggregateFilter(fp.havingExpr, orderableSet); err != nil {
|
||||
return fp, err
|
||||
}
|
||||
return fp, nil
|
||||
@@ -475,7 +473,7 @@ func (b *scopedTraceStatementBuilder) resolveSpanPredicate(ctx context.Context,
|
||||
// span filter + HAVING + ORDER BY + LIMIT/OFFSET, selecting only the aliases ORDER BY
|
||||
// / HAVING reference. Expressions carry $n markers bound to sb, so each can appear
|
||||
// several times and every occurrence resolves to the same arg.
|
||||
func (b *scopedTraceStatementBuilder) buildMatchedCTE(sb *sqlbuilder.SelectBuilder, start, end, startBucket, endBucket uint64, resolved []resolvedColumn, orders []listOrder, orderableSet, filterableSet map[string]struct{}, maskExpr string, fp filterParts, resourcePred string, limit, offset int) (string, []any, error) {
|
||||
func (b *scopedTraceStatementBuilder) buildMatchedCTE(sb *sqlbuilder.SelectBuilder, start, end, startBucket, endBucket uint64, resolved []resolvedColumn, orders []listOrder, orderableSet map[string]struct{}, maskExpr string, fp filterParts, resourcePred string, limit, offset int) (string, []any, error) {
|
||||
needed := neededMatchedAliases(orders, fp.havingExpr, orderableSet)
|
||||
selects := []string{"trace_id"}
|
||||
for _, rc := range resolved {
|
||||
@@ -515,8 +513,8 @@ func (b *scopedTraceStatementBuilder) buildMatchedCTE(sb *sqlbuilder.SelectBuild
|
||||
}
|
||||
if strings.TrimSpace(fp.havingExpr) != "" {
|
||||
// the rewriter matches raw key text, so map the trace. form alongside the bare name
|
||||
columnMap := make(map[string]string, len(filterableSet)*2)
|
||||
for a := range filterableSet {
|
||||
columnMap := make(map[string]string, len(orderableSet)*2)
|
||||
for a := range orderableSet {
|
||||
columnMap[a] = quoteAlias(a)
|
||||
columnMap[telemetrytypes.FieldContextTrace.StringValue()+"."+a] = quoteAlias(a)
|
||||
}
|
||||
@@ -605,17 +603,6 @@ func orderableAliasSet(resolved []resolvedColumn) map[string]struct{} {
|
||||
return set
|
||||
}
|
||||
|
||||
// filterableAliasSet is the subset of aliases usable in the trace-level filter.
|
||||
func filterableAliasSet(resolved []resolvedColumn) map[string]struct{} {
|
||||
set := make(map[string]struct{})
|
||||
for _, rc := range resolved {
|
||||
if rc.filterable {
|
||||
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.
|
||||
func neededMatchedAliases(orders []listOrder, havingExpr string, orderableSet map[string]struct{}) map[string]struct{} {
|
||||
@@ -643,19 +630,19 @@ func traceAggregateNames(havingExpr string) []string {
|
||||
return names
|
||||
}
|
||||
|
||||
// validateAggregateFilter rejects a trace-level filter referencing an aggregate that
|
||||
// is not filterable.
|
||||
func validateAggregateFilter(havingExpr string, filterableSet map[string]struct{}) error {
|
||||
// validateAggregateFilter rejects a trace-level filter referencing an aggregate not
|
||||
// computable in the matched pass.
|
||||
func validateAggregateFilter(havingExpr string, orderableSet map[string]struct{}) error {
|
||||
if strings.TrimSpace(havingExpr) == "" {
|
||||
return nil
|
||||
}
|
||||
allowed := make([]string, 0, len(filterableSet))
|
||||
for a := range filterableSet {
|
||||
allowed := make([]string, 0, len(orderableSet))
|
||||
for a := range orderableSet {
|
||||
allowed = append(allowed, a)
|
||||
}
|
||||
sort.Strings(allowed)
|
||||
for _, name := range traceAggregateNames(havingExpr) {
|
||||
if _, ok := filterableSet[name]; !ok {
|
||||
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, ", "))
|
||||
}
|
||||
|
||||
@@ -4,7 +4,6 @@ import (
|
||||
"context"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"slices"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
@@ -15,7 +14,6 @@ import (
|
||||
"github.com/SigNoz/signoz/pkg/factory"
|
||||
"github.com/SigNoz/signoz/pkg/flagger"
|
||||
"github.com/SigNoz/signoz/pkg/querybuilder"
|
||||
"github.com/SigNoz/signoz/pkg/statementbuilder/aistatementbuilder"
|
||||
"github.com/SigNoz/signoz/pkg/telemetryschema/audittelemetryschema"
|
||||
"github.com/SigNoz/signoz/pkg/telemetryschema/logstelemetryschema"
|
||||
"github.com/SigNoz/signoz/pkg/telemetryschema/metertelemetryschema"
|
||||
@@ -161,13 +159,6 @@ func (t *telemetryMetaStore) getTracesKeys(ctx context.Context, fieldKeySelector
|
||||
instrumentationtypes.CodeFunctionName: "getTracesKeys",
|
||||
})
|
||||
|
||||
// The trace field context never matches ingested keys — it names the computed
|
||||
// per-trace aggregates, which enrichWithAITraceAggregateKeys serves. Without this
|
||||
// the tagType condition below has no branch for it and the scan returns every key.
|
||||
fieldKeySelectors = slices.DeleteFunc(slices.Clone(fieldKeySelectors), func(s *telemetrytypes.FieldKeySelector) bool {
|
||||
return s.FieldContext == telemetrytypes.FieldContextTrace
|
||||
})
|
||||
|
||||
if len(fieldKeySelectors) == 0 {
|
||||
return nil, true, nil
|
||||
}
|
||||
@@ -1177,31 +1168,6 @@ func enrichWithIntrinsicMetricKeys(keys map[string][]*telemetrytypes.TelemetryFi
|
||||
return keys
|
||||
}
|
||||
|
||||
// enrichWithAITraceAggregateKeys adds the computed per-trace aggregate keys for
|
||||
// builder_ai_query selectors; they are never ingested, so the scan cannot serve them.
|
||||
func enrichWithAITraceAggregateKeys(keys map[string][]*telemetrytypes.TelemetryFieldKey, selectors []*telemetrytypes.FieldKeySelector) map[string][]*telemetrytypes.TelemetryFieldKey {
|
||||
defs := aistatementbuilder.MetadataFieldKeys()
|
||||
matched := make(map[string]*telemetrytypes.TelemetryFieldKey)
|
||||
for _, selector := range selectors {
|
||||
if selector.QueryType != qbtypes.QueryTypeBuilderAI.StringValue() {
|
||||
continue
|
||||
}
|
||||
if selector.Signal != telemetrytypes.SignalTraces && selector.Signal != telemetrytypes.SignalUnspecified {
|
||||
continue
|
||||
}
|
||||
for _, def := range defs {
|
||||
if selectorMatchesIntrinsicField(selector, *def) {
|
||||
matched[def.Name] = def
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
for name, def := range matched {
|
||||
keys[name] = append(keys[name], def)
|
||||
}
|
||||
return keys
|
||||
}
|
||||
|
||||
// enrichWithGenAIKeys adds keys that can be queried for GenAI signals, even though they have not been ingested yet.
|
||||
func enrichWithGenAIKeys(keys map[string][]*telemetrytypes.TelemetryFieldKey, selectors []*telemetrytypes.FieldKeySelector) map[string][]*telemetrytypes.TelemetryFieldKey {
|
||||
for _, selector := range selectors {
|
||||
@@ -1310,7 +1276,6 @@ func (t *telemetryMetaStore) GetKeys(ctx context.Context, orgID valuer.UUID, fie
|
||||
mapOfKeys = enrichWithIntrinsicMetricKeys(mapOfKeys, selectors)
|
||||
if t.fl.BooleanOrEmpty(ctx, flagger.FeatureEnableAIObservability, featuretypes.NewFlaggerEvaluationContext(orgID)) {
|
||||
mapOfKeys = enrichWithGenAIKeys(mapOfKeys, selectors)
|
||||
mapOfKeys = enrichWithAITraceAggregateKeys(mapOfKeys, selectors)
|
||||
}
|
||||
|
||||
return mapOfKeys, complete, nil
|
||||
@@ -1392,7 +1357,6 @@ func (t *telemetryMetaStore) GetKeysMulti(ctx context.Context, orgID valuer.UUID
|
||||
mapOfKeys = enrichWithIntrinsicMetricKeys(mapOfKeys, fieldKeySelectors)
|
||||
if t.fl.BooleanOrEmpty(ctx, flagger.FeatureEnableAIObservability, featuretypes.NewFlaggerEvaluationContext(orgID)) {
|
||||
mapOfKeys = enrichWithGenAIKeys(mapOfKeys, fieldKeySelectors)
|
||||
mapOfKeys = enrichWithAITraceAggregateKeys(mapOfKeys, fieldKeySelectors)
|
||||
}
|
||||
|
||||
return mapOfKeys, complete, nil
|
||||
|
||||
@@ -7,6 +7,7 @@ import (
|
||||
"github.com/SigNoz/signoz/pkg/errors"
|
||||
v3 "github.com/SigNoz/signoz/pkg/query-service/model/v3"
|
||||
"github.com/SigNoz/signoz/pkg/types"
|
||||
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
|
||||
"github.com/SigNoz/signoz/pkg/valuer"
|
||||
"github.com/uptrace/bun"
|
||||
)
|
||||
@@ -31,11 +32,12 @@ func (enum *Signal) UnmarshalJSON(data []byte) error {
|
||||
}
|
||||
|
||||
var (
|
||||
SignalTraces = Signal{valuer.NewString("traces")}
|
||||
SignalLogs = Signal{valuer.NewString("logs")}
|
||||
SignalApiMonitoring = Signal{valuer.NewString("api_monitoring")}
|
||||
SignalExceptions = Signal{valuer.NewString("exceptions")}
|
||||
SignalMeter = Signal{valuer.NewString("meter")}
|
||||
SignalTraces = Signal{valuer.NewString("traces")}
|
||||
SignalLogs = Signal{valuer.NewString("logs")}
|
||||
SignalApiMonitoring = Signal{valuer.NewString("api_monitoring")}
|
||||
SignalExceptions = Signal{valuer.NewString("exceptions")}
|
||||
SignalMeter = Signal{valuer.NewString("meter")}
|
||||
SignalAiObservability = Signal{valuer.NewString("ai_observability")}
|
||||
)
|
||||
|
||||
// NewSignal creates a Signal from a string.
|
||||
@@ -51,6 +53,8 @@ func NewSignal(s string) (Signal, error) {
|
||||
return SignalExceptions, nil
|
||||
case "meter":
|
||||
return SignalMeter, nil
|
||||
case "ai_observability":
|
||||
return SignalAiObservability, nil
|
||||
default:
|
||||
return Signal{}, errors.Newf(errors.TypeInvalidInput, errors.CodeInvalidInput, "invalid signal: %s", s)
|
||||
}
|
||||
@@ -187,6 +191,29 @@ func NewDefaultQuickFilter(orgID valuer.UUID) ([]*StorableQuickFilter, error) {
|
||||
{"key": "host.name", "dataType": "float64", "type": "Sum"},
|
||||
}
|
||||
|
||||
// AI observability (builder_ai_query trace explorer), grouped like the common LLM
|
||||
// observability sidebars: core narrowing (error/env/service/operation kind), then
|
||||
// the LLM identity (provider/model/tool/agent), then the per-trace aggregates
|
||||
// (fieldContext trace) as numeric threshold filters — the range treatment
|
||||
// duration_nano gets in the traces defaults.
|
||||
aiObservabilityFilters := []map[string]interface{}{
|
||||
{"key": "hasError", "dataType": "bool", "type": "tag"},
|
||||
{"key": "deployment.environment", "dataType": "string", "type": "resource"},
|
||||
{"key": "service.name", "dataType": "string", "type": "resource"},
|
||||
{"key": telemetrytypes.GenAIOperationName, "dataType": "string", "type": "tag"},
|
||||
{"key": telemetrytypes.GenAIProviderName, "dataType": "string", "type": "tag"},
|
||||
{"key": telemetrytypes.GenAIRequestModel, "dataType": "string", "type": "tag"},
|
||||
{"key": telemetrytypes.GenAIToolName, "dataType": "string", "type": "tag"},
|
||||
{"key": telemetrytypes.GenAIAgentName, "dataType": "string", "type": "tag"},
|
||||
{"key": "estimated_total_cost", "dataType": "float64", "type": "trace"},
|
||||
{"key": "input_tokens", "dataType": "float64", "type": "trace"},
|
||||
{"key": "output_tokens", "dataType": "float64", "type": "trace"},
|
||||
{"key": "total_tokens", "dataType": "float64", "type": "trace"},
|
||||
{"key": "llm_call_count", "dataType": "float64", "type": "trace"},
|
||||
{"key": "tool_call_count", "dataType": "float64", "type": "trace"},
|
||||
{"key": "distinct_tool_count", "dataType": "float64", "type": "trace"},
|
||||
}
|
||||
|
||||
tracesJSON, err := json.Marshal(tracesFilters)
|
||||
if err != nil {
|
||||
return nil, errors.Wrapf(err, errors.TypeInternal, errors.CodeInternal, "failed to marshal traces filters")
|
||||
@@ -212,6 +239,11 @@ func NewDefaultQuickFilter(orgID valuer.UUID) ([]*StorableQuickFilter, error) {
|
||||
return nil, errors.Wrapf(err, errors.TypeInternal, errors.CodeInternal, "failed to marshal meter filters")
|
||||
}
|
||||
|
||||
aiObservabilityJSON, err := json.Marshal(aiObservabilityFilters)
|
||||
if err != nil {
|
||||
return nil, errors.Wrapf(err, errors.TypeInternal, errors.CodeInternal, "failed to marshal ai observability filters")
|
||||
}
|
||||
|
||||
timeRightNow := time.Now()
|
||||
|
||||
return []*StorableQuickFilter{
|
||||
@@ -275,5 +307,17 @@ func NewDefaultQuickFilter(orgID valuer.UUID) ([]*StorableQuickFilter, error) {
|
||||
UpdatedAt: timeRightNow,
|
||||
},
|
||||
},
|
||||
{
|
||||
Identifiable: types.Identifiable{
|
||||
ID: valuer.GenerateUUID(),
|
||||
},
|
||||
OrgID: orgID,
|
||||
Filter: string(aiObservabilityJSON),
|
||||
Signal: SignalAiObservability,
|
||||
TimeAuditable: types.TimeAuditable{
|
||||
CreatedAt: timeRightNow,
|
||||
UpdatedAt: timeRightNow,
|
||||
},
|
||||
},
|
||||
}, nil
|
||||
}
|
||||
|
||||
@@ -238,7 +238,6 @@ type FieldKeySelector struct {
|
||||
EndUnixMilli int64 `json:"endUnixMilli"`
|
||||
Signal Signal `json:"signal"`
|
||||
Source Source `json:"source"`
|
||||
QueryType string `json:"queryType"`
|
||||
FieldContext FieldContext `json:"fieldContext"`
|
||||
FieldDataType FieldDataType `json:"fieldDataType"`
|
||||
Name string `json:"name"`
|
||||
@@ -262,7 +261,6 @@ type GettableFieldKeys struct {
|
||||
type PostableFieldKeysParams struct {
|
||||
Signal Signal `query:"signal"`
|
||||
Source Source `query:"source"`
|
||||
Type string `query:"type"`
|
||||
Limit int `query:"limit"`
|
||||
StartUnixMilli int64 `query:"startUnixMilli"`
|
||||
EndUnixMilli int64 `query:"endUnixMilli"`
|
||||
@@ -299,7 +297,6 @@ func NewFieldKeySelectorFromPostableFieldKeysParams(params PostableFieldKeysPara
|
||||
|
||||
req.Signal = params.Signal
|
||||
req.Source = params.Source
|
||||
req.QueryType = params.Type
|
||||
req.FieldContext = params.FieldContext
|
||||
req.FieldDataType = params.FieldDataType
|
||||
req.SelectorMatchType = FieldSelectorMatchTypeFuzzy
|
||||
|
||||
@@ -3,10 +3,11 @@ package telemetrytypes
|
||||
// OpenTelemetry gen_ai semantic-convention attribute keys. Single source of truth
|
||||
// shared by the AI query builder and the LLM pricing pipeline.
|
||||
const (
|
||||
GenAIRequestModel = "gen_ai.request.model"
|
||||
GenAIToolName = "gen_ai.tool.name"
|
||||
GenAIAgentName = "gen_ai.agent.name"
|
||||
GenAIProviderName = "gen_ai.provider.name"
|
||||
GenAIRequestModel = "gen_ai.request.model"
|
||||
GenAIOperationName = "gen_ai.operation.name"
|
||||
GenAIToolName = "gen_ai.tool.name"
|
||||
GenAIAgentName = "gen_ai.agent.name"
|
||||
GenAIProviderName = "gen_ai.provider.name"
|
||||
|
||||
GenAIUsageInputTokens = "gen_ai.usage.input_tokens"
|
||||
GenAIUsageOutputTokens = "gen_ai.usage.output_tokens"
|
||||
@@ -25,10 +26,11 @@ const (
|
||||
// on, surfaced by the metadata store even before ingestion so the AI gate/columns
|
||||
// resolve on a fresh install.
|
||||
var GenAIFieldDefinitions = map[string]TelemetryFieldKey{
|
||||
GenAIRequestModel: {Name: GenAIRequestModel, Signal: SignalTraces, FieldContext: FieldContextAttribute, FieldDataType: FieldDataTypeString},
|
||||
GenAIToolName: {Name: GenAIToolName, Signal: SignalTraces, FieldContext: FieldContextAttribute, FieldDataType: FieldDataTypeString},
|
||||
GenAIAgentName: {Name: GenAIAgentName, Signal: SignalTraces, FieldContext: FieldContextAttribute, FieldDataType: FieldDataTypeString},
|
||||
GenAIProviderName: {Name: GenAIProviderName, Signal: SignalTraces, FieldContext: FieldContextAttribute, FieldDataType: FieldDataTypeString},
|
||||
GenAIRequestModel: {Name: GenAIRequestModel, Signal: SignalTraces, FieldContext: FieldContextAttribute, FieldDataType: FieldDataTypeString},
|
||||
GenAIOperationName: {Name: GenAIOperationName, Signal: SignalTraces, FieldContext: FieldContextAttribute, FieldDataType: FieldDataTypeString},
|
||||
GenAIToolName: {Name: GenAIToolName, Signal: SignalTraces, FieldContext: FieldContextAttribute, FieldDataType: FieldDataTypeString},
|
||||
GenAIAgentName: {Name: GenAIAgentName, Signal: SignalTraces, FieldContext: FieldContextAttribute, FieldDataType: FieldDataTypeString},
|
||||
GenAIProviderName: {Name: GenAIProviderName, Signal: SignalTraces, FieldContext: FieldContextAttribute, FieldDataType: FieldDataTypeString},
|
||||
|
||||
GenAIUsageInputTokens: {Name: GenAIUsageInputTokens, Signal: SignalTraces, FieldContext: FieldContextAttribute, FieldDataType: FieldDataTypeFloat64},
|
||||
GenAIUsageOutputTokens: {Name: GenAIUsageOutputTokens, Signal: SignalTraces, FieldContext: FieldContextAttribute, FieldDataType: FieldDataTypeFloat64},
|
||||
|
||||
@@ -1,79 +0,0 @@
|
||||
"""Fields metadata API with type="builder_ai_query": gen_ai attributes and per-trace
|
||||
aggregate keys are served pre-ingestion, behind the conftest's AI observability flag."""
|
||||
|
||||
from collections.abc import Callable
|
||||
from http import HTTPStatus
|
||||
|
||||
import pytest
|
||||
import requests
|
||||
|
||||
from fixtures import types
|
||||
from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
|
||||
|
||||
AI_TRACE_AGGREGATES = {
|
||||
"llm_call_count",
|
||||
"tool_call_count",
|
||||
"distinct_tool_count",
|
||||
"input_tokens",
|
||||
"output_tokens",
|
||||
"total_tokens",
|
||||
"estimated_total_cost",
|
||||
"max_llm_duration_nano",
|
||||
}
|
||||
|
||||
|
||||
@pytest.fixture(name="get_keys")
|
||||
def get_keys_fixture(
|
||||
signoz: types.SigNoz,
|
||||
create_user_admin: None, # pylint: disable=unused-argument
|
||||
get_token: Callable[[str, str], str],
|
||||
) -> Callable[[dict], dict]:
|
||||
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
|
||||
|
||||
def get_keys(params: dict) -> dict:
|
||||
response = requests.get(
|
||||
signoz.self.host_configs["8080"].get("/api/v1/fields/keys"),
|
||||
timeout=5,
|
||||
headers={"authorization": f"Bearer {token}"},
|
||||
params={"signal": "traces", **params},
|
||||
)
|
||||
assert response.status_code == HTTPStatus.OK, response.text
|
||||
assert response.json()["status"] == "success"
|
||||
return response.json()["data"]["keys"]
|
||||
|
||||
return get_keys
|
||||
|
||||
|
||||
def trace_context_names(keys: dict) -> set:
|
||||
return {name for name, variants in keys.items() if any(k["fieldContext"] == "trace" for k in variants)}
|
||||
|
||||
|
||||
def test_ai_fields_trace_context_lists_only_aggregates(get_keys: Callable[[dict], dict]) -> None:
|
||||
"""fieldContext=trace (the order-by picker request) returns exactly the
|
||||
filterable aggregates — the ingested-key scan must not leak into it."""
|
||||
keys = get_keys({"type": "builder_ai_query", "fieldContext": "trace"})
|
||||
assert set(keys.keys()) == AI_TRACE_AGGREGATES, keys
|
||||
assert trace_context_names(keys) == AI_TRACE_AGGREGATES
|
||||
|
||||
|
||||
def test_ai_fields_trace_prefix_search(get_keys: Callable[[dict], dict]) -> None:
|
||||
"""`trace.output` in the filter bar parses into the trace context and suggests
|
||||
the matching aggregate."""
|
||||
keys = get_keys({"type": "builder_ai_query", "searchText": "trace.output"})
|
||||
assert trace_context_names(keys) == {"output_tokens"}, keys
|
||||
|
||||
|
||||
def test_ai_fields_bare_prefix_suggests_both_classes(get_keys: Callable[[dict], dict]) -> None:
|
||||
"""A bare prefix suggests the aggregate and the gen_ai span attribute side by side."""
|
||||
keys = get_keys({"type": "builder_ai_query", "searchText": "output_tok"})
|
||||
assert "output_tokens" in trace_context_names(keys), keys
|
||||
assert "gen_ai.usage.output_tokens" in keys, keys
|
||||
assert any(k["fieldContext"] == "attribute" for k in keys["gen_ai.usage.output_tokens"])
|
||||
|
||||
|
||||
def test_ai_fields_aggregates_require_ai_query_type(get_keys: Callable[[dict], dict]) -> None:
|
||||
"""Without type=builder_ai_query the aggregates are not suggested; the gen_ai
|
||||
attributes still are (flag-gated, not query-type-gated)."""
|
||||
keys = get_keys({"searchText": "output_tok"})
|
||||
assert not trace_context_names(keys), keys
|
||||
assert "gen_ai.usage.output_tokens" in keys, keys
|
||||
@@ -1,35 +0,0 @@
|
||||
import pytest
|
||||
from testcontainers.core.container import Network
|
||||
|
||||
from fixtures import types
|
||||
from fixtures.signoz import create_signoz
|
||||
|
||||
|
||||
@pytest.fixture(name="signoz", scope="package")
|
||||
def signoz_ai(
|
||||
network: Network,
|
||||
zeus: types.TestContainerDocker,
|
||||
gateway: types.TestContainerDocker,
|
||||
sqlstore: types.TestContainerSQL,
|
||||
clickhouse: types.TestContainerClickhouse,
|
||||
request: pytest.FixtureRequest,
|
||||
pytestconfig: pytest.Config,
|
||||
) -> types.SigNoz:
|
||||
"""
|
||||
Package-scoped SigNoz with AI observability enabled: the metadata store surfaces
|
||||
the gen_ai semconv keys and the trace-aggregate suggestion keys only behind this
|
||||
flag, so the querierai suite runs against a flag-enabled instance.
|
||||
"""
|
||||
return create_signoz(
|
||||
network=network,
|
||||
zeus=zeus,
|
||||
gateway=gateway,
|
||||
sqlstore=sqlstore,
|
||||
clickhouse=clickhouse,
|
||||
request=request,
|
||||
pytestconfig=pytestconfig,
|
||||
cache_key="signoz-ai",
|
||||
env_overrides={
|
||||
"SIGNOZ_FLAGGER_CONFIG_BOOLEAN_ENABLE__AI__OBSERVABILITY": True,
|
||||
},
|
||||
)
|
||||
Reference in New Issue
Block a user