Compare commits

...

7 Commits

Author SHA1 Message Date
nityanandagohain
2e813dcf2c fix: test cleanup 2026-06-02 17:45:42 +05:30
nityanandagohain
c4d719ed01 Merge remote-tracking branch 'origin/main' into issue_5131_2 2026-06-02 17:38:51 +05:30
Gaurav Tewari
184724003a chore: remove query status (#11476)
* chore: remove confusing query status

* chore: remove extra things

---------

Co-authored-by: Gaurav Tewari <tewarig@users.noreply.github.com>
2026-06-02 11:48:40 +00:00
Vinicius Lourenço
a4d3f10da8 chore(codeowners): add pulse for alerts and infra monitoring pages (#11508) 2026-06-02 11:44:38 +00:00
Nityananda Gohain
a71ac2ada6 fix: add adjustkeys in trace operator cte builder (#11349)
* fix: add adjustkeys in trace operator cte builder

* fix: more fixes

* fix: cleanup

* fix: move tests to trace operator file

* fix: address comments

* fix: lint issues

---------

Co-authored-by: Srikanth Chekuri <srikanth.chekuri92@gmail.com>
2026-06-02 10:59:15 +00:00
Vikrant Gupta
0963ff08cd feat(web): disable all integrations by default (#11539) 2026-06-02 09:32:20 +00:00
nityanandagohain
3fee26e65b fix: skip resource filter if it exceeds a threshold 2026-06-01 17:30:55 +05:30
29 changed files with 1445 additions and 382 deletions

19
.github/CODEOWNERS vendored
View File

@@ -169,3 +169,22 @@ go.mod @therealpandey
## Dashboard V2
/frontend/src/pages/DashboardPageV2/ @SigNoz/pulse-frontend
/frontend/src/pages/DashboardsListPageV2/ @SigNoz/pulse-frontend
## Infrastructure Monitoring
/frontend/src/pages/InfrastructureMonitoring/ @SigNoz/pulse-frontend
/frontend/src/container/InfraMonitoringHosts/ @SigNoz/pulse-frontend
/frontend/src/container/InfraMonitoringK8s/ @SigNoz/pulse-frontend
## Alerts
/frontend/src/pages/AlertList/ @SigNoz/pulse-frontend
/frontend/src/pages/AlertDetails/ @SigNoz/pulse-frontend
/frontend/src/pages/CreateAlert/ @SigNoz/pulse-frontend
/frontend/src/pages/EditRules/ @SigNoz/pulse-frontend
/frontend/src/container/AlertHistory/ @SigNoz/pulse-frontend
/frontend/src/container/CreateAlertRule/ @SigNoz/pulse-frontend
/frontend/src/container/CreateAlertV2/ @SigNoz/pulse-frontend
/frontend/src/container/EditAlertV2/ @SigNoz/pulse-frontend
/frontend/src/container/FormAlertRules/ @SigNoz/pulse-frontend
/frontend/src/container/ListAlertRules/ @SigNoz/pulse-frontend
/frontend/src/container/TriggeredAlerts/ @SigNoz/pulse-frontend
/frontend/src/container/AnomalyAlertEvaluationView/ @SigNoz/pulse-frontend

View File

@@ -52,6 +52,7 @@ jobs:
- rootuser
- serviceaccount
- querier_json_body
- querier_skip_resource_fingerprint
- ttl
sqlstore-provider:
- postgres

View File

@@ -64,16 +64,16 @@ web:
settings:
posthog:
# Whether to enable PostHog in web.
enabled: true
enabled: false
appcues:
# Whether to enable Appcues in web.
enabled: true
enabled: false
sentry:
# Whether to enable Sentry in web.
enabled: true
enabled: false
pylon:
# Whether to enable Pylon in web.
enabled: true
enabled: false
##################### Cache #####################
cache:

View File

@@ -1 +0,0 @@
<svg width="14" height="14" fill="none" xmlns="http://www.w3.org/2000/svg"><g clip-path="url(#prefix__clip0_4062_7291)" stroke-width="1.167" stroke-linecap="round" stroke-linejoin="round"><path d="M7 12.833A5.833 5.833 0 107 1.167a5.833 5.833 0 000 11.666z" fill="#E5484D" stroke="#E5484D"/><path d="M8.75 5.25l-3.5 3.5M5.25 5.25l3.5 3.5" stroke="#121317"/></g><defs><clipPath id="prefix__clip0_4062_7291"><path fill="#fff" d="M0 0h14v14H0z"/></clipPath></defs></svg>

Before

Width:  |  Height:  |  Size: 467 B

View File

@@ -9,8 +9,6 @@ import { useOptionsMenu } from 'container/OptionsMenu';
import { ArrowUp10, Minus } from '@signozhq/icons';
import { DataSource, StringOperators } from 'types/common/queryBuilder';
import QueryStatus from './QueryStatus';
function LogsActionsContainer({
listQuery,
selectedPanelType,
@@ -18,10 +16,6 @@ function LogsActionsContainer({
handleToggleFrequencyChart,
orderBy,
setOrderBy,
isFetching,
isLoading,
isError,
isSuccess,
}: {
listQuery: any;
selectedPanelType: PANEL_TYPES;
@@ -29,10 +23,6 @@ function LogsActionsContainer({
handleToggleFrequencyChart: () => void;
orderBy: string;
setOrderBy: (value: string) => void;
isFetching: boolean;
isLoading: boolean;
isError: boolean;
isSuccess: boolean;
}): JSX.Element {
const { options, config } = useOptionsMenu({
storageKey: LOCALSTORAGE.LOGS_LIST_OPTIONS,
@@ -106,17 +96,6 @@ function LogsActionsContainer({
</div>
</>
)}
{(selectedPanelType === PANEL_TYPES.TIME_SERIES ||
selectedPanelType === PANEL_TYPES.TABLE) && (
<div className="query-stats">
<QueryStatus
loading={isLoading || isFetching}
error={isError}
success={isSuccess}
/>
</div>
)}
</div>
</div>
</div>

View File

@@ -155,40 +155,6 @@
}
}
.query-stats {
display: flex;
align-items: center;
gap: 12px;
align-self: flex-end;
.rows {
color: var(--l2-foreground);
font-family: 'Geist Mono';
font-size: 12px;
font-style: normal;
font-weight: 400;
line-height: 18px; /* 150% */
letter-spacing: 0.36px;
}
.divider {
width: 1px;
height: 14px;
background: var(--l3-background);
}
.time {
color: var(--l2-foreground);
font-family: 'Geist Mono';
font-size: 12px;
font-style: normal;
font-weight: 400;
line-height: 18px; /* 150% */
letter-spacing: 0.36px;
}
}
.ant-btn {
border: none;
}

View File

@@ -1,4 +0,0 @@
.query-status {
display: flex;
align-items: center;
}

View File

@@ -1,49 +0,0 @@
import React, { useMemo } from 'react';
import { Color } from '@signozhq/design-tokens';
import { LoaderCircle, CircleCheck } from '@signozhq/icons';
import { Spin } from 'antd';
import solidXCircleUrl from '@/assets/Icons/solid-x-circle.svg';
import './QueryStatus.styles.scss';
interface IQueryStatusProps {
loading: boolean;
error: boolean;
success: boolean;
}
export default function QueryStatus(
props: IQueryStatusProps,
): React.ReactElement {
const { loading, error, success } = props;
const content = useMemo((): React.ReactElement => {
if (loading) {
return (
<Spin
spinning
size="small"
indicator={<LoaderCircle className="animate-spin" size="md" />}
/>
);
}
if (error) {
return (
<img
src={solidXCircleUrl}
alt="header"
className="error"
style={{ height: '14px', width: '14px' }}
/>
);
}
if (success) {
return (
<CircleCheck className="success" size={14} fill={Color.BG_ROBIN_500} />
);
}
return <div />;
}, [error, loading, success]);
return <div className="query-status">{content}</div>;
}

View File

@@ -160,7 +160,7 @@ function LogsExplorerViewsContainer({
'custom',
);
const { data, isLoading, isFetching, isError, isSuccess, error } =
const { data, isLoading, isFetching, isError, error } =
useGetExplorerQueryRange(
requestData,
selectedPanelType,
@@ -437,10 +437,6 @@ function LogsExplorerViewsContainer({
handleToggleFrequencyChart={handleToggleFrequencyChart}
orderBy={orderBy}
setOrderBy={setOrderBy}
isFetching={isFetching}
isLoading={isLoading}
isError={isError}
isSuccess={isSuccess}
/>
)}

View File

@@ -7,6 +7,12 @@ import (
"github.com/SigNoz/signoz/pkg/factory"
)
type SkipResourceFingerprint struct {
Enabled bool `yaml:"enabled" mapstructure:"enabled"`
// If count of fingerprint is above threshold, skip the fingerprint subquery and filter on main table instead.
Threshold uint64 `yaml:"threshold" mapstructure:"threshold"`
}
// Config represents the configuration for the querier.
type Config struct {
// CacheTTL is the TTL for cached query results
@@ -15,6 +21,8 @@ type Config struct {
FluxInterval time.Duration `yaml:"flux_interval" mapstructure:"flux_interval"`
// MaxConcurrentQueries is the maximum number of concurrent queries for missing ranges
MaxConcurrentQueries int `yaml:"max_concurrent_queries" mapstructure:"max_concurrent_queries"`
// SkipResourceFingerprint configures when the resource fingerprint subquery is skipped in favor of main-table filtering.
SkipResourceFingerprint SkipResourceFingerprint `yaml:"skip_resource_fingerprint" mapstructure:"skip_resource_fingerprint"`
}
// NewConfigFactory creates a new config factory for querier.
@@ -28,6 +36,10 @@ func newConfig() factory.Config {
CacheTTL: 168 * time.Hour,
FluxInterval: 5 * time.Minute,
MaxConcurrentQueries: 4,
SkipResourceFingerprint: SkipResourceFingerprint{
Enabled: false,
Threshold: 100000,
},
}
}
@@ -42,6 +54,9 @@ func (c Config) Validate() error {
if c.MaxConcurrentQueries <= 0 {
return errors.NewInvalidInputf(errors.CodeInvalidInput, "max_concurrent_queries must be positive, got %v", c.MaxConcurrentQueries)
}
if c.SkipResourceFingerprint.Enabled && c.SkipResourceFingerprint.Threshold == 0 {
return errors.NewInvalidInputf(errors.CodeInvalidInput, "skip_resource_fingerprint.threshold must be > 0 when enabled")
}
return nil
}

View File

@@ -88,6 +88,8 @@ func newProvider(
traceAggExprRewriter,
telemetryStore,
flagger,
cfg.SkipResourceFingerprint.Enabled,
cfg.SkipResourceFingerprint.Threshold,
)
// Create trace operator statement builder
@@ -121,6 +123,9 @@ func newProvider(
telemetrylogs.DefaultFullTextColumn,
telemetrylogs.GetBodyJSONKey,
flagger,
telemetryStore,
cfg.SkipResourceFingerprint.Enabled,
cfg.SkipResourceFingerprint.Threshold,
)
// Create audit statement builder

View File

@@ -89,6 +89,9 @@ func prepareQuerierForLogs(t *testing.T, telemetryStore telemetrystore.Telemetry
telemetrylogs.DefaultFullTextColumn,
telemetrylogs.GetBodyJSONKey,
fl,
nil,
false,
100000,
)
return querier.New(
@@ -134,6 +137,8 @@ func prepareQuerierForTraces(t *testing.T, telemetryStore telemetrystore.Telemet
traceAggExprRewriter,
telemetryStore,
fl,
false,
100000,
)
return querier.New(

View File

@@ -1205,6 +1205,9 @@ func buildJSONTestStatementBuilder(t *testing.T, addIndexes bool) (*logQueryStat
DefaultFullTextColumn,
GetBodyJSONKey,
fl,
nil,
false,
100000,
)
return statementBuilder, mockMetadataStore

View File

@@ -11,6 +11,7 @@ import (
"github.com/SigNoz/signoz/pkg/flagger"
"github.com/SigNoz/signoz/pkg/querybuilder"
"github.com/SigNoz/signoz/pkg/telemetryresourcefilter"
"github.com/SigNoz/signoz/pkg/telemetrystore"
"github.com/SigNoz/signoz/pkg/types/featuretypes"
qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
@@ -19,13 +20,14 @@ import (
)
type logQueryStatementBuilder struct {
logger *slog.Logger
metadataStore telemetrytypes.MetadataStore
fm qbtypes.FieldMapper
cb qbtypes.ConditionBuilder
resourceFilterStmtBuilder qbtypes.StatementBuilder[qbtypes.LogAggregation]
aggExprRewriter qbtypes.AggExprRewriter
fl flagger.Flagger
logger *slog.Logger
metadataStore telemetrytypes.MetadataStore
fm qbtypes.FieldMapper
cb qbtypes.ConditionBuilder
resourceFilterResolver *telemetryresourcefilter.ResourceFingerprintResolver[qbtypes.LogAggregation]
aggExprRewriter qbtypes.AggExprRewriter
fl flagger.Flagger
skipResourceFingerprintEnabled bool
fullTextColumn *telemetrytypes.TelemetryFieldKey
jsonKeyToKey qbtypes.JsonKeyToFieldFunc
@@ -42,10 +44,13 @@ func NewLogQueryStatementBuilder(
fullTextColumn *telemetrytypes.TelemetryFieldKey,
jsonKeyToKey qbtypes.JsonKeyToFieldFunc,
fl flagger.Flagger,
telemetryStore telemetrystore.TelemetryStore,
skipResourceFingerprintEnable bool,
skipResourceFingerprintThreshold uint64,
) *logQueryStatementBuilder {
logsSettings := factory.NewScopedProviderSettings(settings, "github.com/SigNoz/signoz/pkg/telemetrylogs")
resourceFilterStmtBuilder := telemetryresourcefilter.New[qbtypes.LogAggregation](
resourceFilterResolver := telemetryresourcefilter.NewResolver[qbtypes.LogAggregation](
settings,
DBName,
LogsResourceV2TableName,
@@ -55,18 +60,21 @@ func NewLogQueryStatementBuilder(
fullTextColumn,
jsonKeyToKey,
fl,
telemetryStore,
skipResourceFingerprintThreshold,
)
return &logQueryStatementBuilder{
logger: logsSettings.Logger(),
metadataStore: metadataStore,
fm: fieldMapper,
cb: conditionBuilder,
resourceFilterStmtBuilder: resourceFilterStmtBuilder,
aggExprRewriter: aggExprRewriter,
fl: fl,
fullTextColumn: fullTextColumn,
jsonKeyToKey: jsonKeyToKey,
logger: logsSettings.Logger(),
metadataStore: metadataStore,
fm: fieldMapper,
cb: conditionBuilder,
resourceFilterResolver: resourceFilterResolver,
aggExprRewriter: aggExprRewriter,
fl: fl,
skipResourceFingerprintEnabled: skipResourceFingerprintEnable,
fullTextColumn: fullTextColumn,
jsonKeyToKey: jsonKeyToKey,
}
}
@@ -271,9 +279,11 @@ func (b *logQueryStatementBuilder) buildListQuery(
bodyJSONEnabled = b.fl.BooleanOrEmpty(ctx, flagger.FeatureUseJSONBody, featuretypes.NewFlaggerEvaluationContext(valuer.UUID{}))
)
if frag, args, err := b.maybeAttachResourceFilter(ctx, sb, query, start, end, variables); err != nil {
frag, args, skipResourceFilter, err := b.maybeAttachResourceFilter(ctx, sb, query, start, end, variables)
if err != nil {
return nil, err
} else if frag != "" {
}
if frag != "" {
cteFragments = append(cteFragments, frag)
cteArgs = append(cteArgs, args)
}
@@ -315,7 +325,7 @@ func (b *logQueryStatementBuilder) buildListQuery(
sb.From(fmt.Sprintf("%s.%s", DBName, LogsV2TableName))
// Add filter conditions
preparedWhereClause, err := b.addFilterCondition(ctx, sb, start, end, query, keys, variables)
preparedWhereClause, err := b.addFilterCondition(ctx, sb, start, end, query, keys, variables, skipResourceFilter)
if err != nil {
return nil, err
@@ -373,9 +383,11 @@ func (b *logQueryStatementBuilder) buildTimeSeriesQuery(
bodyJSONEnabled = b.fl.BooleanOrEmpty(ctx, flagger.FeatureUseJSONBody, featuretypes.NewFlaggerEvaluationContext(valuer.UUID{}))
)
if frag, args, err := b.maybeAttachResourceFilter(ctx, sb, query, start, end, variables); err != nil {
frag, args, skipResourceFilter, err := b.maybeAttachResourceFilter(ctx, sb, query, start, end, variables)
if err != nil {
return nil, err
} else if frag != "" {
}
if frag != "" {
cteFragments = append(cteFragments, frag)
cteArgs = append(cteArgs, args)
}
@@ -419,7 +431,7 @@ func (b *logQueryStatementBuilder) buildTimeSeriesQuery(
// Add FROM clause
sb.From(fmt.Sprintf("%s.%s", DBName, LogsV2TableName))
preparedWhereClause, err := b.addFilterCondition(ctx, sb, start, end, query, keys, variables)
preparedWhereClause, err := b.addFilterCondition(ctx, sb, start, end, query, keys, variables, skipResourceFilter)
if err != nil {
return nil, err
@@ -531,9 +543,11 @@ func (b *logQueryStatementBuilder) buildScalarQuery(
bodyJSONEnabled = b.fl.BooleanOrEmpty(ctx, flagger.FeatureUseJSONBody, featuretypes.NewFlaggerEvaluationContext(valuer.UUID{}))
)
if frag, args, err := b.maybeAttachResourceFilter(ctx, sb, query, start, end, variables); err != nil {
frag, args, skipResourceFilter, err := b.maybeAttachResourceFilter(ctx, sb, query, start, end, variables)
if err != nil {
return nil, err
} else if frag != "" && !skipResourceCTE {
}
if frag != "" && !skipResourceCTE {
cteFragments = append(cteFragments, frag)
cteArgs = append(cteArgs, args)
}
@@ -576,7 +590,7 @@ func (b *logQueryStatementBuilder) buildScalarQuery(
sb.From(fmt.Sprintf("%s.%s", DBName, LogsV2TableName))
// Add filter conditions
preparedWhereClause, err := b.addFilterCondition(ctx, sb, start, end, query, keys, variables)
preparedWhereClause, err := b.addFilterCondition(ctx, sb, start, end, query, keys, variables, skipResourceFilter)
if err != nil {
return nil, err
@@ -640,6 +654,7 @@ func (b *logQueryStatementBuilder) addFilterCondition(
query qbtypes.QueryBuilderQuery[qbtypes.LogAggregation],
keys map[string][]*telemetrytypes.TelemetryFieldKey,
variables map[string]qbtypes.VariableItem,
skipResourceFilter bool,
) (querybuilder.PreparedWhereClause, error) {
var preparedWhereClause querybuilder.PreparedWhereClause
@@ -656,7 +671,7 @@ func (b *logQueryStatementBuilder) addFilterCondition(
ConditionBuilder: b.cb,
FieldKeys: keys,
BodyJSONEnabled: bodyJSONEnabled,
SkipResourceFilter: true,
SkipResourceFilter: skipResourceFilter,
FullTextColumn: b.fullTextColumn,
JsonKeyToKey: b.jsonKeyToKey,
Variables: variables,
@@ -707,33 +722,30 @@ func (b *logQueryStatementBuilder) maybeAttachResourceFilter(
query qbtypes.QueryBuilderQuery[qbtypes.LogAggregation],
start, end uint64,
variables map[string]qbtypes.VariableItem,
) (cteSQL string, cteArgs []any, err error) {
) (cteSQL string, cteArgs []any, skipResourceFilter bool, err error) {
stmt, err := b.buildResourceFilterCTE(ctx, query, start, end, variables)
if b.skipResourceFingerprintEnabled {
decision, err := b.resourceFilterResolver.Resolve(ctx, query, start, end, variables)
if err != nil {
return "", nil, true, err
}
switch decision {
case qbtypes.ResourceFilterResolveKindNoOp:
return "", nil, true, nil
case qbtypes.ResourceFilterResolveKindFallback:
return "", nil, false, nil
}
}
stmt, err := b.resourceFilterResolver.StatementBuilder().Build(
ctx, start, end, qbtypes.RequestTypeRaw, query, variables,
)
if err != nil {
return "", nil, err
return "", nil, true, err
}
if stmt == nil {
return "", nil, nil
return "", nil, true, nil
}
sb.Where("resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter)")
return fmt.Sprintf("__resource_filter AS (%s)", stmt.Query), stmt.Args, nil
}
func (b *logQueryStatementBuilder) buildResourceFilterCTE(
ctx context.Context,
query qbtypes.QueryBuilderQuery[qbtypes.LogAggregation],
start, end uint64,
variables map[string]qbtypes.VariableItem,
) (*qbtypes.Statement, error) {
return b.resourceFilterStmtBuilder.Build(
ctx,
start,
end,
qbtypes.RequestTypeRaw,
query,
variables,
)
return fmt.Sprintf("__resource_filter AS (%s)", stmt.Query), stmt.Args, true, nil
}

View File

@@ -2,19 +2,36 @@ package telemetrylogs
import (
"context"
"regexp"
"testing"
"time"
cmock "github.com/SigNoz/clickhouse-go-mock"
"github.com/SigNoz/signoz/pkg/errors"
"github.com/SigNoz/signoz/pkg/flagger/flaggertest"
"github.com/SigNoz/signoz/pkg/instrumentation/instrumentationtest"
"github.com/SigNoz/signoz/pkg/querybuilder"
"github.com/SigNoz/signoz/pkg/telemetrystore"
"github.com/SigNoz/signoz/pkg/telemetrystore/telemetrystoretest"
qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
"github.com/SigNoz/signoz/pkg/types/telemetrytypes/telemetrytypestest"
"github.com/stretchr/testify/require"
)
type regexQueryMatcher struct{}
func (m *regexQueryMatcher) Match(expectedSQL, actualSQL string) error {
re, err := regexp.Compile(expectedSQL)
if err != nil {
return err
}
if !re.MatchString(actualSQL) {
return errors.NewInvalidInputf(errors.CodeInvalidInput, "expected query to match %s, got %s", expectedSQL, actualSQL)
}
return nil
}
func TestStatementBuilderTimeSeries(t *testing.T) {
// Create a test release time
releaseTime := time.Date(2024, 1, 15, 10, 0, 0, 0, time.UTC)
@@ -212,6 +229,9 @@ func TestStatementBuilderTimeSeries(t *testing.T) {
DefaultFullTextColumn,
GetBodyJSONKey,
fl,
nil,
false,
100000,
)
for _, c := range cases {
@@ -353,6 +373,9 @@ func TestStatementBuilderListQuery(t *testing.T) {
DefaultFullTextColumn,
GetBodyJSONKey,
fl,
nil,
false,
100000,
)
for _, c := range cases {
@@ -499,6 +522,9 @@ func TestStatementBuilderListQueryResourceTests(t *testing.T) {
DefaultFullTextColumn,
GetBodyJSONKey,
fl,
nil,
false,
100000,
)
for _, c := range cases {
@@ -575,6 +601,9 @@ func TestStatementBuilderTimeSeriesBodyGroupBy(t *testing.T) {
DefaultFullTextColumn,
GetBodyJSONKey,
fl,
nil,
false,
100000,
)
for _, c := range cases {
@@ -670,6 +699,9 @@ func TestStatementBuilderListQueryServiceCollision(t *testing.T) {
DefaultFullTextColumn,
GetBodyJSONKey,
fl,
nil,
false,
100000,
)
for _, c := range cases {
@@ -894,6 +926,9 @@ func TestAdjustKey(t *testing.T) {
DefaultFullTextColumn,
GetBodyJSONKey,
fl,
nil,
false,
100000,
)
for _, c := range cases {
@@ -1039,6 +1074,9 @@ func TestStmtBuilderBodyField(t *testing.T) {
DefaultFullTextColumn,
GetBodyJSONKey,
fl,
nil,
false,
100000,
)
q, err := statementBuilder.Build(context.Background(), 1747947419000, 1747983448000, c.requestType, c.query, nil)
@@ -1138,6 +1176,9 @@ func TestStmtBuilderBodyFullTextSearch(t *testing.T) {
DefaultFullTextColumn,
GetBodyJSONKey,
fl,
nil,
false,
100000,
)
q, err := statementBuilder.Build(context.Background(), 1747947419000, 1747983448000, c.requestType, c.query, nil)
@@ -1157,3 +1198,110 @@ func TestStmtBuilderBodyFullTextSearch(t *testing.T) {
})
}
}
// TestSkipResourceFingerprintLogs exercises the three resolver outcomes for
// logs: use-CTE (count < threshold), fallback (count >= threshold), and the
// legacy path (feature disabled).
func TestSkipResourceFingerprintLogs(t *testing.T) {
const (
startMs = uint64(1747947419000)
endMs = uint64(1747983448000)
threshold = uint64(10)
)
query := qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]{
Signal: telemetrytypes.SignalLogs,
Filter: &qbtypes.Filter{
Expression: "service.name = 'redis-manual'",
},
Limit: 5,
}
t.Run("disabled uses the legacy CTE", func(t *testing.T) {
sb := newSkipResourceFingerprintLogsBuilder(t, nil, false, threshold)
stmt, err := sb.Build(context.Background(), startMs, endMs, qbtypes.RequestTypeRaw, query, nil)
require.NoError(t, err)
require.Contains(t, stmt.Query, "__resource_filter AS (SELECT fingerprint")
require.Contains(t, stmt.Query, "resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter)")
})
t.Run("CTE attached when count below threshold", func(t *testing.T) {
mockStore := telemetrystoretest.New(telemetrystore.Config{}, &regexQueryMatcher{})
mock := mockStore.Mock()
mock.ExpectQueryRow(`SELECT count\(\) FROM \(SELECT fingerprint FROM signoz_logs\.distributed_logs_v2_resource`).
WillReturnRow(cmock.NewRow([]cmock.ColumnType{
{Name: "count", Type: "UInt64"},
}, []any{uint64(2)}))
sb := newSkipResourceFingerprintLogsBuilder(t, mockStore, true, threshold)
stmt, err := sb.Build(context.Background(), startMs, endMs, qbtypes.RequestTypeRaw, query, nil)
require.NoError(t, err)
require.Contains(t, stmt.Query, "__resource_filter AS (SELECT fingerprint")
require.Contains(t, stmt.Query, "resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter)")
require.NoError(t, mock.ExpectationsWereMet())
})
t.Run("fallback when count at or above threshold", func(t *testing.T) {
mockStore := telemetrystoretest.New(telemetrystore.Config{}, &regexQueryMatcher{})
mock := mockStore.Mock()
mock.ExpectQueryRow(`SELECT count\(\) FROM \(SELECT fingerprint FROM signoz_logs\.distributed_logs_v2_resource`).
WillReturnRow(cmock.NewRow([]cmock.ColumnType{
{Name: "count", Type: "UInt64"},
}, []any{threshold}))
sb := newSkipResourceFingerprintLogsBuilder(t, mockStore, true, threshold)
stmt, err := sb.Build(context.Background(), startMs, endMs, qbtypes.RequestTypeRaw, query, nil)
require.NoError(t, err)
require.NotContains(t, stmt.Query, "__resource_filter AS")
require.NotContains(t, stmt.Query, "resource_fingerprint")
require.Contains(t, stmt.Query, "service.name")
require.NoError(t, mock.ExpectationsWereMet())
})
}
func newSkipResourceFingerprintLogsBuilder(
t *testing.T,
telemetryStore telemetrystore.TelemetryStore,
skipEnable bool,
threshold uint64,
) *logQueryStatementBuilder {
t.Helper()
fl := flaggertest.New(t)
fm := NewFieldMapper(fl)
cb := NewConditionBuilder(fm, fl)
mockMetadataStore := telemetrytypestest.NewMockMetadataStore()
mockMetadataStore.KeysMap = buildCompleteFieldKeyMap(time.Date(2024, 1, 15, 10, 0, 0, 0, time.UTC))
aggExprRewriter := querybuilder.NewAggExprRewriter(
instrumentationtest.New().ToProviderSettings(),
DefaultFullTextColumn,
fm,
cb,
GetBodyJSONKey,
fl,
)
return NewLogQueryStatementBuilder(
instrumentationtest.New().ToProviderSettings(),
mockMetadataStore,
fm,
cb,
aggExprRewriter,
DefaultFullTextColumn,
GetBodyJSONKey,
fl,
telemetryStore,
skipEnable,
threshold,
)
}

View File

@@ -0,0 +1,77 @@
package telemetryresourcefilter
import (
"context"
"github.com/SigNoz/signoz/pkg/factory"
"github.com/SigNoz/signoz/pkg/flagger"
"github.com/SigNoz/signoz/pkg/telemetrystore"
qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
)
type ResourceFingerprintResolver[T any] struct {
stmtBuilder *resourceFilterStatementBuilder[T]
telemetryStore telemetrystore.TelemetryStore
threshold uint64
}
func NewResolver[T any](
settings factory.ProviderSettings,
dbName string,
tableName string,
signal telemetrytypes.Signal,
source telemetrytypes.Source,
metadataStore telemetrytypes.MetadataStore,
fullTextColumn *telemetrytypes.TelemetryFieldKey,
jsonKeyToKey qbtypes.JsonKeyToFieldFunc,
fl flagger.Flagger,
telemetryStore telemetrystore.TelemetryStore,
threshold uint64,
) *ResourceFingerprintResolver[T] {
return &ResourceFingerprintResolver[T]{
stmtBuilder: New[T](
settings,
dbName,
tableName,
signal,
source,
metadataStore,
fullTextColumn,
jsonKeyToKey,
fl,
),
telemetryStore: telemetryStore,
threshold: threshold,
}
}
func (r *ResourceFingerprintResolver[T]) StatementBuilder() qbtypes.StatementBuilder[T] {
return r.stmtBuilder
}
func (r *ResourceFingerprintResolver[T]) Resolve(
ctx context.Context,
query qbtypes.QueryBuilderQuery[T],
start, end uint64,
variables map[string]qbtypes.VariableItem,
) (qbtypes.ResourceFilterResolveKind, error) {
countStmt, err := r.stmtBuilder.BuildCount(ctx, start, end, query, variables)
if err != nil {
return qbtypes.ResourceFilterResolveKindNoOp, err
}
if countStmt == nil {
return qbtypes.ResourceFilterResolveKindNoOp, nil
}
var count uint64
row := r.telemetryStore.ClickhouseDB().QueryRow(ctx, countStmt.Query, countStmt.Args...)
if err := row.Scan(&count); err != nil {
return qbtypes.ResourceFilterResolveKindNoOp, err
}
if count >= r.threshold {
return qbtypes.ResourceFilterResolveKindFallback, nil
}
return qbtypes.ResourceFilterResolveKindUseCTE, nil
}

View File

@@ -127,6 +127,25 @@ func (b *resourceFilterStatementBuilder[T]) Build(
}, nil
}
// BuildCount returns a statement that counts the distinct fingerprints matching
// the resource filter. Returns (nil, nil) when the filter is a no-op.
func (b *resourceFilterStatementBuilder[T]) BuildCount(
ctx context.Context,
start uint64,
end uint64,
query qbtypes.QueryBuilderQuery[T],
variables map[string]qbtypes.VariableItem,
) (*qbtypes.Statement, error) {
inner, err := b.Build(ctx, start, end, qbtypes.RequestTypeRaw, query, variables)
if err != nil || inner == nil {
return nil, err
}
return &qbtypes.Statement{
Query: fmt.Sprintf("SELECT count() FROM (%s)", inner.Query),
Args: inner.Args,
}, nil
}
// addConditions adds both filter and time conditions to the query.
// Returns true (isNoOp) when the filter expression evaluated to no resource conditions,
// meaning the CTE would select all fingerprints and should be skipped entirely.

View File

@@ -24,13 +24,13 @@ var (
)
type traceQueryStatementBuilder struct {
logger *slog.Logger
metadataStore telemetrytypes.MetadataStore
fm qbtypes.FieldMapper
cb qbtypes.ConditionBuilder
resourceFilterStmtBuilder qbtypes.StatementBuilder[qbtypes.TraceAggregation]
aggExprRewriter qbtypes.AggExprRewriter
telemetryStore telemetrystore.TelemetryStore
logger *slog.Logger
metadataStore telemetrytypes.MetadataStore
fm qbtypes.FieldMapper
cb qbtypes.ConditionBuilder
resourceFilterResolver *telemetryresourcefilter.ResourceFingerprintResolver[qbtypes.TraceAggregation]
aggExprRewriter qbtypes.AggExprRewriter
skipResourceFingerprintEnabled bool
}
var _ qbtypes.StatementBuilder[qbtypes.TraceAggregation] = (*traceQueryStatementBuilder)(nil)
@@ -43,10 +43,12 @@ func NewTraceQueryStatementBuilder(
aggExprRewriter qbtypes.AggExprRewriter,
telemetryStore telemetrystore.TelemetryStore,
flagger flagger.Flagger,
skipResourceFingerprintEnable bool,
skipResourceFingerprintThreshold uint64,
) *traceQueryStatementBuilder {
tracesSettings := factory.NewScopedProviderSettings(settings, "github.com/SigNoz/signoz/pkg/telemetrytraces")
resourceFilterStmtBuilder := telemetryresourcefilter.New[qbtypes.TraceAggregation](
resourceFilterResolver := telemetryresourcefilter.NewResolver[qbtypes.TraceAggregation](
settings,
DBName,
TracesResourceV3TableName,
@@ -56,16 +58,18 @@ func NewTraceQueryStatementBuilder(
nil,
nil,
flagger,
telemetryStore,
skipResourceFingerprintThreshold,
)
return &traceQueryStatementBuilder{
logger: tracesSettings.Logger(),
metadataStore: metadataStore,
fm: fieldMapper,
cb: conditionBuilder,
resourceFilterStmtBuilder: resourceFilterStmtBuilder,
aggExprRewriter: aggExprRewriter,
telemetryStore: telemetryStore,
logger: tracesSettings.Logger(),
metadataStore: metadataStore,
fm: fieldMapper,
cb: conditionBuilder,
resourceFilterResolver: resourceFilterResolver,
aggExprRewriter: aggExprRewriter,
skipResourceFingerprintEnabled: skipResourceFingerprintEnable,
}
}
@@ -124,8 +128,10 @@ func (b *traceQueryStatementBuilder) Build(
-------------------------------- End of tech debt ----------------------------
*/
query = b.adjustKeys(ctx, keys, query, requestType)
for _, action := range adjustTraceKeys(keys, &query, requestType) {
// TODO: change to debug level once we are confident about the behavior
b.logger.InfoContext(ctx, "key adjustment action", slog.String("action", action))
}
// Create SQL builder
q := sqlbuilder.NewSelectBuilder()
@@ -193,24 +199,30 @@ func getKeySelectors(query qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation])
return keySelectors
}
func (b *traceQueryStatementBuilder) adjustKeys(ctx context.Context, keys map[string][]*telemetrytypes.TelemetryFieldKey, query qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation], requestType qbtypes.RequestType) qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation] {
// add deprecated fields only during statement building
// why?
// 1. to not fail filter expression that use deprecated cols
// 2. this could have been moved to metadata fetching itself, however, that
// would mean, they also show up in suggestions we we don't want to do
// 3. reason for not doing a simple append is to keep intrinsic/calculated field first so that it gets
// priority in multi_if sql expression
// mergeDeprecatedTraceKeys prepends deprecated intrinsic/calculated trace field
// definitions to the keys map. We do this during statement building, not at
// metadata fetch time, because:
// 1. Filter expressions that reference deprecated columns must continue to
// resolve — otherwise they fail with "key not found".
// 2. Doing it at metadata fetch time would also surface deprecated keys in
// autocomplete suggestions, which we don't want.
// 3. We prepend (not append) so the intrinsic/calculated entry wins ordering
// in the multi_if SQL expression.
func mergeDeprecatedTraceKeys(keys map[string][]*telemetrytypes.TelemetryFieldKey) {
for fieldKeyName, fieldKey := range IntrinsicFieldsDeprecated {
keys[fieldKeyName] = append([]*telemetrytypes.TelemetryFieldKey{&fieldKey}, keys[fieldKeyName]...)
}
for fieldKeyName, fieldKey := range CalculatedFieldsDeprecated {
keys[fieldKeyName] = append([]*telemetrytypes.TelemetryFieldKey{&fieldKey}, keys[fieldKeyName]...)
}
}
func adjustTraceKeys(keys map[string][]*telemetrytypes.TelemetryFieldKey, query *qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation], requestType qbtypes.RequestType) []string {
mergeDeprecatedTraceKeys(keys)
// Adjust keys for alias expressions in aggregations
actions := querybuilder.AdjustKeysForAliasExpressions(&query, requestType)
actions := querybuilder.AdjustKeysForAliasExpressions(query, requestType)
/*
Check if user is using multiple contexts or data types for same field name
@@ -228,7 +240,7 @@ func (b *traceQueryStatementBuilder) adjustKeys(ctx context.Context, keys map[st
and make it just http.status_code and remove the duplicate entry.
*/
actions = append(actions, querybuilder.AdjustDuplicateKeys(&query)...)
actions = append(actions, querybuilder.AdjustDuplicateKeys(query)...)
/*
Now adjust each key to have correct context and data type
@@ -236,24 +248,20 @@ func (b *traceQueryStatementBuilder) adjustKeys(ctx context.Context, keys map[st
Reason for doing this is to not create an unexpected behavior for users
*/
for idx := range query.SelectFields {
actions = append(actions, b.adjustKey(&query.SelectFields[idx], keys)...)
actions = append(actions, adjustTraceKey(&query.SelectFields[idx], keys)...)
}
for idx := range query.GroupBy {
actions = append(actions, b.adjustKey(&query.GroupBy[idx].TelemetryFieldKey, keys)...)
actions = append(actions, adjustTraceKey(&query.GroupBy[idx].TelemetryFieldKey, keys)...)
}
for idx := range query.Order {
actions = append(actions, b.adjustKey(&query.Order[idx].Key.TelemetryFieldKey, keys)...)
actions = append(actions, adjustTraceKey(&query.Order[idx].Key.TelemetryFieldKey, keys)...)
}
for _, action := range actions {
// TODO: change to debug level once we are confident about the behavior
b.logger.InfoContext(ctx, "key adjustment action", slog.String("action", action))
}
return query
return actions
}
func (b *traceQueryStatementBuilder) adjustKey(key *telemetrytypes.TelemetryFieldKey, keys map[string][]*telemetrytypes.TelemetryFieldKey) []string {
// adjustTraceKey resolves a single TelemetryFieldKey against the keys map.
func adjustTraceKey(key *telemetrytypes.TelemetryFieldKey, keys map[string][]*telemetrytypes.TelemetryFieldKey) []string {
// for recording actions taken
actions := []string{}
@@ -302,9 +310,11 @@ func (b *traceQueryStatementBuilder) buildListQuery(
cteArgs [][]any
)
if frag, args, err := b.maybeAttachResourceFilter(ctx, sb, query, start, end, variables); err != nil {
frag, args, skipResourceFilter, err := b.maybeAttachResourceFilter(ctx, sb, query, start, end, variables)
if err != nil {
return nil, err
} else if frag != "" {
}
if frag != "" {
cteFragments = append(cteFragments, frag)
cteArgs = append(cteArgs, args)
}
@@ -322,7 +332,7 @@ func (b *traceQueryStatementBuilder) buildListQuery(
sb.From(fmt.Sprintf("%s.%s", DBName, SpanIndexV3TableName))
// Add filter conditions
preparedWhereClause, err := b.addFilterCondition(ctx, sb, start, end, query, keys, variables)
preparedWhereClause, err := b.addFilterCondition(ctx, sb, start, end, query, keys, variables, skipResourceFilter)
if err != nil {
return nil, err
}
@@ -383,15 +393,17 @@ func (b *traceQueryStatementBuilder) buildTraceQuery(
cteArgs [][]any
)
if frag, args, err := b.maybeAttachResourceFilter(ctx, distSB, query, start, end, variables); err != nil {
frag, args, skipResourceFilter, err := b.maybeAttachResourceFilter(ctx, distSB, query, start, end, variables)
if err != nil {
return nil, err
} else if frag != "" {
}
if frag != "" {
cteFragments = append(cteFragments, frag)
cteArgs = append(cteArgs, args)
}
// Add filter conditions
preparedWhereClause, err := b.addFilterCondition(ctx, distSB, start, end, query, keys, variables)
preparedWhereClause, err := b.addFilterCondition(ctx, distSB, start, end, query, keys, variables, skipResourceFilter)
if err != nil {
return nil, err
}
@@ -492,9 +504,11 @@ func (b *traceQueryStatementBuilder) buildTimeSeriesQuery(
cteArgs [][]any
)
if frag, args, err := b.maybeAttachResourceFilter(ctx, sb, query, start, end, variables); err != nil {
frag, args, skipResourceFilter, err := b.maybeAttachResourceFilter(ctx, sb, query, start, end, variables)
if err != nil {
return nil, err
} else if frag != "" {
}
if frag != "" {
cteFragments = append(cteFragments, frag)
cteArgs = append(cteArgs, args)
}
@@ -535,7 +549,7 @@ func (b *traceQueryStatementBuilder) buildTimeSeriesQuery(
}
sb.From(fmt.Sprintf("%s.%s", DBName, SpanIndexV3TableName))
preparedWhereClause, err := b.addFilterCondition(ctx, sb, start, end, query, keys, variables)
preparedWhereClause, err := b.addFilterCondition(ctx, sb, start, end, query, keys, variables, skipResourceFilter)
if err != nil {
return nil, err
}
@@ -644,9 +658,11 @@ func (b *traceQueryStatementBuilder) buildScalarQuery(
cteArgs [][]any
)
if frag, args, err := b.maybeAttachResourceFilter(ctx, sb, query, start, end, variables); err != nil {
frag, args, skipResourceFilter, err := b.maybeAttachResourceFilter(ctx, sb, query, start, end, variables)
if err != nil {
return nil, err
} else if frag != "" && !skipResourceCTE {
}
if frag != "" && !skipResourceCTE {
cteFragments = append(cteFragments, frag)
cteArgs = append(cteArgs, args)
}
@@ -688,7 +704,7 @@ func (b *traceQueryStatementBuilder) buildScalarQuery(
sb.From(fmt.Sprintf("%s.%s", DBName, SpanIndexV3TableName))
// Add filter conditions
preparedWhereClause, err := b.addFilterCondition(ctx, sb, start, end, query, keys, variables)
preparedWhereClause, err := b.addFilterCondition(ctx, sb, start, end, query, keys, variables, skipResourceFilter)
if err != nil {
return nil, err
}
@@ -751,6 +767,7 @@ func (b *traceQueryStatementBuilder) addFilterCondition(
query qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation],
keys map[string][]*telemetrytypes.TelemetryFieldKey,
variables map[string]qbtypes.VariableItem,
skipResourceFilter bool,
) (querybuilder.PreparedWhereClause, error) {
var preparedWhereClause querybuilder.PreparedWhereClause
@@ -764,7 +781,7 @@ func (b *traceQueryStatementBuilder) addFilterCondition(
FieldMapper: b.fm,
ConditionBuilder: b.cb,
FieldKeys: keys,
SkipResourceFilter: true,
SkipResourceFilter: skipResourceFilter,
Variables: variables,
StartNs: start,
EndNs: end,
@@ -805,34 +822,30 @@ func (b *traceQueryStatementBuilder) maybeAttachResourceFilter(
query qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation],
start, end uint64,
variables map[string]qbtypes.VariableItem,
) (cteSQL string, cteArgs []any, err error) {
) (cteSQL string, cteArgs []any, skipResourceFilter bool, err error) {
stmt, err := b.buildResourceFilterCTE(ctx, query, start, end, variables)
if b.skipResourceFingerprintEnabled {
decision, err := b.resourceFilterResolver.Resolve(ctx, query, start, end, variables)
if err != nil {
return "", nil, true, err
}
switch decision {
case qbtypes.ResourceFilterResolveKindNoOp:
return "", nil, true, nil
case qbtypes.ResourceFilterResolveKindFallback:
return "", nil, false, nil
}
}
stmt, err := b.resourceFilterResolver.StatementBuilder().Build(
ctx, start, end, qbtypes.RequestTypeRaw, query, variables,
)
if err != nil {
return "", nil, err
return "", nil, true, err
}
if stmt == nil {
return "", nil, nil
return "", nil, true, nil
}
sb.Where("resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter)")
return fmt.Sprintf("__resource_filter AS (%s)", stmt.Query), stmt.Args, nil
}
func (b *traceQueryStatementBuilder) buildResourceFilterCTE(
ctx context.Context,
query qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation],
start, end uint64,
variables map[string]qbtypes.VariableItem,
) (*qbtypes.Statement, error) {
return b.resourceFilterStmtBuilder.Build(
ctx,
start,
end,
qbtypes.RequestTypeRaw,
query,
variables,
)
return fmt.Sprintf("__resource_filter AS (%s)", stmt.Query), stmt.Args, true, nil
}

View File

@@ -2,19 +2,36 @@ package telemetrytraces
import (
"context"
"regexp"
"testing"
"time"
cmock "github.com/SigNoz/clickhouse-go-mock"
"github.com/SigNoz/signoz/pkg/errors"
"github.com/SigNoz/signoz/pkg/flagger/flaggertest"
"github.com/SigNoz/signoz/pkg/instrumentation/instrumentationtest"
"github.com/SigNoz/signoz/pkg/querybuilder"
"github.com/SigNoz/signoz/pkg/telemetrystore"
"github.com/SigNoz/signoz/pkg/telemetrystore/telemetrystoretest"
qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
"github.com/SigNoz/signoz/pkg/types/telemetrytypes/telemetrytypestest"
"github.com/stretchr/testify/require"
)
type regexQueryMatcher struct{}
func (m *regexQueryMatcher) Match(expectedSQL, actualSQL string) error {
re, err := regexp.Compile(expectedSQL)
if err != nil {
return err
}
if !re.MatchString(actualSQL) {
return errors.NewInvalidInputf(errors.CodeInvalidInput, "expected query to match %s, got %s", expectedSQL, actualSQL)
}
return nil
}
func TestStatementBuilder(t *testing.T) {
cases := []struct {
name string
@@ -367,6 +384,8 @@ func TestStatementBuilder(t *testing.T) {
aggExprRewriter,
nil,
fl,
false,
100000,
)
vars := map[string]qbtypes.VariableItem{
@@ -662,6 +681,8 @@ func TestStatementBuilderListQuery(t *testing.T) {
aggExprRewriter,
nil,
fl,
false,
100000,
)
for _, c := range cases {
@@ -771,6 +792,8 @@ func TestStatementBuilderListQueryWithCorruptData(t *testing.T) {
aggExprRewriter,
nil,
fl,
false,
100000,
)
q, err := statementBuilder.Build(context.Background(), 1747947419000, 1747983448000, c.requestType, c.query, nil)
@@ -923,6 +946,8 @@ func TestStatementBuilderTraceQuery(t *testing.T) {
aggExprRewriter,
nil,
fl,
false,
100000,
)
for _, c := range cases {
@@ -1125,28 +1150,13 @@ func TestAdjustKey(t *testing.T) {
},
}
fm := NewFieldMapper()
cb := NewConditionBuilder(fm)
mockMetadataStore := telemetrytypestest.NewMockMetadataStore()
fl := flaggertest.New(t)
aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, fm, cb, nil, fl)
statementBuilder := NewTraceQueryStatementBuilder(
instrumentationtest.New().ToProviderSettings(),
mockMetadataStore,
fm,
cb,
aggExprRewriter,
nil,
fl,
)
for _, c := range cases {
t.Run(c.name, func(t *testing.T) {
// Create a copy of the input key to avoid modifying the original
key := c.inputKey
// Call adjustKey
statementBuilder.adjustKey(&key, c.keysMap)
adjustTraceKey(&key, c.keysMap)
// Verify the key was adjusted as expected
require.Equal(t, c.expectedKey.Name, key.Name, "key name should match")
@@ -1399,21 +1409,6 @@ func TestAdjustKeys(t *testing.T) {
},
}
fm := NewFieldMapper()
cb := NewConditionBuilder(fm)
mockMetadataStore := telemetrytypestest.NewMockMetadataStore()
fl := flaggertest.New(t)
aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, fm, cb, nil, fl)
statementBuilder := NewTraceQueryStatementBuilder(
instrumentationtest.New().ToProviderSettings(),
mockMetadataStore,
fm,
cb,
aggExprRewriter,
nil,
fl,
)
for _, c := range cases {
t.Run(c.name, func(t *testing.T) {
// Create a deep copy of the keys map to avoid modifying the original
@@ -1424,7 +1419,7 @@ func TestAdjustKeys(t *testing.T) {
}
// Call adjustKeys
c.query = statementBuilder.adjustKeys(context.Background(), keysMapCopy, c.query, qbtypes.RequestTypeScalar)
adjustTraceKeys(keysMapCopy, &c.query, qbtypes.RequestTypeScalar)
// Verify select fields were adjusted
if c.expectedSelectFields != nil {
@@ -1483,3 +1478,114 @@ func TestAdjustKeys(t *testing.T) {
})
}
}
// TestSkipResourceFingerprint exercises the three resolver outcomes when
// skip_resource_fingerprint is enabled: use-CTE (count < threshold),
// fallback (count >= threshold), and the legacy path (feature disabled).
func TestSkipResourceFingerprint(t *testing.T) {
const (
startMs = uint64(1747947419000)
endMs = uint64(1747983448000)
threshold = uint64(10)
)
query := qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Signal: telemetrytypes.SignalTraces,
Filter: &qbtypes.Filter{
Expression: "service.name = 'redis-manual'",
},
SelectFields: []telemetrytypes.TelemetryFieldKey{
{
Name: "name",
FieldContext: telemetrytypes.FieldContextSpan,
FieldDataType: telemetrytypes.FieldDataTypeString,
},
},
Limit: 5,
}
t.Run("disabled uses the legacy CTE", func(t *testing.T) {
sb := newSkipResourceFingerprintBuilder(t, nil, false, threshold)
stmt, err := sb.Build(context.Background(), startMs, endMs, qbtypes.RequestTypeRaw, query, nil)
require.NoError(t, err)
require.Contains(t, stmt.Query, "__resource_filter AS (SELECT fingerprint")
require.Contains(t, stmt.Query, "resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter)")
})
t.Run("CTE attached when count below threshold", func(t *testing.T) {
mockStore := telemetrystoretest.New(telemetrystore.Config{}, &regexQueryMatcher{})
mock := mockStore.Mock()
// Only the count query runs against the telemetry store; the CTE
// itself is embedded as SQL in the main query (no extra round trip).
mock.ExpectQueryRow(`SELECT count\(\) FROM \(SELECT fingerprint FROM signoz_traces\.distributed_traces_v3_resource`).
WillReturnRow(cmock.NewRow([]cmock.ColumnType{
{Name: "count", Type: "UInt64"},
}, []any{uint64(2)}))
sb := newSkipResourceFingerprintBuilder(t, mockStore, true, threshold)
stmt, err := sb.Build(context.Background(), startMs, endMs, qbtypes.RequestTypeRaw, query, nil)
require.NoError(t, err)
require.Contains(t, stmt.Query, "__resource_filter AS (SELECT fingerprint")
require.Contains(t, stmt.Query, "resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter)")
require.NoError(t, mock.ExpectationsWereMet())
})
t.Run("fallback when count at or above threshold", func(t *testing.T) {
mockStore := telemetrystoretest.New(telemetrystore.Config{}, &regexQueryMatcher{})
mock := mockStore.Mock()
mock.ExpectQueryRow(`SELECT count\(\) FROM \(SELECT fingerprint FROM signoz_traces\.distributed_traces_v3_resource`).
WillReturnRow(cmock.NewRow([]cmock.ColumnType{
{Name: "count", Type: "UInt64"},
}, []any{threshold}))
sb := newSkipResourceFingerprintBuilder(t, mockStore, true, threshold)
stmt, err := sb.Build(context.Background(), startMs, endMs, qbtypes.RequestTypeRaw, query, nil)
require.NoError(t, err)
require.NotContains(t, stmt.Query, "__resource_filter AS")
require.NotContains(t, stmt.Query, "resource_fingerprint")
// resource conditions are pushed onto the main table via the
// resource.`service.name` / resources_string lookup
require.Contains(t, stmt.Query, "service.name")
require.NoError(t, mock.ExpectationsWereMet())
})
}
func newSkipResourceFingerprintBuilder(
t *testing.T,
telemetryStore telemetrystore.TelemetryStore,
skipEnable bool,
threshold uint64,
) *traceQueryStatementBuilder {
t.Helper()
fm := NewFieldMapper()
cb := NewConditionBuilder(fm)
mockMetadataStore := telemetrytypestest.NewMockMetadataStore()
mockMetadataStore.KeysMap = buildCompleteFieldKeyMap()
fl := flaggertest.New(t)
aggExprRewriter := querybuilder.NewAggExprRewriter(
instrumentationtest.New().ToProviderSettings(), nil, fm, cb, nil, fl,
)
return NewTraceQueryStatementBuilder(
instrumentationtest.New().ToProviderSettings(),
mockMetadataStore,
fm,
cb,
aggExprRewriter,
telemetryStore,
fl,
skipEnable,
threshold,
)
}

View File

@@ -216,6 +216,13 @@ func (b *traceOperatorCTEBuilder) buildQueryCTE(ctx context.Context, queryName s
}
b.stmtBuilder.logger.DebugContext(ctx, "Retrieved keys for query", slog.String("query_name", queryName), slog.Int("keys_count", len(keys)))
// The CTE only selects spans matching the filter. Aggregations, group by
// and order by run later in buildFinalQuery, so RequestTypeRaw is fine here.
for _, action := range adjustTraceKeys(keys, query, qbtypes.RequestTypeRaw) {
// TODO: change to debug level once we are confident about the behavior
b.stmtBuilder.logger.InfoContext(ctx, "key adjustment action", slog.String("action", action))
}
// Build resource filter CTE for this specific query
resourceFilterCTEName := fmt.Sprintf("__resource_filter_%s", cteName)
resourceStmt, err := b.buildResourceFilterCTE(ctx, *query)
@@ -417,21 +424,28 @@ func (b *traceOperatorCTEBuilder) buildNotCTE(leftCTE, rightCTE string) (string,
}
func (b *traceOperatorCTEBuilder) buildFinalQuery(ctx context.Context, selectFromCTE string, requestType qbtypes.RequestType) (*qbtypes.Statement, error) {
keySelectors := b.getKeySelectors()
keys, _, err := b.stmtBuilder.metadataStore.GetKeysMulti(ctx, keySelectors)
if err != nil {
return nil, err
}
b.adjustOperatorKeys(ctx, keys, requestType)
switch requestType {
case qbtypes.RequestTypeRaw:
return b.buildListQuery(ctx, selectFromCTE)
return b.buildListQuery(ctx, selectFromCTE, keys)
case qbtypes.RequestTypeTimeSeries:
return b.buildTimeSeriesQuery(ctx, selectFromCTE)
return b.buildTimeSeriesQuery(ctx, selectFromCTE, keys)
case qbtypes.RequestTypeTrace:
return b.buildTraceQuery(ctx, selectFromCTE)
return b.buildTraceQuery(ctx, selectFromCTE, keys)
case qbtypes.RequestTypeScalar:
return b.buildScalarQuery(ctx, selectFromCTE)
return b.buildScalarQuery(ctx, selectFromCTE, keys)
default:
return nil, errors.NewInvalidInputf(errors.CodeInvalidInput, "unsupported request type: %s", requestType)
}
}
func (b *traceOperatorCTEBuilder) buildListQuery(ctx context.Context, selectFromCTE string) (*qbtypes.Statement, error) {
func (b *traceOperatorCTEBuilder) buildListQuery(ctx context.Context, selectFromCTE string, keys map[string][]*telemetrytypes.TelemetryFieldKey) (*qbtypes.Statement, error) {
sb := sqlbuilder.NewSelectBuilder()
// Select core fields
@@ -453,22 +467,6 @@ func (b *traceOperatorCTEBuilder) buildListQuery(ctx context.Context, selectFrom
"parent_span_id": true,
}
// Get keys for selectFields
keySelectors := b.getKeySelectors()
for _, field := range b.operator.SelectFields {
keySelectors = append(keySelectors, &telemetrytypes.FieldKeySelector{
Name: field.Name,
Signal: telemetrytypes.SignalTraces,
FieldContext: field.FieldContext,
FieldDataType: field.FieldDataType,
})
}
keys, _, err := b.stmtBuilder.metadataStore.GetKeysMulti(ctx, keySelectors)
if err != nil {
return nil, err
}
// Add selectFields using ColumnExpressionFor since we now have all base table columns
for _, field := range b.operator.SelectFields {
if selectedFields[field.Name] {
@@ -518,6 +516,44 @@ func (b *traceOperatorCTEBuilder) buildListQuery(ctx context.Context, selectFrom
}, nil
}
// adjustOperatorKeys runs the same key adjustments as adjustTraceKeys, but on
// the operator's own fields. The operator has a different struct shape than
// QueryBuilderQuery, so we copy the relevant fields into a temp query, run
// the shared helpers, and copy the results back.
func (b *traceOperatorCTEBuilder) adjustOperatorKeys(ctx context.Context, keys map[string][]*telemetrytypes.TelemetryFieldKey, requestType qbtypes.RequestType) {
mergeDeprecatedTraceKeys(keys)
tmp := qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Aggregations: b.operator.Aggregations,
SelectFields: b.operator.SelectFields,
GroupBy: b.operator.GroupBy,
Order: b.operator.Order,
}
actions := querybuilder.AdjustKeysForAliasExpressions(&tmp, requestType)
actions = append(actions, querybuilder.AdjustDuplicateKeys(&tmp)...)
for idx := range tmp.SelectFields {
actions = append(actions, adjustTraceKey(&tmp.SelectFields[idx], keys)...)
}
for idx := range tmp.GroupBy {
actions = append(actions, adjustTraceKey(&tmp.GroupBy[idx].TelemetryFieldKey, keys)...)
}
for idx := range tmp.Order {
actions = append(actions, adjustTraceKey(&tmp.Order[idx].Key.TelemetryFieldKey, keys)...)
}
// Copy back the slices the helpers can rewrite.
b.operator.Aggregations = tmp.Aggregations
b.operator.SelectFields = tmp.SelectFields
b.operator.GroupBy = tmp.GroupBy
b.operator.Order = tmp.Order
for _, action := range actions {
b.stmtBuilder.logger.InfoContext(ctx, "key adjustment action", slog.String("action", action))
}
}
func (b *traceOperatorCTEBuilder) getKeySelectors() []*telemetrytypes.FieldKeySelector {
var keySelectors []*telemetrytypes.FieldKeySelector
@@ -545,6 +581,15 @@ func (b *traceOperatorCTEBuilder) getKeySelectors() []*telemetrytypes.FieldKeySe
})
}
for _, sf := range b.operator.SelectFields {
keySelectors = append(keySelectors, &telemetrytypes.FieldKeySelector{
Name: sf.Name,
Signal: telemetrytypes.SignalTraces,
FieldContext: sf.FieldContext,
FieldDataType: sf.FieldDataType,
})
}
for i := range keySelectors {
keySelectors[i].Signal = telemetrytypes.SignalTraces
}
@@ -552,7 +597,7 @@ func (b *traceOperatorCTEBuilder) getKeySelectors() []*telemetrytypes.FieldKeySe
return keySelectors
}
func (b *traceOperatorCTEBuilder) buildTimeSeriesQuery(ctx context.Context, selectFromCTE string) (*qbtypes.Statement, error) {
func (b *traceOperatorCTEBuilder) buildTimeSeriesQuery(ctx context.Context, selectFromCTE string, keys map[string][]*telemetrytypes.TelemetryFieldKey) (*qbtypes.Statement, error) {
sb := sqlbuilder.NewSelectBuilder()
sb.Select(fmt.Sprintf(
@@ -560,12 +605,6 @@ func (b *traceOperatorCTEBuilder) buildTimeSeriesQuery(ctx context.Context, sele
int64(b.operator.StepInterval.Seconds()),
))
keySelectors := b.getKeySelectors()
keys, _, err := b.stmtBuilder.metadataStore.GetKeysMulti(ctx, keySelectors)
if err != nil {
return nil, err
}
var allGroupByArgs []any
for _, gb := range b.operator.GroupBy {
@@ -644,8 +683,7 @@ func (b *traceOperatorCTEBuilder) buildTimeSeriesQuery(ctx context.Context, sele
combinedArgs := append(allGroupByArgs, allAggChArgs...)
// Add HAVING clause if specified
err = b.addHavingClause(sb)
if err != nil {
if err := b.addHavingClause(sb); err != nil {
return nil, err
}
@@ -672,17 +710,11 @@ func (b *traceOperatorCTEBuilder) buildTraceSummaryCTE(selectFromCTE string) {
b.addCTE("trace_summary", sql, args, []string{"all_spans", selectFromCTE})
}
func (b *traceOperatorCTEBuilder) buildTraceQuery(ctx context.Context, selectFromCTE string) (*qbtypes.Statement, error) {
func (b *traceOperatorCTEBuilder) buildTraceQuery(ctx context.Context, selectFromCTE string, keys map[string][]*telemetrytypes.TelemetryFieldKey) (*qbtypes.Statement, error) {
b.buildTraceSummaryCTE(selectFromCTE)
sb := sqlbuilder.NewSelectBuilder()
keySelectors := b.getKeySelectors()
keys, _, err := b.stmtBuilder.metadataStore.GetKeysMulti(ctx, keySelectors)
if err != nil {
return nil, err
}
var allGroupByArgs []any
for _, gb := range b.operator.GroupBy {
@@ -764,8 +796,7 @@ func (b *traceOperatorCTEBuilder) buildTraceQuery(ctx context.Context, selectFro
sb.GroupBy(groupByKeys...)
}
err = b.addHavingClause(sb)
if err != nil {
if err := b.addHavingClause(sb); err != nil {
return nil, err
}
@@ -821,15 +852,9 @@ func (b *traceOperatorCTEBuilder) buildTraceQuery(ctx context.Context, selectFro
}, nil
}
func (b *traceOperatorCTEBuilder) buildScalarQuery(ctx context.Context, selectFromCTE string) (*qbtypes.Statement, error) {
func (b *traceOperatorCTEBuilder) buildScalarQuery(ctx context.Context, selectFromCTE string, keys map[string][]*telemetrytypes.TelemetryFieldKey) (*qbtypes.Statement, error) {
sb := sqlbuilder.NewSelectBuilder()
keySelectors := b.getKeySelectors()
keys, _, err := b.stmtBuilder.metadataStore.GetKeysMulti(ctx, keySelectors)
if err != nil {
return nil, err
}
var allGroupByArgs []any
for _, gb := range b.operator.GroupBy {
@@ -911,8 +936,7 @@ func (b *traceOperatorCTEBuilder) buildScalarQuery(ctx context.Context, selectFr
combinedArgs := append(allGroupByArgs, allAggChArgs...)
// Add HAVING clause if specified
err = b.addHavingClause(sb)
if err != nil {
if err := b.addHavingClause(sb); err != nil {
return nil, err
}

View File

@@ -14,6 +14,24 @@ import (
"github.com/stretchr/testify/require"
)
func newTestTraceOperatorStatementBuilder(t *testing.T) *traceOperatorStatementBuilder {
t.Helper()
fm := NewFieldMapper()
cb := NewConditionBuilder(fm)
mockMetadataStore := telemetrytypestest.NewMockMetadataStore()
mockMetadataStore.KeysMap = buildCompleteFieldKeyMap()
fl := flaggertest.New(t)
aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, fm, cb, nil, fl)
traceStmtBuilder := NewTraceQueryStatementBuilder(
instrumentationtest.New().ToProviderSettings(),
mockMetadataStore, fm, cb, aggExprRewriter, nil, fl, false, 100000,
)
return NewTraceOperatorStatementBuilder(
instrumentationtest.New().ToProviderSettings(),
mockMetadataStore, fm, cb, traceStmtBuilder, aggExprRewriter, fl,
)
}
func TestTraceOperatorStatementBuilder(t *testing.T) {
cases := []struct {
name string
@@ -463,32 +481,7 @@ func TestTraceOperatorStatementBuilder(t *testing.T) {
},
}
fm := NewFieldMapper()
cb := NewConditionBuilder(fm)
mockMetadataStore := telemetrytypestest.NewMockMetadataStore()
mockMetadataStore.KeysMap = buildCompleteFieldKeyMap()
fl := flaggertest.New(t)
aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, fm, cb, nil, fl)
traceStmtBuilder := NewTraceQueryStatementBuilder(
instrumentationtest.New().ToProviderSettings(),
mockMetadataStore,
fm,
cb,
aggExprRewriter,
nil,
fl,
)
statementBuilder := NewTraceOperatorStatementBuilder(
instrumentationtest.New().ToProviderSettings(),
mockMetadataStore,
fm,
cb,
traceStmtBuilder,
aggExprRewriter,
fl,
)
statementBuilder := newTestTraceOperatorStatementBuilder(t)
for _, c := range cases {
t.Run(c.name, func(t *testing.T) {
@@ -579,32 +572,7 @@ func TestTraceOperatorStatementBuilderErrors(t *testing.T) {
},
}
fm := NewFieldMapper()
cb := NewConditionBuilder(fm)
mockMetadataStore := telemetrytypestest.NewMockMetadataStore()
mockMetadataStore.KeysMap = buildCompleteFieldKeyMap()
fl := flaggertest.New(t)
aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, fm, cb, nil, fl)
traceStmtBuilder := NewTraceQueryStatementBuilder(
instrumentationtest.New().ToProviderSettings(),
mockMetadataStore,
fm,
cb,
aggExprRewriter,
nil,
fl,
)
statementBuilder := NewTraceOperatorStatementBuilder(
instrumentationtest.New().ToProviderSettings(),
mockMetadataStore,
fm,
cb,
traceStmtBuilder,
aggExprRewriter,
fl,
)
statementBuilder := newTestTraceOperatorStatementBuilder(t)
for _, c := range cases {
t.Run(c.name, func(t *testing.T) {
@@ -626,3 +594,142 @@ func TestTraceOperatorStatementBuilderErrors(t *testing.T) {
})
}
}
func TestTraceOperatorStatementBuilderAdjustsKeys(t *testing.T) {
cases := []struct {
name string
requestType qbtypes.RequestType
operator qbtypes.QueryBuilderTraceOperator
builderFilter string
wantSQL string
wantArgs []any
}{
{
name: "deprecated duration filter in referenced builder query",
requestType: qbtypes.RequestTypeRaw,
operator: qbtypes.QueryBuilderTraceOperator{
Expression: "A",
Limit: 10,
},
builderFilter: "durationNano = '3s'",
wantSQL: "duration_nano = ?",
wantArgs: []any{int64(3000000000)},
},
{
name: "context-prefixed aggregation alias in order by",
requestType: qbtypes.RequestTypeScalar,
operator: qbtypes.QueryBuilderTraceOperator{
Expression: "A",
Aggregations: []qbtypes.TraceAggregation{
{
Expression: "count()",
Alias: "span.count_",
},
},
Order: []qbtypes.OrderBy{
{
Key: qbtypes.OrderByKey{
TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{
Name: "count_",
FieldContext: telemetrytypes.FieldContextSpan,
},
},
Direction: qbtypes.OrderDirectionDesc,
},
},
},
wantSQL: "ORDER BY __result_0 desc",
},
}
statementBuilder := newTestTraceOperatorStatementBuilder(t)
for _, c := range cases {
t.Run(c.name, func(t *testing.T) {
err := c.operator.ParseExpression()
require.NoError(t, err)
filter := c.builderFilter
if filter == "" {
filter = "service.name = 'frontend'"
}
q, err := statementBuilder.Build(
context.Background(),
1747947419000,
1747983448000,
c.requestType,
c.operator,
&qbtypes.CompositeQuery{
Queries: []qbtypes.QueryEnvelope{
{
Type: qbtypes.QueryTypeBuilder,
Spec: qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Name: "A",
Signal: telemetrytypes.SignalTraces,
Filter: &qbtypes.Filter{Expression: filter},
},
},
},
},
)
require.NoError(t, err)
require.Contains(t, q.Query, c.wantSQL)
for _, arg := range c.wantArgs {
require.Contains(t, q.Args, arg)
}
})
}
}
// TestTraceOperatorStatementBuilderDeduplicatesKeys checks that a trace
// operator with the same field name listed twice in GroupBy (once with a
// context, once without) ends up with a single column in the outer SELECT
// and a single entry in GROUP BY.
func TestTraceOperatorStatementBuilderDeduplicatesKeys(t *testing.T) {
statementBuilder := newTestTraceOperatorStatementBuilder(t)
operator := qbtypes.QueryBuilderTraceOperator{
Expression: "A",
Aggregations: []qbtypes.TraceAggregation{
{Expression: "count()"},
},
GroupBy: []qbtypes.GroupByKey{
{TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{
Name: "http.method",
FieldContext: telemetrytypes.FieldContextAttribute,
}},
// Same name, no context — should be merged with the entry above.
{TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{
Name: "http.method",
}},
},
}
require.NoError(t, operator.ParseExpression())
q, err := statementBuilder.Build(
context.Background(),
1747947419000,
1747983448000,
qbtypes.RequestTypeScalar,
operator,
&qbtypes.CompositeQuery{
Queries: []qbtypes.QueryEnvelope{
{
Type: qbtypes.QueryTypeBuilder,
Spec: qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Name: "A",
Signal: telemetrytypes.SignalTraces,
Filter: &qbtypes.Filter{Expression: "service.name = 'frontend'"},
},
},
},
},
)
require.NoError(t, err)
require.Contains(t, q.Query,
"SELECT toString(multiIf(mapContains(attributes_string, 'http.method') = ?, attributes_string['http.method'], NULL)) AS `http.method`, count() AS __result_0 FROM A GROUP BY `http.method` ORDER BY __result_0 DESC")
}

View File

@@ -5,13 +5,13 @@ import (
"strings"
"testing"
"github.com/SigNoz/signoz/pkg/flagger/flaggertest"
"github.com/SigNoz/signoz/pkg/instrumentation/instrumentationtest"
"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/types/telemetrytypes/telemetrytypestest"
"github.com/stretchr/testify/assert"
"github.com/SigNoz/signoz/pkg/flagger/flaggertest"
"github.com/stretchr/testify/require"
)
@@ -44,8 +44,10 @@ func TestTraceTimeRangeOptimization(t *testing.T) {
fm,
cb,
aggExprRewriter,
nil, // telemetryStore is nil - optimization won't happen but code path is tested
nil, // telemetryStore is nil - adaptive path is disabled
fl,
false,
100000,
)
tests := []struct {

View File

@@ -0,0 +1,9 @@
package querybuildertypesv5
type ResourceFilterResolveKind int
const (
ResourceFilterResolveKindNoOp ResourceFilterResolveKind = iota
ResourceFilterResolveKindUseCTE
ResourceFilterResolveKindFallback
)

View File

@@ -54,16 +54,16 @@ func newConfig() factory.Config {
Directory: "/etc/signoz/web",
Settings: SettingsConfig{
Posthog: PosthogConfig{
Enabled: true,
Enabled: false,
},
Appcues: AppcuesConfig{
Enabled: true,
Enabled: false,
},
Sentry: SentryConfig{
Enabled: true,
Enabled: false,
},
Pylon: PylonConfig{
Enabled: true,
Enabled: false,
},
},
}

View File

@@ -459,6 +459,57 @@ def find_named_result(
)
def assert_scalar_value(
response: requests.Response,
name: str,
expected: Any,
*,
row: int = 0,
col: int = 0,
) -> None:
"""Assert that the named scalar result has `expected` at data[row][col]."""
result = find_named_result(response.json()["data"]["data"]["results"], name)
assert result is not None, f"no result for query {name}"
assert result["data"][row][col] == expected, f"expected {expected} at [{row}][{col}], got {result['data'][row][col]}"
def assert_grouped_scalar(
response: requests.Response,
name: str,
*,
expected_groups: int,
expected_columns: int,
last_col_value: Any | None = None,
) -> None:
"""Assert grouped scalar result has the expected column count and group count.
If `last_col_value` is set and there is exactly one group, also assert the
last column of that single row equals it (a common aggregation-value check)."""
result = find_named_result(response.json()["data"]["data"]["results"], name)
assert result is not None, f"no result for query {name}"
columns = result["columns"]
rows = result["data"]
assert len(columns) == expected_columns, f"expected {expected_columns} columns, got {len(columns)}: {columns}"
assert len(rows) == expected_groups, f"expected {expected_groups} groups, got {len(rows)}: {rows}"
if last_col_value is not None and expected_groups == 1:
assert rows[0][-1] == last_col_value, f"expected last col {last_col_value}, got row {rows[0]}"
def assert_raw_row_subset(
response: requests.Response,
name: str,
expected: dict[str, Any],
*,
row: int = 0,
) -> None:
"""Assert that the named raw result's rows[row]['data'] is a superset of `expected`."""
result = find_named_result(response.json()["data"]["data"]["results"], name)
assert result is not None, f"no result for query {name}"
rows = result["rows"]
assert rows is not None, f"no rows for query {name}"
data = rows[row]["data"]
assert expected.items() <= data.items(), f"expected subset {expected}, got data {data}"
def build_scalar_query(
name: str,
signal: str,

View File

@@ -25,13 +25,22 @@ returnSpansFrom="A"
from collections.abc import Callable
from datetime import UTC, datetime, timedelta
from http import HTTPStatus
from typing import Any
import pytest
import requests
from fixtures import types
from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
from fixtures.querier import get_rows
from fixtures.querier import (
assert_grouped_scalar,
assert_raw_row_subset,
assert_scalar_value,
format_timestamp,
generate_traces_with_corrupt_metadata,
get_rows,
make_query_request,
)
from fixtures.traces import TraceIdGenerator, Traces, TracesKind, TracesStatusCode
@@ -434,3 +443,173 @@ def test_trace_operator(
)
assert response.status_code == HTTPStatus.OK, f"HTTP {response.status_code}: {response.text}"
assert case["validate"](response), f"validation failed: {response.json()}"
def _expected_trace_subset(trace: Traces) -> dict[str, Any]:
return {
"duration_nano": trace.duration_nano,
"name": trace.name,
"parent_span_id": trace.parent_span_id,
"span_id": trace.span_id,
"timestamp": format_timestamp(trace.timestamp),
"trace_id": trace.trace_id,
}
@pytest.mark.parametrize(
"payload_factory,request_type,assert_result",
[
# Case 1: CTE filter uses the deprecated intrinsic field `durationNano`.
pytest.param(
lambda traces: [
{
"type": "builder_query",
"spec": {
"name": "A",
"signal": "traces",
"filter": {"expression": 'durationNano = "3s"'},
},
},
{
"type": "builder_query",
"spec": {
"name": "B",
"signal": "traces",
"filter": {"expression": 'durationNano = "5s"'},
},
},
{
"type": "builder_trace_operator",
"spec": {
"name": "C",
"expression": "A => B",
"limit": 1,
},
},
],
"raw",
lambda response, traces: assert_raw_row_subset(response, "C", _expected_trace_subset(traces[0])),
id="deprecated-intrinsic-filter",
),
# Case 2: CTE filter uses the deprecated calculated field `responseStatusCode`.
pytest.param(
lambda traces: [
{
"type": "builder_query",
"spec": {
"name": "A",
"signal": "traces",
"filter": {"expression": 'responseStatusCode = "200"'},
},
},
{
"type": "builder_query",
"spec": {
"name": "B",
"signal": "traces",
"filter": {"expression": 'durationNano = "5s"'},
},
},
{
"type": "builder_trace_operator",
"spec": {
"name": "C",
"expression": "A => B",
"limit": 1,
},
},
],
"raw",
lambda response, traces: assert_raw_row_subset(response, "C", _expected_trace_subset(traces[0])),
id="deprecated-calculated-filter",
),
# Case 3: order by uses `count_` with fieldContext `span`, which has
# to be rewritten to the aggregation alias `span.count_`.
pytest.param(
lambda traces: [
{
"type": "builder_query",
"spec": {
"name": "A",
"signal": "traces",
"aggregations": [{"expression": "count()"}],
},
},
{
"type": "builder_trace_operator",
"spec": {
"name": "C",
"expression": "A",
"aggregations": [{"expression": "count()", "alias": "span.count_"}],
"order": [{"key": {"name": "count_", "fieldContext": "span"}, "direction": "desc"}],
},
},
],
"scalar",
lambda response, traces: assert_scalar_value(response, "C", len(traces)),
id="context-prefixed-aggregation-alias-order",
),
# Case 4: group by lists `cloud.provider` twice (once with a resource
# context, once without).
pytest.param(
lambda traces: [
{
"type": "builder_query",
"spec": {
"name": "A",
"signal": "traces",
"disabled": True,
"aggregations": [{"expression": "count()"}],
},
},
{
"type": "builder_trace_operator",
"spec": {
"name": "C",
"expression": "A",
"aggregations": [{"expression": "count()"}],
"groupBy": [
{"name": "cloud.provider", "fieldContext": "resource"},
{"name": "cloud.provider"},
],
},
},
],
"scalar",
lambda response, traces: assert_grouped_scalar(response, "C", expected_groups=1, expected_columns=2, last_col_value=len(traces)),
id="duplicate-group-by-deduplicated",
),
],
)
def test_trace_operator_with_adjusted_keys(
signoz: types.SigNoz,
create_user_admin: None, # pylint: disable=unused-argument
get_token: Callable[[str, str], str],
insert_traces: Callable[[list[Traces]], None],
payload_factory: Callable[[list[Traces]], list[dict[str, Any]]],
request_type: str,
assert_result: Callable[[requests.Response, list[Traces]], None],
) -> None:
"""
Trace operators build a CTE per referenced builder query and an outer
query on top. Both layers need the same key adjustment as regular trace
queries, otherwise deprecated keys and context-prefixed aliases don't
resolve.
"""
traces = generate_traces_with_corrupt_metadata()
insert_traces(traces)
payload = payload_factory(traces)
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
response = make_query_request(
signoz,
token,
start_ms=int((datetime.now(tz=UTC) - timedelta(minutes=5)).timestamp() * 1000),
end_ms=int(datetime.now(tz=UTC).timestamp() * 1000),
request_type=request_type,
queries=payload,
)
assert response.status_code == HTTPStatus.OK, response.text
assert_result(response, traces)

View File

@@ -0,0 +1,180 @@
"""
End-to-end coverage for the skip_resource_fingerprint querier optimization.
The conftest in this package boots SigNoz with:
- skip_resource_fingerprint.enabled = true
- skip_resource_fingerprint.threshold = 2
With that configuration the two non-trivial resolver branches are reachable
from a single SigNoz instance:
- count < 2 -> the resolver attaches the fingerprint CTE (same shape as the
legacy path; cheap because the fingerprint set is small).
- count >= 2 -> fallback path: no fingerprint subquery, resource conditions
are evaluated directly on the main spans table.
These tests assert end-to-end correctness — the optimization must be
semantically transparent.
"""
from collections.abc import Callable
from datetime import UTC, datetime, timedelta
from http import HTTPStatus
from fixtures import types
from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
from fixtures.querier import (
Aggregation,
BuilderQuery,
OrderBy,
TelemetryFieldKey,
get_rows,
make_query_request,
)
from fixtures.traces import TraceIdGenerator, Traces, TracesKind, TracesStatusCode
def _span(
*,
timestamp: datetime,
service_name: str,
name: str = "span",
duration_seconds: float = 1.0,
extra_resources: dict | None = None,
attributes: dict | None = None,
) -> Traces:
resources = {"service.name": service_name}
if extra_resources:
resources.update(extra_resources)
return Traces(
timestamp=timestamp,
duration=timedelta(seconds=duration_seconds),
trace_id=TraceIdGenerator.trace_id(),
span_id=TraceIdGenerator.span_id(),
parent_span_id="",
name=name,
kind=TracesKind.SPAN_KIND_SERVER,
status_code=TracesStatusCode.STATUS_CODE_OK,
status_message="",
resources=resources,
attributes=attributes or {},
)
def test_skip_resource_fingerprint_use_cte_path(
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 filter that matches a single unique resource fingerprint (count = 1 < 2)
keeps the legacy CTE attached. The query should still return only the rows
belonging to that resource.
"""
now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
insert_traces(
[
_span(timestamp=now - timedelta(seconds=10), service_name="skip-cte-svc", name="span-1"),
_span(timestamp=now - timedelta(seconds=8), service_name="skip-cte-svc", name="span-2"),
# Noise from a different resource — must be filtered out.
_span(timestamp=now - timedelta(seconds=6), service_name="skip-cte-noise", name="span-noise"),
]
)
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
response = make_query_request(
signoz,
token,
start_ms=int((datetime.now(tz=UTC) - timedelta(minutes=5)).timestamp() * 1000),
end_ms=int(datetime.now(tz=UTC).timestamp() * 1000),
request_type="raw",
queries=[
BuilderQuery(
signal="traces",
limit=50,
order=[OrderBy(TelemetryFieldKey("timestamp"), "asc")],
filter_expression="service.name = 'skip-cte-svc'",
select_fields=[
TelemetryFieldKey("service.name", "string", "resource"),
TelemetryFieldKey("name", field_context="span"),
],
aggregations=[Aggregation("count()")],
).to_dict()
],
)
assert response.status_code == HTTPStatus.OK
rows = get_rows(response)
assert len(rows) == 2, f"expected only the 2 'skip-cte-svc' spans, got {len(rows)}"
names = [row["data"]["name"] for row in rows]
assert names == ["span-1", "span-2"]
services = {row["data"]["service.name"] for row in rows}
assert services == {"skip-cte-svc"}
def test_skip_resource_fingerprint_fallback_path(
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 filter that matches multiple unique resource fingerprints (count >= 2)
drives the resolver down the fallback path: no fingerprint subquery, and
the resource condition is evaluated directly on the main spans table.
The result must still be correct (no over- or under-matching).
"""
now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
# 3 services share the same deployment.environment, so the resource filter
# selects 3 fingerprints, exceeding our threshold of 2.
fallback_env = {"deployment.environment": "skip-fallback"}
insert_traces(
[
_span(timestamp=now - timedelta(seconds=10), service_name="skip-fb-svc-a", extra_resources=fallback_env),
_span(timestamp=now - timedelta(seconds=9), service_name="skip-fb-svc-b", extra_resources=fallback_env),
_span(timestamp=now - timedelta(seconds=8), service_name="skip-fb-svc-c", extra_resources=fallback_env),
# Noise without the fallback env — must be filtered out.
_span(
timestamp=now - timedelta(seconds=7),
service_name="skip-fb-other",
extra_resources={"deployment.environment": "skip-other"},
),
]
)
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
response = make_query_request(
signoz,
token,
start_ms=int((datetime.now(tz=UTC) - timedelta(minutes=5)).timestamp() * 1000),
end_ms=int(datetime.now(tz=UTC).timestamp() * 1000),
request_type="raw",
queries=[
BuilderQuery(
signal="traces",
limit=50,
order=[OrderBy(TelemetryFieldKey("timestamp"), "asc")],
filter_expression="deployment.environment = 'skip-fallback'",
select_fields=[
TelemetryFieldKey("service.name", "string", "resource"),
],
aggregations=[Aggregation("count()")],
).to_dict()
],
)
assert response.status_code == HTTPStatus.OK
rows = get_rows(response)
assert len(rows) == 3, f"expected 3 spans tagged with skip-fallback, got {len(rows)}"
services = sorted(row["data"]["service.name"] for row in rows)
assert services == ["skip-fb-svc-a", "skip-fb-svc-b", "skip-fb-svc-c"]

View File

@@ -0,0 +1,164 @@
"""
End-to-end coverage for the skip_resource_fingerprint optimization on logs.
The conftest boots SigNoz with threshold=2, so:
- count < 2 -> resolver attaches the fingerprint CTE (same shape as legacy).
- count >= 2 -> fallback: resource conditions evaluated on the main logs table.
Both branches must return the same correct rows.
"""
from collections.abc import Callable
from datetime import UTC, datetime, timedelta
from http import HTTPStatus
from fixtures import types
from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
from fixtures.logs import Logs
from fixtures.querier import (
Aggregation,
BuilderQuery,
OrderBy,
TelemetryFieldKey,
get_rows,
make_query_request,
)
def _log(
*,
timestamp: datetime,
service_name: str,
body: str,
extra_resources: dict | None = None,
) -> Logs:
resources = {"service.name": service_name}
if extra_resources:
resources.update(extra_resources)
return Logs(
timestamp=timestamp,
resources=resources,
attributes={},
body=body,
severity_text="INFO",
)
def test_skip_resource_fingerprint_logs_use_cte_path(
signoz: types.SigNoz,
create_user_admin: None, # pylint: disable=unused-argument
get_token: Callable[[str, str], str],
insert_logs: Callable[[list[Logs]], None],
) -> None:
"""
A filter matching a single resource fingerprint (count = 1 < 2) keeps the
legacy CTE attached. The result must only include rows for that resource.
"""
now = datetime.now(tz=UTC)
insert_logs(
[
_log(timestamp=now - timedelta(seconds=10), service_name="skip-logs-cte-svc", body="log-1"),
_log(timestamp=now - timedelta(seconds=8), service_name="skip-logs-cte-svc", body="log-2"),
# Noise from a different resource — must not appear.
_log(timestamp=now - timedelta(seconds=6), service_name="skip-logs-cte-noise", body="noise"),
]
)
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
response = make_query_request(
signoz,
token,
start_ms=int((datetime.now(tz=UTC) - timedelta(minutes=5)).timestamp() * 1000),
end_ms=int(datetime.now(tz=UTC).timestamp() * 1000),
request_type="raw",
queries=[
BuilderQuery(
signal="logs",
limit=50,
order=[OrderBy(TelemetryFieldKey("timestamp"), "asc")],
filter_expression="service.name = 'skip-logs-cte-svc'",
select_fields=[
TelemetryFieldKey("service.name", "string", "resource"),
TelemetryFieldKey("body"),
],
aggregations=[Aggregation("count()")],
).to_dict()
],
)
assert response.status_code == HTTPStatus.OK
rows = get_rows(response)
assert len(rows) == 2, f"expected 2 'skip-logs-cte-svc' rows, got {len(rows)}"
bodies = [row["data"]["body"] for row in rows]
assert bodies == ["log-1", "log-2"]
services = {row["data"]["service.name"] for row in rows}
assert services == {"skip-logs-cte-svc"}
def test_skip_resource_fingerprint_logs_fallback_path(
signoz: types.SigNoz,
create_user_admin: None, # pylint: disable=unused-argument
get_token: Callable[[str, str], str],
insert_logs: Callable[[list[Logs]], None],
) -> None:
"""
A filter matching multiple resource fingerprints (count >= 2) drives the
fallback path: no CTE, resource conditions evaluated on the main logs
table. Result must still be correct (no over- or under-matching).
"""
now = datetime.now(tz=UTC)
fallback_env = {"deployment.environment": "skip-logs-fallback"}
insert_logs(
[
_log(timestamp=now - timedelta(seconds=10), service_name="skip-logs-fb-svc-a", body="a", extra_resources=fallback_env),
_log(timestamp=now - timedelta(seconds=9), service_name="skip-logs-fb-svc-b", body="b", extra_resources=fallback_env),
_log(timestamp=now - timedelta(seconds=8), service_name="skip-logs-fb-svc-c", body="c", extra_resources=fallback_env),
# Noise without the fallback env — must be filtered out.
_log(
timestamp=now - timedelta(seconds=7),
service_name="skip-logs-fb-other",
body="noise",
extra_resources={"deployment.environment": "skip-logs-other"},
),
]
)
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
response = make_query_request(
signoz,
token,
start_ms=int((datetime.now(tz=UTC) - timedelta(minutes=5)).timestamp() * 1000),
end_ms=int(datetime.now(tz=UTC).timestamp() * 1000),
request_type="raw",
queries=[
BuilderQuery(
signal="logs",
limit=50,
order=[OrderBy(TelemetryFieldKey("timestamp"), "asc")],
filter_expression="deployment.environment = 'skip-logs-fallback'",
select_fields=[
TelemetryFieldKey("service.name", "string", "resource"),
TelemetryFieldKey("body"),
],
aggregations=[Aggregation("count()")],
).to_dict()
],
)
assert response.status_code == HTTPStatus.OK
rows = get_rows(response)
assert len(rows) == 3, f"expected 3 fallback rows, got {len(rows)}"
services = sorted(row["data"]["service.name"] for row in rows)
assert services == ["skip-logs-fb-svc-a", "skip-logs-fb-svc-b", "skip-logs-fb-svc-c"]
bodies = sorted(row["data"]["body"] for row in rows)
assert bodies == ["a", "b", "c"]

View File

@@ -0,0 +1,37 @@
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_skip_resource_fingerprint(
network: Network,
migrator: types.Operation, # pylint: disable=unused-argument
zeus: types.TestContainerDocker,
gateway: types.TestContainerDocker,
sqlstore: types.TestContainerSQL,
clickhouse: types.TestContainerClickhouse,
request: pytest.FixtureRequest,
pytestconfig: pytest.Config,
) -> types.SigNoz:
"""
Package-scoped SigNoz instance with the skip_resource_fingerprint
optimization enabled and a low threshold so both the materialized and
fallback resolver paths are exercised by sibling tests.
"""
return create_signoz(
network=network,
zeus=zeus,
gateway=gateway,
sqlstore=sqlstore,
clickhouse=clickhouse,
request=request,
pytestconfig=pytestconfig,
cache_key="signoz-skip-resource-fingerprint",
env_overrides={
"SIGNOZ_QUERIER_SKIP__RESOURCE__FINGERPRINT_ENABLED": True,
"SIGNOZ_QUERIER_SKIP__RESOURCE__FINGERPRINT_THRESHOLD": 2,
},
)