Compare commits

..

3 Commits

Author SHA1 Message Date
nityanandagohain
decf6760d3 fix: update integration tests 2026-08-05 12:21:27 +05:30
nityanandagohain
1e20dc7a87 fix: remove if condition for json parser 2026-08-05 11:57:07 +05:30
Ashwin Bhatkal
a6ac14344e fix(dashboard): count panel stats from the v2 spec (#12396)
* fix(dashboard): count panel stats from the v2 spec

The panel counters walked a top-level `widgets` array and returned early
when the key was missing. A v2 dashboard stores only `metadata` and
`spec`, with panels as a map under `spec.panels`, so every v2 row hit
that early return and all `dashboard.panels.*` stats stayed at zero.
`dashboard.count` was unaffected — it is a row count.

Read the v2 spec instead: count the panels under `spec.panels` and take
each panel's signal from its query envelope, reusing the typed read path
and `QueryEnvelope.GetSignal`. Signal-less queries (promql, clickhouse
sql, formulas) count towards the panel total only. v1 rows are no longer
parsed for panel stats and contribute to `dashboard.count` alone.

* test(dashboard): drop the constant name arg from the stats query helper

statsBuilderQuery only ever received "A", which go-lint flags via unparam.
A panel holds a single query, so the name never mattered to the assertions;
the composite test still names its sub-queries through statsBuilderQuerySpec.

* refactor(dashboard): move v2 panel stats to a perses_ file

All v2 code lives in perses_-prefixed files until the v1 code goes away.
Pure move of the stats block out of dashboard.go, tests alongside it.

* fix(dashboard): count create-v2 stats off the postable spec

CreateV2 already holds the postable dashboard, so decoding the storable
back into a v2 dashboard just to count its panels was a needless type
conversion on the create path.

Split the panel walk into addPanelStats over a DashboardSpec, and add
NewStatsFromPostableDashboardV2 for the create path. The storable variant
keeps its signature for the periodic collectors, which only have rows.
2026-08-05 04:16:02 +00:00
16 changed files with 398 additions and 356 deletions

View File

@@ -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:

View File

@@ -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

View File

@@ -20,7 +20,6 @@ func (m *module) CreateV2(ctx context.Context, orgID valuer.UUID, createdBy stri
}
dashboard := postable.NewDashboardV2(orgID, createdBy, source)
var storableDashboard *dashboardtypes.StorableDashboard
err := m.store.RunInTx(ctx, func(ctx context.Context) error {
resolvedTags, err := m.tagModule.SyncTags(ctx, orgID, coretypes.KindDashboard, dashboard.ID, postable.Tags)
@@ -33,14 +32,13 @@ func (m *module) CreateV2(ctx context.Context, orgID valuer.UUID, createdBy stri
if err != nil {
return err
}
storableDashboard = storable
return m.store.Create(ctx, storable)
})
if err != nil {
return nil, err
}
m.analytics.TrackUser(ctx, orgID.String(), creator.String(), "Dashboard Created", dashboardtypes.NewStatsFromStorableDashboards([]*dashboardtypes.StorableDashboard{storableDashboard}))
m.analytics.TrackUser(ctx, orgID.String(), creator.String(), "Dashboard Created", dashboardtypes.NewStatsFromPostableDashboardV2(postable))
return dashboard, nil
}

View File

@@ -239,16 +239,9 @@ func processJSONParser(parent *pipelinetypes.PipelineOperator) ([]pipelinetypes.
return nil, errors.NewInternalf(CodeInvalidOperatorType, "operator type received %s", parent.Type)
}
parseFromNotNilCheck, err := fieldNotNilCheck(parent.ParseFrom)
if err != nil {
return nil, errors.WrapInvalidInputf(err, CodeFieldNilCheckType,
"couldn't generate nil check for parseFrom of json parser op %s: %s", parent.Name, err,
)
}
parent.If = fmt.Sprintf(
`%s && ((type(%s) == "string" && isJSON(%s) && type(fromJSON(unquote(%s))) == "map" ) || type(%s) == "map")`,
parseFromNotNilCheck, parent.ParseFrom, parent.ParseFrom, parent.ParseFrom, parent.ParseFrom,
)
// on_error: send_quiet replaces the expensive isJSON `if` check;
// parse failures pass the record through unchanged without noisy logs.
parent.OnError = signozstanzahelper.SendOnErrorQuiet
if parent.EnableFlattening {
parent.MaxFlatteningDepth = constants.MaxJSONFlatteningDepth
}
@@ -298,7 +291,7 @@ func processJSONParser(parent *pipelinetypes.PipelineOperator) ([]pipelinetypes.
}
// JSONMapping: host
err = generateMoveOperators(mapping[pipelinetypes.Host], `resource["host.name"]`)
err := generateMoveOperators(mapping[pipelinetypes.Host], `resource["host.name"]`)
if err != nil {
return nil, err
}

View File

@@ -324,6 +324,17 @@ func TestNoCollectorErrorsFromProcessorsForMismatchedLogs(t *testing.T) {
makeTestLog("mismatching log", map[string]string{
"test_json": "bad json",
}),
}, {
"json parser should quietly ignore log with non JSON body",
pipelinetypes.PipelineOperator{
ID: "json",
Type: "json_parser",
Enabled: true,
Name: "json parser",
ParseFrom: "body",
ParseTo: "attributes",
},
makeTestLog("plain text log", map[string]string{}),
}, {
"move parser should ignore non matching logs",
pipelinetypes.PipelineOperator{
@@ -894,8 +905,8 @@ func TestProcessJSONParser_WithFlatteningAndMapping(t *testing.T) {
require.Equal(t, 1, parentOp.MaxFlatteningDepth)
require.Nil(t, parentOp.Mapping) // Mapping should be removed
require.Nil(t, parent.Mapping) // Mapping should be removed
require.Contains(t, parentOp.If, `isJSON(body)`)
require.Contains(t, parentOp.If, `type(body)`)
require.Empty(t, parentOp.If)
require.Equal(t, signozstanzahelper.SendOnErrorQuiet, parentOp.OnError)
require.Equal(t, 1+totalOps, len(ops))
@@ -951,7 +962,8 @@ func TestProcessJSONParser_WithoutMapping(t *testing.T) {
require.True(t, op.EnableFlattening)
require.True(t, op.EnablePaths)
require.Equal(t, "parsed", op.PathPrefix)
require.Contains(t, op.If, `isJSON(body)`)
require.Empty(t, op.If)
require.Equal(t, signozstanzahelper.SendOnErrorQuiet, op.OnError)
}
func TestProcessJSONParser_Simple(t *testing.T) {
@@ -975,7 +987,8 @@ func TestProcessJSONParser_Simple(t *testing.T) {
require.False(t, op.EnableFlattening)
require.False(t, op.EnablePaths)
require.Equal(t, "", op.PathPrefix)
require.Contains(t, op.If, `isJSON(body)`)
require.Empty(t, op.If)
require.Equal(t, signozstanzahelper.SendOnErrorQuiet, op.OnError)
}
func TestProcessJSONParser_InvalidType(t *testing.T) {

View File

@@ -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
}

View File

@@ -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

View File

@@ -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, ", "))
}

View File

@@ -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

View File

@@ -176,69 +176,6 @@ func NewGettableDashboardFromDashboard(dashboard *Dashboard) (*GettableDashboard
}, nil
}
func NewStatsFromStorableDashboards(dashboards []*StorableDashboard) map[string]any {
stats := make(map[string]any)
stats["dashboard.panels.count"] = int64(0)
stats["dashboard.panels.traces.count"] = int64(0)
stats["dashboard.panels.metrics.count"] = int64(0)
stats["dashboard.panels.logs.count"] = int64(0)
for _, dashboard := range dashboards {
addStatsFromStorableDashboard(dashboard, stats)
}
stats["dashboard.count"] = int64(len(dashboards))
return stats
}
func addStatsFromStorableDashboard(dashboard *StorableDashboard, stats map[string]any) {
if dashboard.Data == nil {
return
}
if dashboard.Data["widgets"] == nil {
return
}
widgets, ok := dashboard.Data["widgets"]
if !ok {
return
}
data, ok := widgets.([]interface{})
if !ok {
return
}
for _, widget := range data {
sData, ok := widget.(map[string]interface{})
if ok && sData["query"] != nil {
stats["dashboard.panels.count"] = stats["dashboard.panels.count"].(int64) + 1
query, ok := sData["query"].(map[string]interface{})
if ok && query["queryType"] == "builder" && query["builder"] != nil {
builderData, ok := query["builder"].(map[string]interface{})
if ok && builderData["queryData"] != nil {
builderQueryData, ok := builderData["queryData"].([]interface{})
if ok {
for _, queryData := range builderQueryData {
data, ok := queryData.(map[string]interface{})
if ok {
switch data["dataSource"] {
case "traces":
stats["dashboard.panels.traces.count"] = stats["dashboard.panels.traces.count"].(int64) + 1
case "metrics":
stats["dashboard.panels.metrics.count"] = stats["dashboard.panels.metrics.count"].(int64) + 1
case "logs":
stats["dashboard.panels.logs.count"] = stats["dashboard.panels.logs.count"].(int64) + 1
}
}
}
}
}
}
}
}
}
func (storableDashboardData *StorableDashboardData) GetWidgetIds() []string {
data := *storableDashboardData
widgetIds := []string{}

View File

@@ -0,0 +1,90 @@
package dashboardtypes
import (
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
)
const (
statKeyDashboardCount = "dashboard.count"
statKeyPanelCount = "dashboard.panels.count"
statKeyPanelTracesCount = "dashboard.panels.traces.count"
statKeyPanelMetricsCount = "dashboard.panels.metrics.count"
statKeyPanelLogsCount = "dashboard.panels.logs.count"
)
// panelSignalStatKeys maps a builder query's signal to the stat it contributes
// to. Signal-less queries (promql, clickhouse sql, formulas) count towards the
// panel total only.
var panelSignalStatKeys = map[telemetrytypes.Signal]string{
telemetrytypes.SignalTraces: statKeyPanelTracesCount,
telemetrytypes.SignalMetrics: statKeyPanelMetricsCount,
telemetrytypes.SignalLogs: statKeyPanelLogsCount,
}
// NewStatsFromStorableDashboards reports the stats of stored dashboards. Rows that
// do not decode as v2 contribute to dashboard.count only.
func NewStatsFromStorableDashboards(dashboards []*StorableDashboard) map[string]any {
stats := newPanelStats()
for _, dashboard := range dashboards {
if dashboard == nil {
continue
}
dashboardV2, err := dashboard.ToDashboardV2(nil)
if err != nil {
continue
}
addPanelStats(&dashboardV2.Spec, stats)
}
stats[statKeyDashboardCount] = int64(len(dashboards))
return stats
}
// NewStatsFromPostableDashboardV2 reports the stats of a dashboard as it is
// created, straight off the postable spec — the create path has no reason to make
// a storable round-trip just to be counted.
func NewStatsFromPostableDashboardV2(postable PostableDashboardV2) map[string]any {
stats := newPanelStats()
addPanelStats(&postable.Spec, stats)
stats[statKeyDashboardCount] = int64(1)
return stats
}
func newPanelStats() map[string]any {
return map[string]any{
statKeyPanelCount: int64(0),
statKeyPanelTracesCount: int64(0),
statKeyPanelMetricsCount: int64(0),
statKeyPanelLogsCount: int64(0),
}
}
// addPanelStats counts the panels of a v2 spec, and each panel's queries against
// the signal they read.
func addPanelStats(spec *DashboardSpec, stats map[string]any) {
for _, panel := range spec.Panels {
if panel == nil {
continue
}
incrementStat(stats, statKeyPanelCount)
for _, query := range panel.Spec.Queries {
composite, err := query.Spec.Plugin.buildV5CompositeQueryFromPlugin()
if err != nil {
continue
}
for _, envelope := range composite.Queries {
if key, ok := panelSignalStatKeys[envelope.GetSignal()]; ok {
incrementStat(stats, key)
}
}
}
}
}
func incrementStat(stats map[string]any, key string) {
count, _ := stats[key].(int64)
stats[key] = count + 1
}

View File

@@ -0,0 +1,220 @@
package dashboardtypes
import (
"encoding/json"
"testing"
"github.com/SigNoz/signoz/pkg/types"
"github.com/SigNoz/signoz/pkg/valuer"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
func statsSpecJSON(panelsJSON string) string {
return `{
"display": {"name": "Stats Dashboard"},
"variables": [],
"panels": {` + panelsJSON + `},
"layouts": [],
"links": []
}`
}
// newStatsStorableV2 builds a stored v2 row from a panels JSON fragment, going
// through the untyped data blob the way a row read off the DB does.
func newStatsStorableV2(t *testing.T, panelsJSON string) *StorableDashboard {
t.Helper()
raw := `{
"metadata": {"schemaVersion": "` + SchemaVersion + `"},
"spec": ` + statsSpecJSON(panelsJSON) + `
}`
var data StorableDashboardData
require.NoError(t, json.Unmarshal([]byte(raw), &data))
return &StorableDashboard{
Identifiable: types.Identifiable{ID: valuer.GenerateUUID()},
OrgID: valuer.GenerateUUID(),
Source: SourceUser,
Name: "stats-dashboard",
Data: data,
}
}
func newStatsPostableV2(t *testing.T, panelsJSON string) PostableDashboardV2 {
t.Helper()
var spec DashboardSpec
require.NoError(t, json.Unmarshal([]byte(statsSpecJSON(panelsJSON)), &spec))
return PostableDashboardV2{
DashboardV2MetadataBase: DashboardV2MetadataBase{SchemaVersion: SchemaVersion},
Name: "stats-dashboard",
Spec: spec,
}
}
func statsPanel(queriesJSON string) string {
return `{
"kind": "Panel",
"spec": {
"links": [],
"plugin": {"kind": "signoz/TimeSeriesPanel", "spec": {}},
"queries": [` + queriesJSON + `]
}
}`
}
// A panel holds a single query, so its name never matters to the assertions.
func statsBuilderQuery(signal string) string {
return `{
"kind": "time_series",
"spec": {"plugin": {"kind": "signoz/BuilderQuery", "spec": ` + statsBuilderQuerySpec("A", signal) + `}}
}`
}
func statsBuilderQuerySpec(name, signal string) string {
aggregations := `[{"expression": "count()"}]`
if signal == "metrics" {
aggregations = `[{"metricName": "m", "timeAggregation": "rate", "spaceAggregation": "sum"}]`
}
return `{"name": "` + name + `", "signal": "` + signal + `", "aggregations": ` + aggregations + `}`
}
func TestNewStatsFromStorableDashboardsCountsV2Panels(t *testing.T) {
dashboard := newStatsStorableV2(t, `
"p1": `+statsPanel(statsBuilderQuery("logs"))+`,
"p2": `+statsPanel(statsBuilderQuery("metrics"))+`,
"p3": `+statsPanel(statsBuilderQuery("traces"))+`
`)
stats := NewStatsFromStorableDashboards([]*StorableDashboard{dashboard})
assert.Equal(t, int64(1), stats[statKeyDashboardCount])
assert.Equal(t, int64(3), stats[statKeyPanelCount])
assert.Equal(t, int64(1), stats[statKeyPanelLogsCount])
assert.Equal(t, int64(1), stats[statKeyPanelMetricsCount])
assert.Equal(t, int64(1), stats[statKeyPanelTracesCount])
}
// A panel carries exactly one query envelope, so multi-signal panels arrive as a
// composite: the panel counts once and every builder sub-query counts its signal.
func TestNewStatsFromStorableDashboardsCountsCompositeSubQueries(t *testing.T) {
composite := `{
"kind": "time_series",
"spec": {"plugin": {"kind": "signoz/CompositeQuery", "spec": {"queries": [
{"type": "builder_query", "spec": ` + statsBuilderQuerySpec("A", "traces") + `},
{"type": "builder_query", "spec": ` + statsBuilderQuerySpec("B", "logs") + `}
]}}}
}`
dashboard := newStatsStorableV2(t, `"p1": `+statsPanel(composite))
stats := NewStatsFromStorableDashboards([]*StorableDashboard{dashboard})
assert.Equal(t, int64(1), stats[statKeyPanelCount])
assert.Equal(t, int64(1), stats[statKeyPanelTracesCount])
assert.Equal(t, int64(1), stats[statKeyPanelLogsCount])
}
// promql and clickhouse queries carry no signal, so they land in the panel total
// and nowhere else.
func TestNewStatsFromStorableDashboardsIgnoresSignallessQueries(t *testing.T) {
promql := `{
"kind": "time_series",
"spec": {"plugin": {"kind": "signoz/PromQLQuery", "spec": {"name": "A", "query": "up"}}}
}`
dashboard := newStatsStorableV2(t, `"p1": `+statsPanel(promql))
stats := NewStatsFromStorableDashboards([]*StorableDashboard{dashboard})
assert.Equal(t, int64(1), stats[statKeyPanelCount])
assert.Equal(t, int64(0), stats[statKeyPanelTracesCount])
assert.Equal(t, int64(0), stats[statKeyPanelMetricsCount])
assert.Equal(t, int64(0), stats[statKeyPanelLogsCount])
}
func TestNewStatsFromStorableDashboardsAggregatesAcrossDashboards(t *testing.T) {
first := newStatsStorableV2(t, `"p1": `+statsPanel(statsBuilderQuery("logs")))
second := newStatsStorableV2(t, `
"p1": `+statsPanel(statsBuilderQuery("logs"))+`,
"p2": `+statsPanel(statsBuilderQuery("traces"))+`
`)
stats := NewStatsFromStorableDashboards([]*StorableDashboard{first, second})
assert.Equal(t, int64(2), stats[statKeyDashboardCount])
assert.Equal(t, int64(3), stats[statKeyPanelCount])
assert.Equal(t, int64(2), stats[statKeyPanelLogsCount])
assert.Equal(t, int64(1), stats[statKeyPanelTracesCount])
}
// v1 rows are counted as dashboards but contribute no panel stats — the counters
// read the v2 spec only.
func TestNewStatsFromStorableDashboardsSkipsNonV2Rows(t *testing.T) {
v1 := &StorableDashboard{
Identifiable: types.Identifiable{ID: valuer.GenerateUUID()},
OrgID: valuer.GenerateUUID(),
Source: SourceUser,
Name: "legacy-dashboard",
Data: StorableDashboardData{
"title": "Legacy Title",
"version": "v5",
"widgets": []any{
map[string]any{"query": map[string]any{
"queryType": "builder",
"builder": map[string]any{
"queryData": []any{map[string]any{"dataSource": "logs"}},
},
}},
},
},
}
empty := &StorableDashboard{
Identifiable: types.Identifiable{ID: valuer.GenerateUUID()},
OrgID: valuer.GenerateUUID(),
Source: SourceUser,
Name: "bare",
}
stats := NewStatsFromStorableDashboards([]*StorableDashboard{v1, empty})
assert.Equal(t, int64(2), stats[statKeyDashboardCount])
assert.Equal(t, int64(0), stats[statKeyPanelCount])
assert.Equal(t, int64(0), stats[statKeyPanelLogsCount])
}
// The create path counts off the postable spec, so it never round-trips through a
// storable to be counted.
func TestNewStatsFromPostableDashboardV2(t *testing.T) {
postable := newStatsPostableV2(t, `
"p1": `+statsPanel(statsBuilderQuery("logs"))+`,
"p2": `+statsPanel(statsBuilderQuery("traces"))+`
`)
stats := NewStatsFromPostableDashboardV2(postable)
assert.Equal(t, int64(1), stats[statKeyDashboardCount])
assert.Equal(t, int64(2), stats[statKeyPanelCount])
assert.Equal(t, int64(1), stats[statKeyPanelLogsCount])
assert.Equal(t, int64(1), stats[statKeyPanelTracesCount])
assert.Equal(t, int64(0), stats[statKeyPanelMetricsCount])
}
func TestNewStatsFromPostableDashboardV2WithNoPanels(t *testing.T) {
stats := NewStatsFromPostableDashboardV2(newStatsPostableV2(t, ``))
assert.Equal(t, int64(1), stats[statKeyDashboardCount])
assert.Equal(t, int64(0), stats[statKeyPanelCount])
assert.Equal(t, int64(0), stats[statKeyPanelLogsCount])
}
func TestNewStatsFromStorableDashboardsWithNoDashboards(t *testing.T) {
stats := NewStatsFromStorableDashboards(nil)
assert.Equal(t, int64(0), stats[statKeyDashboardCount])
assert.Equal(t, int64(0), stats[statKeyPanelCount])
assert.Equal(t, int64(0), stats[statKeyPanelTracesCount])
assert.Equal(t, int64(0), stats[statKeyPanelMetricsCount])
assert.Equal(t, int64(0), stats[statKeyPanelLogsCount])
}

View File

@@ -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

View File

@@ -359,15 +359,28 @@ def test_preview_logs_pipelines_success(
) -> None:
"""
Setup:
Create a preview request with a pipeline and sample logs.
Preview a json_parser pipeline with one JSON log and one plain-text log.
Tests:
1. Send preview request with valid pipeline configuration
2. Verify the preview processes logs correctly
3. Verify the response contains processed logs
1. JSON body gets parsed into attributes
2. Non-JSON body passes through unchanged instead of being dropped
"""
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
empty_log_fields = {
"id": "",
"trace_id": "",
"span_id": "",
"trace_flags": 0,
"severity_text": "",
"severity_number": 0,
"attributes_string": {},
"attributes_int": {},
"attributes_float": {},
"attributes_bool": {},
"resources_string": {},
}
preview_payload = {
"pipelines": [
{
@@ -396,29 +409,25 @@ def test_preview_logs_pipelines_success(
{
"type": "json_parser",
"id": "json-parser-preview",
"orderId": 1,
"enabled": True,
"parse_from": "body",
"parse_to": "attributes",
"on_error": "send",
}
],
}
],
"logs": [
{
"body": '{"level": "info", "message": "Test log message", "timestamp": "2024-01-01T00:00:00Z"}',
"body": '{"level": "info", "message": "json log"}',
"timestamp": 1704067200000000000, # nanoseconds, not milliseconds
"id": "",
"trace_id": "",
"span_id": "",
"trace_flags": 0,
"severity_text": "",
"severity_number": 0,
"attributes_string": {},
"attributes_int": {},
"attributes_float": {},
"attributes_bool": {},
"resources_string": {"service.name": "test-service"},
}
**empty_log_fields,
},
{
"body": "plain text log that is not json",
"timestamp": 1704067201000000000,
**empty_log_fields,
},
],
}
@@ -435,13 +444,16 @@ def test_preview_logs_pipelines_success(
assert response.status_code == HTTPStatus.OK
response_data = response.json()
assert response_data["status"] == "success"
assert "data" in response_data
assert "logs" in response_data["data"]
assert len(response_data["data"]["logs"]) == 1
logs = response_data["data"]["logs"]
assert len(logs) == 2
# Verify the log was processed
processed_log = response_data["data"]["logs"][0]
assert "attributes_string" in processed_log or "attributes" in processed_log
json_log = next(log for log in logs if log["body"].startswith("{"))
assert json_log["attributes_string"]["level"] == "info"
assert json_log["attributes_string"]["message"] == "json log"
plain_log = next(log for log in logs if not log["body"].startswith("{"))
assert plain_log["body"] == "plain text log that is not json"
assert plain_log["attributes_string"] == {}
def test_create_multiple_pipelines_success(

View File

@@ -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

View File

@@ -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,
},
)