Compare commits

...

9 Commits

Author SHA1 Message Date
nityanandagohain
d90f025875 fix: support for related values in ai field values 2026-08-28 13:21:40 +05:30
Vikrant Gupta
1b8155fea9 chore(featureflag): make features endpoint open access (#12704)
#### Description

- Changes `GET /api/v1/features` from `ViewAccess` to `OpenAccess` in
both editions so every authenticated user, including those on custom
roles, can read feature flags.
- Feature flags describe the org's plan, not the caller's privileges,
and the frontend needs them to boot. With #12700 making the active
license readable by every authenticated user, the flags must be readable
too — otherwise custom-role users load the license but hang on the flags
fetch.
- Applies the same change to the flagger endpoint `GET /api/v2/features`
so the v2 client behaves identically when the frontend migrates to it.

#### Issues closed by this PR

Closes: https://github.com/SigNoz/platform-pod/issues/2653
2026-08-27 09:58:46 +00:00
Nikhil Soni
dac406eb93 feat: add support for quering scope fields in traces (#10920)
### 📄 Summary


Add support for instrumentation scope for traces. PR on collector -
https://github.com/SigNoz/signoz-otel-collector/pull/811


#### Screenshots / Screen Recordings (if applicable)

<img width="468" height="298" alt="image"
src="https://github.com/user-attachments/assets/03335ca3-a1c5-428b-9bfa-cd1796f735df"
/>

#### Issues closed by this PR

https://github.com/SigNoz/signoz/issues/5319
2026-08-27 09:55:43 +00:00
Gaurav Tewari
f69bce998a feat: add useBarChart instead of graph (#12517)
#### Description

- The logs frequency chart now draws with the uPlotV2 `BarChart` instead
of the Chart.js `Graph`, putting it on the same chart stack as the rest
of the product. Covers both Logs Explorer and Live Logs, which share the
component.
- Chart setup moves into a `useLogsExplorerChartConfig` hook built on
the shared `buildBaseConfig`. Severity colours, labels, drag-to-zoom and
timezone handling all behave as before; stacking goes through
`stack`/`StackMode`.
- Timestamps now convert ns → s, since uPlot's x scale is in seconds
where Chart.js wanted ms.
- Removes the `.ant-card-body` rules from three stylesheets. They
stopped matching anything when #8904 dropped the antd Card wrapper.
- Unblocked by #12627, which removed the last-minute trim from the time
scale. My earlier attempt at that (#12528) is closed.

#### Issues closed by this PR

Closes -
https://github.com/orgs/SigNoz/projects/39/views/20?pane=issue&itemId=128310734&issue=SigNoz%7Csignoz%7C9059

#### Screenshots / Screen Recordings



https://github.com/user-attachments/assets/b27f75ae-085d-410b-a5b8-491bac590fd0


#### Additional Information

---------

Co-authored-by: Gaurav Tewari <tewarig@users.noreply.github.com>
2026-08-27 07:09:12 +00:00
Srikanth Chekuri
ae9f58ba44 test: add integration tests for semconv family resolution (#12444)
Some checks failed
build-staging / prepare (push) Has been cancelled
build-staging / js-build (push) Has been cancelled
build-staging / go-build (push) Has been cancelled
build-staging / staging (push) Has been cancelled
cacheci / tests (push) Has been cancelled
Release Drafter / update_release_draft (push) Has been cancelled
#### Description

- Adds integration tests that pin the semconv family matrix on a live
stack, in a dedicated package that runs SigNoz with
`resolve_semconv_families` on, plus a second instance with the flag at
its default. Part of #6143.
- The fleet has one identity per state: OLD (old spelling only), NEW
(current only), BOTH (current `staging` and old `production` — the
conflict row), NEITHER (keyless).
- 36 filter cells (nine operators × both spellings × both contexts): the
result sets do not depend on the requested spelling, the current
spelling wins on the conflict row, and negative operators keep keyless
rows exactly like a single key.
- Singles: group-by merges the fleet and echoes the requested spelling
in the group column; a bare name with the family under two contexts
warns and keeps the resource side; logs stay literal with the flag on;
everything stays literal with the flag off.
- The suite is registered in the `integrationci` matrix. Verified: 42
passed against the live stack, on postgres and on sqlite.

#### Additional Information

- Stack: #12441 (merged) → #12442 (merged) → #12443 (merged) →
**#12444**. Rebased on main after the #12443 merge.
- The earlier `13_semconv_evolution.py` in this branch is replaced: its
`!=` expectations omitted keyless rows, which contradicts the contract
pinned by `queriercommon/06_keyless_semantics.py`.
2026-08-27 02:40:49 +00:00
Nityananda Gohain
a930da0901 feat: support for ts/scalar for llm spans (#12121)
Some checks failed
build-staging / prepare (push) Has been cancelled
build-staging / js-build (push) Has been cancelled
build-staging / go-build (push) Has been cancelled
build-staging / staging (push) Has been cancelled
cacheci / tests (push) Has been cancelled
Release Drafter / update_release_draft (push) Has been cancelled
## Pull Request

---

### 📄 Summary

follow-up for #12027. Span-list trace-aggregate filtering ships in
#12122.

Adds `scalar` and `time_series` request types to `builder_ai_query`. The
`trace.` prefix selects the aggregation domain: trace aggregates use a
native CTE pipeline, while span aggregates delegate to the standard
traces builder with the qualification gate applied.

Trace-level filters qualify entire traces across both domains using the
standard filter pipeline. Grouping, `HAVING`, ordering, and limits match
the traces builder, including whole-window ranking for grouped time
series and top-N limits for scalar queries.

`count(trace.trace_id)` counts every AI trace, matching the trace list;
token aggregates average over traces that have token data (standard
`NULL` semantics, same as span-attribute aggregations elsewhere).

Includes SQL golden tests, rewrite unit tests, and integration coverage
for both domains, qualification, grouping, limits, bucketing, variables,
and targeted `400` errors.


#### Issues closed by this PR
Fixes https://github.com/SigNoz/engineering-pod/issues/5602
Fixes https://github.com/SigNoz/engineering-pod/issues/5603

---

###  Change Type
_Select all that apply_

- [x]  Feature
- [ ] 🐛 Bug fix
- [ ] ♻️ Refactor
- [ ] 🛠️ Infra / Tooling
- [ ] 🧪 Test-only

---

### 🧪 Testing Strategy
> How was this change validated?

- Tests added/updated:  
- Manual verification:
- Edge cases covered:

---

### ⚠️ Risk & Impact Assessment
> What could break? How do we recover?

- Blast radius: None
- Potential regressions:
- Rollback plan:



---

### 📋 Checklist
- [x] Tests added or explicitly not required
- [x] Manually tested
- [ ] Breaking changes documented
- [ ] Backward compatibility considered

---

## 👀 Notes for Reviewers

Still in testing phase
---
2026-08-26 13:55:57 +00:00
Gaurav Tewari
5314cb7828 feat(ai-observability): AI o11y explorer (#12682)
<!--A few plain bullets saying what changed and why, for a reviewer
skimming it - not a wall of text, not a restatement of the diff, not
generated boilerplate.-->
#### Description
- forks the Traces Explorer into the AI Observability Explorer tab,
replacing the "Explorer coming soon" placeholder. All four views land:
list, trace, timeseries and table.
- the copied code is kept identical to the traces original on purpose —
same variable names, same `LOCALSTORAGE` keys, same analytics events,
same `DataSource.TRACES`. Only the folder layout differs. Divergence
(GenAI columns, AI query surface) comes in follow-ups, so this stays a
clean base to diff against.
- shared modules are imported, not duplicated:
`TracesExplorer/TracesTable`, `TracesExplorer/Controls`,
`TracesExplorer/explorerUtils`, `TracesExplorer/ListView/utils` and
`pages/TracesExplorer/aiActions`.
- both table views therefore use the shared TanStack table, so the AI
explorer starts out with resizable/reorderable columns rather than the
old antd `ResizeTable`.

<!--Reference issues using `Closes #issue-number` to enable automatic
closure on merge. -->
#### Issues closed by this PR

Closes -
https://github.com/orgs/SigNoz/projects/39/views/20?pane=issue&itemId=223107878&issue=SigNoz%7Cengineering-pod%7C5843

<!--If applicable, include screenshots or screen recordings that clearly
show the behavior before the change and the result after the change. -->
#### Screenshots / Screen Recordings


https://github.com/user-attachments/assets/6f3bb336-f555-4345-aeca-ba061401f5af





<!--Anything reviewers should keep in mind while reviewing -->
#### Additional Information
- **Stacked on #12672.** The first three commits are that PR
cherry-picked, since the trace view fork depends on its `FieldCell`
trace_id handling and the optional `columnStorageKey` /
`respectColumnOrder` props. Review only the last commit here; rebase
drops the rest once #12672 lands.
- `LLMObservability.test.tsx` now stubs `Explorer` the same way it
already stubs `DashboardContainer` — the real toolbar calls
`useNavigationType`, which needs a data router that integration test
does not mount.

Co-authored-by: Gaurav Tewari <tewarig@users.noreply.github.com>
2026-08-26 13:45:26 +00:00
Nityananda Gohain
02c5555a48 fix: add ai_observability to saved views (#12675)
Some checks failed
build-staging / prepare (push) Has been cancelled
build-staging / js-build (push) Has been cancelled
build-staging / go-build (push) Has been cancelled
build-staging / staging (push) Has been cancelled
cacheci / tests (push) Has been cancelled
Release Drafter / update_release_draft (push) Has been cancelled
#### Description
* adds `ai_observability` to saved view for ai explorer

<!--Reference issues using `Closes #issue-number` to enable automatic
closure on merge. -->
#### Issues closed by this PR
Closes https://github.com/SigNoz/engineering-pod/issues/5955
2026-08-26 11:28:54 +00:00
Naman Verma
0bdc7bf6a1 fix: resolve aggregate column for exp histograms before samples table (#12640)
<!--A few plain bullets saying what changed and why, for a reviewer
skimming it - not a wall of text, not a restatement of the diff, not
generated boilerplate.-->
#### Description

Column for exponential histograms is not decided by samples tables so it
should not run for exp histogrms

<!--Reference issues using `Closes #issue-number` to enable automatic
closure on merge. -->
#### Issues closed by this PR

Closes https://github.com/SigNoz/pulse-pod/issues/295
2026-08-26 10:15:09 +00:00
78 changed files with 6168 additions and 417 deletions

View File

@@ -62,6 +62,7 @@ jobs:
- role
- rootuser
- savedview
- semconvfamilies
- serviceaccount
- spanmapper
- querier_json_body

View File

@@ -8010,6 +8010,7 @@ components:
- logs
- metrics
- meter
- ai_observability
type: string
SavedviewtypesUpdatableSavedView:
properties:
@@ -8800,6 +8801,7 @@ components:
- span
- trace
- resource
- scope
- attribute
- body
- ""
@@ -9220,6 +9222,10 @@ paths:
name: name
schema:
type: string
- in: query
name: existingQuery
schema:
type: string
responses:
"200":
content:
@@ -15467,10 +15473,8 @@ paths:
$ref: '#/components/schemas/RenderErrorResponse'
description: Internal Server Error
security:
- api_key:
- VIEWER
- tokenizer:
- VIEWER
- api_key: []
- tokenizer: []
summary: Get features
tags:
- features

View File

@@ -67,7 +67,7 @@ func (ah *APIHandler) RegisterRoutes(router *mux.Router, am *middleware.AuthZ) {
// note: add ee override methods first
// routes available only in ee version
router.HandleFunc("/api/v1/features", am.ViewAccess(ah.getFeatureFlags)).Methods(http.MethodGet)
router.HandleFunc("/api/v1/features", am.OpenAccess(ah.getFeatureFlags)).Methods(http.MethodGet)
// base overrides
router.HandleFunc("/api/v1/version", am.OpenAccess(ah.getVersion)).Methods(http.MethodGet)

View File

@@ -3492,6 +3492,7 @@ export enum TelemetrytypesFieldContextDTO {
span = 'span',
trace = 'trace',
resource = 'resource',
scope = 'scope',
attribute = 'attribute',
body = 'body',
'' = '',
@@ -9021,6 +9022,7 @@ export enum SavedviewtypesSourceDTO {
logs = 'logs',
metrics = 'metrics',
meter = 'meter',
ai_observability = 'ai_observability',
}
export interface SavedviewtypesSavedViewSpecDTO {
display?: SavedviewtypesDisplayDTO;
@@ -10246,6 +10248,11 @@ export type GetAIObservabilityFieldsValuesParams = {
* @description undefined
*/
name?: string;
/**
* @type string
* @description undefined
*/
existingQuery?: string;
};
export type GetAIObservabilityFieldsValues200 = {

View File

@@ -10,6 +10,7 @@ const fieldContextToSuggestionMap: Record<
[TelemetrytypesFieldContextDTO.attribute]: 'attribute',
// no maps for the following values on suggestion context
[TelemetrytypesFieldContextDTO.trace]: undefined,
[TelemetrytypesFieldContextDTO.scope]: undefined,
[TelemetrytypesFieldContextDTO.body]: undefined,
[TelemetrytypesFieldContextDTO.metric]: undefined,
[TelemetrytypesFieldContextDTO.log]: undefined,

View File

@@ -1,11 +0,0 @@
.explorer {
display: flex;
flex-direction: column;
gap: var(--spacing-4);
padding: var(--spacing-2) var(--spacing-0);
}
.placeholder {
color: var(--l2-foreground);
font-size: var(--periscope-font-size-base);
}

View File

@@ -0,0 +1,106 @@
.trace-explorer-header {
.trace-explorer-run-query {
display: flex;
flex-direction: row-reverse;
align-items: center;
margin: 8px 16px;
gap: 8px;
}
.filter-outlined-btn {
border-radius: 0px 2px 2px 0px;
border-top: 1px solid var(--l1-border);
border-right: 1px solid var(--l1-border);
border-bottom: 1px solid var(--l1-border);
background: var(--l2-background);
box-shadow: 0px 0px 8px 0px rgba(0, 0, 0, 0.1);
}
}
.trace-explorer-header.single-child {
justify-content: flex-end;
}
.traces-explorer-views {
padding: 8px;
padding-bottom: 60px;
margin-bottom: 24px;
.ant-tabs-tabpane {
padding: 0 8px;
}
}
.qb-search-view-container {
padding: 8px;
display: flex;
flex-direction: row;
align-items: center;
gap: 8px;
.ant-select-selector {
border-radius: 2px;
border: 1px solid var(--l1-border) !important;
background: var(--l2-background) !important;
height: 34px !important;
box-sizing: border-box !important;
}
}
.trace-explorer-list-view {
flex: 1;
}
.trace-explorer-traces-view {
flex: 1;
}
.trace-explorer-table-view {
flex: 1;
}
.trace-explorer-time-series-view {
flex: 1;
}
.trace-explorer-page {
display: flex;
// Meant to fix the query builder colors
--input-background: var(--l2-background);
--input-hover-background: var(--l2-background);
--input-focus-background: var(--l2-background);
--input-border-color: var(--l2-border);
--input-hover-border-color: var(--internal-ant-border-color-hover);
--input-focus-border-color: var(--internal-ant-border-color-hover);
.filter {
width: 260px;
height: 100%;
min-height: 100vh;
border-right: 0px;
border: 1px solid var(--l1-border);
background-color: var(--l1-background);
> .ant-card-body {
padding: 0;
width: 258px;
}
}
.trace-explorer {
width: 100%;
background: var(--l1-background);
> .ant-card-body {
padding: 0;
}
border-color: var(--l1-border);
}
.trace-explorer.filters-expanded {
width: calc(100% - 260px);
}
}

View File

@@ -1,13 +1,373 @@
import styles from './Explorer.module.scss';
import { useCallback, useEffect, useMemo, useRef, useState } from 'react';
import { useQueryClient } from 'react-query';
import { useSearchParams } from 'react-router-dom-v5-compat';
import * as Sentry from '@sentry/react';
import { Card } from 'antd';
import logEvent from 'api/common/logEvent';
import cx from 'classnames';
import ExplorerCard from 'components/ExplorerCard/ExplorerCard';
import QueryCancelledPlaceholder from 'components/QueryCancelledPlaceholder';
import QuickFilters from 'components/QuickFilters/QuickFilters';
import { QuickFiltersSource, SignalType } from 'components/QuickFilters/types';
import WarningPopover from 'components/WarningPopover/WarningPopover';
import { AVAILABLE_EXPORT_PANEL_TYPES } from 'constants/panelTypes';
import { initialQueriesMap, PANEL_TYPES } from 'constants/queryBuilder';
import { usePageActions } from 'container/AIAssistant/pageActions/usePageActions';
import ExplorerOptionWrapper from 'container/ExplorerOptions/ExplorerOptionWrapper';
import { useOptionsMenu } from 'container/OptionsMenu';
import LeftToolbarActions from 'container/QueryBuilder/components/ToolbarActions/LeftToolbarActions';
import RightToolbarActions from 'container/QueryBuilder/components/ToolbarActions/RightToolbarActions';
import Toolbar from 'container/Toolbar/Toolbar';
import {
getExportQueryData,
getQueryByPanelType,
} from 'container/TracesExplorer/explorerUtils';
import { ExportDashboard } from 'hooks/dashboard/useExportDashboards';
import { useGetExportToDashboardLink } from 'hooks/dashboard/useGetExportToDashboardLink';
import { useGetPanelTypesQueryParam } from 'hooks/queryBuilder/useGetPanelTypesQueryParam';
import { useQueryBuilder } from 'hooks/queryBuilder/useQueryBuilder';
import { useShareBuilderUrl } from 'hooks/queryBuilder/useShareBuilderUrl';
import {
ICurrentQueryData,
useHandleExplorerTabChange,
} from 'hooks/useHandleExplorerTabChange';
import { useIsAIAssistantEnabled } from 'hooks/useIsAIAssistantEnabled';
import { useSafeNavigate } from 'hooks/useSafeNavigate';
import { isEmpty } from 'lodash-es';
import ErrorBoundaryFallback from 'pages/ErrorBoundaryFallback/ErrorBoundaryFallback';
import { ExplorerViews } from 'pages/LogsExplorer/utils';
import {
tracesAddFilterAction,
tracesChangeViewAction,
tracesRunQueryAction,
tracesSaveViewAction,
} from 'pages/TracesExplorer/aiActions';
import { Warning } from 'types/api';
import { Query } from 'types/api/queryBuilder/queryBuilderData';
import { DataSource } from 'types/common/queryBuilder';
import {
explorerViewToPanelType,
getExplorerViewFromUrl,
} from 'utils/explorerUtils';
import { v4 } from 'uuid';
import { TOOLBAR_VIEWS } from './constants';
import ListView from './ListView/ListView';
import { defaultSelectedColumns } from './ListView/configs';
import QuerySection from './QuerySection/QuerySection';
import TableView from './TableView/TableView';
import TimeSeriesView from './TimeSeriesView/TimeSeriesView';
import TracesView from './TracesView/TracesView';
import './Explorer.styles.scss';
// Shell for the AI Observability Explorer tab. Owns the
// /ai-observability/explorer route and is intentionally empty for now: the
// query builder + results surface land in a follow-up.
function Explorer(): JSX.Element {
const {
panelType,
updateAllQueriesOperators,
handleRunQuery,
stagedQuery,
handleSetConfig,
currentQuery,
handleSetQueryData,
redirectWithQueryBuilderData,
} = useQueryBuilder();
const isAIAssistantEnabled = useIsAIAssistantEnabled();
const { options } = useOptionsMenu({
dataSource: DataSource.TRACES,
aggregateOperator: 'noop',
initialOptions: {
selectColumns: defaultSelectedColumns,
},
});
const [searchParams] = useSearchParams();
const queryClient = useQueryClient();
const listQueryKeyRef = useRef<any>();
// Get panel type from URL
const panelTypesFromUrl = useGetPanelTypesQueryParam(PANEL_TYPES.LIST);
const [isLoadingQueries, setIsLoadingQueries] = useState<boolean>(false);
const [isCancelled, setIsCancelled] = useState(false);
useEffect(() => {
if (isLoadingQueries) {
setIsCancelled(false);
}
}, [isLoadingQueries]);
const handleCancelQuery = useCallback(() => {
if (listQueryKeyRef.current) {
queryClient.cancelQueries(listQueryKeyRef.current);
}
setIsCancelled(true);
// Reset loading state — the active view unmounts when cancelled, so no
// child will call setIsLoadingQueries(false) otherwise.
setIsLoadingQueries(false);
}, [queryClient]);
const [selectedView, setSelectedView] = useState<ExplorerViews>(() =>
getExplorerViewFromUrl(searchParams, panelTypesFromUrl),
);
const [warning, setWarning] = useState<Warning | undefined>();
const [isOpen, setOpen] = useState<boolean>(true);
const defaultQuery = useMemo(
(): Query =>
updateAllQueriesOperators(
initialQueriesMap.traces,
PANEL_TYPES.LIST,
DataSource.TRACES,
),
[updateAllQueriesOperators],
);
const { handleExplorerTabChange } = useHandleExplorerTabChange();
const { safeNavigate } = useSafeNavigate();
const getExportToDashboardLink = useGetExportToDashboardLink();
const handleChangeSelectedView = useCallback(
(view: ExplorerViews, querySearchParameters?: ICurrentQueryData): void => {
handleSetConfig(explorerViewToPanelType[view], DataSource.TRACES);
setSelectedView(view);
handleExplorerTabChange(
explorerViewToPanelType[view],
querySearchParameters,
);
},
[handleExplorerTabChange, handleSetConfig],
);
// ─── AI Assistant page actions (only when license feature is on) ───────────
const aiActions = useMemo(
() =>
isAIAssistantEnabled
? [
tracesRunQueryAction({
currentQuery,
handleSetQueryData,
redirectWithQueryBuilderData,
}),
tracesAddFilterAction({
currentQuery,
handleSetQueryData,
redirectWithQueryBuilderData,
}),
tracesChangeViewAction({
onChangeView: (view) => handleChangeSelectedView(view as ExplorerViews),
}),
tracesSaveViewAction({
// POC stub — logs a save request; wire to real API when available
onSaveView: async (name) => {
// eslint-disable-next-line no-console
console.info('[AI Assistant] Save view requested:', name);
},
}),
]
: [],
// eslint-disable-next-line react-hooks/exhaustive-deps
[
isAIAssistantEnabled,
currentQuery,
handleSetQueryData,
redirectWithQueryBuilderData,
handleChangeSelectedView,
],
);
usePageActions('traces-explorer', aiActions);
// ───────────────────────────────────────────────────────────────────────────
const exportDefaultQuery = useMemo(
() =>
getQueryByPanelType(
stagedQuery || initialQueriesMap.traces,
panelType || PANEL_TYPES.LIST,
),
[stagedQuery, panelType],
);
const handleExport = useCallback(
(dashboard: ExportDashboard | null, isNewDashboard?: boolean): void => {
if (!dashboard || !panelType) {
return;
}
const panelTypeParam = AVAILABLE_EXPORT_PANEL_TYPES.includes(panelType)
? panelType
: PANEL_TYPES.TIME_SERIES;
const widgetId = v4();
const query = getExportQueryData(
exportDefaultQuery,
panelTypeParam,
options,
);
logEvent('Traces Explorer: Add to dashboard successful', {
panelType,
isNewDashboard,
dashboardName: dashboard?.title,
});
const dashboardEditView = getExportToDashboardLink({
query,
panelType: panelTypeParam,
dashboardId: dashboard.id,
widgetId,
});
if (dashboardEditView) {
safeNavigate(dashboardEditView);
}
},
[
exportDefaultQuery,
panelType,
safeNavigate,
options,
getExportToDashboardLink,
],
);
useShareBuilderUrl({ defaultValue: defaultQuery });
const logEventCalledRef = useRef(false);
useEffect(() => {
if (!logEventCalledRef.current) {
logEvent('Traces Explorer: Page visited', {});
logEventCalledRef.current = true;
}
}, []);
const isFilterApplied = useMemo(() => {
// if any of the non-disabled queries has filters applied, return true
const result = stagedQuery?.builder?.queryData?.filter(
(item) => !isEmpty(item.filters?.items) && !item.disabled,
);
return !!result?.length;
}, [stagedQuery]);
return (
<div className={styles.explorer} data-testid="llm-observability-explorer">
<div className={styles.placeholder}>Explorer coming soon.</div>
</div>
<Sentry.ErrorBoundary fallback={<ErrorBoundaryFallback />}>
<div
className="trace-explorer-page"
data-testid="llm-observability-explorer"
>
<Card className="filter" hidden={!isOpen}>
<QuickFilters
className="qf-traces-explorer"
source={QuickFiltersSource.TRACES_EXPLORER}
signal={SignalType.TRACES}
handleFilterVisibilityChange={(): void => {
setOpen(!isOpen);
}}
/>
</Card>
<div
className={cx('trace-explorer', {
'filters-expanded': isOpen,
})}
>
<div className="trace-explorer-header">
<Toolbar
showAutoRefresh
leftActions={
<LeftToolbarActions
showFilter={isOpen}
handleFilterVisibilityChange={(): void => setOpen(!isOpen)}
items={TOOLBAR_VIEWS}
selectedView={selectedView}
onChangeSelectedView={handleChangeSelectedView}
/>
}
warningElement={
!isEmpty(warning) ? <WarningPopover warningData={warning} /> : <div />
}
rightActions={
<RightToolbarActions
onStageRunQuery={(): void => {
setIsCancelled(false);
handleRunQuery();
}}
isLoadingQueries={isLoadingQueries}
handleCancelQuery={handleCancelQuery}
/>
}
/>
</div>
<ExplorerCard sourcepage={DataSource.TRACES}>
<div className="query-section-container">
<QuerySection />
</div>
</ExplorerCard>
<div className="traces-explorer-views">
{isCancelled && (
<QueryCancelledPlaceholder subText='Click "Run Query" to load traces.' />
)}
{!isCancelled && selectedView === ExplorerViews.LIST && (
<div className="trace-explorer-list-view">
<ListView
isFilterApplied={isFilterApplied}
setWarning={setWarning}
setIsLoadingQueries={setIsLoadingQueries}
queryKeyRef={listQueryKeyRef}
/>
</div>
)}
{!isCancelled && selectedView === ExplorerViews.TRACE && (
<div className="trace-explorer-traces-view">
<TracesView
isFilterApplied={isFilterApplied}
setWarning={setWarning}
setIsLoadingQueries={setIsLoadingQueries}
queryKeyRef={listQueryKeyRef}
/>
</div>
)}
{!isCancelled && selectedView === ExplorerViews.TIMESERIES && (
<div className="trace-explorer-time-series-view">
<TimeSeriesView
dataSource={DataSource.TRACES}
isFilterApplied={isFilterApplied}
setWarning={setWarning}
setIsLoadingQueries={setIsLoadingQueries}
queryKeyRef={listQueryKeyRef}
/>
</div>
)}
{!isCancelled && selectedView === ExplorerViews.TABLE && (
<div className="trace-explorer-table-view">
<TableView
setWarning={setWarning}
setIsLoadingQueries={setIsLoadingQueries}
queryKeyRef={listQueryKeyRef}
/>
</div>
)}
</div>
<ExplorerOptionWrapper
disabled={!stagedQuery}
query={exportDefaultQuery}
sourcepage={DataSource.TRACES}
onExport={handleExport}
handleChangeSelectedView={handleChangeSelectedView}
/>
</div>
</div>
</Sentry.ErrorBoundary>
);
}

View File

@@ -0,0 +1,8 @@
.container {
display: flex;
flex-direction: column;
flex: 1;
min-height: 0;
height: calc(100vh - 240px);
min-height: 400px;
}

View File

@@ -0,0 +1,34 @@
.trace-explorer-controls {
display: flex;
justify-content: flex-end;
align-items: center;
gap: 8px;
.order-by-container {
display: flex;
align-items: center;
gap: 8px;
.order-by-label {
color: var(--muted-foreground);
font-size: 12px;
font-style: normal;
font-weight: 400;
line-height: 16px; /* 133.333% */
display: flex;
align-items: center;
gap: 4px;
}
.order-by-select {
width: 100px;
.ant-select-selector {
border: none;
box-shadow: none;
background-color: transparent;
}
}
}
}

View File

@@ -0,0 +1,272 @@
import {
Dispatch,
memo,
MutableRefObject,
SetStateAction,
useCallback,
useEffect,
useMemo,
useState,
} from 'react';
import { QueryKey } from 'react-query';
// eslint-disable-next-line no-restricted-imports
import { useSelector } from 'react-redux';
import logEvent from 'api/common/logEvent';
import DownloadOptionsMenu from 'components/DownloadOptionsMenu/DownloadOptionsMenu';
import ListViewOrderBy from 'components/OrderBy/ListViewOrderBy';
import type { TableColumnDef } from 'components/TanStackTableView/types';
import { ENTITY_VERSION_V5 } from 'constants/app';
import { QueryParams } from 'constants/query';
import { initialQueriesMap, PANEL_TYPES } from 'constants/queryBuilder';
import { REACT_QUERY_KEY } from 'constants/reactQueryKeys';
import { useOptionsMenu } from 'container/OptionsMenu';
import { CustomTimeType } from 'container/TopNav/DateTimeSelectionV2/types';
import TraceExplorerControls from 'container/TracesExplorer/Controls';
import { getListViewQuery } from 'container/TracesExplorer/explorerUtils';
import {
getTraceLink,
transformSpanRows,
} from 'container/TracesExplorer/ListView/utils';
import {
getFieldColumn,
TracesTableRow,
} from 'container/TracesExplorer/TracesTable/getFieldColumn';
import TracesTable from 'container/TracesExplorer/TracesTable/TracesTable';
import { useGetQueryRange } from 'hooks/queryBuilder/useGetQueryRange';
import { useQueryBuilder } from 'hooks/queryBuilder/useQueryBuilder';
import { Pagination } from 'hooks/queryPagination';
import { getDefaultPaginationConfig } from 'hooks/queryPagination/utils';
import useUrlQueryData from 'hooks/useUrlQueryData';
import { ArrowUp10, Minus } from '@signozhq/icons';
import { AppState } from 'store/reducers';
import { Warning } from 'types/api';
import { DataSource } from 'types/common/queryBuilder';
import { GlobalReducer } from 'types/reducer/globalTime';
import {
defaultSelectedColumns,
PER_PAGE_OPTIONS,
TIMESTAMP_FIELD,
} from './configs';
import './ListView.styles.scss';
import styles from './ListView.module.scss';
interface ListViewProps {
isFilterApplied: boolean;
setWarning: Dispatch<SetStateAction<Warning | undefined>>;
setIsLoadingQueries: Dispatch<SetStateAction<boolean>>;
queryKeyRef?: MutableRefObject<QueryKey | undefined>;
}
function ListView({
isFilterApplied,
setWarning,
setIsLoadingQueries,
queryKeyRef,
}: ListViewProps): JSX.Element {
const { stagedQuery, panelType: panelTypeFromQueryBuilder } =
useQueryBuilder();
const panelType = panelTypeFromQueryBuilder || PANEL_TYPES.LIST;
const [orderBy, setOrderBy] = useState<string>('timestamp:desc');
const {
selectedTime: globalSelectedTime,
maxTime,
minTime,
loading: timeRangeUpdateLoading,
} = useSelector<AppState, GlobalReducer>((state) => state.globalTime);
const { options, config } = useOptionsMenu({
dataSource: DataSource.TRACES,
aggregateOperator: 'count',
initialOptions: {
selectColumns: defaultSelectedColumns,
},
});
const { queryData: paginationQueryData } = useUrlQueryData<Pagination>(
QueryParams.pagination,
);
const paginationConfig =
paginationQueryData ?? getDefaultPaginationConfig(PER_PAGE_OPTIONS);
const requestQuery = useMemo(
() => getListViewQuery(stagedQuery || initialQueriesMap.traces, orderBy),
[stagedQuery, orderBy],
);
// Stable sorted-name signature for the queryKey.
// - Drag updates selectColumns; raw queryKey would churn on reorder.
// - Trace API fetches only listed columns → add/remove must refetch.
// - Sorted-name signature: stable on reorder, changes on add/remove.
const selectColumnsSignature = useMemo(
() =>
(options?.selectColumns ?? [])
.map((c) => c.name)
.sort()
.join(','),
[options?.selectColumns],
);
const queryKey = useMemo(
() => [
REACT_QUERY_KEY.GET_QUERY_RANGE,
globalSelectedTime,
maxTime,
minTime,
stagedQuery,
panelType,
paginationConfig,
selectColumnsSignature,
orderBy,
],
[
stagedQuery,
panelType,
globalSelectedTime,
paginationConfig,
selectColumnsSignature,
maxTime,
minTime,
orderBy,
],
);
if (queryKeyRef) {
queryKeyRef.current = queryKey;
}
const { data, isFetching, isLoading, isError, error } = useGetQueryRange(
{
query: requestQuery,
graphType: panelType,
selectedTime: 'GLOBAL_TIME' as const,
globalSelectedInterval: globalSelectedTime as CustomTimeType,
params: {
dataSource: 'traces',
},
tableParams: {
pagination: paginationConfig,
selectColumns: options?.selectColumns,
},
},
ENTITY_VERSION_V5,
{
queryKey,
enabled:
// don't make api call while the time range state in redux is loading
!timeRangeUpdateLoading &&
!!stagedQuery &&
panelType === PANEL_TYPES.LIST &&
!!options?.selectColumns?.length,
},
);
useEffect(() => {
if (data?.payload) {
setWarning(data?.warning);
}
// eslint-disable-next-line react-hooks/exhaustive-deps
}, [data?.payload, data?.warning]);
useEffect(() => {
if (isLoading || isFetching) {
setIsLoadingQueries(true);
} else {
setIsLoadingQueries(false);
}
}, [isLoading, isFetching, setIsLoadingQueries]);
const queryTableDataResult = data?.payload?.data?.newResult?.data?.result;
const queryTableData = useMemo(
() => queryTableDataResult || [],
[queryTableDataResult],
);
const columns = useMemo<TableColumnDef<TracesTableRow>[]>(() => {
const fields = [
TIMESTAMP_FIELD,
...(options?.selectColumns ?? []).filter(
(field) => field.name !== TIMESTAMP_FIELD.name,
),
];
return fields.map((field) => getFieldColumn(field));
}, [options?.selectColumns]);
const rows = useMemo(
() => transformSpanRows(queryTableData),
[queryTableData],
);
const handleColumnOrderChange = useCallback(
(reordered: TableColumnDef<TracesTableRow>[]): void => {
config?.addColumn?.onReorder(reordered.map((column) => column.id));
},
[config],
);
const handleOrderChange = useCallback((value: string) => {
setOrderBy(value);
}, []);
useEffect(() => {
if (!isLoading && !isFetching && !isError && rows.length !== 0) {
void logEvent('Traces Explorer: Data present', {
panelType,
});
}
}, [isLoading, isFetching, isError, rows, panelType]);
return (
<div className={styles.container}>
<div className="trace-explorer-controls">
<div className="order-by-container">
<div className="order-by-label">
Order by <Minus size={14} /> <ArrowUp10 size={14} />
</div>
<ListViewOrderBy
value={orderBy}
onChange={handleOrderChange}
dataSource={DataSource.TRACES}
/>
</div>
<DownloadOptionsMenu
dataSource={DataSource.TRACES}
selectedColumns={options?.selectColumns}
/>
<TraceExplorerControls
isLoading={isFetching}
totalCount={rows.length}
config={config}
perPageOptions={PER_PAGE_OPTIONS}
/>
</div>
<TracesTable
data={rows}
columns={columns}
panelType="LIST"
getRowHref={getTraceLink}
isLoading={isLoading}
isFetching={isFetching}
isError={isError}
error={error}
isFilterApplied={isFilterApplied}
onColumnOrderChange={handleColumnOrderChange}
onColumnRemove={config?.addColumn?.onRemove}
/>
</div>
);
}
ListView.defaultProps = {
queryKeyRef: undefined,
};
export default memo(ListView);

View File

@@ -0,0 +1,19 @@
import type { TelemetryFieldKey } from 'api/v5/v5';
import { DEFAULT_PER_PAGE_OPTIONS } from 'hooks/queryPagination';
export const defaultSelectedColumns: string[] = [
'service.name',
'name',
'duration_nano',
'http_method',
'response_status_code',
'timestamp',
];
export const PER_PAGE_OPTIONS: number[] = [10, ...DEFAULT_PER_PAGE_OPTIONS];
// Pinned timestamp column
export const TIMESTAMP_FIELD = {
name: 'timestamp',
fieldContext: 'span',
} as TelemetryFieldKey;

View File

@@ -0,0 +1,61 @@
import { memo, useCallback, useMemo } from 'react';
import { QueryBuilderV2 } from 'components/QueryBuilderV2/QueryBuilderV2';
import { PANEL_TYPES } from 'constants/queryBuilder';
import ExplorerOrderBy from 'container/ExplorerOrderBy';
import { OrderByFilterProps } from 'container/QueryBuilder/filters/OrderByFilter/OrderByFilter.interfaces';
import { QueryBuilderProps } from 'container/QueryBuilder/QueryBuilder.interfaces';
import { useGetPanelTypesQueryParam } from 'hooks/queryBuilder/useGetPanelTypesQueryParam';
import { DataSource } from 'types/common/queryBuilder';
function QuerySection(): JSX.Element {
const panelTypes = useGetPanelTypesQueryParam(PANEL_TYPES.LIST);
const filterConfigs: QueryBuilderProps['filterConfigs'] = useMemo(() => {
const isList = panelTypes === PANEL_TYPES.LIST;
const config: QueryBuilderProps['filterConfigs'] = {
stepInterval: { isHidden: false, isDisabled: false },
limit: { isHidden: isList, isDisabled: true },
having: { isHidden: isList, isDisabled: true },
};
return config;
}, [panelTypes]);
const renderOrderBy = useCallback(
({ query, onChange }: OrderByFilterProps) => (
<ExplorerOrderBy query={query} onChange={onChange} />
),
[],
);
const queryComponents = useMemo((): QueryBuilderProps['queryComponents'] => {
const shouldRenderCustomOrderBy =
panelTypes === PANEL_TYPES.LIST || panelTypes === PANEL_TYPES.TRACE;
return {
...(shouldRenderCustomOrderBy ? { renderOrderBy } : {}),
};
}, [panelTypes, renderOrderBy]);
const isListViewPanel = useMemo(
() => panelTypes === PANEL_TYPES.LIST || panelTypes === PANEL_TYPES.TRACE,
[panelTypes],
);
return (
<QueryBuilderV2
isListViewPanel={isListViewPanel}
showTraceOperator
config={{ initialDataSource: DataSource.TRACES, queryVariant: 'static' }}
queryComponents={queryComponents}
panelType={panelTypes}
filterConfigs={filterConfigs}
showOnlyWhereClause={
panelTypes === PANEL_TYPES.LIST || panelTypes === PANEL_TYPES.TRACE
}
version="v3" // setting this to v3 as we this is rendered in logs explorer
/>
);
}
export default memo(QuerySection);

View File

@@ -0,0 +1,7 @@
.traces-table-view-header {
display: flex;
justify-content: flex-end;
align-items: center;
padding: 12px;
flex-shrink: 0;
}

View File

@@ -0,0 +1,130 @@
import {
Dispatch,
memo,
MutableRefObject,
SetStateAction,
useEffect,
useMemo,
} from 'react';
// eslint-disable-next-line no-restricted-imports
import { useSelector } from 'react-redux';
import { Space } from 'antd';
import ErrorInPlace from 'components/ErrorInPlace/ErrorInPlace';
import ExportMenu from 'components/ExportMenu/ExportMenu';
import { ENTITY_VERSION_V5 } from 'constants/app';
import { initialQueriesMap, PANEL_TYPES } from 'constants/queryBuilder';
import { REACT_QUERY_KEY } from 'constants/reactQueryKeys';
import { QueryTable } from 'container/QueryTable';
import { useGetQueryRange } from 'hooks/queryBuilder/useGetQueryRange';
import { useQueryBuilder } from 'hooks/queryBuilder/useQueryBuilder';
import { AppState } from 'store/reducers';
import { Warning } from 'types/api';
import APIError from 'types/api/error';
import { QueryDataV3 } from 'types/api/widgets/getQuery';
import { DataSource } from 'types/common/queryBuilder';
import { GlobalReducer } from 'types/reducer/globalTime';
import './TableView.styles.scss';
function TableView({
setWarning,
setIsLoadingQueries,
queryKeyRef,
}: {
setWarning: Dispatch<SetStateAction<Warning | undefined>>;
setIsLoadingQueries: Dispatch<SetStateAction<boolean>>;
queryKeyRef?: MutableRefObject<any>;
}): JSX.Element {
const { stagedQuery, panelType } = useQueryBuilder();
const {
selectedTime: globalSelectedTime,
maxTime,
minTime,
} = useSelector<AppState, GlobalReducer>((state) => state.globalTime);
const queryKey = useMemo(
() => [
REACT_QUERY_KEY.GET_QUERY_RANGE,
globalSelectedTime,
maxTime,
minTime,
stagedQuery,
],
[globalSelectedTime, maxTime, minTime, stagedQuery],
);
if (queryKeyRef) {
queryKeyRef.current = queryKey;
}
const { data, isLoading, isFetching, isError, error } = useGetQueryRange(
{
query: stagedQuery || initialQueriesMap.traces,
graphType: panelType || PANEL_TYPES.TABLE,
selectedTime: 'GLOBAL_TIME',
globalSelectedInterval: globalSelectedTime,
params: {
dataSource: 'traces',
},
},
ENTITY_VERSION_V5,
{
queryKey,
enabled: !!stagedQuery && panelType === PANEL_TYPES.TABLE,
},
);
useEffect(() => {
if (isLoading || isFetching) {
setIsLoadingQueries(true);
} else {
setIsLoadingQueries(false);
}
}, [isLoading, isFetching, setIsLoadingQueries]);
const queryTableData = useMemo(
() =>
data?.payload?.data?.newResult?.data?.result ||
data?.payload.data.result ||
[],
[data],
);
useEffect(() => {
if (data?.payload) {
setWarning(data.warning);
}
// eslint-disable-next-line react-hooks/exhaustive-deps
}, [data?.payload, data?.warning]);
return (
<Space.Compact block direction="vertical">
{isError && error && <ErrorInPlace error={error as APIError} />}
{!isError && data && (
<div className="traces-table-view-header">
<ExportMenu
dataSource={DataSource.TRACES}
data={data}
query={stagedQuery || initialQueriesMap.traces}
fileName="traces-table"
/>
</div>
)}
{!isError && (
<QueryTable
query={stagedQuery || initialQueriesMap.traces}
queryTableData={queryTableData as QueryDataV3[]}
loading={isLoading}
sticky
/>
)}
</Space.Compact>
);
}
TableView.defaultProps = {
queryKeyRef: undefined,
};
export default memo(TableView);

View File

@@ -0,0 +1,8 @@
.trace-explorer-time-series-view-container {
&-header {
display: flex;
justify-content: flex-start;
align-items: center;
padding: 12px;
}
}

View File

@@ -0,0 +1,147 @@
import {
Dispatch,
MutableRefObject,
SetStateAction,
useEffect,
useMemo,
} from 'react';
// eslint-disable-next-line no-restricted-imports
import { useSelector } from 'react-redux';
import { ENTITY_VERSION_V5 } from 'constants/app';
import { initialQueriesMap, PANEL_TYPES } from 'constants/queryBuilder';
import { REACT_QUERY_KEY } from 'constants/reactQueryKeys';
import TimeSeriesView from 'container/TimeSeriesView/TimeSeriesView';
import { convertDataValueToMs } from 'container/TimeSeriesView/utils';
import { useGetQueryRange } from 'hooks/queryBuilder/useGetQueryRange';
import { useQueryBuilder } from 'hooks/queryBuilder/useQueryBuilder';
import useUrlYAxisUnit from 'hooks/useUrlYAxisUnit';
import { AppState } from 'store/reducers';
import { Warning } from 'types/api';
import APIError from 'types/api/error';
import { DataSource } from 'types/common/queryBuilder';
import { GlobalReducer } from 'types/reducer/globalTime';
import './TimeSeriesView.styles.scss';
function TimeSeriesViewContainer({
dataSource = DataSource.TRACES,
isFilterApplied,
setWarning,
setIsLoadingQueries,
queryKeyRef,
}: TimeSeriesViewProps): JSX.Element {
const { stagedQuery, currentQuery, panelType } = useQueryBuilder();
const isValidToConvertToMs = useMemo(() => {
const isValid: boolean[] = [];
currentQuery.builder.queryData.forEach(
({ aggregateAttribute, aggregateOperator }) => {
const isExistDurationNanoAttribute =
aggregateAttribute?.key === 'durationNano' ||
aggregateAttribute?.key === 'duration_nano';
const isCountOperator =
aggregateOperator === 'count' || aggregateOperator === 'count_distinct';
isValid.push(!isCountOperator && isExistDurationNanoAttribute);
},
);
return isValid.every(Boolean);
}, [currentQuery]);
const defaultUnit = isValidToConvertToMs ? 'ms' : 'short';
const { yAxisUnit, onUnitChange } = useUrlYAxisUnit(defaultUnit);
const {
selectedTime: globalSelectedTime,
maxTime,
minTime,
} = useSelector<AppState, GlobalReducer>((state) => state.globalTime);
const queryKey = useMemo(
() => [
REACT_QUERY_KEY.GET_QUERY_RANGE,
globalSelectedTime,
maxTime,
minTime,
stagedQuery,
],
[globalSelectedTime, maxTime, minTime, stagedQuery],
);
if (queryKeyRef) {
queryKeyRef.current = queryKey;
}
const { data, isLoading, isFetching, isError, error } = useGetQueryRange(
{
query: stagedQuery || initialQueriesMap[dataSource],
graphType: panelType || PANEL_TYPES.TIME_SERIES,
selectedTime: 'GLOBAL_TIME',
globalSelectedInterval: globalSelectedTime,
params: {
dataSource,
},
},
// ENTITY_VERSION_V4,
ENTITY_VERSION_V5,
{
queryKey,
enabled: !!stagedQuery && panelType === PANEL_TYPES.TIME_SERIES,
},
);
useEffect(() => {
if (data?.payload) {
setWarning(data?.warning);
}
// eslint-disable-next-line react-hooks/exhaustive-deps
}, [data?.payload, data?.warning]);
const responseData = useMemo(
() => (isValidToConvertToMs ? convertDataValueToMs(data) : data),
[data, isValidToConvertToMs],
);
useEffect(() => {
if (isLoading || isFetching) {
setIsLoadingQueries(true);
} else {
setIsLoadingQueries(false);
}
}, [isLoading, isFetching, setIsLoadingQueries]);
return (
<div className="trace-explorer-time-series-view-container">
<TimeSeriesView
isFilterApplied={isFilterApplied}
isError={isError}
error={error as APIError}
isLoading={isLoading || isFetching}
data={responseData}
yAxisUnit={yAxisUnit}
onYAxisUnitChange={onUnitChange}
dataSource={dataSource}
setWarning={setWarning}
allowExport
/>
</div>
);
}
interface TimeSeriesViewProps {
dataSource?: DataSource;
isFilterApplied: boolean;
setWarning: Dispatch<SetStateAction<Warning | undefined>>;
setIsLoadingQueries: Dispatch<SetStateAction<boolean>>;
queryKeyRef?: MutableRefObject<any>;
}
TimeSeriesViewContainer.defaultProps = {
dataSource: DataSource.TRACES,
queryKeyRef: undefined,
};
export default TimeSeriesViewContainer;

View File

@@ -0,0 +1,15 @@
.container {
display: flex;
flex-direction: column;
flex: 1;
min-height: 0;
// Page chain isn't a flex column, so anchor the virtualized table against the viewport.
height: calc(100vh - 240px);
min-height: 400px;
}
.actionsContainer {
display: flex;
justify-content: space-between;
align-items: center;
}

View File

@@ -0,0 +1,190 @@
import {
Dispatch,
memo,
MutableRefObject,
SetStateAction,
useEffect,
useMemo,
} from 'react';
import { QueryKey } from 'react-query';
// eslint-disable-next-line no-restricted-imports
import { useSelector } from 'react-redux';
import { Typography } from '@signozhq/ui/typography';
import logEvent from 'api/common/logEvent';
import DownloadOptionsMenu from 'components/DownloadOptionsMenu/DownloadOptionsMenu';
import { ENTITY_VERSION_V5 } from 'constants/app';
import { QueryParams } from 'constants/query';
import { initialQueriesMap, PANEL_TYPES } from 'constants/queryBuilder';
import { REACT_QUERY_KEY } from 'constants/reactQueryKeys';
import TraceExplorerControls from 'container/TracesExplorer/Controls';
import { getListViewQuery } from 'container/TracesExplorer/explorerUtils';
import { getTraceLink } from 'container/TracesExplorer/ListView/utils';
import { TracesTableRow } from 'container/TracesExplorer/TracesTable/getFieldColumn';
import TracesTable from 'container/TracesExplorer/TracesTable/TracesTable';
import { useGetQueryRange } from 'hooks/queryBuilder/useGetQueryRange';
import { useQueryBuilder } from 'hooks/queryBuilder/useQueryBuilder';
import { Pagination } from 'hooks/queryPagination';
import useUrlQueryData from 'hooks/useUrlQueryData';
import { AppState } from 'store/reducers';
import { Warning } from 'types/api';
import { DataSource } from 'types/common/queryBuilder';
import { GlobalReducer } from 'types/reducer/globalTime';
import DOCLINKS from 'utils/docLinks';
import { columns, PER_PAGE_OPTIONS } from './configs';
import styles from './TracesView.module.scss';
interface TracesViewProps {
isFilterApplied: boolean;
setWarning: Dispatch<SetStateAction<Warning | undefined>>;
setIsLoadingQueries: Dispatch<SetStateAction<boolean>>;
queryKeyRef?: MutableRefObject<QueryKey | undefined>;
}
function TracesView({
isFilterApplied,
setWarning,
setIsLoadingQueries,
queryKeyRef,
}: TracesViewProps): JSX.Element {
const { stagedQuery, panelType } = useQueryBuilder();
const {
selectedTime: globalSelectedTime,
maxTime,
minTime,
} = useSelector<AppState, GlobalReducer>((state) => state.globalTime);
const { queryData: paginationQueryData } = useUrlQueryData<Pagination>(
QueryParams.pagination,
);
const transformedQuery = useMemo(
() => getListViewQuery(stagedQuery || initialQueriesMap.traces),
[stagedQuery],
);
const queryKey = useMemo(
() => [
REACT_QUERY_KEY.GET_QUERY_RANGE,
globalSelectedTime,
maxTime,
minTime,
stagedQuery,
panelType,
paginationQueryData,
],
[
globalSelectedTime,
maxTime,
minTime,
stagedQuery,
panelType,
paginationQueryData,
],
);
if (queryKeyRef) {
queryKeyRef.current = queryKey;
}
const { data, isLoading, isFetching, isError, error } = useGetQueryRange(
{
query: transformedQuery,
graphType: panelType || PANEL_TYPES.TRACE,
selectedTime: 'GLOBAL_TIME',
globalSelectedInterval: globalSelectedTime,
params: {
dataSource: 'traces',
},
tableParams: {
pagination: paginationQueryData,
},
},
ENTITY_VERSION_V5,
{
queryKey,
enabled: !!stagedQuery && panelType === PANEL_TYPES.TRACE,
},
);
useEffect(() => {
if (data?.payload) {
setWarning(data?.warning);
}
// eslint-disable-next-line react-hooks/exhaustive-deps
}, [data?.payload, data?.warning]);
const responseData = data?.payload?.data?.newResult?.data?.result[0]?.list;
const rows = useMemo<TracesTableRow[]>(
() =>
(responseData ?? []).map((item) => {
const row = item.data;
return { ...row, id: row.trace_id };
}) as TracesTableRow[],
[responseData],
);
useEffect(() => {
if (isLoading || isFetching) {
setIsLoadingQueries(true);
} else {
setIsLoadingQueries(false);
}
}, [isLoading, isFetching, setIsLoadingQueries]);
useEffect(() => {
if (!isLoading && !isFetching && !isError && rows.length !== 0) {
void logEvent('Traces Explorer: Data present', {
panelType: 'TRACE',
});
}
}, [isLoading, isFetching, isError, rows.length]);
return (
<div className={styles.container}>
<div className={styles.actionsContainer}>
<Typography>
This tab only shows Root Spans. More details
<Typography.Link href={DOCLINKS.TRACES_DETAILS_LINK} target="_blank">
{' '}
here
</Typography.Link>
</Typography>
<div className="trace-explorer-controls">
<DownloadOptionsMenu
dataSource={DataSource.TRACES}
panelType={PANEL_TYPES.TRACE}
/>
<TraceExplorerControls
isLoading={isLoading}
totalCount={rows.length}
perPageOptions={PER_PAGE_OPTIONS}
/>
</div>
</div>
<TracesTable
data={rows}
columns={columns}
respectColumnOrder
panelType="TRACE"
getRowHref={getTraceLink}
isLoading={isLoading}
isFetching={isFetching}
isError={isError}
error={error}
isFilterApplied={isFilterApplied}
/>
</div>
);
}
TracesView.defaultProps = {
queryKeyRef: undefined,
};
export default memo(TracesView);

View File

@@ -0,0 +1,25 @@
import { TelemetryFieldKey } from 'api/v5/v5';
import type { TableColumnDef } from 'components/TanStackTableView/types';
import {
getFieldColumn,
TracesTableRow,
} from 'container/TracesExplorer/TracesTable/getFieldColumn';
import { DEFAULT_PER_PAGE_OPTIONS } from 'hooks/queryPagination';
export const PER_PAGE_OPTIONS: number[] = [10, ...DEFAULT_PER_PAGE_OPTIONS];
const TRACE_FIELDS = [
{ name: 'service.name', fieldContext: 'resource' },
{ name: 'name' },
{ name: 'duration_nano' },
{ name: 'span_count' },
{ name: 'trace_id' },
] as TelemetryFieldKey[];
export const columns: TableColumnDef<TracesTableRow>[] = TRACE_FIELDS.map(
(field) => ({
...getFieldColumn(field),
enableRemove: false,
canBeHidden: false,
}),
);

View File

@@ -0,0 +1,36 @@
export const TOOLBAR_VIEWS = {
list: {
name: 'list',
label: 'List',
show: true,
key: 'list',
},
timeseries: {
name: 'timeseries',
label: 'Timeseries',
disabled: false,
show: true,
key: 'timeseries',
},
trace: {
name: 'trace',
label: 'Trace',
disabled: false,
show: true,
key: 'trace',
},
table: {
name: 'table',
label: 'Table',
disabled: false,
show: true,
key: 'table',
},
clickhouse: {
name: 'clickhouse',
label: 'Clickhouse',
disabled: false,
show: false,
key: 'clickhouse',
},
};

View File

@@ -18,6 +18,12 @@ jest.mock('pages/DashboardPageV2/DashboardContainer', () => ({
default: (): JSX.Element => <div data-testid="llm-overview-dashboard" />,
}));
// Same data-router gap as the dashboard above: the Explorer toolbar calls useNavigationType.
jest.mock('container/LLMObservability/Explorer/Explorer', () => ({
__esModule: true,
default: (): JSX.Element => <div data-testid="llm-observability-explorer" />,
}));
function setupList(items = mockRules): void {
server.use(
rest.get(LLM_PRICING_ENDPOINT, (_req, res, ctx) =>

View File

@@ -4,13 +4,6 @@
padding: 0px 8px;
.logs-frequency-chart {
.ant-card-body {
height: 140px;
min-height: 140px;
padding: 0 16px 22px 16px;
font-family: 'Geist Mono';
}
margin-bottom: 0px;
}
}

View File

@@ -3,13 +3,6 @@
min-height: 200px;
border-bottom: 1px solid var(--l1-border);
.ant-card-body {
height: 200px;
min-height: 200px;
padding: 0 16px 16px 16px;
font-family: 'Geist Mono';
}
.logs-frequency-chart-loading {
height: 100%;
display: flex;

View File

@@ -1,25 +1,29 @@
import { memo, useCallback, useMemo } from 'react';
import { memo, useCallback, useMemo, useRef } from 'react';
// eslint-disable-next-line no-restricted-imports
import { useDispatch, useSelector } from 'react-redux';
import { useLocation } from 'react-router-dom';
import Graph from 'components/Graph';
import Spinner from 'components/Spinner';
import { QueryParams } from 'constants/query';
import { themeColors } from 'constants/theme';
import BarChart from 'container/DashboardContainer/visualization/charts/BarChart/BarChart';
import { useResizeObserver } from 'hooks/useDimensions';
import { useSafeNavigate } from 'hooks/useSafeNavigate';
import useUrlQuery from 'hooks/useUrlQuery';
import getChartData, { GetChartDataProps } from 'lib/getChartData';
import GetMinMax from 'lib/getMinMax';
import { colors } from 'lib/getRandomColor';
import { LegendPosition } from 'lib/uPlotV2/components/types';
import { StackMode } from 'lib/uPlotV2/config/types';
import { useTimezone } from 'providers/Timezone';
import { UpdateTimeInterval } from 'store/actions';
import { AppState } from 'store/reducers';
import { GlobalReducer } from 'types/reducer/globalTime';
import { LogsExplorerChartProps } from './LogsExplorerChart.interfaces';
import { getColorsForSeverityLabels } from './utils';
import { useLogsExplorerChartConfig } from './useLogsExplorerChartConfig';
import './LogsExplorerChart.styles.scss';
// Axis and tooltip format separately; both need this or only one abbreviates.
const Y_AXIS_UNIT = 'short';
function LogsExplorerChart({
data,
isLoading,
@@ -37,24 +41,6 @@ function LogsExplorerChart({
const { minTime, maxTime } = useSelector<AppState, GlobalReducer>(
(state) => state.globalTime,
);
const handleCreateDatasets: Required<GetChartDataProps>['createDataset'] =
useCallback(
(element, index, allLabels) => ({
data: element,
backgroundColor: isLogsExplorerViews
? getColorsForSeverityLabels(allLabels[index], index)
: colors[index % colors.length] || themeColors.red,
borderColor: isLogsExplorerViews
? getColorsForSeverityLabels(allLabels[index], index)
: colors[index % colors.length] || themeColors.red,
...(isLabelEnabled
? {
label: allLabels[index],
}
: {}),
}),
[isLabelEnabled, isLogsExplorerViews],
);
const onDragSelect = useCallback(
(start: number, end: number): void => {
@@ -86,44 +72,47 @@ function LogsExplorerChart({
[dispatch, location.pathname, safeNavigate, urlQuery, isShowingLiveLogs],
);
const graphData = useMemo(
() =>
getChartData({
queryData: [
{
queryData: data,
},
],
createDataset: handleCreateDatasets,
}),
[data, handleCreateDatasets],
);
// Convert nanosecond timestamps to milliseconds for Chart.js
const { chartMinTime, chartMaxTime } = useMemo(
// uPlot plots the series on a seconds-based x scale
const { minTimeScale, maxTimeScale } = useMemo(
() => ({
chartMinTime: minTime ? Math.floor(minTime / 1e6) : undefined,
chartMaxTime: maxTime ? Math.floor(maxTime / 1e6) : undefined,
minTimeScale: minTime ? Math.floor(minTime / 1e9) : undefined,
maxTimeScale: maxTime ? Math.floor(maxTime / 1e9) : undefined,
}),
[minTime, maxTime],
);
const { timezone } = useTimezone();
const graphRef = useRef<HTMLDivElement>(null);
const dimensions = useResizeObserver(graphRef);
const { config, chartData } = useLogsExplorerChartConfig({
data,
isLogsExplorerViews,
isLabelEnabled,
onDragSelect,
minTimeScale,
maxTimeScale,
yAxisUnit: Y_AXIS_UNIT,
});
return (
<div className={`${className} logs-frequency-chart-container`}>
<div ref={graphRef} className={`${className} logs-frequency-chart-container`}>
{isLoading ? (
<div className="logs-frequency-chart-loading">
<Spinner size="default" height="100%" />
</div>
) : (
<Graph
name="logsExplorerChart"
data={graphData.data}
isStacked={isLogsExplorerViews}
type="bar"
animate
onDragSelect={onDragSelect}
minTime={chartMinTime}
maxTime={chartMaxTime}
<BarChart
config={config}
data={chartData}
width={dimensions.width}
height={dimensions.height}
stack={isLogsExplorerViews ? StackMode.Normal : StackMode.None}
showLegend={isLabelEnabled}
legendConfig={{ position: LegendPosition.BOTTOM }}
timezone={timezone}
data-testid="logs-frequency-chart"
yAxisUnit={Y_AXIS_UNIT}
/>
)}
</div>

View File

@@ -0,0 +1,105 @@
import { useMemo } from 'react';
import { PANEL_TYPES } from 'constants/queryBuilder';
import { themeColors } from 'constants/theme';
import { buildBaseConfig } from 'container/DashboardContainer/visualization/panels/utils/baseConfigBuilder';
import { useIsDarkMode } from 'hooks/useDarkMode';
import getLabelName from 'lib/getLabelName';
import { colors } from 'lib/getRandomColor';
import { getUPlotChartData } from 'lib/uPlotLib/utils/getUplotChartData';
import { DrawStyle } from 'lib/uPlotV2/config/types';
import { UPlotConfigBuilder } from 'lib/uPlotV2/config/UPlotConfigBuilder';
import { useTimezone } from 'providers/Timezone';
import { MetricRangePayloadProps } from 'types/api/metrics/getQueryRange';
import { QueryData } from 'types/api/widgets/getQuery';
import uPlot from 'uplot';
import { getColorsForSeverityLabels } from './utils';
export interface UseLogsExplorerChartConfigParams {
data: QueryData[];
isLogsExplorerViews?: boolean;
isLabelEnabled?: boolean;
onDragSelect: (start: number, end: number) => void;
minTimeScale?: number;
maxTimeScale?: number;
yAxisUnit?: string;
}
export interface UseLogsExplorerChartConfigResult {
config: UPlotConfigBuilder;
chartData: uPlot.AlignedData;
}
export function useLogsExplorerChartConfig({
data,
isLogsExplorerViews = false,
isLabelEnabled = true,
onDragSelect,
minTimeScale,
maxTimeScale,
yAxisUnit,
}: UseLogsExplorerChartConfigParams): UseLogsExplorerChartConfigResult {
const isDarkMode = useIsDarkMode();
const { timezone } = useTimezone();
// getUPlotChartData / buildBaseConfig both consume the legacy query-range payload
// shape, so the raw series list is wrapped instead of being plotted directly.
const apiResponse = useMemo(
() =>
({
data: { result: data, resultType: '' },
}) as unknown as MetricRangePayloadProps,
[data],
);
const chartData = useMemo(() => getUPlotChartData(apiResponse), [apiResponse]);
const config = useMemo(() => {
const builder = buildBaseConfig({
id: 'logs-explorer-frequency-chart',
isDarkMode,
onDragSelect,
timezone,
minTimeScale,
maxTimeScale,
yAxisUnit,
panelType: PANEL_TYPES.BAR,
});
data.forEach((series, index) => {
const label = getLabelName(
series.metric,
series.queryName || '',
series.legend || '',
);
const color = isLogsExplorerViews
? getColorsForSeverityLabels(label, index)
: colors[index % colors.length] || themeColors.red;
builder.addSeries({
scaleKey: 'y',
drawStyle: DrawStyle.Bar,
// No group-by yields query name "A"; use ' ' not '' so uPlot does not default the label to "Value".
label: isLabelEnabled && label.trim() ? label : ' ',
lineColor: color,
colorMapping: {},
isDarkMode,
});
});
return builder;
}, [
data,
isDarkMode,
isLabelEnabled,
isLogsExplorerViews,
maxTimeScale,
minTimeScale,
onDragSelect,
timezone,
yAxisUnit,
]);
return { config, chartData };
}

View File

@@ -217,13 +217,6 @@
padding: 0px 8px;
.logs-frequency-chart {
.ant-card-body {
height: 140px;
min-height: 140px;
padding: 0 16px 22px 16px;
font-family: 'Geist Mono';
}
margin-bottom: 0px;
}
}

View File

@@ -4,13 +4,12 @@ import (
"net/http"
"github.com/SigNoz/signoz/pkg/http/handler"
"github.com/SigNoz/signoz/pkg/types"
"github.com/SigNoz/signoz/pkg/types/featuretypes"
"github.com/gorilla/mux"
)
func (provider *provider) addFlaggerRoutes(router *mux.Router) error {
if err := router.Handle("/api/v2/features", handler.New(provider.authzMiddleware.ViewAccess(provider.flaggerHandler.GetFeatures), handler.OpenAPIDef{
if err := router.Handle("/api/v2/features", handler.New(provider.authzMiddleware.OpenAccess(provider.flaggerHandler.GetFeatures), handler.OpenAPIDef{
ID: "GetFeatures",
Tags: []string{"features"},
Summary: "Get features",
@@ -22,7 +21,7 @@ func (provider *provider) addFlaggerRoutes(router *mux.Router) error {
SuccessStatusCode: http.StatusOK,
ErrorStatusCodes: []int{},
Deprecated: false,
SecuritySchemes: newSecuritySchemes(types.RoleViewer),
SecuritySchemes: newScopedSecuritySchemes(nil),
})).Methods(http.MethodGet).GetError(); err != nil {
return err
}

View File

@@ -5,7 +5,6 @@ import (
"net/http"
"time"
"github.com/SigNoz/signoz/pkg/errors"
"github.com/SigNoz/signoz/pkg/http/binding"
"github.com/SigNoz/signoz/pkg/http/render"
"github.com/SigNoz/signoz/pkg/modules/aiobservability"
@@ -66,13 +65,6 @@ func (handler *handler) GetFieldsValues(rw http.ResponseWriter, req *http.Reques
ctx, cancel := context.WithTimeout(req.Context(), 10*time.Second)
defer cancel()
// binding ignores query params the struct does not declare, so an unsupported
// existingQuery would silently return values it did not narrow
if req.URL.Query().Has("existingQuery") {
render.Error(rw, errors.New(errors.TypeInvalidInput, errors.CodeInvalidInput, "existingQuery is not supported"))
return
}
var params aiobservabilitytypes.PostableFieldValueParams
if err := binding.Query.BindQuery(req.URL.Query(), &params); err != nil {
render.Error(rw, err)
@@ -84,18 +76,29 @@ func (handler *handler) GetFieldsValues(rw http.ResponseWriter, req *http.Reques
render.Error(rw, err)
return
}
orgID := valuer.MustNewUUID(claims.OrgID)
params.ExistingQuery = aitelemetryschema.ScopedExistingQuery(params.ExistingQuery)
fieldValueSelector := aiobservabilitytypes.NewFieldValueSelectorFromPostableFieldValueParams(params)
values := &telemetrytypes.TelemetryFieldValues{}
complete := true
// the trace context names the computed per-trace aggregates, which are never ingested
if fieldValueSelector.FieldContext != telemetrytypes.FieldContextTrace {
values, complete, err = handler.telemetryMetadataStore.GetAllValues(ctx, valuer.MustNewUUID(claims.OrgID), fieldValueSelector)
values, complete, err = handler.telemetryMetadataStore.GetAllValues(ctx, orgID, fieldValueSelector)
if err != nil {
render.Error(rw, err)
return
}
// related values are best-effort: on failure the plain values still serve
// the filter bar
relatedValues, relatedComplete, err := handler.telemetryMetadataStore.GetRelatedValues(ctx, orgID, fieldValueSelector)
if err != nil {
relatedValues = []string{}
}
values.RelatedValues = relatedValues
complete = complete && relatedComplete
}
render.Success(rw, http.StatusOK, &telemetrytypes.GettableFieldValues{

View File

@@ -82,6 +82,12 @@ func (q *builderQuery[T]) Fingerprint() string {
return ""
}
// AI trace aggregations qualify and rank traces on whole-window per-trace
// values, which do not decompose into cacheable time buckets.
if q.queryType == qbtypes.QueryTypeBuilderAI {
return ""
}
// Create a deterministic fingerprint for builder queries
// This needs to include all fields that affect the query results
parts := []string{q.queryType.StringValue()}

View File

@@ -117,8 +117,7 @@ func TestBuilderQueryFingerprintQueryType(t *testing.T) {
}
assert.True(t, strings.HasPrefix(regular.Fingerprint(), qbtypes.QueryTypeBuilder.StringValue()+"&"))
assert.True(t, strings.HasPrefix(ai.Fingerprint(), qbtypes.QueryTypeBuilderAI.StringValue()+"&"))
assert.NotEqual(t, regular.Fingerprint(), ai.Fingerprint())
assert.Empty(t, ai.Fingerprint())
}
func TestMakeBucketsOrder(t *testing.T) {

View File

@@ -470,7 +470,7 @@ func (q *querier) resolveMetricMetadata(ctx context.Context, orgID valuer.UUID,
continue
}
// Type is resolved now; validate aggregation compatibility against it.
if err := spec.Aggregations[i].ValidateForType(); err != nil {
if err := spec.Aggregations[i].ValidateForTypeAndTemporality(); err != nil {
return nil, nil, err
}
if reducedMetricsSet[spec.Aggregations[i].MetricName] {

View File

@@ -439,7 +439,7 @@ func (aH *APIHandler) RegisterRoutes(router *mux.Router, am *middleware.AuthZ) {
router.HandleFunc("/api/v2/traces/fields", am.EditAccess(aH.updateTraceField)).Methods(http.MethodPost)
router.HandleFunc("/api/v1/version", am.OpenAccess(aH.getVersion)).Methods(http.MethodGet)
router.HandleFunc("/api/v1/features", am.ViewAccess(aH.getFeatureFlags)).Methods(http.MethodGet)
router.HandleFunc("/api/v1/features", am.OpenAccess(aH.getFeatureFlags)).Methods(http.MethodGet)
router.HandleFunc("/api/v1/health", am.OpenAccess(aH.getHealth)).Methods(http.MethodGet)
router.HandleFunc("/api/v1/listErrors", am.ViewAccess(aH.listErrors)).Methods(http.MethodPost)
@@ -1497,7 +1497,7 @@ func (aH *APIHandler) getFeatureFlags(w http.ResponseWriter, r *http.Request) {
claims, err := authtypes.ClaimsFromContext(r.Context())
if err != nil {
aH.HandleError(w, err, http.StatusInternalServerError)
aH.HandleError(w, err, http.StatusUnauthorized)
return
}

View File

@@ -107,8 +107,8 @@ func SynthesizeKeys(field *telemetrytypes.TelemetryFieldKey, value any) []*telem
fieldContext = telemetrytypes.FieldContextAttribute
}
fieldDataType := field.FieldDataType
// Resource values are strings; pin the type so operand coercion applies.
if fieldContext == telemetrytypes.FieldContextResource &&
// Resource and scope values are strings; pin the type so operand coercion applies.
if (fieldContext == telemetrytypes.FieldContextResource || fieldContext == telemetrytypes.FieldContextScope) &&
fieldDataType == telemetrytypes.FieldDataTypeUnspecified {
fieldDataType = telemetrytypes.FieldDataTypeString
}

View File

@@ -56,6 +56,17 @@ func QueryStringToKeysSelectors(query string) []*telemetrytypes.FieldKeySelector
FieldDataType: key.FieldDataType,
})
}
// todo(tushar): consider reverting changes done to this method in below PR to avoid scope specific checks
// https://github.com/SigNoz/signoz/issues/11374
if key.FieldContext == telemetrytypes.FieldContextScope {
keys = append(keys, &telemetrytypes.FieldKeySelector{
Name: key.FieldContext.StringValue() + "." + key.Name,
Signal: key.Signal,
FieldContext: telemetrytypes.FieldContextUnspecified, // this allows 'scope.' prefix for keys with other context as well
FieldDataType: key.FieldDataType,
})
}
}
}

View File

@@ -72,6 +72,44 @@ func TestQueryToKeys(t *testing.T) {
},
},
},
{
query: `scope.version = '1.0.0'`,
expectedKeys: []telemetrytypes.FieldKeySelector{
{
Name: "version",
Signal: telemetrytypes.SignalUnspecified,
FieldContext: telemetrytypes.FieldContextScope,
FieldDataType: telemetrytypes.FieldDataTypeUnspecified,
},
{
Name: "scope.version",
Signal: telemetrytypes.SignalUnspecified,
FieldContext: telemetrytypes.FieldContextUnspecified,
FieldDataType: telemetrytypes.FieldDataTypeUnspecified,
},
},
},
{
// A scope attribute whose own name carries a `scope.` prefix. `scope.prefixed`
// normalizes to {prefixed, scope}; the second selector re-adds the prefix so the
// metadata fetch can target the attribute's exact key `scope.prefixed` rather than
// relying on the broad `%prefixed%` match.
query: `scope.prefixed = 'x'`,
expectedKeys: []telemetrytypes.FieldKeySelector{
{
Name: "prefixed",
Signal: telemetrytypes.SignalUnspecified,
FieldContext: telemetrytypes.FieldContextScope,
FieldDataType: telemetrytypes.FieldDataTypeUnspecified,
},
{
Name: "scope.prefixed",
Signal: telemetrytypes.SignalUnspecified,
FieldContext: telemetrytypes.FieldContextUnspecified,
FieldDataType: telemetrytypes.FieldDataTypeUnspecified,
},
},
},
}
for _, testCase := range testCases {

View File

@@ -1,8 +1,6 @@
package aistatementbuilder
import (
"strings"
"github.com/SigNoz/signoz/pkg/factory"
"github.com/SigNoz/signoz/pkg/flagger"
"github.com/SigNoz/signoz/pkg/statementbuilder"
@@ -27,11 +25,8 @@ func NewFactory(
// Scope describes gen_ai for the scoped trace builder: an AI trace has >=1 gen_ai
// LLM, tool, or agent span, and its list adds AI/LLM per-trace metrics.
func Scope() scopedtraces.TraceScope {
gateKeyNames := []string{aiobservabilitytypes.GenAIRequestModel, aiobservabilitytypes.GenAIToolName, aiobservabilitytypes.GenAIAgentName}
gateExprs := make([]string, 0, len(gateKeyNames))
gateKeys := make([]*telemetrytypes.TelemetryFieldKey, 0, len(gateKeyNames))
for _, name := range gateKeyNames {
gateExprs = append(gateExprs, name+" EXISTS")
gateKeys := make([]*telemetrytypes.TelemetryFieldKey, 0, len(aiobservabilitytypes.GenAISpanGateKeys))
for _, name := range aiobservabilitytypes.GenAISpanGateKeys {
gateKeys = append(gateKeys, &telemetrytypes.TelemetryFieldKey{
Name: name,
Signal: telemetrytypes.SignalTraces,
@@ -79,7 +74,7 @@ func Scope() scopedtraces.TraceScope {
}
return scopedtraces.TraceScope{
FilterExpression: strings.Join(gateExprs, " OR "),
FilterExpression: aiobservabilitytypes.GenAISpanFilterExpression(),
FieldKeys: gateKeys,
Columns: columns,
DefaultOrderAlias: "last_activity_time",

View File

@@ -994,8 +994,8 @@ func TestBuild_TraceList_MultiVariantGateKey(t *testing.T) {
assert.Contains(t, got, "mapContains(attributes_string, 'gen_ai.tool.name') OR mapContains(attributes_number, 'gen_ai.tool.name')")
}
// `trace.` marks a trace-level aggregate; `tracefield.` routes trace-level too but is
// not a rewritable alias, so the HAVING rewriter rejects it.
// A `trace.`-prefixed aggregate in the filter box and the same condition in the
// explicit Having box build the same query; output-only aggregates are rejected.
func TestBuild_TraceList_TraceContextPrefix(t *testing.T) {
b := newTestBuilder(t)
build := func(q qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]) (*qbtypes.Statement, error) {
@@ -1003,19 +1003,14 @@ func TestBuild_TraceList_TraceContextPrefix(t *testing.T) {
return b.Build(context.Background(), valuer.UUID{}, testStartMs, testEndMs, qbtypes.RequestTypeTrace, q, nil)
}
_, err := build(qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
viaTrace, err := build(qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Filter: &qbtypes.Filter{Expression: "trace.output_tokens > 1000"}})
require.NoError(t, err)
_, err = build(qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Filter: &qbtypes.Filter{Expression: "tracefield.output_tokens > 1000"}})
require.Error(t, err)
assert.Contains(t, err.Error(), "Invalid references in `Having` expression: [tracefield.output_tokens]")
_, err = build(qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Having: &qbtypes.Having{Expression: "tracefield.output_tokens > 1000"}})
require.Error(t, err)
assert.Contains(t, err.Error(), "Invalid references in `Having` expression: [tracefield.output_tokens]")
viaHaving, err := build(qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Having: &qbtypes.Having{Expression: "trace.output_tokens > 1000"}})
require.NoError(t, err)
assert.Equal(t, viaTrace.Query, viaHaving.Query)
_, err = build(qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Filter: &qbtypes.Filter{Expression: "trace.span_count > 3"}})
@@ -1023,7 +1018,8 @@ func TestBuild_TraceList_TraceContextPrefix(t *testing.T) {
assert.Contains(t, err.Error(), "cannot be used")
}
// Query variables in a trace-level condition are substituted into the HAVING.
// Query variables in a trace-level condition resolve like span filters: bound args,
// list/IN handling, dynamic __all__ dropping the condition.
func TestBuild_TraceList_VariableInAggregateFilter(t *testing.T) {
b := newTestBuilder(t)
build := func(expr string, vars map[string]qbtypes.VariableItem) (*qbtypes.Statement, error) {
@@ -1035,17 +1031,18 @@ func TestBuild_TraceList_VariableInAggregateFilter(t *testing.T) {
}, vars)
}
// scalar variable -> literal in HAVING
// scalar variable -> bound arg via the filter pipeline
stmt, err := build("trace.output_tokens > $threshold",
map[string]qbtypes.VariableItem{"threshold": {Value: 700}})
require.NoError(t, err)
assert.Contains(t, stmt.Query, "HAVING output_tokens > 700")
assert.Contains(t, stmt.Query, "HAVING output_tokens > ?")
assert.Contains(t, stmt.Args, float64(700))
// list variable with IN
stmt, err = build("trace.llm_call_count IN $counts",
map[string]qbtypes.VariableItem{"counts": {Value: []any{1, 2}}})
require.NoError(t, err)
assert.Contains(t, stmt.Query, "HAVING llm_call_count IN")
assert.Contains(t, stmt.Query, "HAVING llm_call_count IN (?, ?)")
// dynamic __all__ -> condition dropped, no HAVING at all
stmt, err = build("trace.output_tokens > $threshold",
@@ -1053,7 +1050,7 @@ func TestBuild_TraceList_VariableInAggregateFilter(t *testing.T) {
require.NoError(t, err)
assert.NotContains(t, stmt.Query, "HAVING")
// unresolved variable -> rejected, not compared as a literal
// unresolved variable -> rejected, though only as an unknown aggregate today
_, err = build("trace.output_tokens > $missing", map[string]qbtypes.VariableItem{"other": {Value: 1}})
require.Error(t, err)
}

View File

@@ -0,0 +1,753 @@
package aistatementbuilder
import (
"context"
"testing"
"time"
qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
"github.com/SigNoz/signoz/pkg/valuer"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
// The builder assumes at least one aggregation; request validation is what enforces it.
func TestBuild_Aggregation_NoAggregations_RejectedByRequestValidation(t *testing.T) {
for _, rt := range []qbtypes.RequestType{qbtypes.RequestTypeScalar, qbtypes.RequestTypeTimeSeries} {
req := qbtypes.QueryRangeRequest{
Start: testStartMs,
End: testEndMs,
RequestType: rt,
CompositeQuery: qbtypes.CompositeQuery{
Queries: []qbtypes.QueryEnvelope{{
Type: qbtypes.QueryTypeBuilderAI,
Spec: qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Name: "A",
Signal: telemetrytypes.SignalTraces,
StepInterval: qbtypes.Step{Duration: 60 * time.Second},
},
}},
},
}
require.ErrorContains(t, req.Validate(), "at least one aggregation is required", rt.StringValue())
}
}
// Traces without token spans yield NULL, which the outer avg skips.
func TestBuild_FullSQL_Scalar_TraceAgg(t *testing.T) {
b := newTestBuilder(t)
stmt, err := b.Build(context.Background(), valuer.UUID{}, testStartMs, testEndMs, qbtypes.RequestTypeScalar,
qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Signal: telemetrytypes.SignalTraces,
Aggregations: []qbtypes.TraceAggregation{{Expression: "avg(trace.output_tokens)"}},
}, nil)
require.NoError(t, err)
assertSQLEqual(t, `
WITH __scoped_traces AS (
SELECT trace_id,
sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.output_tokens'), toFloat64(attributes_number['gen_ai.usage.output_tokens']), NULL)) AS output_tokens
FROM signoz_traces.distributed_signoz_index_v3
WHERE timestamp >= '1747947419000000000'
AND timestamp < '1747983448000000000'
AND ts_bucket_start >= 1747945619
AND ts_bucket_start <= 1747983448
AND (mapContains(attributes_string, 'gen_ai.request.model') OR mapContains(attributes_string, 'gen_ai.tool.name') OR mapContains(attributes_string, 'gen_ai.agent.name'))
GROUP BY trace_id
)
SELECT avg(output_tokens) AS __result_0
FROM __scoped_traces
ORDER BY __result_0 DESC
SETTINGS distributed_product_mode='allow', max_memory_usage=10000000000
`, stmt)
}
// A span-level filter is ANDed into the per-trace scan's WHERE, next to the gate mask.
func TestBuild_FullSQL_Scalar_SpanFilter(t *testing.T) {
b := newTestBuilder(t)
stmt, err := b.Build(context.Background(), valuer.UUID{}, testStartMs, testEndMs, qbtypes.RequestTypeScalar,
qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Signal: telemetrytypes.SignalTraces,
Aggregations: []qbtypes.TraceAggregation{{Expression: "avg(trace.output_tokens)"}},
Filter: &qbtypes.Filter{Expression: "gen_ai.request.model = 'gpt-4o-mini'"},
}, nil)
require.NoError(t, err)
assertSQLEqual(t, `
WITH __scoped_traces AS (
SELECT trace_id,
sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.output_tokens'), toFloat64(attributes_number['gen_ai.usage.output_tokens']), NULL)) AS output_tokens
FROM signoz_traces.distributed_signoz_index_v3
WHERE timestamp >= '1747947419000000000'
AND timestamp < '1747983448000000000'
AND ts_bucket_start >= 1747945619
AND ts_bucket_start <= 1747983448
AND (mapContains(attributes_string, 'gen_ai.request.model') OR mapContains(attributes_string, 'gen_ai.tool.name') OR mapContains(attributes_string, 'gen_ai.agent.name'))
AND (attributes_string['gen_ai.request.model'] = 'gpt-4o-mini' AND mapContains(attributes_string, 'gen_ai.request.model'))
GROUP BY trace_id
)
SELECT avg(output_tokens) AS __result_0
FROM __scoped_traces
ORDER BY __result_0 DESC
SETTINGS distributed_product_mode='allow', max_memory_usage=10000000000
`, stmt)
}
// A trace-level filter qualifies first: __qualified holds the trace ids whose
// whole-window value passes, and the per-trace scan is constrained to them.
func TestBuild_FullSQL_Scalar_TraceFilter(t *testing.T) {
b := newTestBuilder(t)
stmt, err := b.Build(context.Background(), valuer.UUID{}, testStartMs, testEndMs, qbtypes.RequestTypeScalar,
qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Signal: telemetrytypes.SignalTraces,
Aggregations: []qbtypes.TraceAggregation{{Expression: "avg(trace.output_tokens)"}},
Filter: &qbtypes.Filter{Expression: "trace.output_tokens > 1000"},
}, nil)
require.NoError(t, err)
assertSQLEqual(t, `
WITH __qualified AS (
SELECT trace_id,
sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.output_tokens'), toFloat64(attributes_number['gen_ai.usage.output_tokens']), NULL)) AS output_tokens
FROM signoz_traces.distributed_signoz_index_v3
WHERE timestamp >= '1747947419000000000'
AND timestamp < '1747983448000000000'
AND ts_bucket_start >= 1747945619
AND ts_bucket_start <= 1747983448
AND (mapContains(attributes_string, 'gen_ai.request.model') OR mapContains(attributes_string, 'gen_ai.tool.name') OR mapContains(attributes_string, 'gen_ai.agent.name'))
GROUP BY trace_id
HAVING output_tokens > 1000
),
__scoped_traces AS (
SELECT trace_id,
sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.output_tokens'), toFloat64(attributes_number['gen_ai.usage.output_tokens']), NULL)) AS output_tokens
FROM signoz_traces.distributed_signoz_index_v3
WHERE timestamp >= '1747947419000000000'
AND timestamp < '1747983448000000000'
AND ts_bucket_start >= 1747945619
AND ts_bucket_start <= 1747983448
AND (mapContains(attributes_string, 'gen_ai.request.model') OR mapContains(attributes_string, 'gen_ai.tool.name') OR mapContains(attributes_string, 'gen_ai.agent.name'))
AND trace_id GLOBAL IN (SELECT trace_id FROM __qualified)
GROUP BY trace_id
)
SELECT avg(output_tokens) AS __result_0
FROM __scoped_traces
ORDER BY __result_0 DESC
SETTINGS distributed_product_mode='allow', max_memory_usage=10000000000
`, stmt)
}
// Grouping by an intrinsic: the positional alias keeps `toString(name) AS name` (a cyclic
// alias) from forming, and an order key on the dimension resolves to that alias.
func TestBuild_FullSQL_Scalar_GroupByIntrinsic(t *testing.T) {
b := newTestBuilder(t)
stmt, err := b.Build(context.Background(), valuer.UUID{}, testStartMs, testEndMs, qbtypes.RequestTypeScalar,
qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Signal: telemetrytypes.SignalTraces,
Aggregations: []qbtypes.TraceAggregation{{Expression: "avg(trace.output_tokens)"}},
GroupBy: []qbtypes.GroupByKey{{TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{Name: "name"}}},
Order: []qbtypes.OrderBy{{Key: qbtypes.OrderByKey{TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{Name: "name"}}, Direction: qbtypes.OrderDirectionAsc}},
}, nil)
require.NoError(t, err)
assertSQLEqual(t, `
WITH __scoped_traces AS (
SELECT trace_id,
toString(multiIf(name <> '', toString(name), NULL)) AS __GROUP_BY_KEY_0_name,
sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.output_tokens'), toFloat64(attributes_number['gen_ai.usage.output_tokens']), NULL)) AS output_tokens
FROM signoz_traces.distributed_signoz_index_v3
WHERE timestamp >= '1747947419000000000'
AND timestamp < '1747983448000000000'
AND ts_bucket_start >= 1747945619
AND ts_bucket_start <= 1747983448
AND (mapContains(attributes_string, 'gen_ai.request.model') OR mapContains(attributes_string, 'gen_ai.tool.name') OR mapContains(attributes_string, 'gen_ai.agent.name'))
GROUP BY trace_id, __GROUP_BY_KEY_0_name
)
SELECT __GROUP_BY_KEY_0_name, avg(output_tokens) AS __result_0
FROM __scoped_traces
GROUP BY __GROUP_BY_KEY_0_name
ORDER BY __GROUP_BY_KEY_0_name asc
SETTINGS distributed_product_mode='allow', max_memory_usage=10000000000
`, stmt)
}
// Every dimension at once; the HAVING on the alias is rewritten to __result_0.
func TestBuild_FullSQL_Scalar_FullCombo(t *testing.T) {
b := newTestBuilder(t)
stmt, err := b.Build(context.Background(), valuer.UUID{}, testStartMs, testEndMs, qbtypes.RequestTypeScalar,
qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Signal: telemetrytypes.SignalTraces,
Aggregations: []qbtypes.TraceAggregation{
{Expression: "avg(trace.output_tokens)", Alias: "avg_out"},
{Expression: "count(trace.trace_id)"},
},
Filter: &qbtypes.Filter{Expression: "gen_ai.request.model = 'gpt-4o-mini' AND trace.total_tokens > 100"},
GroupBy: []qbtypes.GroupByKey{{TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{Name: "gen_ai.request.model"}}},
Having: &qbtypes.Having{Expression: "avg_out > 50"},
Order: []qbtypes.OrderBy{{Key: qbtypes.OrderByKey{TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{Name: "avg_out"}}, Direction: qbtypes.OrderDirectionDesc}},
Limit: 5,
}, nil)
require.NoError(t, err)
assertSQLEqual(t, `
WITH __qualified AS (
SELECT trace_id,
coalesce(sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.input_tokens'), toFloat64(attributes_number['gen_ai.usage.input_tokens']), NULL)), 0) + coalesce(sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.output_tokens'), toFloat64(attributes_number['gen_ai.usage.output_tokens']), NULL)), 0) AS total_tokens
FROM signoz_traces.distributed_signoz_index_v3
WHERE timestamp >= '1747947419000000000'
AND timestamp < '1747983448000000000'
AND ts_bucket_start >= 1747945619
AND ts_bucket_start <= 1747983448
AND (mapContains(attributes_string, 'gen_ai.request.model') OR mapContains(attributes_string, 'gen_ai.tool.name') OR mapContains(attributes_string, 'gen_ai.agent.name'))
GROUP BY trace_id
HAVING total_tokens > 100
),
__scoped_traces AS (
SELECT trace_id,
toString(multiIf(mapContains(attributes_string, 'gen_ai.request.model'), attributes_string['gen_ai.request.model'], NULL)) AS __GROUP_BY_KEY_0_gen_ai.request.model,
sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.output_tokens'), toFloat64(attributes_number['gen_ai.usage.output_tokens']), NULL)) AS output_tokens
FROM signoz_traces.distributed_signoz_index_v3
WHERE timestamp >= '1747947419000000000'
AND timestamp < '1747983448000000000'
AND ts_bucket_start >= 1747945619
AND ts_bucket_start <= 1747983448
AND (mapContains(attributes_string, 'gen_ai.request.model') OR mapContains(attributes_string, 'gen_ai.tool.name') OR mapContains(attributes_string, 'gen_ai.agent.name'))
AND (attributes_string['gen_ai.request.model'] = 'gpt-4o-mini' AND mapContains(attributes_string, 'gen_ai.request.model'))
AND trace_id GLOBAL IN (SELECT trace_id FROM __qualified)
GROUP BY trace_id, __GROUP_BY_KEY_0_gen_ai.request.model
)
SELECT __GROUP_BY_KEY_0_gen_ai.request.model, avg(output_tokens) AS __result_0, count(trace_id) AS __result_1
FROM __scoped_traces
GROUP BY __GROUP_BY_KEY_0_gen_ai.request.model
HAVING __result_0 > 50
ORDER BY __result_0 desc
LIMIT 5
SETTINGS distributed_product_mode='allow', max_memory_usage=10000000000
`, stmt)
}
// Time series: the per-trace scan buckets by span time, the outer aggregation per bucket.
func TestBuild_FullSQL_TimeSeries_TraceAgg(t *testing.T) {
b := newTestBuilder(t)
stmt, err := b.Build(context.Background(), valuer.UUID{}, testStartMs, testEndMs, qbtypes.RequestTypeTimeSeries,
qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Signal: telemetrytypes.SignalTraces,
StepInterval: qbtypes.Step{Duration: 60 * time.Second},
Aggregations: []qbtypes.TraceAggregation{{Expression: "avg(trace.output_tokens)"}},
}, nil)
require.NoError(t, err)
assertSQLEqual(t, `
WITH __scoped_traces AS (
SELECT trace_id,
toStartOfInterval(timestamp, INTERVAL 60 SECOND) AS ts,
sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.output_tokens'), toFloat64(attributes_number['gen_ai.usage.output_tokens']), NULL)) AS output_tokens
FROM signoz_traces.distributed_signoz_index_v3
WHERE timestamp >= '1747947419000000000'
AND timestamp < '1747983448000000000'
AND ts_bucket_start >= 1747945619
AND ts_bucket_start <= 1747983448
AND (mapContains(attributes_string, 'gen_ai.request.model') OR mapContains(attributes_string, 'gen_ai.tool.name') OR mapContains(attributes_string, 'gen_ai.agent.name'))
GROUP BY trace_id, ts
)
SELECT ts, avg(output_tokens) AS __result_0
FROM __scoped_traces
GROUP BY ts
SETTINGS distributed_product_mode='allow', max_memory_usage=10000000000
`, stmt)
}
// A grouped, limited time series ranks groups on unbucketed whole-window values
// (__scoped_traces_total), so a non-composable aggregate like avg ranks exactly.
func TestBuild_FullSQL_TimeSeries_GroupLimit(t *testing.T) {
b := newTestBuilder(t)
stmt, err := b.Build(context.Background(), valuer.UUID{}, testStartMs, testEndMs, qbtypes.RequestTypeTimeSeries,
qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Signal: telemetrytypes.SignalTraces,
StepInterval: qbtypes.Step{Duration: 60 * time.Second},
Aggregations: []qbtypes.TraceAggregation{{Expression: "sum(trace.output_tokens)", Alias: "total_out"}},
GroupBy: []qbtypes.GroupByKey{{TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{Name: "gen_ai.request.model"}}},
Having: &qbtypes.Having{Expression: "total_out > 500"},
Order: []qbtypes.OrderBy{{Key: qbtypes.OrderByKey{TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{Name: "total_out"}}, Direction: qbtypes.OrderDirectionDesc}},
Limit: 3,
}, nil)
require.NoError(t, err)
assertSQLEqual(t, `
WITH __scoped_traces_total AS (
SELECT trace_id,
toString(multiIf(mapContains(attributes_string, 'gen_ai.request.model'), attributes_string['gen_ai.request.model'], NULL)) AS __GROUP_BY_KEY_0_gen_ai.request.model,
sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.output_tokens'), toFloat64(attributes_number['gen_ai.usage.output_tokens']), NULL)) AS output_tokens
FROM signoz_traces.distributed_signoz_index_v3
WHERE timestamp >= '1747947419000000000'
AND timestamp < '1747983448000000000'
AND ts_bucket_start >= 1747945619
AND ts_bucket_start <= 1747983448
AND (mapContains(attributes_string, 'gen_ai.request.model') OR mapContains(attributes_string, 'gen_ai.tool.name') OR mapContains(attributes_string, 'gen_ai.agent.name'))
GROUP BY trace_id, __GROUP_BY_KEY_0_gen_ai.request.model
),
__limit_cte AS (
SELECT __GROUP_BY_KEY_0_gen_ai.request.model, sum(output_tokens) AS __result_0
FROM __scoped_traces_total
GROUP BY __GROUP_BY_KEY_0_gen_ai.request.model
ORDER BY __result_0 desc
LIMIT 3
),
__scoped_traces AS (
SELECT trace_id,
toStartOfInterval(timestamp, INTERVAL 60 SECOND) AS ts,
toString(multiIf(mapContains(attributes_string, 'gen_ai.request.model'), attributes_string['gen_ai.request.model'], NULL)) AS __GROUP_BY_KEY_0_gen_ai.request.model,
sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.output_tokens'), toFloat64(attributes_number['gen_ai.usage.output_tokens']), NULL)) AS output_tokens
FROM signoz_traces.distributed_signoz_index_v3
WHERE timestamp >= '1747947419000000000'
AND timestamp < '1747983448000000000'
AND ts_bucket_start >= 1747945619
AND ts_bucket_start <= 1747983448
AND (mapContains(attributes_string, 'gen_ai.request.model') OR mapContains(attributes_string, 'gen_ai.tool.name') OR mapContains(attributes_string, 'gen_ai.agent.name'))
AND (toString(multiIf(mapContains(attributes_string, 'gen_ai.request.model'), attributes_string['gen_ai.request.model'], NULL))) GLOBAL IN (SELECT __GROUP_BY_KEY_0_gen_ai.request.model FROM __limit_cte)
GROUP BY trace_id, ts, __GROUP_BY_KEY_0_gen_ai.request.model
)
SELECT ts, __GROUP_BY_KEY_0_gen_ai.request.model, sum(output_tokens) AS __result_0
FROM __scoped_traces
GROUP BY ts, __GROUP_BY_KEY_0_gen_ai.request.model
HAVING __result_0 > 500
ORDER BY ts desc
SETTINGS distributed_product_mode='allow', max_memory_usage=10000000000
`, stmt)
}
// A span-level scalar delegates to the trace builder, constrained by __trace_scope;
// the shape is the delegate's own, hence no SETTINGS suffix.
func TestBuild_FullSQL_Scalar_SpanAgg_TraceScoped(t *testing.T) {
b := newTestBuilder(t)
stmt, err := b.Build(context.Background(), valuer.UUID{}, testStartMs, testEndMs, qbtypes.RequestTypeScalar,
qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Signal: telemetrytypes.SignalTraces,
Aggregations: []qbtypes.TraceAggregation{{Expression: "sum(gen_ai.usage.output_tokens)"}},
Filter: &qbtypes.Filter{Expression: "trace.output_tokens > 1000"},
}, nil)
require.NoError(t, err)
assertSQLEqual(t, `
WITH __trace_scope AS (
SELECT trace_id,
sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.output_tokens'), toFloat64(attributes_number['gen_ai.usage.output_tokens']), NULL)) AS output_tokens
FROM signoz_traces.distributed_signoz_index_v3
WHERE timestamp >= '1747947419000000000'
AND timestamp < '1747983448000000000'
AND ts_bucket_start >= 1747945619
AND ts_bucket_start <= 1747983448
AND (mapContains(attributes_string, 'gen_ai.request.model') OR mapContains(attributes_string, 'gen_ai.tool.name') OR mapContains(attributes_string, 'gen_ai.agent.name'))
GROUP BY trace_id
HAVING output_tokens > 1000
)
SELECT sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.output_tokens'), toFloat64(attributes_number['gen_ai.usage.output_tokens']), NULL)) AS __result_0
FROM signoz_traces.distributed_signoz_index_v3
WHERE trace_id GLOBAL IN (SELECT trace_id FROM __trace_scope)
AND (mapContains(attributes_string, 'gen_ai.request.model') OR mapContains(attributes_string, 'gen_ai.tool.name') OR mapContains(attributes_string, 'gen_ai.agent.name'))
AND timestamp >= '1747947419000000000'
AND timestamp < '1747983448000000000'
AND ts_bucket_start >= 1747945619
AND ts_bucket_start <= 1747983448
ORDER BY __result_0 DESC
`, stmt)
}
// Two group keys make the top-N prune a 2-tuple GLOBAL IN, and the qualification plus
// span predicate apply to the ranking scan and the main scan alike.
func TestBuild_FullSQL_TimeSeries_GroupLimit_MultiKey(t *testing.T) {
b := newTestBuilder(t)
stmt, err := b.Build(context.Background(), valuer.UUID{}, testStartMs, testEndMs, qbtypes.RequestTypeTimeSeries,
qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Signal: telemetrytypes.SignalTraces,
StepInterval: qbtypes.Step{Duration: 60 * time.Second},
Aggregations: []qbtypes.TraceAggregation{
{Expression: "sum(trace.output_tokens)"},
{Expression: "count(trace.trace_id)"},
},
Filter: &qbtypes.Filter{Expression: "gen_ai.request.model = 'gpt-4o-mini' AND trace.total_tokens > 100"},
GroupBy: []qbtypes.GroupByKey{
{TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{Name: "gen_ai.request.model"}},
{TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{Name: "gen_ai.user.id"}},
},
Limit: 2,
}, nil)
require.NoError(t, err)
assertSQLEqual(t, `
WITH __qualified AS (
SELECT trace_id,
coalesce(sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.input_tokens'), toFloat64(attributes_number['gen_ai.usage.input_tokens']), NULL)), 0) + coalesce(sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.output_tokens'), toFloat64(attributes_number['gen_ai.usage.output_tokens']), NULL)), 0) AS total_tokens
FROM signoz_traces.distributed_signoz_index_v3
WHERE timestamp >= '1747947419000000000'
AND timestamp < '1747983448000000000'
AND ts_bucket_start >= 1747945619
AND ts_bucket_start <= 1747983448
AND (mapContains(attributes_string, 'gen_ai.request.model') OR mapContains(attributes_string, 'gen_ai.tool.name') OR mapContains(attributes_string, 'gen_ai.agent.name'))
GROUP BY trace_id
HAVING total_tokens > 100
),
__scoped_traces_total AS (
SELECT trace_id,
toString(multiIf(mapContains(attributes_string, 'gen_ai.request.model'), attributes_string['gen_ai.request.model'], NULL)) AS __GROUP_BY_KEY_0_gen_ai.request.model,
toString(multiIf(mapContains(attributes_string, 'gen_ai.user.id'), attributes_string['gen_ai.user.id'], NULL)) AS __GROUP_BY_KEY_1_gen_ai.user.id,
sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.output_tokens'), toFloat64(attributes_number['gen_ai.usage.output_tokens']), NULL)) AS output_tokens
FROM signoz_traces.distributed_signoz_index_v3
WHERE timestamp >= '1747947419000000000'
AND timestamp < '1747983448000000000'
AND ts_bucket_start >= 1747945619
AND ts_bucket_start <= 1747983448
AND (mapContains(attributes_string, 'gen_ai.request.model') OR mapContains(attributes_string, 'gen_ai.tool.name') OR mapContains(attributes_string, 'gen_ai.agent.name'))
AND (attributes_string['gen_ai.request.model'] = 'gpt-4o-mini' AND mapContains(attributes_string, 'gen_ai.request.model'))
AND trace_id GLOBAL IN (SELECT trace_id FROM __qualified)
GROUP BY trace_id, __GROUP_BY_KEY_0_gen_ai.request.model, __GROUP_BY_KEY_1_gen_ai.user.id
),
__limit_cte AS (
SELECT __GROUP_BY_KEY_0_gen_ai.request.model, __GROUP_BY_KEY_1_gen_ai.user.id, sum(output_tokens) AS __result_0, count(trace_id) AS __result_1
FROM __scoped_traces_total
GROUP BY __GROUP_BY_KEY_0_gen_ai.request.model, __GROUP_BY_KEY_1_gen_ai.user.id
ORDER BY __result_0 DESC
LIMIT 2
),
__scoped_traces AS (
SELECT trace_id,
toStartOfInterval(timestamp, INTERVAL 60 SECOND) AS ts,
toString(multiIf(mapContains(attributes_string, 'gen_ai.request.model'), attributes_string['gen_ai.request.model'], NULL)) AS __GROUP_BY_KEY_0_gen_ai.request.model,
toString(multiIf(mapContains(attributes_string, 'gen_ai.user.id'), attributes_string['gen_ai.user.id'], NULL)) AS __GROUP_BY_KEY_1_gen_ai.user.id,
sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.output_tokens'), toFloat64(attributes_number['gen_ai.usage.output_tokens']), NULL)) AS output_tokens
FROM signoz_traces.distributed_signoz_index_v3
WHERE timestamp >= '1747947419000000000'
AND timestamp < '1747983448000000000'
AND ts_bucket_start >= 1747945619
AND ts_bucket_start <= 1747983448
AND (mapContains(attributes_string, 'gen_ai.request.model') OR mapContains(attributes_string, 'gen_ai.tool.name') OR mapContains(attributes_string, 'gen_ai.agent.name'))
AND (attributes_string['gen_ai.request.model'] = 'gpt-4o-mini' AND mapContains(attributes_string, 'gen_ai.request.model'))
AND trace_id GLOBAL IN (SELECT trace_id FROM __qualified)
AND (toString(multiIf(mapContains(attributes_string, 'gen_ai.request.model'), attributes_string['gen_ai.request.model'], NULL)), toString(multiIf(mapContains(attributes_string, 'gen_ai.user.id'), attributes_string['gen_ai.user.id'], NULL))) GLOBAL IN (SELECT __GROUP_BY_KEY_0_gen_ai.request.model, __GROUP_BY_KEY_1_gen_ai.user.id FROM __limit_cte)
GROUP BY trace_id, ts, __GROUP_BY_KEY_0_gen_ai.request.model, __GROUP_BY_KEY_1_gen_ai.user.id
)
SELECT ts, __GROUP_BY_KEY_0_gen_ai.request.model, __GROUP_BY_KEY_1_gen_ai.user.id, sum(output_tokens) AS __result_0, count(trace_id) AS __result_1
FROM __scoped_traces
GROUP BY ts, __GROUP_BY_KEY_0_gen_ai.request.model, __GROUP_BY_KEY_1_gen_ai.user.id
SETTINGS distributed_product_mode='allow', max_memory_usage=10000000000
`, stmt)
}
// A time-series limit without group-by has nothing to rank: it is ignored, matching
// the trace builder — the query equals its unlimited form.
func TestBuild_TimeSeries_LimitWithoutGroupByIgnored(t *testing.T) {
b := newTestBuilder(t)
build := func(limit int) *qbtypes.Statement {
stmt, err := b.Build(context.Background(), valuer.UUID{}, testStartMs, testEndMs, qbtypes.RequestTypeTimeSeries,
qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Signal: telemetrytypes.SignalTraces,
StepInterval: qbtypes.Step{Duration: 60 * time.Second},
Aggregations: []qbtypes.TraceAggregation{{Expression: "avg(trace.output_tokens)"}},
Limit: limit,
}, nil)
require.NoError(t, err)
return stmt
}
assert.Equal(t, build(0).Query, build(5).Query)
}
// ---------------------------------------------------------------------------
// Behavior / branch tests not covered by the goldens above
// ---------------------------------------------------------------------------
// Mixing span- and trace-level aggregations across one query is rejected.
func TestBuild_Aggregation_MixedDomainsRejected(t *testing.T) {
b := newTestBuilder(t)
_, err := b.Build(context.Background(), valuer.UUID{}, testStartMs, testEndMs, qbtypes.RequestTypeScalar,
qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Signal: telemetrytypes.SignalTraces,
Aggregations: []qbtypes.TraceAggregation{
{Expression: "avg(trace.output_tokens)"},
{Expression: "sum(gen_ai.usage.output_tokens)"},
},
}, nil)
require.ErrorContains(t, err, "cannot be mixed")
}
// Output-only aggregates are rejected in trace-level filters on the aggregation
// path too (the raw and trace-list paths are covered elsewhere).
func TestBuild_Aggregation_OutputOnlyFilterRejected(t *testing.T) {
b := newTestBuilder(t)
_, err := b.Build(context.Background(), valuer.UUID{}, testStartMs, testEndMs, qbtypes.RequestTypeScalar,
qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Signal: telemetrytypes.SignalTraces,
Aggregations: []qbtypes.TraceAggregation{{Expression: "count()"}},
Filter: &qbtypes.Filter{Expression: "trace.span_count > 3"},
}, nil)
require.ErrorContains(t, err, `aggregate "span_count" cannot be used`)
}
// Trace-level columns are rejected as group-by keys; order keys never reach the builder,
// since request validation only admits group keys and aggregation aliases/expressions.
func TestBuild_Aggregation_GroupByOrderValidation(t *testing.T) {
b := newTestBuilder(t)
ctx := context.Background()
_, err := b.Build(ctx, valuer.UUID{}, testStartMs, testEndMs, qbtypes.RequestTypeScalar,
qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Signal: telemetrytypes.SignalTraces,
Aggregations: []qbtypes.TraceAggregation{{Expression: "avg(trace.output_tokens)"}},
GroupBy: []qbtypes.GroupByKey{{TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{Name: "trace.llm_call_count"}}},
}, nil)
require.ErrorContains(t, err, `grouping by trace-level aggregate "trace.llm_call_count" is not supported`)
req := qbtypes.QueryRangeRequest{
Start: testStartMs,
End: testEndMs,
RequestType: qbtypes.RequestTypeScalar,
CompositeQuery: qbtypes.CompositeQuery{
Queries: []qbtypes.QueryEnvelope{{
Type: qbtypes.QueryTypeBuilderAI,
Spec: qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Name: "A",
Signal: telemetrytypes.SignalTraces,
Aggregations: []qbtypes.TraceAggregation{{Expression: "avg(trace.output_tokens)"}},
Order: []qbtypes.OrderBy{{Key: qbtypes.OrderByKey{TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{Name: "trace.total_tokens"}}, Direction: qbtypes.OrderDirectionDesc}},
},
}},
},
}
require.ErrorContains(t, req.Validate(), "invalid order by key")
_, err = b.Build(ctx, valuer.UUID{}, testStartMs, testEndMs, qbtypes.RequestTypeScalar,
qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Signal: telemetrytypes.SignalTraces,
Aggregations: []qbtypes.TraceAggregation{{Expression: "avg(trace.output_tokens)", Alias: "avg_out"}},
Order: []qbtypes.OrderBy{{Key: qbtypes.OrderByKey{TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{Name: "avg_out"}}, Direction: qbtypes.OrderDirectionAsc}},
}, nil)
require.NoError(t, err)
}
// Variables in trace-level conditions resolve as bound args; a dynamic __all__ drops the
// condition, and an unresolved $var is rejected only as an unknown aggregate today.
func TestBuild_FullSQL_Aggregation_VariablesInTraceFilter(t *testing.T) {
b := newTestBuilder(t)
ctx := context.Background()
q := qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Signal: telemetrytypes.SignalTraces,
Aggregations: []qbtypes.TraceAggregation{{Expression: "avg(trace.output_tokens)"}},
Filter: &qbtypes.Filter{Expression: "trace.output_tokens > $threshold"},
}
stmt, err := b.Build(ctx, valuer.UUID{}, testStartMs, testEndMs, qbtypes.RequestTypeScalar, q,
map[string]qbtypes.VariableItem{"threshold": {Type: qbtypes.TextBoxVariableType, Value: float64(1000)}})
require.NoError(t, err)
assertSQLEqual(t, `
WITH __qualified AS (
SELECT trace_id,
sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.output_tokens'), toFloat64(attributes_number['gen_ai.usage.output_tokens']), NULL)) AS output_tokens
FROM signoz_traces.distributed_signoz_index_v3
WHERE timestamp >= '1747947419000000000'
AND timestamp < '1747983448000000000'
AND ts_bucket_start >= 1747945619
AND ts_bucket_start <= 1747983448
AND (mapContains(attributes_string, 'gen_ai.request.model') OR mapContains(attributes_string, 'gen_ai.tool.name') OR mapContains(attributes_string, 'gen_ai.agent.name'))
GROUP BY trace_id
HAVING output_tokens > 1000
),
__scoped_traces AS (
SELECT trace_id,
sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.output_tokens'), toFloat64(attributes_number['gen_ai.usage.output_tokens']), NULL)) AS output_tokens
FROM signoz_traces.distributed_signoz_index_v3
WHERE timestamp >= '1747947419000000000'
AND timestamp < '1747983448000000000'
AND ts_bucket_start >= 1747945619
AND ts_bucket_start <= 1747983448
AND (mapContains(attributes_string, 'gen_ai.request.model') OR mapContains(attributes_string, 'gen_ai.tool.name') OR mapContains(attributes_string, 'gen_ai.agent.name'))
AND trace_id GLOBAL IN (SELECT trace_id FROM __qualified)
GROUP BY trace_id
)
SELECT avg(output_tokens) AS __result_0
FROM __scoped_traces
ORDER BY __result_0 DESC
SETTINGS distributed_product_mode='allow', max_memory_usage=10000000000
`, stmt)
// an unresolved $var is only rejected as an unknown aggregate today; a targeted
// "unknown variable" error is a separate concern
_, err = b.Build(ctx, valuer.UUID{}, testStartMs, testEndMs, qbtypes.RequestTypeScalar, q, nil)
require.ErrorContains(t, err, `aggregate "$threshold" cannot be used`)
// __all__ drops the condition: the query equals its unfiltered form
stmt, err = b.Build(ctx, valuer.UUID{}, testStartMs, testEndMs, qbtypes.RequestTypeScalar, q,
map[string]qbtypes.VariableItem{"threshold": {Type: qbtypes.DynamicVariableType, Value: "__all__"}})
require.NoError(t, err)
unfiltered := q
unfiltered.Filter = nil
want, err := b.Build(ctx, valuer.UUID{}, testStartMs, testEndMs, qbtypes.RequestTypeScalar, unfiltered, nil)
require.NoError(t, err)
assert.Equal(t, want.Query, stmt.Query)
// list variables render as IN with bound args; the scan selects only trace_id
// since no aggregation touches a per-trace column
stmt, err = b.Build(ctx, valuer.UUID{}, testStartMs, testEndMs, qbtypes.RequestTypeScalar,
qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Signal: telemetrytypes.SignalTraces,
Aggregations: []qbtypes.TraceAggregation{{Expression: "count(trace.trace_id)"}},
Filter: &qbtypes.Filter{Expression: "trace.llm_call_count IN $counts"},
}, map[string]qbtypes.VariableItem{
"counts": {Type: qbtypes.QueryVariableType, Value: []any{float64(1), float64(2)}},
})
require.NoError(t, err)
assertSQLEqual(t, `
WITH __qualified AS (
SELECT trace_id,
countIf(mapContains(attributes_string, 'gen_ai.request.model')) AS llm_call_count
FROM signoz_traces.distributed_signoz_index_v3
WHERE timestamp >= '1747947419000000000'
AND timestamp < '1747983448000000000'
AND ts_bucket_start >= 1747945619
AND ts_bucket_start <= 1747983448
AND (mapContains(attributes_string, 'gen_ai.request.model') OR mapContains(attributes_string, 'gen_ai.tool.name') OR mapContains(attributes_string, 'gen_ai.agent.name'))
GROUP BY trace_id
HAVING llm_call_count IN (1, 2)
),
__scoped_traces AS (
SELECT trace_id
FROM signoz_traces.distributed_signoz_index_v3
WHERE timestamp >= '1747947419000000000'
AND timestamp < '1747983448000000000'
AND ts_bucket_start >= 1747945619
AND ts_bucket_start <= 1747983448
AND (mapContains(attributes_string, 'gen_ai.request.model') OR mapContains(attributes_string, 'gen_ai.tool.name') OR mapContains(attributes_string, 'gen_ai.agent.name'))
AND trace_id GLOBAL IN (SELECT trace_id FROM __qualified)
GROUP BY trace_id
)
SELECT count(trace_id) AS __result_0
FROM __scoped_traces
ORDER BY __result_0 DESC
SETTINGS distributed_product_mode='allow', max_memory_usage=10000000000
`, stmt)
}
// Resource conditions on the native path: the __resource_filter CTE prunes the
// qualification scan and the per-trace scan by fingerprint.
func TestBuild_FullSQL_Aggregation_ResourceFilter_Native(t *testing.T) {
b := newTestBuilder(t)
stmt, err := b.Build(context.Background(), valuer.UUID{}, testStartMs, testEndMs, qbtypes.RequestTypeScalar,
qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Signal: telemetrytypes.SignalTraces,
Aggregations: []qbtypes.TraceAggregation{{Expression: "avg(trace.output_tokens)"}},
Filter: &qbtypes.Filter{Expression: "service.name = 'api' AND trace.output_tokens > 1000"},
}, nil)
require.NoError(t, err)
assertSQLEqual(t, `
WITH __resource_filter AS (
SELECT fingerprint
FROM signoz_traces.distributed_traces_v3_resource
WHERE (simpleJSONExtractString(labels, 'service.name') = 'api' AND labels LIKE '%service.name%' AND labels LIKE '%service.name":"api%')
AND seen_at_ts_bucket_start >= 1747945619
AND seen_at_ts_bucket_start <= 1747983448
GROUP BY fingerprint
),
__qualified AS (
SELECT trace_id,
sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.output_tokens'), toFloat64(attributes_number['gen_ai.usage.output_tokens']), NULL)) AS output_tokens
FROM signoz_traces.distributed_signoz_index_v3
WHERE timestamp >= '1747947419000000000'
AND timestamp < '1747983448000000000'
AND ts_bucket_start >= 1747945619
AND ts_bucket_start <= 1747983448
AND (mapContains(attributes_string, 'gen_ai.request.model') OR mapContains(attributes_string, 'gen_ai.tool.name') OR mapContains(attributes_string, 'gen_ai.agent.name'))
AND resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter)
GROUP BY trace_id
HAVING output_tokens > 1000
),
__scoped_traces AS (
SELECT trace_id,
sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.output_tokens'), toFloat64(attributes_number['gen_ai.usage.output_tokens']), NULL)) AS output_tokens
FROM signoz_traces.distributed_signoz_index_v3
WHERE timestamp >= '1747947419000000000'
AND timestamp < '1747983448000000000'
AND ts_bucket_start >= 1747945619
AND ts_bucket_start <= 1747983448
AND (mapContains(attributes_string, 'gen_ai.request.model') OR mapContains(attributes_string, 'gen_ai.tool.name') OR mapContains(attributes_string, 'gen_ai.agent.name'))
AND resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter)
AND trace_id GLOBAL IN (SELECT trace_id FROM __qualified)
GROUP BY trace_id
)
SELECT avg(output_tokens) AS __result_0
FROM __scoped_traces
ORDER BY __result_0 DESC
SETTINGS distributed_product_mode='allow', max_memory_usage=10000000000
`, stmt)
}
// On the delegated path __trace_scope and the main query share one __resource_filter
// CTE, so the resource table is scanned once.
func TestBuild_FullSQL_Aggregation_ResourceFilter_Delegated(t *testing.T) {
b := newTestBuilder(t)
stmt, err := b.Build(context.Background(), valuer.UUID{}, testStartMs, testEndMs, qbtypes.RequestTypeScalar,
qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Signal: telemetrytypes.SignalTraces,
Aggregations: []qbtypes.TraceAggregation{{Expression: "sum(gen_ai.usage.output_tokens)"}},
Filter: &qbtypes.Filter{Expression: "service.name = 'api' AND trace.output_tokens > 1000"},
}, nil)
require.NoError(t, err)
assertSQLEqual(t, `
WITH __resource_filter AS (
SELECT fingerprint
FROM signoz_traces.distributed_traces_v3_resource
WHERE ((simpleJSONExtractString(labels, 'service.name') = 'api' AND labels LIKE '%service.name%' AND labels LIKE '%service.name":"api%'))
AND seen_at_ts_bucket_start >= 1747945619
AND seen_at_ts_bucket_start <= 1747983448
GROUP BY fingerprint
),
__trace_scope AS (
SELECT trace_id,
sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.output_tokens'), toFloat64(attributes_number['gen_ai.usage.output_tokens']), NULL)) AS output_tokens
FROM signoz_traces.distributed_signoz_index_v3
WHERE timestamp >= '1747947419000000000'
AND timestamp < '1747983448000000000'
AND ts_bucket_start >= 1747945619
AND ts_bucket_start <= 1747983448
AND (mapContains(attributes_string, 'gen_ai.request.model') OR mapContains(attributes_string, 'gen_ai.tool.name') OR mapContains(attributes_string, 'gen_ai.agent.name'))
AND resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter)
GROUP BY trace_id
HAVING output_tokens > 1000
)
SELECT sum(multiIf(mapContains(attributes_number, 'gen_ai.usage.output_tokens'), toFloat64(attributes_number['gen_ai.usage.output_tokens']), NULL)) AS __result_0
FROM signoz_traces.distributed_signoz_index_v3
WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter)
AND trace_id GLOBAL IN (SELECT trace_id FROM __trace_scope)
AND (((mapContains(attributes_string, 'gen_ai.request.model') OR mapContains(attributes_string, 'gen_ai.tool.name') OR mapContains(attributes_string, 'gen_ai.agent.name'))) AND ((multiIf(resource.service.name IS NOT NULL, resource.service.name::String, mapContains(resources_string, 'service.name'), resources_string['service.name'], NULL) = 'api' AND multiIf(resource.service.name IS NOT NULL, resource.service.name::String, mapContains(resources_string, 'service.name'), resources_string['service.name'], NULL) IS NOT NULL)))
AND timestamp >= '1747947419000000000'
AND timestamp < '1747983448000000000'
AND ts_bucket_start >= 1747945619
AND ts_bucket_start <= 1747983448
ORDER BY __result_0 DESC
`, stmt)
}
// rate() divides by the window (scalar) / step (series). Per AggreFuncMap it counts
// per-trace rows per second; it does not sum the column.
func TestBuild_Aggregation_RateDividesByInterval(t *testing.T) {
b := newTestBuilder(t)
ctx := context.Background()
q := qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Signal: telemetrytypes.SignalTraces,
Aggregations: []qbtypes.TraceAggregation{{Expression: "rate(trace.llm_call_count)"}},
}
stmt, err := b.Build(ctx, valuer.UUID{}, testStartMs, testEndMs, qbtypes.RequestTypeScalar, q, nil)
require.NoError(t, err)
assert.Contains(t, stmt.Query, "count(llm_call_count)/36029 AS __result_0") // (end-start) seconds
q.StepInterval = qbtypes.Step{Duration: 60 * time.Second}
stmt, err = b.Build(ctx, valuer.UUID{}, testStartMs, testEndMs, qbtypes.RequestTypeTimeSeries, q, nil)
require.NoError(t, err)
assert.Contains(t, stmt.Query, "count(llm_call_count)/60 AS __result_0")
// a sub-second window clamps the divisor instead of truncating it to zero
stmt, err = b.Build(ctx, valuer.UUID{}, testStartMs, testStartMs+500, qbtypes.RequestTypeScalar, q, nil)
require.NoError(t, err)
assert.Contains(t, stmt.Query, "count(llm_call_count)/1 AS __result_0")
}

View File

@@ -428,20 +428,24 @@ func (b *StatementBuilder) buildTemporalAggDeltaFastPath(
sb.SelectMore(fmt.Sprintf("`%s`", GroupByColumnAlias(i, g.Name)))
}
aggCol, err := metricstelemetryschema.AggregationColumnForSamplesTable(
samplesTable, query.Aggregations[0].Temporality, query.Aggregations[0].TimeAggregation,
)
if err != nil {
return "", nil, err
}
if query.Aggregations[0].TimeAggregation == metrictypes.TimeAggregationRate {
// TODO(srikanthccv): should it be step interval or use [start_time_unix_nano](https://github.com/open-telemetry/opentelemetry-proto/blob/d3fb76d70deb0874692bd0ebe03148580d85f3bb/opentelemetry/proto/metrics/v1/metrics.proto#L400C11-L400C31)?
aggCol = fmt.Sprintf("%s/%d", aggCol, stepSec)
}
var aggCol string
if query.Aggregations[0].SpaceAggregation.IsPercentile() &&
query.Aggregations[0].Type == metrictypes.ExpHistogramType {
// merging sketches already spans every series in the step, so neither a
// samples-table value column nor the rate divisor applies
aggCol = fmt.Sprintf("quantilesDDMerge(0.01, %f)(sketch)[1]", query.Aggregations[0].SpaceAggregation.Percentile())
} else {
col, err := metricstelemetryschema.AggregationColumnForSamplesTable(
samplesTable, query.Aggregations[0].Temporality, query.Aggregations[0].TimeAggregation,
)
if err != nil {
return "", nil, err
}
aggCol = col
if query.Aggregations[0].TimeAggregation == metrictypes.TimeAggregationRate {
// TODO(srikanthccv): should it be step interval or use [start_time_unix_nano](https://github.com/open-telemetry/opentelemetry-proto/blob/d3fb76d70deb0874692bd0ebe03148580d85f3bb/opentelemetry/proto/metrics/v1/metrics.proto#L400C11-L400C31)?
aggCol = fmt.Sprintf("%s/%d", aggCol, stepSec)
}
}
sb.SelectMore(fmt.Sprintf("%s AS value", aggCol))

View File

@@ -126,6 +126,64 @@ func TestStatementBuilder(t *testing.T) {
},
expectedErr: nil,
},
{
name: "test_exp_histogram_percentile_delta",
requestType: qbtypes.RequestTypeTimeSeries,
query: qbtypes.QueryBuilderQuery[qbtypes.MetricAggregation]{
Signal: telemetrytypes.SignalMetrics,
StepInterval: qbtypes.Step{Duration: 30 * time.Second},
Aggregations: []qbtypes.MetricAggregation{
{
MetricName: "signoz_latency",
Type: metrictypes.ExpHistogramType,
Temporality: metrictypes.Delta,
SpaceAggregation: metrictypes.SpaceAggregationPercentile95,
},
},
GroupBy: []qbtypes.GroupByKey{
{
TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{
Name: "service.name",
},
},
},
},
expected: qbtypes.Statement{
Query: "WITH __spatial_aggregation_cte AS (SELECT toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(30)) AS ts, `__GROUP_BY_KEY_0_service.name`, quantilesDDMerge(0.01, 0.950000)(sketch)[1] AS value FROM signoz_metrics.distributed_exp_hist AS points INNER JOIN (SELECT fingerprint, JSONExtractString(labels, 'service.name') AS `__GROUP_BY_KEY_0_service.name` FROM signoz_metrics.time_series_v4_6hrs WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli <= ? AND LOWER(temporality) LIKE LOWER(?) GROUP BY fingerprint, `__GROUP_BY_KEY_0_service.name`) AS filtered_time_series ON points.fingerprint = filtered_time_series.fingerprint WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli < ? GROUP BY ts, `__GROUP_BY_KEY_0_service.name`) SELECT * FROM __spatial_aggregation_cte ORDER BY `__GROUP_BY_KEY_0_service.name`, ts",
Args: []any{"signoz_latency", uint64(1747936800000), uint64(1747983420000), "delta", "signoz_latency", uint64(1747947390000), uint64(1747983420000)},
},
expectedErr: nil,
},
{
// the sketch merge spans the whole step, so `rate` must not add a /step divisor
name: "test_exp_histogram_percentile_delta_rate_time_aggregation",
requestType: qbtypes.RequestTypeTimeSeries,
query: qbtypes.QueryBuilderQuery[qbtypes.MetricAggregation]{
Signal: telemetrytypes.SignalMetrics,
StepInterval: qbtypes.Step{Duration: 30 * time.Second},
Aggregations: []qbtypes.MetricAggregation{
{
MetricName: "signoz_latency",
Type: metrictypes.ExpHistogramType,
Temporality: metrictypes.Delta,
TimeAggregation: metrictypes.TimeAggregationRate,
SpaceAggregation: metrictypes.SpaceAggregationPercentile95,
},
},
GroupBy: []qbtypes.GroupByKey{
{
TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{
Name: "service.name",
},
},
},
},
expected: qbtypes.Statement{
Query: "WITH __spatial_aggregation_cte AS (SELECT toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(30)) AS ts, `__GROUP_BY_KEY_0_service.name`, quantilesDDMerge(0.01, 0.950000)(sketch)[1] AS value FROM signoz_metrics.distributed_exp_hist AS points INNER JOIN (SELECT fingerprint, JSONExtractString(labels, 'service.name') AS `__GROUP_BY_KEY_0_service.name` FROM signoz_metrics.time_series_v4_6hrs WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli <= ? AND LOWER(temporality) LIKE LOWER(?) GROUP BY fingerprint, `__GROUP_BY_KEY_0_service.name`) AS filtered_time_series ON points.fingerprint = filtered_time_series.fingerprint WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli < ? GROUP BY ts, `__GROUP_BY_KEY_0_service.name`) SELECT * FROM __spatial_aggregation_cte ORDER BY `__GROUP_BY_KEY_0_service.name`, ts",
Args: []any{"signoz_latency", uint64(1747936800000), uint64(1747983420000), "delta", "signoz_latency", uint64(1747947390000), uint64(1747983420000)},
},
expectedErr: nil,
},
{
name: "test_histogram_percentile1",
requestType: qbtypes.RequestTypeTimeSeries,

View File

@@ -4,7 +4,6 @@ import (
"context"
"fmt"
"log/slog"
"sort"
"strings"
"github.com/SigNoz/signoz/pkg/errors"
@@ -19,7 +18,6 @@ import (
qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
"github.com/SigNoz/signoz/pkg/valuer"
qbvariables "github.com/SigNoz/signoz/pkg/variables"
"github.com/huandu/go-sqlbuilder"
)
@@ -117,6 +115,8 @@ func (b *scopedTraceStatementBuilder) Build(
return b.buildTraceListQuery(ctx, orgID, querybuilder.ToNanoSecs(start), querybuilder.ToNanoSecs(end), query, variables)
case qbtypes.RequestTypeRaw:
return b.buildDelegated(ctx, orgID, start, end, requestType, query, variables)
case qbtypes.RequestTypeScalar, qbtypes.RequestTypeTimeSeries:
return b.buildAggregation(ctx, orgID, start, end, requestType, query, variables)
default:
return nil, ErrUnsupportedRequestType
}
@@ -145,6 +145,63 @@ func (b *scopedTraceStatementBuilder) buildDelegated(
return b.traceStmtBuilder.Build(ctx, orgID, start, end, requestType, gated, variables)
}
// traceScopedStatementBuilder is the delegate's optional capability of constraining a
// query to a set of trace ids (implemented by the traces statement builder).
// traceScopeResource is the __resource_filter CTE traceScope's predicate references,
// shared with the delegate's own resource filter so the table is scanned once.
type traceScopedStatementBuilder interface {
BuildTraceScoped(ctx context.Context, orgID valuer.UUID, start, end uint64, requestType qbtypes.RequestType, query qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation], variables map[string]qbtypes.VariableItem, traceScope, traceScopeResource *qbtypes.Statement) (*qbtypes.Statement, error)
}
// buildDelegatedAggregation serves span-level scalar/time-series through the standard
// trace builder, with the gate ANDed into the span-level filter part; a trace-level
// part becomes a qualification the delegate constrains trace_id by.
func (b *scopedTraceStatementBuilder) buildDelegatedAggregation(
ctx context.Context,
orgID valuer.UUID,
start, end uint64,
requestType qbtypes.RequestType,
query qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation],
variables map[string]qbtypes.VariableItem,
) (*qbtypes.Statement, error) {
var spanExpr, traceExpr string
var err error
if query.Filter != nil && strings.TrimSpace(query.Filter.Expression) != "" {
spanExpr, traceExpr, err = querybuilder.SplitFilterForAggregates(query.Filter.Expression, b.aggregateAliasSet())
if err != nil {
return nil, err
}
}
gate := b.scope.FilterExpression
expr := gate
if strings.TrimSpace(spanExpr) != "" {
expr = fmt.Sprintf("(%s) AND (%s)", gate, spanExpr)
}
// shallow copy; only Filter is replaced, caller's query untouched
gated := query
gated.Filter = &qbtypes.Filter{Expression: expr}
if strings.TrimSpace(traceExpr) == "" {
return b.traceStmtBuilder.Build(ctx, orgID, start, end, requestType, gated, variables)
}
scoped, ok := b.traceStmtBuilder.(traceScopedStatementBuilder)
if !ok {
return nil, errors.NewInternalf(errors.CodeInternal, "trace statement builder does not support trace-scoped queries")
}
scope, scopeResource, err := b.buildQualifiedStatement(ctx, orgID, querybuilder.ToNanoSecs(start), querybuilder.ToNanoSecs(end), traceExpr, query, variables)
if err != nil {
return nil, err
}
if scope == nil {
// every trace-level condition was dropped by variable resolution
return b.traceStmtBuilder.Build(ctx, orgID, start, end, requestType, gated, variables)
}
return scoped.BuildTraceScoped(ctx, orgID, start, end, requestType, gated, variables, scope, scopeResource)
}
// buildTraceListQuery wires the CTE pipeline (start/end are nanoseconds):
// matched (windowed, mask-pruned top-N trace_ids) → ranked (their [start,end] from
// the summary table) → buckets (ts_bucket_start prune) → enrichment (every per-trace
@@ -166,9 +223,13 @@ func (b *scopedTraceStatementBuilder) buildTraceListQuery(
limit = 100
}
filterExpr := ""
if query.Filter != nil {
filterExpr = query.Filter.Expression
}
// Condition args bind into the builder an expression is embedded in, so the
// matched and enrichment passes each resolve against their own builder.
keys, err := b.fetchKeys(ctx, orgID)
keys, err := b.fetchKeys(ctx, orgID, spanFilterSelectors(filterExpr)...)
if err != nil {
return nil, err
}
@@ -186,23 +247,17 @@ func (b *scopedTraceStatementBuilder) buildTraceListQuery(
if err != nil {
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(), keys, 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)
if err != nil {
return nil, err
}
matchedFrag, matchedArgs := b.buildMatchedCTE(matchedSB, start, end, startBucket, endBucket, resolved, orders, maskExpr, fp, resourcePred, limit, query.Offset)
rankedFrag, rankedArgs := b.buildRankedCTE(start, end)
adj := querybuilder.BucketAdjustment // 30-min bucket width in seconds
@@ -258,9 +313,10 @@ func (b *scopedTraceStatementBuilder) maybeAttachResourceFilter(
"resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter)", nil
}
func (b *scopedTraceStatementBuilder) fetchKeys(ctx context.Context, orgID valuer.UUID) (map[string][]*telemetrytypes.TelemetryFieldKey, error) {
func (b *scopedTraceStatementBuilder) fetchKeys(ctx context.Context, orgID valuer.UUID, extra ...*telemetrytypes.FieldKeySelector) (map[string][]*telemetrytypes.TelemetryFieldKey, error) {
fields := b.resolverFieldKeys()
selectors := make([]*telemetrytypes.FieldKeySelector, 0, len(fields))
selectors := make([]*telemetrytypes.FieldKeySelector, 0, len(fields)+len(extra))
selectors = append(selectors, extra...)
for _, k := range fields {
selectors = append(selectors, &telemetrytypes.FieldKeySelector{
Name: k.Name,
@@ -329,10 +385,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) {
@@ -342,7 +397,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
}
@@ -384,29 +439,30 @@ func (b *scopedTraceStatementBuilder) resolveListOrders(order []qbtypes.OrderBy,
return orders, nil
}
// filterParts is the user filter split into a span-level predicate and a trace-level
// HAVING expression.
// filterParts is the user filter split into a span-level predicate and the resolved
// trace-level HAVING (nil when there is none).
type filterParts struct {
spanPred string
hasSpanFilter bool
havingExpr string
having *traceHaving
warnings []string
warningsURL string
}
// 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) {
// splitFilter splits query.Filter into a span-level predicate and a trace-level
// HAVING (explicit query.Having ANDed on before resolution); args bind into sb.
// keys must cover the filter's span-level selectors.
func (b *scopedTraceStatementBuilder) splitFilter(ctx context.Context, orgID valuer.UUID, query qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation], classifySet map[string]struct{}, keys map[string][]*telemetrytypes.TelemetryFieldKey, start, end uint64, variables map[string]qbtypes.VariableItem, sb *sqlbuilder.SelectBuilder) (filterParts, error) {
var fp filterParts
havingExpr := ""
if query.Filter != nil && strings.TrimSpace(query.Filter.Expression) != "" {
spanExpr, traceExpr, err := querybuilder.SplitFilterForAggregates(query.Filter.Expression, classifySet)
if err != nil {
return fp, err
}
fp.havingExpr = traceExpr
havingExpr = traceExpr
if strings.TrimSpace(spanExpr) != "" {
pred, warnings, url, err := b.resolveSpanPredicate(ctx, orgID, start, end, spanExpr, variables, sb)
pred, warnings, url, err := b.resolveSpanPredicate(ctx, orgID, start, end, spanExpr, keys, variables, sb)
if err != nil {
return fp, err
}
@@ -419,37 +475,23 @@ func (b *scopedTraceStatementBuilder) splitFilter(ctx context.Context, orgID val
}
}
if query.Having != nil && strings.TrimSpace(query.Having.Expression) != "" {
if fp.havingExpr != "" {
fp.havingExpr = fmt.Sprintf("(%s) AND (%s)", fp.havingExpr, query.Having.Expression)
if havingExpr != "" {
havingExpr = fmt.Sprintf("(%s) AND (%s)", havingExpr, query.Having.Expression)
} else {
fp.havingExpr = query.Having.Expression
havingExpr = query.Having.Expression
}
}
// the HAVING is a plain text rewrite, so substitute variables here
if strings.TrimSpace(fp.havingExpr) != "" && len(variables) > 0 {
replaced, err := qbvariables.ReplaceVariablesInExpression(fp.havingExpr, variables)
if err != nil {
return fp, err
}
fp.havingExpr = replaced
}
if err := validateAggregateFilter(fp.havingExpr, filterableSet); err != nil {
having, err := b.resolveTraceHaving(ctx, havingExpr, variables, sb)
if err != nil {
return fp, err
}
fp.having = having
return fp, nil
}
// resolveSpanPredicate resolves a span-level filter expression to a bare boolean
// predicate, args bound into sb.
func (b *scopedTraceStatementBuilder) resolveSpanPredicate(ctx context.Context, orgID valuer.UUID, start, end uint64, expr string, variables map[string]qbtypes.VariableItem, sb *sqlbuilder.SelectBuilder) (string, []string, string, error) {
selectors := querybuilder.QueryStringToKeysSelectors(expr)
for i := range selectors {
selectors[i].Signal = telemetrytypes.SignalTraces
}
keys, _, err := b.metadataStore.GetKeysMulti(ctx, orgID, querybuilder.ExpandKeySelectorsForFamilies(ctx, orgID, b.fl, selectors))
if err != nil {
return "", nil, "", err
}
// predicate, args bound into sb; keys must cover the expression's selectors.
func (b *scopedTraceStatementBuilder) resolveSpanPredicate(ctx context.Context, orgID valuer.UUID, start, end uint64, expr string, keys map[string][]*telemetrytypes.TelemetryFieldKey, variables map[string]qbtypes.VariableItem, sb *sqlbuilder.SelectBuilder) (string, []string, string, error) {
prepared, err := querybuilder.PrepareWhereClause(expr, querybuilder.FilterExprVisitorOpts{
Context: ctx,
OrgID: orgID,
@@ -478,8 +520,8 @@ 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) {
needed := neededMatchedAliases(orders, fp.havingExpr, orderableSet)
func (b *scopedTraceStatementBuilder) buildMatchedCTE(sb *sqlbuilder.SelectBuilder, start, end, startBucket, endBucket uint64, resolved []resolvedColumn, orders []listOrder, maskExpr string, fp filterParts, resourcePred string, limit, offset int) (string, []any) {
needed := neededMatchedAliases(orders, fp.having)
selects := []string{"trace_id"}
for _, rc := range resolved {
if _, ok := needed[rc.alias]; !ok {
@@ -516,22 +558,8 @@ func (b *scopedTraceStatementBuilder) buildMatchedCTE(sb *sqlbuilder.SelectBuild
having = append(having, "countIf("+maskExpr+") > 0")
having = append(having, "countIf("+fp.spanPred+") > 0")
}
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[a] = quoteAlias(a)
columnMap[telemetrytypes.FieldContextTrace.StringValue()+"."+a] = quoteAlias(a)
}
hv, err := querybuilder.NewHavingExpressionRewriter().Rewrite(fp.havingExpr, columnMap)
if err != nil {
return "", nil, err
}
if hv != "" {
// escape user text so a literal $ isn't read as an arg marker; the countIf
// entries hold live $n markers and must stay unescaped
having = append(having, sqlbuilder.Escape(hv))
}
if fp.having != nil {
having = append(having, fp.having.pred)
}
if len(having) > 0 {
sb.Having(strings.Join(having, " AND "))
@@ -544,7 +572,7 @@ func (b *scopedTraceStatementBuilder) buildMatchedCTE(sb *sqlbuilder.SelectBuild
}
sql, args := sb.BuildWithFlavor(sqlbuilder.ClickHouse)
return fmt.Sprintf("matched AS (%s)", sql), args, nil
return fmt.Sprintf("matched AS (%s)", sql), args
}
// buildRankedCTE builds `ranked`: [start,end] bounds per matched trace from the
@@ -585,8 +613,9 @@ func (b *scopedTraceStatementBuilder) buildEnrichmentSelect(sb *sqlbuilder.Selec
return sb.BuildWithFlavor(sqlbuilder.ClickHouse)
}
// aggregateAliasSet is every trace-level column alias, used to classify filter keys;
// SpanLevel columns are filtered span-level, so skip them.
// aggregateAliasSet recognises trace-level keys — display-only aliases included, so one
// gets a targeted error instead of falling through as a span attribute (what a predicate
// may actually use is filterableColumnSet). SpanLevel columns are filtered span-level.
func (b *scopedTraceStatementBuilder) aggregateAliasSet() map[string]struct{} {
set := make(map[string]struct{}, len(b.scope.Columns))
for _, c := range b.scope.Columns {
@@ -597,70 +626,36 @@ func (b *scopedTraceStatementBuilder) aggregateAliasSet() map[string]struct{} {
return set
}
// orderableAliasSet is the subset of aliases computable in the matched pass.
func orderableAliasSet(resolved []resolvedColumn) map[string]struct{} {
set := make(map[string]struct{})
for _, rc := range resolved {
if rc.orderable {
set[rc.alias] = struct{}{}
}
}
return set
}
// 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{} {
// in ORDER BY plus those the resolved trace-level HAVING touches.
func neededMatchedAliases(orders []listOrder, having *traceHaving) map[string]struct{} {
needed := make(map[string]struct{})
for _, o := range orders {
needed[o.alias] = struct{}{}
}
for _, name := range traceAggregateNames(havingExpr) {
if _, ok := orderableSet[name]; ok {
if having != nil {
for name := range having.used {
needed[name] = struct{}{}
}
}
return needed
}
// traceAggregateNames extracts the aggregate names a trace-level HAVING references;
// only unspecified- and trace-context selectors name aggregates.
func traceAggregateNames(havingExpr string) []string {
var names []string
for _, sel := range querybuilder.QueryStringToKeysSelectors(havingExpr) {
if sel.FieldContext == telemetrytypes.FieldContextUnspecified || sel.FieldContext == telemetrytypes.FieldContextTrace {
names = append(names, sel.Name)
}
}
return names
}
// validateAggregateFilter rejects a trace-level filter referencing an aggregate that
// is not filterable.
// validateAggregateFilter rejects filters on aggregates that are not filterable
// (e.g. span_count) upfront, since inside the where-clause visitor the error would
// surface only as a detail of a combined one. Only unspecified- and trace-context
// selectors name aggregates.
func validateAggregateFilter(havingExpr string, filterableSet map[string]struct{}) error {
if strings.TrimSpace(havingExpr) == "" {
return nil
}
allowed := make([]string, 0, len(filterableSet))
for a := range filterableSet {
allowed = append(allowed, a)
}
sort.Strings(allowed)
for _, name := range traceAggregateNames(havingExpr) {
if _, ok := filterableSet[name]; !ok {
for _, sel := range querybuilder.QueryStringToKeysSelectors(havingExpr) {
if sel.FieldContext != telemetrytypes.FieldContextUnspecified && sel.FieldContext != telemetrytypes.FieldContextTrace {
continue
}
if _, ok := filterableSet[sel.Name]; !ok {
return errors.NewInvalidInputf(errors.CodeInvalidInput,
"aggregate %q cannot be used in the trace-list filter; filterable aggregates: %s", name, strings.Join(allowed, ", "))
"aggregate %q cannot be used in a trace-level filter; filterable aggregates: %s", sel.Name, strings.Join(sortedAliases(filterableSet), ", "))
}
}
return nil
@@ -675,6 +670,19 @@ func orderClause(orders []listOrder) []string {
return append(out, "trace_id DESC")
}
// spanFilterSelectors are the metadata selectors for every key a filter expression
// references, for batching into a single GetKeysMulti fetch.
func spanFilterSelectors(expr string) []*telemetrytypes.FieldKeySelector {
if strings.TrimSpace(expr) == "" {
return nil
}
selectors := querybuilder.QueryStringToKeysSelectors(expr)
for i := range selectors {
selectors[i].Signal = telemetrytypes.SignalTraces
}
return selectors
}
// quoteAlias backticks an alias containing characters special to the SQL builder.
func quoteAlias(alias string) string {
if strings.ContainsAny(alias, ".$`") {

View File

@@ -0,0 +1,783 @@
package scopedtracesstatementbuilder
import (
"context"
"fmt"
"sort"
"strings"
chparser "github.com/AfterShip/clickhouse-sql-parser/parser"
"github.com/SigNoz/signoz/pkg/errors"
"github.com/SigNoz/signoz/pkg/querybuilder"
"github.com/SigNoz/signoz/pkg/telemetryschema/tracestelemetryschema"
qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
"github.com/SigNoz/signoz/pkg/valuer"
"github.com/huandu/go-sqlbuilder"
)
// The per-trace values these aggregations read are window-clipped and span-filtered,
// unlike the list's enrichment pass over every span of the whole trace, so the same
// column reads differently in each.
// traceAggregation is one aggregation rewritten to run over the per-trace scan.
type traceAggregation struct {
expr string // rewritten SQL over the per-trace column aliases
used map[string]struct{} // per-trace aliases referenced
isRate bool
}
// buildAggregation routes by aggregation domain: bare keys delegate to the standard
// trace builder, trace.-prefixed aggregates run over the per-trace scan.
func (b *scopedTraceStatementBuilder) buildAggregation(
ctx context.Context,
orgID valuer.UUID,
start, end uint64,
requestType qbtypes.RequestType,
query qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation],
variables map[string]qbtypes.VariableItem,
) (*qbtypes.Statement, error) {
traceAggs, err := b.classifyAggregations(query.Aggregations)
if err != nil {
return nil, err
}
if err := b.validateGroupBy(query); err != nil {
return nil, err
}
if len(traceAggs) == 0 {
return b.buildDelegatedAggregation(ctx, orgID, start, end, requestType, query, variables)
}
return b.buildTraceAggregationQuery(ctx, orgID, querybuilder.ToNanoSecs(start), querybuilder.ToNanoSecs(end), requestType, query, variables, traceAggs)
}
// classifyAggregations returns the rewritten trace-domain aggregations, nil when all
// are span-domain; mixing the two domains is rejected.
func (b *scopedTraceStatementBuilder) classifyAggregations(aggs []qbtypes.TraceAggregation) ([]traceAggregation, error) {
// permission, not recognition: unknown names are reported against exactly this set
traceCols := b.orderableColumnSet()
var out []traceAggregation
spanCount := 0
for _, agg := range aggs {
ta, isTrace, err := rewriteTraceAggregation(agg.Expression, traceCols)
if err != nil {
return nil, err
}
if isTrace {
out = append(out, *ta)
} else {
spanCount++
}
}
if len(out) > 0 && spanCount > 0 {
return nil, errors.NewInvalidInputf(errors.CodeInvalidInput,
"span-level and trace-level (trace.) aggregations cannot be mixed in one query")
}
return out, nil
}
// orderableColumnSet is what a trace-level aggregation may use;
// recognising a key as trace-level is aggregateAliasSet's job.
func (b *scopedTraceStatementBuilder) orderableColumnSet() map[string]struct{} {
set := make(map[string]struct{})
for _, c := range b.scope.Columns {
if c.Orderable {
set[c.Alias] = struct{}{}
}
}
return set
}
// filterableColumnSet is what a trace-level filter predicate may use.
func (b *scopedTraceStatementBuilder) filterableColumnSet() map[string]struct{} {
set := make(map[string]struct{})
for _, c := range b.scope.Columns {
if c.Filterable {
set[c.Alias] = struct{}{}
}
}
return set
}
// validateGroupBy rejects trace-level columns as group-by keys with a targeted error
// (not the field mapper's generic "field not found"). Order keys need no check here:
// request validation only admits group keys and aggregation aliases/expressions.
func (b *scopedTraceStatementBuilder) validateGroupBy(query qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]) error {
// recognition, not permission: a display-only alias must be named here to be rejected
// rather than reaching the field mapper as a span attribute
aliases := b.aggregateAliasSet()
for _, gb := range query.GroupBy {
key := gb.TelemetryFieldKey
key.Normalize()
// a bare name may be a span column sharing the alias (duration_nano, timestamp)
if key.FieldContext != telemetrytypes.FieldContextTrace {
continue
}
if _, ok := aliases[key.Name]; ok {
return errors.NewInvalidInputf(errors.CodeInvalidInput,
"grouping by trace-level aggregate %q is not supported; group by span attributes instead (e.g. service.name)", gb.Name)
}
}
return nil
}
// rewriteTraceAggregation rewrites an aggregation over trace.-prefixed columns to run
// on the per-trace scan (trace.output_tokens → output_tokens, functions mapped via
// AggreFuncMap); a pure span-level expression returns isTrace=false for the delegate.
func rewriteTraceAggregation(expr string, traceCols map[string]struct{}) (*traceAggregation, bool, error) {
p := chparser.NewParser("SELECT " + expr)
stmts, err := p.ParseStmts()
if err != nil {
return nil, false, errors.WrapInvalidInputf(err, errors.CodeInvalidInput, "failed to parse aggregation expression %q", expr)
}
if len(stmts) == 0 {
return nil, false, errors.NewInvalidInputf(errors.CodeInvalidInput, "invalid aggregation expression %q", expr)
}
sel, ok := stmts[0].(*chparser.SelectQuery)
if !ok || len(sel.SelectItems) == 0 {
return nil, false, errors.NewInvalidInputf(errors.CodeInvalidInput, "invalid aggregation expression %q", expr)
}
v := &traceAggVisitor{traceCols: traceCols, used: make(map[string]struct{})}
if err := sel.SelectItems[0].Accept(v); err != nil {
return nil, false, err
}
if !v.hasTrace {
return nil, false, nil
}
if v.hasSpan {
return nil, false, errors.NewInvalidInputf(errors.CodeInvalidInput,
"aggregation %q mixes trace-level (trace.) and span-level columns; use one domain per aggregation", expr)
}
// the interval divides the rendered expression as a whole, so a second aggregation
// alongside the rate would be divided too
if v.isRate && v.aggCount > 1 {
return nil, false, errors.NewInvalidInputf(errors.CodeInvalidInput,
"aggregation %q combines a rate with another aggregation; the rate interval would divide both, so give each its own aggregation", expr)
}
return &traceAggregation{expr: chparser.Format(sel.SelectItems[0]), used: v.used, isRate: v.isRate}, true, nil
}
// traceAggVisitor classifies column references and rewrites trace.-prefixed ones in
// place; the ancestor stack tells a column identifier from a path segment, function
// name, or alias, and rejects trace. columns inside *If combinators.
type traceAggVisitor struct {
chparser.DefaultASTVisitor
traceCols map[string]struct{}
used map[string]struct{}
stack []chparser.Expr
aggCount int
hasTrace bool
hasSpan bool
isRate bool
}
func (v *traceAggVisitor) Enter(expr chparser.Expr) { v.stack = append(v.stack, expr) }
func (v *traceAggVisitor) Leave(expr chparser.Expr) { v.stack = v.stack[:len(v.stack)-1] }
// parent is the node enclosing the one currently being visited (the visited node
// itself is the stack top).
func (v *traceAggVisitor) parent() chparser.Expr {
if len(v.stack) < 2 {
return nil
}
return v.stack[len(v.stack)-2]
}
func (v *traceAggVisitor) enclosingCombinator() (string, bool) {
for _, e := range v.stack {
fn, ok := e.(*chparser.FunctionExpr)
if !ok {
continue
}
if agg, known := querybuilder.AggreFuncMap[valuer.NewString(strings.ToLower(fn.Name.Name))]; known && agg.FuncCombinator {
return fn.Name.Name, true
}
}
return "", false
}
// enclosingAggregate walks the ancestor stack; AggreFuncMap holds only aggregates and
// VisitFunctionExpr rejects any name missing from it, so a known name is enough.
func (v *traceAggVisitor) enclosingAggregate() bool {
for _, e := range v.stack {
fn, ok := e.(*chparser.FunctionExpr)
if !ok {
continue
}
if _, known := querybuilder.AggreFuncMap[valuer.NewString(strings.ToLower(fn.Name.Name))]; known {
return true
}
}
return false
}
// VisitPath classifies a dotted reference (trace.output_tokens); trace-level ones are
// rewritten in place to the bare per-trace alias.
func (v *traceAggVisitor) VisitPath(p *chparser.Path) error {
col, isTrace := traceColumnFromPath(p)
if !isTrace {
v.hasSpan = true
return nil
}
if err := v.acceptTraceColumn(chparser.Format(p), col); err != nil {
return err
}
p.Fields = p.Fields[len(p.Fields)-1:]
p.Fields[0].Name = col
return nil
}
// VisitIdent classifies a plain identifier (a backquoted `trace.output_tokens` is
// trace-level); path segments, function names, and aliases are structural, not columns.
func (v *traceAggVisitor) VisitIdent(i *chparser.Ident) error {
switch parent := v.parent().(type) {
case *chparser.Path:
return nil // segments are classified whole by VisitPath
case *chparser.FunctionExpr:
if parent.Name == i {
return nil
}
case *chparser.ColumnExpr:
if parent.Alias == i {
return nil
}
}
key := telemetrytypes.GetFieldKeyFromKeyText(i.Name)
if key.FieldContext != telemetrytypes.FieldContextTrace || key.Name == "" {
v.hasSpan = true
return nil
}
if err := v.acceptTraceColumn(i.Name, key.Name); err != nil {
return err
}
i.Name = key.Name
return nil
}
func (v *traceAggVisitor) acceptTraceColumn(ref, col string) error {
if name, in := v.enclosingCombinator(); in {
return errors.NewInvalidInputf(errors.CodeInvalidInput,
"%q over trace-level (trace.) columns is not supported; put the trace-level condition in the filter expression instead", name)
}
// trace_id is always selected by the per-trace scan (count(trace.trace_id)
// counts traces); everything else must be a scope column.
if col != "trace_id" {
if _, known := v.traceCols[col]; !known {
return errors.NewInvalidInputf(errors.CodeInvalidInput,
"unknown trace-level aggregation column %q; usable columns: %s", ref, strings.Join(sortedAliases(v.traceCols), ", "))
}
v.used[col] = struct{}{}
}
// ungrouped, a bare per-trace column would make the outer SELECT emit one row per
// trace instead of one aggregated row
if !v.enclosingAggregate() {
return errors.NewInvalidInputf(errors.CodeInvalidInput,
"trace-level column %q must be inside an aggregation function (e.g. avg(%s))", ref, ref)
}
v.hasTrace = true
return nil
}
// VisitFunctionExpr validates and maps the function name. Children were already
// visited (post-order), so classification is complete for this subtree.
func (v *traceAggVisitor) VisitFunctionExpr(fn *chparser.FunctionExpr) error {
name := strings.ToLower(fn.Name.Name)
aggFunc, ok := querybuilder.AggreFuncMap[valuer.NewString(name)]
if !ok {
return errors.NewInvalidInputf(errors.CodeInvalidInput, "unrecognized function: %s", name)
}
if fn.Params != nil && fn.Params.Items != nil && len(fn.Params.Items.Items) > 0 && aggFunc.FuncCombinator {
// combinator predicates over span columns stay span-level (countIf(has_error=true))
v.hasSpan = true
return nil
}
fn.Name.Name = aggFunc.FuncName
v.aggCount++
if aggFunc.Rate {
v.isRate = true
}
return nil
}
// traceColumnFromPath returns the per-trace column a dotted reference names
// (trace.output_tokens -> output_tokens, trace.a.b -> a.b).
func traceColumnFromPath(p *chparser.Path) (string, bool) {
key := telemetrytypes.GetFieldKeyFromKeyText(chparser.Format(p))
if key.FieldContext != telemetrytypes.FieldContextTrace || key.Name == "" {
return "", false
}
return key.Name, true
}
func sortedAliases(set map[string]struct{}) []string {
out := make([]string, 0, len(set))
for a := range set {
out = append(out, a)
}
sort.Strings(out)
return out
}
// ---------------------------------------------------------------------------
// Qualification + per-trace scan
// ---------------------------------------------------------------------------
// buildQualifiedStatement selects the trace ids whose window-clipped aggregates satisfy
// the trace-level filter. The second statement (nil without resource conditions) is the
// __resource_filter CTE the scope's predicate references; the embedder emits it exactly
// once, shared with its own resource filter. start/end are ns; both statements are nil
// when variable resolution dropped every condition.
func (b *scopedTraceStatementBuilder) buildQualifiedStatement(
ctx context.Context,
orgID valuer.UUID,
start, end uint64,
traceExpr string,
query qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation],
variables map[string]qbtypes.VariableItem,
) (*qbtypes.Statement, *qbtypes.Statement, error) {
keys, err := b.fetchKeys(ctx, orgID)
if err != nil {
return nil, nil, err
}
sb := sqlbuilder.NewSelectBuilder()
maskExpr, resolved, err := b.resolveFor(ctx, orgID, start, end, keys, sb)
if err != nil {
return nil, nil, err
}
having, err := b.resolveTraceHaving(ctx, traceExpr, variables, sb)
if err != nil {
return nil, nil, err
}
if having == nil {
return nil, nil, nil
}
// nil when the filter has no resource-attribute conditions
resourceStmt, err := b.resourceFilterStmtBuilder.Build(ctx, orgID, start, end, qbtypes.RequestTypeRaw, query, variables)
if err != nil {
return nil, nil, err
}
var resourcePred string
if resourceStmt != nil {
resourcePred = "resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter)"
}
sql, args := b.buildPerTraceScan(sb, start, end, resolved, maskExpr, perTraceScanOpts{
needed: having.used,
havingPred: having.pred,
resourcePred: resourcePred,
})
return &qbtypes.Statement{Query: sql, Args: args}, resourceStmt, nil
}
// groupColumn holds a resolved, arg-free span-attribute expression.
type groupColumn struct {
alias string
expr string
}
// groupByColumnAlias prefixes the i-th group-by dimension so the alias cannot shadow the
// span column its expression reads; the querier (stripKeyAlias) strips it back off.
func groupByColumnAlias(i int, name string) string {
return fmt.Sprintf("__GROUP_BY_KEY_%d_%s", i, name)
}
// orderColumn is the SQL identifier a non-aggregation order key sorts by: the
// positional alias when the key names a group-by dimension, else the key itself.
func orderColumn(orderKey string, groupBy []qbtypes.GroupByKey) string {
for i := range groupBy {
if groupBy[i].Name == orderKey {
return groupByColumnAlias(i, groupBy[i].Name)
}
}
return orderKey
}
// perTraceScanOpts parametrize one windowed, mask-pruned GROUP BY trace_id scan.
// All expressions are already resolved against the scan's builder.
type perTraceScanOpts struct {
stepSeconds int64 // >0 → bucket per-trace values by time (ts column)
groupCols []groupColumn
needed map[string]struct{} // per-trace aliases to select
spanPred string // resolved span-level filter, ANDed per span
resourcePred string // resource-fingerprint prune (CTE reference or inline subquery)
qualified bool // constrain to __qualified
limitPred string // top-N group prune (GLOBAL IN __limit_cte)
havingPred string // resolved HAVING predicate over the selected aliases
}
func (b *scopedTraceStatementBuilder) buildPerTraceScan(sb *sqlbuilder.SelectBuilder, start, end uint64, resolved []resolvedColumn, maskExpr string, o perTraceScanOpts) (string, []any) {
startBucket := start/querybuilder.NsToSeconds - querybuilder.BucketAdjustment
endBucket := end / querybuilder.NsToSeconds
selects := []string{"trace_id"}
if o.stepSeconds > 0 {
selects = append(selects, fmt.Sprintf("toStartOfInterval(timestamp, INTERVAL %d SECOND) AS ts", o.stepSeconds))
}
for _, gc := range o.groupCols {
selects = append(selects, fmt.Sprintf("toString(%s) AS `%s`", gc.expr, gc.alias))
}
for _, rc := range resolved {
if _, ok := o.needed[rc.alias]; !ok {
continue
}
selects = append(selects, rc.expr+" AS "+quoteAlias(rc.alias))
}
sb.Select(selects...)
sb.From(fmt.Sprintf("%s.%s", tracestelemetryschema.DBName, tracestelemetryschema.SpanIndexV3TableName))
where := []string{
sb.GE("timestamp", fmt.Sprintf("%d", start)),
sb.L("timestamp", fmt.Sprintf("%d", end)),
sb.GE("ts_bucket_start", startBucket),
sb.LE("ts_bucket_start", endBucket),
maskExpr,
}
if strings.TrimSpace(o.spanPred) != "" {
where = append(where, o.spanPred)
}
if o.resourcePred != "" {
where = append(where, o.resourcePred)
}
if o.qualified {
where = append(where, "trace_id GLOBAL IN (SELECT trace_id FROM __qualified)")
}
if o.limitPred != "" {
where = append(where, o.limitPred)
}
sb.Where(where...)
groupBy := []string{"trace_id"}
if o.stepSeconds > 0 {
groupBy = append(groupBy, "ts")
}
for _, gc := range o.groupCols {
groupBy = append(groupBy, "`"+gc.alias+"`")
}
sb.GroupBy(groupBy...)
if strings.TrimSpace(o.havingPred) != "" {
sb.Having(o.havingPred)
}
return sb.BuildWithFlavor(sqlbuilder.ClickHouse)
}
// groupBySelectors are the metadata selectors for the group-by keys, for batching
// into a single GetKeysMulti fetch.
func groupBySelectors(groupBy []qbtypes.GroupByKey) []*telemetrytypes.FieldKeySelector {
selectors := make([]*telemetrytypes.FieldKeySelector, 0, len(groupBy))
for i := range groupBy {
selectors = append(selectors, &telemetrytypes.FieldKeySelector{
Name: groupBy[i].Name,
Signal: telemetrytypes.SignalTraces,
FieldContext: groupBy[i].FieldContext,
FieldDataType: groupBy[i].FieldDataType,
SelectorMatchType: telemetrytypes.FieldSelectorMatchTypeExact,
})
}
return selectors
}
// resolveGroupColumns resolves group-by keys through the field mapper for selection
// inside the per-trace scan; keys must cover the group-by selectors.
func (b *scopedTraceStatementBuilder) resolveGroupColumns(ctx context.Context, orgID valuer.UUID, start, end uint64, groupBy []qbtypes.GroupByKey, keys map[string][]*telemetrytypes.TelemetryFieldKey) ([]groupColumn, error) {
if len(groupBy) == 0 {
return nil, nil
}
out := make([]groupColumn, 0, len(groupBy))
for i := range groupBy {
expr, err := b.fm.ColumnExpressionFor(ctx, orgID, start, end, &groupBy[i].TelemetryFieldKey, telemetrytypes.FieldDataTypeString, keys)
if err != nil {
return nil, err
}
out = append(out, groupColumn{alias: groupByColumnAlias(i, groupBy[i].Name), expr: sqlbuilder.Escape(expr)})
}
return out, nil
}
// ---------------------------------------------------------------------------
// Native trace-domain aggregation query
// ---------------------------------------------------------------------------
// scanContext is one per-scan resolution: a fresh builder with the mask, columns,
// span predicate, and optionally the trace-level HAVING resolved against it.
type scanContext struct {
sb *sqlbuilder.SelectBuilder
maskExpr string
resolved []resolvedColumn
spanPred string
having *traceHaving
warnings []string
warnURL string
}
func (b *scopedTraceStatementBuilder) newScanContext(
ctx context.Context,
orgID valuer.UUID,
start, end uint64,
keys map[string][]*telemetrytypes.TelemetryFieldKey,
spanExpr, traceExpr string,
variables map[string]qbtypes.VariableItem,
) (*scanContext, error) {
sc := &scanContext{sb: sqlbuilder.NewSelectBuilder()}
var err error
sc.maskExpr, sc.resolved, err = b.resolveFor(ctx, orgID, start, end, keys, sc.sb)
if err != nil {
return nil, err
}
if strings.TrimSpace(spanExpr) != "" {
pred, warns, url, err := b.resolveSpanPredicate(ctx, orgID, start, end, spanExpr, keys, variables, sc.sb)
if err != nil {
return nil, err
}
sc.spanPred, sc.warnings, sc.warnURL = pred, warns, url
}
if strings.TrimSpace(traceExpr) != "" {
sc.having, err = b.resolveTraceHaving(ctx, traceExpr, variables, sc.sb)
if err != nil {
return nil, err
}
}
return sc, nil
}
// buildTraceAggregationQuery aggregates over the per-trace scan: __qualified (when the
// filter has a trace-level part) → __scoped_traces → outer aggregation. start/end are ns.
func (b *scopedTraceStatementBuilder) buildTraceAggregationQuery(
ctx context.Context,
orgID valuer.UUID,
start, end uint64,
requestType qbtypes.RequestType,
query qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation],
variables map[string]qbtypes.VariableItem,
traceAggs []traceAggregation,
) (*qbtypes.Statement, error) {
var spanExpr, traceExpr string
var err error
if query.Filter != nil && strings.TrimSpace(query.Filter.Expression) != "" {
// the broad set so a condition on a display-only alias still lands in the
// trace-level part, where resolveTraceHaving rejects it by name
spanExpr, traceExpr, err = querybuilder.SplitFilterForAggregates(query.Filter.Expression, b.aggregateAliasSet())
if err != nil {
return nil, err
}
}
keys, err := b.fetchKeys(ctx, orgID, append(spanFilterSelectors(spanExpr), groupBySelectors(query.GroupBy)...)...)
if err != nil {
return nil, err
}
resourceFrag, resourceArgs, resourcePred, err := b.maybeAttachResourceFilter(ctx, orgID, query, start, end, variables)
if err != nil {
return nil, err
}
var cteFragments []string
var cteArgs [][]any
if resourceFrag != "" {
cteFragments = append(cteFragments, resourceFrag)
cteArgs = append(cteArgs, resourceArgs)
}
// __qualified: its own scan resolution, HAVING = the trace-level filter part
qualified := false
if strings.TrimSpace(traceExpr) != "" {
qsc, err := b.newScanContext(ctx, orgID, start, end, keys, "", traceExpr, variables)
if err != nil {
return nil, err
}
if qsc.having != nil {
qsql, qargs := b.buildPerTraceScan(qsc.sb, start, end, qsc.resolved, qsc.maskExpr, perTraceScanOpts{
needed: qsc.having.used,
havingPred: qsc.having.pred,
resourcePred: resourcePred,
})
cteFragments = append(cteFragments, fmt.Sprintf("__qualified AS (%s)", qsql))
cteArgs = append(cteArgs, qargs)
qualified = true
}
}
groupCols, err := b.resolveGroupColumns(ctx, orgID, start, end, query.GroupBy, keys)
if err != nil {
return nil, err
}
groupNames := make([]string, 0, len(groupCols))
for _, gc := range groupCols {
groupNames = append(groupNames, "`"+gc.alias+"`")
}
needed := make(map[string]struct{})
for _, ta := range traceAggs {
for a := range ta.used {
needed[a] = struct{}{}
}
}
// a window or step under one second would truncate to a zero divisor
windowSeconds := max((end-start)/querybuilder.NsToSeconds, 1)
stepSeconds := int64(0)
rateInterval := windowSeconds
if requestType == qbtypes.RequestTypeTimeSeries {
stepSeconds = int64(query.StepInterval.Seconds())
rateInterval = max(uint64(stepSeconds), 1)
}
// outer aggregation over the per-trace rows
sb := sqlbuilder.NewSelectBuilder()
selects := []string{}
if stepSeconds > 0 {
selects = append(selects, "ts")
}
selects = append(selects, groupNames...)
for i, ta := range traceAggs {
selects = append(selects, fmt.Sprintf("%s AS __result_%d", ta.rendered(rateInterval), i))
}
sb.Select(selects...)
sb.From("__scoped_traces")
// grouped, limited time series → rank groups on whole-window per-trace values
// (exact for non-composable aggregates) and prune the main scan to the top-N.
limitPred := ""
if requestType == qbtypes.RequestTypeTimeSeries && query.Limit > 0 && len(groupCols) > 0 {
tsc, err := b.newScanContext(ctx, orgID, start, end, keys, spanExpr, "", variables)
if err != nil {
return nil, err
}
totalSQL, totalArgs := b.buildPerTraceScan(tsc.sb, start, end, tsc.resolved, tsc.maskExpr, perTraceScanOpts{
groupCols: groupCols,
needed: needed,
spanPred: tsc.spanPred,
resourcePred: resourcePred,
qualified: qualified,
})
cteFragments = append(cteFragments, fmt.Sprintf("__scoped_traces_total AS (%s)", totalSQL))
cteArgs = append(cteArgs, totalArgs)
limitSQL, limitArgs := outerLimitSQL(query, traceAggs, groupNames, windowSeconds)
cteFragments = append(cteFragments, fmt.Sprintf("__limit_cte AS (%s)", limitSQL))
cteArgs = append(cteArgs, limitArgs)
exprs := make([]string, 0, len(groupCols))
for _, gc := range groupCols {
exprs = append(exprs, "toString("+gc.expr+")")
}
limitPred = fmt.Sprintf("(%s) GLOBAL IN (SELECT %s FROM __limit_cte)",
strings.Join(exprs, ", "), strings.Join(groupNames, ", "))
}
msc, err := b.newScanContext(ctx, orgID, start, end, keys, spanExpr, "", variables)
if err != nil {
return nil, err
}
perTraceSQL, perTraceArgs := b.buildPerTraceScan(msc.sb, start, end, msc.resolved, msc.maskExpr, perTraceScanOpts{
stepSeconds: stepSeconds,
groupCols: groupCols,
needed: needed,
spanPred: msc.spanPred,
resourcePred: resourcePred,
qualified: qualified,
limitPred: limitPred,
})
cteFragments = append(cteFragments, fmt.Sprintf("__scoped_traces AS (%s)", perTraceSQL))
cteArgs = append(cteArgs, perTraceArgs)
groupBys := []string{}
if stepSeconds > 0 {
groupBys = append(groupBys, "ts")
}
groupBys = append(groupBys, groupNames...)
if len(groupBys) > 0 {
sb.GroupBy(groupBys...)
}
if query.Having != nil && strings.TrimSpace(query.Having.Expression) != "" {
rewritten, err := querybuilder.NewHavingExpressionRewriter().RewriteForTraces(query.Having.Expression, query.Aggregations)
if err != nil {
return nil, err
}
sb.Having(sqlbuilder.Escape(rewritten))
}
if requestType == qbtypes.RequestTypeTimeSeries {
if len(query.Order) != 0 {
for _, orderBy := range query.Order {
if _, ok := traceAggOrderIndex(orderBy, query); !ok {
sb.OrderBy(fmt.Sprintf("`%s` %s", orderColumn(orderBy.Key.Name, query.GroupBy), orderBy.Direction.StringValue()))
}
}
sb.OrderBy("ts desc")
}
} else {
for _, orderBy := range query.Order {
if idx, ok := traceAggOrderIndex(orderBy, query); ok {
sb.OrderBy(fmt.Sprintf("__result_%d %s", idx, orderBy.Direction.StringValue()))
} else {
sb.OrderBy(fmt.Sprintf("`%s` %s", orderColumn(orderBy.Key.Name, query.GroupBy), orderBy.Direction.StringValue()))
}
}
if len(query.Order) == 0 {
sb.OrderBy("__result_0 DESC")
}
if query.Limit > 0 {
sb.Limit(query.Limit)
}
}
mainSQL, mainArgs := sb.BuildWithFlavor(sqlbuilder.ClickHouse)
finalSQL := querybuilder.CombineCTEs(cteFragments) + mainSQL + " SETTINGS distributed_product_mode='allow', max_memory_usage=10000000000"
finalArgs := querybuilder.PrependArgs(cteArgs, mainArgs)
return &qbtypes.Statement{
Query: finalSQL,
Args: finalArgs,
Warnings: msc.warnings,
WarningsDocURL: msc.warnURL,
}, nil
}
// rendered divides a rate aggregation by the interval (step for time series, window
// length for scalar); the divisor applies to the whole expression, which holds only
// because a rate must be the sole aggregation.
func (ta traceAggregation) rendered(rateInterval uint64) string {
if ta.isRate {
return fmt.Sprintf("%s/%d", ta.expr, rateInterval)
}
return ta.expr
}
// outerLimitSQL ranks groups on whole-window per-trace values, so a non-composable
// aggregate (avg) ranks exactly rather than over bucketed rows.
func outerLimitSQL(query qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation], traceAggs []traceAggregation, groupNames []string, windowSeconds uint64) (string, []any) {
sb := sqlbuilder.NewSelectBuilder()
selects := append([]string{}, groupNames...)
for i, ta := range traceAggs {
selects = append(selects, fmt.Sprintf("%s AS __result_%d", ta.rendered(windowSeconds), i))
}
sb.Select(selects...)
sb.From("__scoped_traces_total")
sb.GroupBy(groupNames...)
for _, orderBy := range query.Order {
if idx, ok := traceAggOrderIndex(orderBy, query); ok {
sb.OrderBy(fmt.Sprintf("__result_%d %s", idx, orderBy.Direction.StringValue()))
} else {
sb.OrderBy(fmt.Sprintf("`%s` %s", orderColumn(orderBy.Key.Name, query.GroupBy), orderBy.Direction.StringValue()))
}
}
if len(query.Order) == 0 {
sb.OrderBy("__result_0 DESC")
}
sb.Limit(query.Limit)
return sb.BuildWithFlavor(sqlbuilder.ClickHouse)
}
// traceAggOrderIndex reports whether an order key refers to the i-th aggregation
// (by alias, expression, or index), mirroring the trace builder.
func traceAggOrderIndex(k qbtypes.OrderBy, q qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]) (int, bool) {
for i, agg := range q.Aggregations {
if k.Key.Name == agg.Alias ||
k.Key.Name == agg.Expression ||
k.Key.Name == fmt.Sprintf("%d", i) {
return i, true
}
}
return 0, false
}

View File

@@ -0,0 +1,75 @@
package scopedtracesstatementbuilder
import (
"testing"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
func TestRewriteTraceAggregation(t *testing.T) {
cols := map[string]struct{}{
"input_tokens": {}, "output_tokens": {}, "total_tokens": {}, "llm_call_count": {}, "max_llm_latency_ns": {},
}
cases := []struct {
name string
expr string
isTrace bool
want string // rewritten expr, only checked when isTrace
used []string
wantErr string
}{
{name: "avg trace col", expr: "avg(trace.output_tokens)", isTrace: true, want: "avg(output_tokens)", used: []string{"output_tokens"}},
{name: "sum trace col", expr: "sum(trace.total_tokens)", isTrace: true, want: "sum(total_tokens)", used: []string{"total_tokens"}},
{name: "count traces", expr: "count(trace.trace_id)", isTrace: true, want: "count(trace_id)"},
{name: "p90 trace col", expr: "p90(trace.max_llm_latency_ns)", isTrace: true, want: "quantile(0.90)(max_llm_latency_ns)", used: []string{"max_llm_latency_ns"}},
{name: "arithmetic between trace cols", expr: "avg(trace.output_tokens + trace.input_tokens)", isTrace: true, want: "avg(output_tokens + input_tokens)", used: []string{"output_tokens", "input_tokens"}},
{name: "arithmetic with constant", expr: "sum(trace.output_tokens * 1.5)", isTrace: true, want: "sum(output_tokens * 1.5)", used: []string{"output_tokens"}},
{name: "ratio of two aggregations", expr: "sum(trace.output_tokens)/count(trace.trace_id)", isTrace: true, want: "sum(output_tokens) / count(trace_id)", used: []string{"output_tokens"}},
{name: "backquoted trace col", expr: "avg(`trace.output_tokens`)", isTrace: true, want: "avg(`output_tokens`)", used: []string{"output_tokens"}},
{name: "bare count is span-level", expr: "count()", isTrace: false},
{name: "span attribute is span-level", expr: "sum(gen_ai.usage.output_tokens)", isTrace: false},
{name: "countIf span predicate is span-level", expr: "countIf(has_error = true)", isTrace: false},
{name: "mixed domains in one expression", expr: "sum(trace.output_tokens) + sum(gen_ai.usage.input_tokens)", wantErr: "mixes trace-level"},
{name: "mixed domains in one function", expr: "sum(trace.output_tokens + gen_ai.usage.input_tokens)", wantErr: "mixes trace-level"},
{name: "output-only column rejected", expr: "avg(trace.span_count)", wantErr: "unknown trace-level aggregation column"},
{name: "unknown column rejected", expr: "avg(trace.bogus)", wantErr: "unknown trace-level aggregation column"},
// a dotted column keeps every segment after the prefix, so it is reported whole
{name: "multi segment column rejected by full name", expr: "avg(trace.service.name)", wantErr: `"trace.service.name"`},
{name: "bare trace identifier is span-level", expr: "avg(trace)", isTrace: false},
{name: "countIf over trace col rejected", expr: "countIf(trace.output_tokens > 1000)", wantErr: "not supported"},
{name: "bare trace col rejected", expr: "trace.output_tokens", wantErr: "must be inside an aggregation function"},
{name: "backquoted bare trace col rejected", expr: "`trace.output_tokens`", wantErr: "must be inside an aggregation function"},
{name: "bare trace_id rejected", expr: "trace.trace_id", wantErr: "must be inside an aggregation function"},
{name: "arithmetic outside an aggregation rejected", expr: "trace.output_tokens + trace.input_tokens", wantErr: "must be inside an aggregation function"},
{name: "trace col beside an aggregation rejected", expr: "sum(trace.output_tokens) + trace.input_tokens", wantErr: "must be inside an aggregation function"},
{name: "aggregation scaled by a constant", expr: "sum(trace.output_tokens) * 2", isTrace: true, want: "sum(output_tokens) * 2", used: []string{"output_tokens"}},
{name: "rate over traces", expr: "rate(trace.trace_id)", isTrace: true, want: "count(trace_id)"},
{name: "rate_sum trace col", expr: "rate_sum(trace.output_tokens)", isTrace: true, want: "sum(output_tokens)", used: []string{"output_tokens"}},
// the interval divides the whole rendered expression, so a second aggregation
// alongside a rate would be divided too
{name: "rate mixed with another aggregation rejected", expr: "rate(trace.trace_id) + avg(trace.output_tokens)", wantErr: "combines a rate with another aggregation"},
{name: "ratio of two rates rejected", expr: "rate_sum(trace.output_tokens)/rate_sum(trace.input_tokens)", wantErr: "combines a rate with another aggregation"},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
ta, isTrace, err := rewriteTraceAggregation(tc.expr, cols)
if tc.wantErr != "" {
require.ErrorContains(t, err, tc.wantErr)
return
}
require.NoError(t, err)
require.Equal(t, tc.isTrace, isTrace)
if !tc.isTrace {
return
}
assert.Equal(t, tc.want, ta.expr)
for _, u := range tc.used {
assert.Contains(t, ta.used, u)
}
assert.Len(t, ta.used, len(tc.used))
})
}
}

View File

@@ -0,0 +1,144 @@
package scopedtracesstatementbuilder
import (
"context"
"strings"
"github.com/SigNoz/signoz/pkg/errors"
"github.com/SigNoz/signoz/pkg/querybuilder"
qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
"github.com/SigNoz/signoz/pkg/valuer"
qbvariables "github.com/SigNoz/signoz/pkg/variables"
"github.com/huandu/go-sqlbuilder"
)
// traceHaving is the resolved trace-level filter part: a HAVING predicate over the
// per-trace aliases plus the aliases it references (so scans select only those).
type traceHaving struct {
pred string
used map[string]struct{}
}
// resolveTraceHaving runs a trace-level filter through the standard where-clause
// pipeline against the per-trace aliases, so operators, bound args, and __all__ behave
// as in span filters. Returns nil when nothing is left to filter; args bind into sb.
func (b *scopedTraceStatementBuilder) resolveTraceHaving(ctx context.Context, expr string, variables map[string]qbtypes.VariableItem, sb *sqlbuilder.SelectBuilder) (*traceHaving, error) {
if strings.TrimSpace(expr) == "" {
return nil, nil //nolint:nilnil
}
// replaced before validation so variable literals are not mistaken for aggregate
// names; an unresolved $var is left in place and fails validation as an unknown one
if len(variables) > 0 {
replaced, err := qbvariables.ReplaceVariablesInExpression(expr, variables)
if err != nil {
return nil, err
}
expr = replaced
if strings.TrimSpace(expr) == "" {
return nil, nil //nolint:nilnil
}
}
allowed := b.filterableColumnSet()
// upfront targeted errors; the visitor folds them into a combined "Found N errors"
if err := validateAggregateFilter(expr, allowed); err != nil {
return nil, err
}
// both spellings resolve here: the key parser strips the trace. prefix into
// FieldContextTrace, which matches this entry's context
fieldKeys := make(map[string][]*telemetrytypes.TelemetryFieldKey, len(allowed))
for alias := range allowed {
key := &telemetrytypes.TelemetryFieldKey{Name: alias, FieldContext: telemetrytypes.FieldContextTrace}
fieldKeys[alias] = []*telemetrytypes.TelemetryFieldKey{key}
}
cb := &aliasConditionBuilder{allowed: allowed, used: make(map[string]struct{})}
prepared, err := querybuilder.PrepareWhereClause(expr, querybuilder.FilterExprVisitorOpts{
Context: ctx,
Logger: b.logger,
ConditionBuilder: cb,
FieldKeys: fieldKeys,
Variables: variables,
Builder: sb,
})
if err != nil {
return nil, err
}
if prepared.IsEmpty() {
return nil, nil //nolint:nilnil
}
return &traceHaving{pred: prepared.Expr, used: cb.used}, nil
}
// aliasConditionBuilder renders filter conditions directly against the per-trace
// aliases, recording the ones it touches; a key resolving to no alias is an error.
type aliasConditionBuilder struct {
allowed map[string]struct{}
used map[string]struct{}
}
var _ qbtypes.ConditionBuilder = (*aliasConditionBuilder)(nil)
func (c *aliasConditionBuilder) ConditionFor(
_ context.Context,
_ valuer.UUID,
_, _ uint64,
key *telemetrytypes.TelemetryFieldKey,
keys map[string][]*telemetrytypes.TelemetryFieldKey,
_ qbtypes.ConditionBuilderOptions,
op qbtypes.FilterOperator,
value any,
sb *sqlbuilder.SelectBuilder,
) ([]string, []string, error) {
matching := keys[key.Name]
if len(matching) == 0 {
return nil, nil, errors.NewInvalidInputf(errors.CodeInvalidInput,
"aggregate %q cannot be used in a trace-level filter; filterable aggregates: %s",
key.Name, strings.Join(sortedAliases(c.allowed), ", "))
}
alias := matching[0].Name
c.used[alias] = struct{}{}
col := quoteAlias(alias)
var cond string
switch op {
case qbtypes.FilterOperatorEqual:
cond = sb.E(col, value)
case qbtypes.FilterOperatorNotEqual:
cond = sb.NE(col, value)
case qbtypes.FilterOperatorGreaterThan:
cond = sb.G(col, value)
case qbtypes.FilterOperatorGreaterThanOrEq:
cond = sb.GE(col, value)
case qbtypes.FilterOperatorLessThan:
cond = sb.L(col, value)
case qbtypes.FilterOperatorLessThanOrEq:
cond = sb.LE(col, value)
case qbtypes.FilterOperatorIn, qbtypes.FilterOperatorNotIn:
values, ok := value.([]any)
if !ok {
values = []any{value}
}
if op == qbtypes.FilterOperatorIn {
cond = sb.In(col, values...)
} else {
cond = sb.NotIn(col, values...)
}
case qbtypes.FilterOperatorBetween, qbtypes.FilterOperatorNotBetween:
values, ok := value.([]any)
if !ok || len(values) != 2 {
return nil, nil, errors.NewInvalidInputf(errors.CodeInvalidInput,
"between on trace-level aggregate %q requires exactly two values", alias)
}
if op == qbtypes.FilterOperatorBetween {
cond = sb.Between(col, values[0], values[1])
} else {
cond = sb.NotBetween(col, values[0], values[1])
}
default:
return nil, nil, errors.NewInvalidInputf(errors.CodeInvalidInput,
"trace-level aggregate %q supports only comparison operators (=, !=, <, <=, >, >=, in, between)", alias)
}
return []string{cond}, nil, nil
}

View File

@@ -33,6 +33,12 @@ type traceQueryStatementBuilder struct {
aggExprRewriter qbtypes.AggExprRewriter
fl flagger.Flagger
skipResourceFingerprintEnabled bool
// traceScope, set only on the per-call copy made by BuildTraceScoped, constrains
// queries to spans whose trace_id is in the __trace_scope CTE.
traceScope *qbtypes.Statement
// traceScopeResource is the __resource_filter CTE traceScope's predicate references,
// emitted only when this builder's own resource filter did not already emit it.
traceScopeResource *qbtypes.Statement
}
var _ qbtypes.StatementBuilder[qbtypes.TraceAggregation] = (*traceQueryStatementBuilder)(nil)
@@ -97,6 +103,41 @@ func NewTraceQueryStatementBuilder(
}
}
// BuildTraceScoped is Build constrained to trace_ids selected by traceScope; the
// receiver is copied so the shared builder stays stateless.
func (b *traceQueryStatementBuilder) BuildTraceScoped(
ctx context.Context,
orgID valuer.UUID,
start uint64,
end uint64,
requestType qbtypes.RequestType,
query qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation],
variables map[string]qbtypes.VariableItem,
traceScope, traceScopeResource *qbtypes.Statement,
) (*qbtypes.Statement, error) {
scoped := *b
scoped.traceScope = traceScope
scoped.traceScopeResource = traceScopeResource
return scoped.Build(ctx, orgID, start, end, requestType, query, variables)
}
// attachTraceScope adds the trace-scope condition to sb and returns the CTE fragments
// + args to prepend; resourceEmitted reports whether the query already carries the
// __resource_filter CTE, so the scope's copy is emitted only when it does not.
func (b *traceQueryStatementBuilder) attachTraceScope(sb *sqlbuilder.SelectBuilder, resourceEmitted bool) ([]string, [][]any) {
if b.traceScope == nil {
return nil, nil
}
sb.Where("trace_id GLOBAL IN (SELECT trace_id FROM __trace_scope)")
var frags []string
var args [][]any
if b.traceScopeResource != nil && !resourceEmitted {
frags = append(frags, fmt.Sprintf("__resource_filter AS (%s)", b.traceScopeResource.Query))
args = append(args, b.traceScopeResource.Args)
}
return append(frags, fmt.Sprintf("__trace_scope AS (%s)", b.traceScope.Query)), append(args, b.traceScope.Args)
}
// Build builds a SQL query for traces based on the given parameters.
func (b *traceQueryStatementBuilder) Build(
ctx context.Context,
@@ -163,31 +204,15 @@ func getKeySelectors(query qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation])
}
for idx := range query.GroupBy {
groupBy := query.GroupBy[idx]
keySelectors = append(keySelectors, &telemetrytypes.FieldKeySelector{
Name: groupBy.Name,
Signal: telemetrytypes.SignalTraces,
FieldContext: groupBy.FieldContext,
FieldDataType: groupBy.FieldDataType,
})
keySelectors = append(keySelectors, keySelectorsForField(query.GroupBy[idx].TelemetryFieldKey)...)
}
for idx := range query.SelectFields {
keySelectors = append(keySelectors, &telemetrytypes.FieldKeySelector{
Name: query.SelectFields[idx].Name,
Signal: telemetrytypes.SignalTraces,
FieldContext: query.SelectFields[idx].FieldContext,
FieldDataType: query.SelectFields[idx].FieldDataType,
})
keySelectors = append(keySelectors, keySelectorsForField(query.SelectFields[idx])...)
}
for idx := range query.Order {
keySelectors = append(keySelectors, &telemetrytypes.FieldKeySelector{
Name: query.Order[idx].Key.Name,
Signal: telemetrytypes.SignalTraces,
FieldContext: query.Order[idx].Key.FieldContext,
FieldDataType: query.Order[idx].Key.FieldDataType,
})
keySelectors = append(keySelectors, keySelectorsForField(query.Order[idx].Key.TelemetryFieldKey)...)
}
for idx := range keySelectors {
@@ -198,6 +223,26 @@ func getKeySelectors(query qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation])
return keySelectors
}
func keySelectorsForField(key telemetrytypes.TelemetryFieldKey) []*telemetrytypes.FieldKeySelector {
selectors := []*telemetrytypes.FieldKeySelector{
{
Name: key.Name,
Signal: telemetrytypes.SignalTraces,
FieldContext: key.FieldContext,
FieldDataType: key.FieldDataType,
},
}
if key.FieldContext != telemetrytypes.FieldContextUnspecified {
selectors = append(selectors, &telemetrytypes.FieldKeySelector{
Name: key.FieldContext.StringValue() + "." + key.Name,
Signal: telemetrytypes.SignalTraces,
FieldContext: telemetrytypes.FieldContextUnspecified,
FieldDataType: key.FieldDataType,
})
}
return selectors
}
// mergeDeprecatedTraceKeys prepends deprecated intrinsic/calculated trace field
// definitions to the keys map. We do this during statement building, not at
// metadata fetch time, because:
@@ -269,20 +314,14 @@ func adjustTraceKey(key *telemetrytypes.TelemetryFieldKey, keys map[string][]*te
For example: trace_id (intrinsic), response_status_code (calculated).
*/
// Resolve against the context-qualified name first, then the bare name since that can be instrinsic field e.g. scope.name.
var isIntrinsicOrCalculatedField bool
var intrinsicOrCalculatedField telemetrytypes.TelemetryFieldKey
if _, ok := tracestelemetryschema.IntrinsicFields[key.Name]; ok {
isIntrinsicOrCalculatedField = true
intrinsicOrCalculatedField = tracestelemetryschema.IntrinsicFields[key.Name]
} else if _, ok := tracestelemetryschema.CalculatedFields[key.Name]; ok {
isIntrinsicOrCalculatedField = true
intrinsicOrCalculatedField = tracestelemetryschema.CalculatedFields[key.Name]
} else if _, ok := tracestelemetryschema.IntrinsicFieldsDeprecated[key.Name]; ok {
isIntrinsicOrCalculatedField = true
intrinsicOrCalculatedField = tracestelemetryschema.IntrinsicFieldsDeprecated[key.Name]
} else if _, ok := tracestelemetryschema.CalculatedFieldsDeprecated[key.Name]; ok {
isIntrinsicOrCalculatedField = true
intrinsicOrCalculatedField = tracestelemetryschema.CalculatedFieldsDeprecated[key.Name]
if key.FieldContext != telemetrytypes.FieldContextUnspecified {
intrinsicOrCalculatedField, isIntrinsicOrCalculatedField = lookupIntrinsicOrCalculatedField(key.FieldContext.StringValue() + "." + key.Name)
}
if !isIntrinsicOrCalculatedField {
intrinsicOrCalculatedField, isIntrinsicOrCalculatedField = lookupIntrinsicOrCalculatedField(key.Name)
}
if isIntrinsicOrCalculatedField {
@@ -294,6 +333,24 @@ func adjustTraceKey(key *telemetrytypes.TelemetryFieldKey, keys map[string][]*te
return actions
}
// lookupIntrinsicOrCalculatedField returns the intrinsic or calculated field registered under
// name, across the current and deprecated tables.
func lookupIntrinsicOrCalculatedField(name string) (telemetrytypes.TelemetryFieldKey, bool) {
if f, ok := tracestelemetryschema.IntrinsicFields[name]; ok {
return f, true
}
if f, ok := tracestelemetryschema.CalculatedFields[name]; ok {
return f, true
}
if f, ok := tracestelemetryschema.IntrinsicFieldsDeprecated[name]; ok {
return f, true
}
if f, ok := tracestelemetryschema.CalculatedFieldsDeprecated[name]; ok {
return f, true
}
return telemetrytypes.TelemetryFieldKey{}, false
}
// buildListQuery builds a query for list panel type.
func (b *traceQueryStatementBuilder) buildListQuery(
ctx context.Context,
@@ -521,6 +578,11 @@ func (b *traceQueryStatementBuilder) buildTimeSeriesQuery(
cteArgs = append(cteArgs, args)
}
if scopeFrags, scopeArgs := b.attachTraceScope(sb, frag != ""); len(scopeFrags) > 0 {
cteFragments = append(cteFragments, scopeFrags...)
cteArgs = append(cteArgs, scopeArgs...)
}
sb.SelectMore(fmt.Sprintf(
"toStartOfInterval(timestamp, INTERVAL %d SECOND) AS ts",
int64(query.StepInterval.Seconds()),
@@ -681,6 +743,13 @@ func (b *traceQueryStatementBuilder) buildScalarQuery(
cteArgs = append(cteArgs, args)
}
// skipResourceCTE means this scalar is embedded as a CTE of a time-series query,
// which has already emitted the __trace_scope fragment — add only the condition.
if scopeFrags, scopeArgs := b.attachTraceScope(sb, frag != ""); len(scopeFrags) > 0 && !skipResourceCTE {
cteFragments = append(cteFragments, scopeFrags...)
cteArgs = append(cteArgs, scopeArgs...)
}
allAggChArgs := []any{}
fieldNames := make([]string, 0, len(query.GroupBy))

View File

@@ -374,6 +374,94 @@ func TestStatementBuilder(t *testing.T) {
},
expectedErr: nil,
},
{
name: "scope.name filter and group by",
requestType: qbtypes.RequestTypeTimeSeries,
query: qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Signal: telemetrytypes.SignalTraces,
StepInterval: qbtypes.Step{Duration: 30 * time.Second},
Aggregations: []qbtypes.TraceAggregation{
{
Expression: "count()",
},
},
Filter: &qbtypes.Filter{
Expression: "scope.name = 'opentelemetry-io'",
},
Limit: 10,
GroupBy: []qbtypes.GroupByKey{
{
TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{
Name: "scope.name",
FieldContext: telemetrytypes.FieldContextScope,
},
},
},
},
expected: qbtypes.Statement{
Query: "WITH __limit_cte AS (SELECT toString(multiIf(scope.name::String <> '', scope.name::String, NULL)) AS `__GROUP_BY_KEY_0_scope.name`, count() AS __result_0 FROM signoz_traces.distributed_signoz_index_v3 WHERE (scope.name::String = ? AND scope.name::String <> '') AND timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ? GROUP BY `__GROUP_BY_KEY_0_scope.name` ORDER BY __result_0 DESC LIMIT ?) SELECT toStartOfInterval(timestamp, INTERVAL 30 SECOND) AS ts, toString(multiIf(scope.name::String <> '', scope.name::String, NULL)) AS `__GROUP_BY_KEY_0_scope.name`, count() AS __result_0 FROM signoz_traces.distributed_signoz_index_v3 WHERE (scope.name::String = ? AND scope.name::String <> '') AND timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ? AND (`__GROUP_BY_KEY_0_scope.name`) GLOBAL IN (SELECT `__GROUP_BY_KEY_0_scope.name` FROM __limit_cte) GROUP BY ts, `__GROUP_BY_KEY_0_scope.name`",
Args: []any{"opentelemetry-io", "1747947419000000000", "1747983448000000000", uint64(1747945619), uint64(1747983448), 10, "opentelemetry-io", "1747947419000000000", "1747983448000000000", uint64(1747945619), uint64(1747983448)},
},
expectedErr: nil,
},
{
name: "scope.version filter with scope.name group by",
requestType: qbtypes.RequestTypeTimeSeries,
query: qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Signal: telemetrytypes.SignalTraces,
StepInterval: qbtypes.Step{Duration: 30 * time.Second},
Aggregations: []qbtypes.TraceAggregation{
{
Expression: "count()",
},
},
Filter: &qbtypes.Filter{
Expression: "scope.version = '1.0.0'",
},
Limit: 10,
GroupBy: []qbtypes.GroupByKey{
{
TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{
Name: "scope.name",
FieldContext: telemetrytypes.FieldContextScope,
},
},
},
},
expected: qbtypes.Statement{
Query: "WITH __limit_cte AS (SELECT toString(multiIf(scope.name::String <> '', scope.name::String, NULL)) AS `__GROUP_BY_KEY_0_scope.name`, count() AS __result_0 FROM signoz_traces.distributed_signoz_index_v3 WHERE (scope.version::String = ? AND scope.version::String <> '') AND timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ? GROUP BY `__GROUP_BY_KEY_0_scope.name` ORDER BY __result_0 DESC LIMIT ?) SELECT toStartOfInterval(timestamp, INTERVAL 30 SECOND) AS ts, toString(multiIf(scope.name::String <> '', scope.name::String, NULL)) AS `__GROUP_BY_KEY_0_scope.name`, count() AS __result_0 FROM signoz_traces.distributed_signoz_index_v3 WHERE (scope.version::String = ? AND scope.version::String <> '') AND timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ? AND (`__GROUP_BY_KEY_0_scope.name`) GLOBAL IN (SELECT `__GROUP_BY_KEY_0_scope.name` FROM __limit_cte) GROUP BY ts, `__GROUP_BY_KEY_0_scope.name`",
Args: []any{"1.0.0", "1747947419000000000", "1747983448000000000", uint64(1747945619), uint64(1747983448), 10, "1.0.0", "1747947419000000000", "1747983448000000000", uint64(1747945619), uint64(1747983448)},
},
expectedErr: nil,
},
{
name: "scope.version filter only (no scope field in group by)",
requestType: qbtypes.RequestTypeTimeSeries,
query: qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Signal: telemetrytypes.SignalTraces,
StepInterval: qbtypes.Step{Duration: 30 * time.Second},
Aggregations: []qbtypes.TraceAggregation{
{
Expression: "count()",
},
},
Filter: &qbtypes.Filter{
Expression: "scope.version = '1.0.0'",
},
Limit: 10,
GroupBy: []qbtypes.GroupByKey{
{
TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{
Name: "service.name",
},
},
},
},
expected: qbtypes.Statement{
Query: "WITH __limit_cte AS (SELECT toString(multiIf(multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, mapContains(resources_string, 'service.name'), resources_string['service.name'], NULL) IS NOT NULL, multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, mapContains(resources_string, 'service.name'), resources_string['service.name'], NULL), NULL)) AS `__GROUP_BY_KEY_0_service.name`, count() AS __result_0 FROM signoz_traces.distributed_signoz_index_v3 WHERE (scope.version::String = ? AND scope.version::String <> '') AND timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ? GROUP BY `__GROUP_BY_KEY_0_service.name` ORDER BY __result_0 DESC LIMIT ?) SELECT toStartOfInterval(timestamp, INTERVAL 30 SECOND) AS ts, toString(multiIf(multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, mapContains(resources_string, 'service.name'), resources_string['service.name'], NULL) IS NOT NULL, multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, mapContains(resources_string, 'service.name'), resources_string['service.name'], NULL), NULL)) AS `__GROUP_BY_KEY_0_service.name`, count() AS __result_0 FROM signoz_traces.distributed_signoz_index_v3 WHERE (scope.version::String = ? AND scope.version::String <> '') AND timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ? AND (`__GROUP_BY_KEY_0_service.name`) GLOBAL IN (SELECT `__GROUP_BY_KEY_0_service.name` FROM __limit_cte) GROUP BY ts, `__GROUP_BY_KEY_0_service.name`",
Args: []any{"1.0.0", "1747947419000000000", "1747983448000000000", uint64(1747945619), uint64(1747983448), 10, "1.0.0", "1747947419000000000", "1747983448000000000", uint64(1747945619), uint64(1747983448)},
},
},
}
fl := flaggertest.New(t)
@@ -800,6 +888,143 @@ func TestStatementBuilderListQueryWithCorruptData(t *testing.T) {
},
expectedErr: nil,
},
{
name: "List query with scope filter only (no scope in select or group by)",
requestType: qbtypes.RequestTypeRaw,
keysMap: map[string][]*telemetrytypes.TelemetryFieldKey{
"scope.version": {
{
Name: "scope.version",
Signal: telemetrytypes.SignalTraces,
FieldContext: telemetrytypes.FieldContextScope,
FieldDataType: telemetrytypes.FieldDataTypeString,
},
},
},
query: qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Signal: telemetrytypes.SignalTraces,
StepInterval: qbtypes.Step{Duration: 30 * time.Second},
Filter: &qbtypes.Filter{
Expression: "scope.version = '1.0.0'",
},
Limit: 10,
},
expected: qbtypes.Statement{
Query: "SELECT timestamp AS `__SELECT_KEY_0_timestamp`, trace_id AS `__SELECT_KEY_1_trace_id`, span_id AS `__SELECT_KEY_2_span_id`, trace_state AS `__SELECT_KEY_3_trace_state`, parent_span_id AS `__SELECT_KEY_4_parent_span_id`, flags AS `__SELECT_KEY_5_flags`, name AS `__SELECT_KEY_6_name`, kind AS `__SELECT_KEY_7_kind`, kind_string AS `__SELECT_KEY_8_kind_string`, duration_nano AS `__SELECT_KEY_9_duration_nano`, status_code AS `__SELECT_KEY_10_status_code`, status_message AS `__SELECT_KEY_11_status_message`, status_code_string AS `__SELECT_KEY_12_status_code_string`, events AS `__SELECT_KEY_13_events`, links AS `__SELECT_KEY_14_links`, response_status_code AS `__SELECT_KEY_15_response_status_code`, external_http_url AS `__SELECT_KEY_16_external_http_url`, http_url AS `__SELECT_KEY_17_http_url`, external_http_method AS `__SELECT_KEY_18_external_http_method`, http_method AS `__SELECT_KEY_19_http_method`, http_host AS `__SELECT_KEY_20_http_host`, db_name AS `__SELECT_KEY_21_db_name`, db_operation AS `__SELECT_KEY_22_db_operation`, has_error AS `__SELECT_KEY_23_has_error`, is_remote AS `__SELECT_KEY_24_is_remote`, attributes_string, attributes_number, attributes_bool, resources_string FROM signoz_traces.distributed_signoz_index_v3 WHERE (scope.version::String = ? AND scope.version::String <> '') AND timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ? LIMIT ?",
Args: []any{"1.0.0", "1747947419000000000", "1747983448000000000", uint64(1747945619), uint64(1747983448), 10},
},
},
{
// Regression test: scope.version in selectFields with no metadata (isColumn=true filters it out)
// must still produce scope.version::String, not scope.attributes.version::String
name: "scope.version in selectFields only, no metadata (intrinsic field fallback)",
requestType: qbtypes.RequestTypeRaw,
keysMap: map[string][]*telemetrytypes.TelemetryFieldKey{},
query: qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Signal: telemetrytypes.SignalTraces,
StepInterval: qbtypes.Step{Duration: 30 * time.Second},
Filter: &qbtypes.Filter{},
SelectFields: []telemetrytypes.TelemetryFieldKey{
{Name: "scope.version", FieldContext: telemetrytypes.FieldContextUnspecified},
},
Limit: 10,
},
expected: qbtypes.Statement{
Query: "SELECT timestamp AS `__SELECT_KEY_0_timestamp`, trace_id AS `__SELECT_KEY_1_trace_id`, span_id AS `__SELECT_KEY_2_span_id`, multiIf(scope.version::String <> '', scope.version::String, NULL) AS `__SELECT_KEY_3_scope.version` FROM signoz_traces.distributed_signoz_index_v3 WHERE timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ? LIMIT ?",
Args: []any{"1747947419000000000", "1747983448000000000", uint64(1747945619), uint64(1747983448), 10},
},
},
{
// A scope attribute whose own name literally carries a `scope.` prefix (`scope.prefixed`,
// normalized to {prefixed, scope}) resolves to that attribute in a SELECT even without a
// filter: getKeySelectors emits the reconstructed `scope.prefixed` selector so the metadata
// fetch surfaces it and AdjustKey recovers the full name. Without it the `scope.` prefix is
// lost and it wrongly reads `scope.attributes.prefixed`.
name: "scope-prefixed attribute in selectFields resolves without a filter",
requestType: qbtypes.RequestTypeRaw,
keysMap: map[string][]*telemetrytypes.TelemetryFieldKey{
"scope.prefixed": {
{
Name: "scope.prefixed",
Signal: telemetrytypes.SignalTraces,
FieldContext: telemetrytypes.FieldContextScope,
FieldDataType: telemetrytypes.FieldDataTypeString,
},
},
},
query: qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Signal: telemetrytypes.SignalTraces,
StepInterval: qbtypes.Step{Duration: 30 * time.Second},
Filter: &qbtypes.Filter{},
SelectFields: []telemetrytypes.TelemetryFieldKey{
{Name: "prefixed", FieldContext: telemetrytypes.FieldContextScope},
},
Limit: 10,
},
expected: qbtypes.Statement{
Query: "SELECT timestamp AS `__SELECT_KEY_0_timestamp`, trace_id AS `__SELECT_KEY_1_trace_id`, span_id AS `__SELECT_KEY_2_span_id`, multiIf(scope.attributes.`scope.prefixed` IS NOT NULL, scope.attributes.`scope.prefixed`::String, NULL) AS `__SELECT_KEY_3_scope.prefixed` FROM signoz_traces.distributed_signoz_index_v3 WHERE timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ? LIMIT ?",
Args: []any{"1747947419000000000", "1747983448000000000", uint64(1747945619), uint64(1747983448), 10},
},
},
{
// A scope-context key whose name matches a declared scope path resolves to that
// declared path (scope.name), not the span `name` column and not an undeclared
// scope attribute. getTracesKeys surfaces the declared path as an intrinsic key
// (metadata.go), which shadows the same-named span intrinsic.
name: "scope-context name resolves to the declared scope path",
requestType: qbtypes.RequestTypeRaw,
keysMap: map[string][]*telemetrytypes.TelemetryFieldKey{
"scope.name": {
{
Name: "scope.name",
Signal: telemetrytypes.SignalTraces,
FieldContext: telemetrytypes.FieldContextScope,
FieldDataType: telemetrytypes.FieldDataTypeString,
},
},
"name": {
{
Name: "name",
Signal: telemetrytypes.SignalTraces,
FieldContext: telemetrytypes.FieldContextSpan,
FieldDataType: telemetrytypes.FieldDataTypeString,
},
},
},
query: qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Signal: telemetrytypes.SignalTraces,
StepInterval: qbtypes.Step{Duration: 30 * time.Second},
Filter: &qbtypes.Filter{},
SelectFields: []telemetrytypes.TelemetryFieldKey{
{Name: "name", FieldContext: telemetrytypes.FieldContextScope},
},
Limit: 10,
},
expected: qbtypes.Statement{
Query: "SELECT timestamp AS `__SELECT_KEY_0_timestamp`, trace_id AS `__SELECT_KEY_1_trace_id`, span_id AS `__SELECT_KEY_2_span_id`, multiIf(scope.name::String <> '', scope.name::String, NULL) AS `__SELECT_KEY_3_name` FROM signoz_traces.distributed_signoz_index_v3 WHERE timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ? LIMIT ?",
Args: []any{"1747947419000000000", "1747983448000000000", uint64(1747945619), uint64(1747983448), 10},
},
},
{
// span.scope.name (span context, name "scope.name") resolves to the declared
// scope path scope.name, not a span attribute literally named scope.name.
name: "span-context scope.name in selectFields resolves to the declared scope path",
requestType: qbtypes.RequestTypeRaw,
keysMap: map[string][]*telemetrytypes.TelemetryFieldKey{},
query: qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Signal: telemetrytypes.SignalTraces,
StepInterval: qbtypes.Step{Duration: 30 * time.Second},
Filter: &qbtypes.Filter{},
SelectFields: []telemetrytypes.TelemetryFieldKey{
{Name: "scope.name", FieldContext: telemetrytypes.FieldContextSpan},
},
Limit: 10,
},
expected: qbtypes.Statement{
Query: "SELECT timestamp AS `__SELECT_KEY_0_timestamp`, trace_id AS `__SELECT_KEY_1_trace_id`, span_id AS `__SELECT_KEY_2_span_id`, multiIf(scope.name::String <> '', scope.name::String, NULL) AS `__SELECT_KEY_3_scope.name` FROM signoz_traces.distributed_signoz_index_v3 WHERE timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ? LIMIT ?",
Args: []any{"1747947419000000000", "1747983448000000000", uint64(1747945619), uint64(1747983448), 10},
},
},
}
for _, c := range cases {

View File

@@ -180,7 +180,7 @@ func (t *telemetryMetaStore) getTracesKeys(ctx context.Context, fieldKeySelector
`CASE
// WHEN tagType = 'spanfield' THEN 1
WHEN tagType = 'resource' THEN 2
// WHEN tagType = 'scope' THEN 3
WHEN tagType = 'scope' THEN 3
WHEN tagType = 'tag' THEN 4
ELSE 5
END as priority`,

View File

@@ -0,0 +1,36 @@
package aitelemetryschema
import (
"github.com/SigNoz/signoz/pkg/querybuilder"
"github.com/SigNoz/signoz/pkg/types/aiobservabilitytypes"
)
var (
traceAggregateNames = func() map[string]struct{} {
names := make(map[string]struct{}, len(TraceAggregateFields))
for name := range TraceAggregateFields {
names[name] = struct{}{}
}
return names
}()
genAISpanGate = "(" + aiobservabilitytypes.GenAISpanFilterExpression() + ")"
)
// ScopedExistingQuery narrows value suggestions to gen_ai spans: the caller's
// filter minus its per-trace aggregate atoms (never ingested, so nothing can
// narrow on them), ANDed with the gen_ai span gate. An unparseable filter is
// dropped, matching how the metadata store treats it downstream.
func ScopedExistingQuery(existingQuery string) string {
spanExpr := ""
if existingQuery != "" {
if expr, _, err := querybuilder.SplitFilterForAggregates(existingQuery, traceAggregateNames); err == nil {
spanExpr = expr
}
}
if spanExpr == "" {
return genAISpanGate
}
return genAISpanGate + " AND (" + spanExpr + ")"
}

View File

@@ -0,0 +1,94 @@
package aitelemetryschema
import (
"testing"
"github.com/stretchr/testify/assert"
)
func TestScopedExistingQuery(t *testing.T) {
gate := "(gen_ai.request.model EXISTS OR gen_ai.tool.name EXISTS OR gen_ai.agent.name EXISTS)"
testCases := []struct {
name string
existingQuery string
expected string
}{
{
name: "empty query returns the gate alone",
existingQuery: "",
expected: gate,
},
{
name: "span filter is preserved under the gate",
existingQuery: "service.name = 'checkout'",
expected: gate + " AND (service.name = 'checkout')",
},
{
name: "trace aggregate filter is stripped",
existingQuery: "llm_call_count > 5",
expected: gate,
},
{
name: "mixed filter keeps only the span part",
existingQuery: "llm_call_count > 5 AND gen_ai.request.model = 'gpt-4'",
expected: gate + " AND (gen_ai.request.model = 'gpt-4')",
},
{
name: "trace context filter is stripped",
existingQuery: "trace.total_tokens > 100 AND service.name = 'checkout'",
expected: gate + " AND (service.name = 'checkout')",
},
{
name: "unparseable filter is dropped",
existingQuery: "service.name = ",
expected: gate,
},
{
name: "multiple span conditions survive as one AND chain",
existingQuery: "service.name = 'checkout' AND gen_ai.request.model = 'gpt-4' AND llm_call_count > 5",
expected: gate + " AND (service.name = 'checkout' AND gen_ai.request.model = 'gpt-4')",
},
{
name: "span OR group is kept whole and parenthesized against the AND join",
existingQuery: "service.name = 'a' OR service.name = 'b'",
expected: gate + " AND ((service.name = 'a' OR service.name = 'b'))",
},
{
name: "parenthesized span OR group ANDed with an aggregate keeps only the group",
existingQuery: "(service.name = 'a' OR service.name = 'b') AND llm_call_count > 5",
expected: gate + " AND ((service.name = 'a' OR service.name = 'b'))",
},
{
name: "OR group of trace aggregates is stripped whole",
existingQuery: "llm_call_count > 5 OR total_tokens > 100",
expected: gate,
},
{
name: "OR mixing aggregate and span atoms drops the whole filter",
existingQuery: "llm_call_count > 5 OR service.name = 'checkout'",
expected: gate,
},
{
name: "parenthesized AND group is split, not routed whole",
existingQuery: "(llm_call_count > 5 AND service.name = 'checkout') AND gen_ai.request.model = 'gpt-4'",
expected: gate + " AND (service.name = 'checkout' AND gen_ai.request.model = 'gpt-4')",
},
{
name: "NOT over an aggregate group is stripped",
existingQuery: "NOT (llm_call_count > 5) AND service.name = 'checkout'",
expected: gate + " AND (service.name = 'checkout')",
},
{
name: "NOT over a span group is kept",
existingQuery: "NOT (service.name = 'checkout')",
expected: gate + " AND (NOT (service.name = 'checkout'))",
},
}
for _, testCase := range testCases {
t.Run(testCase.name, func(t *testing.T) {
assert.Equal(t, testCase.expected, ScopedExistingQuery(testCase.existingQuery))
})
}
}

View File

@@ -391,6 +391,96 @@ func TestConditionForResourceWithEvolution(t *testing.T) {
}
}
// TestConditionForScopeIntrinsicFields covers the scope.name/scope.version intrinsic
// fields against the "scope" JSON column. These are *declared* String paths on that
// column, so a row without a scope reads as ” and never NULL: presence must be an
// empty-string check, since "IS NOT NULL" would hold for every row. That also rules
// out treating them as nested attribute keys under scope.attributes, which are
// undeclared (Dynamic) paths and genuinely NULL when absent.
func TestConditionForScopeIntrinsicFields(t *testing.T) {
ctx := context.Background()
fm := NewFieldMapper(flaggertest.New(t))
conditionBuilder := NewConditionBuilder(fm, flaggertest.New(t))
testCases := []struct {
name string
key telemetrytypes.TelemetryFieldKey
operator qbtypes.FilterOperator
value any
expectedSQL string
}{
{
name: "Equal - scope.name",
key: telemetrytypes.TelemetryFieldKey{
Name: "scope.name",
FieldContext: telemetrytypes.FieldContextScope,
FieldDataType: telemetrytypes.FieldDataTypeString,
},
operator: qbtypes.FilterOperatorEqual,
value: "io.signoz.payment",
expectedSQL: "(scope.name::String = ? AND scope.name::String <> '')",
},
{
name: "Equal - scope.version",
key: telemetrytypes.TelemetryFieldKey{
Name: "scope.version",
FieldContext: telemetrytypes.FieldContextScope,
FieldDataType: telemetrytypes.FieldDataTypeString,
},
operator: qbtypes.FilterOperatorEqual,
value: "2.3.1",
expectedSQL: "(scope.version::String = ? AND scope.version::String <> '')",
},
{
name: "Exists - scope.name",
key: telemetrytypes.TelemetryFieldKey{
Name: "scope.name",
FieldContext: telemetrytypes.FieldContextScope,
FieldDataType: telemetrytypes.FieldDataTypeString,
},
operator: qbtypes.FilterOperatorExists,
value: nil,
expectedSQL: "scope.name::String <> ''",
},
{
name: "NotExists - scope.version",
key: telemetrytypes.TelemetryFieldKey{
Name: "scope.version",
FieldContext: telemetrytypes.FieldContextScope,
FieldDataType: telemetrytypes.FieldDataTypeString,
},
operator: qbtypes.FilterOperatorNotExists,
value: nil,
expectedSQL: "scope.version::String = ''",
},
{
// `scope.attribute.name` (normalized to {attribute.name, scope}) addresses the scope
// attribute named `name` — the declared `scope.name` path is never reached this way.
name: "Equal - scope.attribute.name reaches the named scope attribute",
key: telemetrytypes.TelemetryFieldKey{
Name: "attribute.name",
FieldContext: telemetrytypes.FieldContextScope,
FieldDataType: telemetrytypes.FieldDataTypeString,
},
operator: qbtypes.FilterOperatorEqual,
value: "io.signoz.checkout",
expectedSQL: "(scope.attributes.`name`::String = ? AND scope.attributes.`name` IS NOT NULL)",
},
}
for _, tc := range testCases {
sb := sqlbuilder.NewSelectBuilder()
t.Run(tc.name, func(t *testing.T) {
conds, _, err := conditionBuilder.ConditionFor(ctx, valuer.UUID{}, 0, 0, &tc.key, map[string][]*telemetrytypes.TelemetryFieldKey{tc.key.Name: {&tc.key}}, qbtypes.ConditionBuilderOptions{}, tc.operator, tc.value, sb)
require.NoError(t, err)
sb.Where(conds...)
sql, _ := sb.BuildWithFlavor(sqlbuilder.ClickHouse)
assert.Contains(t, sql, tc.expectedSQL)
assert.NotContains(t, sql, "scope.`scope.", "must not double-prefix the scope JSON path")
})
}
}
// TestConditionForSynthesizedKeys covers the KeyNotFound fallback: when a
// referenced attribute key has no metadata match, the builder synthesizes key(s) from
// user input and queries anyway, emitting a warning instead of failing.
@@ -414,6 +504,20 @@ func TestConditionForSynthesizedKeys(t *testing.T) {
assert.Contains(t, args, "timeout")
})
t.Run("scope context with no metadata -> scope attribute", func(t *testing.T) {
sb := sqlbuilder.NewSelectBuilder()
key := telemetrytypes.TelemetryFieldKey{Name: "custom.attr", FieldContext: telemetrytypes.FieldContextScope}
conds, warnings, err := cb.ConditionFor(ctx, valuer.UUID{}, 0, 0, &key, noMatches, qbtypes.ConditionBuilderOptions{}, qbtypes.FilterOperatorEqual, "v", sb)
assert.NoError(t, err, "an undeclared scope attribute must still be filterable")
assert.NotEmpty(t, warnings)
sb.Where(conds...)
sql, _ := sb.BuildWithFlavor(sqlbuilder.ClickHouse)
assert.Contains(t, sql, "scope.attributes.`custom.attr`")
// `scope.` can be part of the attribute's own name, so the literal spelling is a
// candidate too — the caller ORs the two.
assert.Contains(t, sql, "scope.attributes.`scope.custom.attr`")
})
t.Run("bare key with number operand -> attribute number", func(t *testing.T) {
sb := sqlbuilder.NewSelectBuilder()
key := telemetrytypes.TelemetryFieldKey{Name: "http.status"}

View File

@@ -121,6 +121,20 @@ var (
FieldContext: telemetrytypes.FieldContextSpan,
FieldDataType: telemetrytypes.FieldDataTypeString,
},
"scope.name": {
Name: "scope.name",
Description: "Instrumentation scope name",
Signal: telemetrytypes.SignalTraces,
FieldContext: telemetrytypes.FieldContextScope,
FieldDataType: telemetrytypes.FieldDataTypeString,
},
"scope.version": {
Name: "scope.version",
Description: "Instrumentation scope version",
Signal: telemetrytypes.SignalTraces,
FieldContext: telemetrytypes.FieldContextScope,
FieldDataType: telemetrytypes.FieldDataTypeString,
},
}
IntrinsicFieldsDeprecated = map[string]telemetrytypes.TelemetryFieldKey{
"traceID": {

View File

@@ -53,6 +53,7 @@ var (
ValueType: schema.ColumnTypeString,
}},
"resource": {Name: "resource", Type: schema.JSONColumnType{}},
"scope": {Name: "scope", Type: schema.JSONColumnType{}},
"events": {Name: "events", Type: schema.ArrayColumnType{
ElementType: schema.ColumnTypeString,
@@ -181,7 +182,7 @@ func (m *fieldMapper) getColumn(
case telemetrytypes.FieldContextResource:
return []*schema.Column{indexV3Columns["resource"], indexV3Columns["resources_string"]}, nil
case telemetrytypes.FieldContextScope:
return []*schema.Column{}, qbtypes.ErrColumnNotFound
return []*schema.Column{indexV3Columns["scope"]}, nil
case telemetrytypes.FieldContextAttribute:
switch key.FieldDataType {
case telemetrytypes.FieldDataTypeString:
@@ -292,14 +293,25 @@ func (m *fieldMapper) resolveColumnExprs(
switch column.Type.GetType() {
case schema.ColumnTypeEnumJSON:
// json is only supported for resource context as of now
if key.FieldContext != telemetrytypes.FieldContextResource {
return nil, nil, nil, errors.Newf(errors.TypeInvalidInput, errors.CodeInvalidInput, "only resource context fields are supported for json columns, got %s", key.FieldContext.String)
}
// have to add ::string as clickHouse throws an error :- data types Variant/Dynamic are not allowed in GROUP BY
// once clickHouse dependency is updated, we need to check if we can remove it.
exprs = append(exprs, fmt.Sprintf("%s.`%s`::String", columnName, key.Name))
existExprs = append(existExprs, fmt.Sprintf("%s.`%s` IS NOT NULL", columnName, key.Name))
switch key.FieldContext {
case telemetrytypes.FieldContextResource:
exprs = append(exprs, fmt.Sprintf("%s.`%s`::String", columnName, key.Name))
existExprs = append(existExprs, fmt.Sprintf("%s.`%s` IS NOT NULL", columnName, key.Name))
case telemetrytypes.FieldContextScope:
if f, ok := IntrinsicFields[key.Name]; ok && f.FieldContext == telemetrytypes.FieldContextScope {
// declared String paths on the scope column read '' for the missing case
exprs = append(exprs, fmt.Sprintf("%s::String", key.Name))
existExprs = append(existExprs, fmt.Sprintf("%s <> ''", key.Name))
} else {
attributeName := strings.TrimPrefix(key.Name, "attribute.") // literal "attribute" prefix in attribute keys needs double prefix
exprs = append(exprs, fmt.Sprintf("%s.attributes.%s::String", columnName, querybuilder.ClickHouseIdentifier(attributeName)))
existExprs = append(existExprs, fmt.Sprintf("%s.attributes.%s IS NOT NULL", columnName, querybuilder.ClickHouseIdentifier(attributeName)))
}
default:
return nil, nil, nil, errors.NewInternalf(errors.CodeInternal, "only resource and scope context fields are supported for json columns, got %s", key.FieldContext.String)
}
case schema.ColumnTypeEnumString,
schema.ColumnTypeEnumUInt64,
schema.ColumnTypeEnumUInt32,
@@ -341,20 +353,6 @@ func (m *fieldMapper) resolveColumnExprs(
return exprs, existExprs, columns, nil
}
// logicalForResolvedColumn upgrades a directly-resolvable key (the FieldFor
// probe succeeded) to its family when the metadata map proves membership;
// otherwise the key stays a single-member logical field.
func (m *fieldMapper) logicalForResolvedColumn(ctx context.Context, orgID valuer.UUID, field *telemetrytypes.TelemetryFieldKey, keys map[string][]*telemetrytypes.TelemetryFieldKey) *telemetrytypes.LogicalField {
for _, logical := range querybuilder.MatchingLogicalFields(ctx, orgID, m.fl, field, keys) {
if logical.IsFamily() &&
logical.FieldContext == field.FieldContext &&
(field.FieldDataType == telemetrytypes.FieldDataTypeUnspecified || logical.FieldDataType == field.FieldDataType) {
return logical
}
}
return telemetrytypes.SingleLogicalField(field.Name, field)
}
// upgradeToFamilies swaps single-member candidates for their family when the
// metadata map proves membership. Candidate order and every non-family
// candidate stay exactly as the legacy flow produced them; sibling candidates
@@ -419,9 +417,11 @@ func (m *fieldMapper) ColumnExpressionFor(
var candidates []*telemetrytypes.LogicalField
switch _, err := m.FieldFor(ctx, orgID, startNs, endNs, field); {
case err == nil:
// A directly-resolvable key upgrades to its family when the metadata
// map proves membership; otherwise it stays single-member.
candidates = []*telemetrytypes.LogicalField{m.logicalForResolvedColumn(ctx, orgID, field, keys)}
// Every match from metadata is kept, similar to the filter path.
candidates = querybuilder.MatchingLogicalFields(ctx, orgID, m.fl, field, keys)
if len(candidates) == 0 {
candidates = []*telemetrytypes.LogicalField{telemetrytypes.SingleLogicalField(field.Name, field)}
}
case errors.Is(err, qbtypes.ErrColumnNotFound):
// The legacy candidate flow, unchanged: column (when the bare name is
// one) plus metadata matches, else synthesized type-variant keys. The
@@ -595,15 +595,37 @@ func (m *fieldMapper) CandidateKeys(ctx context.Context, _ valuer.UUID, field *t
// honored as-is: the stripped name lives in the attribute maps
stripped := telemetrytypes.NewTelemetryFieldKey(field.Name, telemetrytypes.FieldContextUnspecified, field.FieldDataType)
return querybuilder.SynthesizeKeys(stripped, value)
case telemetrytypes.FieldContextAttribute, telemetrytypes.FieldContextResource:
case telemetrytypes.FieldContextAttribute, telemetrytypes.FieldContextResource, telemetrytypes.FieldContextScope:
// strict context honored as-is: stripped interpretation first, literal spelling second
literal := telemetrytypes.NewTelemetryFieldKey(field.FieldContext.StringValue()+"."+field.Name, field.FieldContext, field.FieldDataType)
return append(querybuilder.SynthesizeKeys(field, value), querybuilder.SynthesizeKeys(literal, value)...)
}
// contexts that don't exist on spans (log, body, scope, …) have nothing to synthesize
// contexts that don't exist on spans (log, body, …) have nothing to synthesize
return nil
}
// scopeJSONExistsExpression renders the existence predicate for the scope JSON column, the one
// signal-specific case the generic querybuilder.ExistsExpression must not carry.
func scopeJSONExistsExpression(key *telemetrytypes.TelemetryFieldKey, fieldExpression string, exists bool) (string, bool) {
if key.FieldContext != telemetrytypes.FieldContextScope {
return "", false
}
// Declared String paths are non-Nullable (absent reads '' not NULL).
if f, ok := IntrinsicFields[key.Name]; ok && f.FieldContext == telemetrytypes.FieldContextScope {
if exists {
return fieldExpression + " <> ''", true
}
return fieldExpression + " = ''", true
}
// Scope attribute: the value expression casts the JSON path to String, which folds a missing
// key's NULL to '', so presence must test the raw path — drop the ::String cast.
path := strings.TrimSuffix(fieldExpression, "::String")
if exists {
return path + " IS NOT NULL", true
}
return path + " IS NULL", true
}
// ExistsFor implements the per-key existence primitive of qbtypes.FieldMapper.
func (m *fieldMapper) ExistsFor(
ctx context.Context,
@@ -620,5 +642,8 @@ func (m *fieldMapper) ExistsFor(
if err != nil {
return "", err
}
if expr, ok := scopeJSONExistsExpression(key, fieldExpression, exists); ok {
return expr, nil
}
return querybuilder.ExistsExpression(columns, key, tsStart, tsEnd, fieldExpression, exists)
}

View File

@@ -84,6 +84,45 @@ func TestGetFieldKeyName(t *testing.T) {
expectedResult: "multiIf(resource.`deployment.environment` IS NOT NULL, resource.`deployment.environment`::String, `resource_string_deployment$$environment_exists`, `resource_string_deployment$$environment`, NULL)",
expectedError: nil,
},
{
name: "Scope field - scope.name",
key: telemetrytypes.TelemetryFieldKey{
Name: "scope.name",
FieldContext: telemetrytypes.FieldContextScope,
},
expectedResult: "scope.name::String",
expectedError: nil,
},
{
name: "Scope field - scope.version",
key: telemetrytypes.TelemetryFieldKey{
Name: "scope.version",
FieldContext: telemetrytypes.FieldContextScope,
},
expectedResult: "scope.version::String",
expectedError: nil,
},
{
name: "Scope field - custom attribute",
key: telemetrytypes.TelemetryFieldKey{
Name: "custom.attr",
FieldContext: telemetrytypes.FieldContextScope,
},
expectedResult: "scope.attributes.`custom.attr`::String",
expectedError: nil,
},
{
// `scope.attribute.name` normalizes to {attribute.name, scope}; the literal
// `attribute.` prefix is dropped so it addresses the scope attribute named `name`
// (which the declared `scope.name` path deliberately does not).
name: "Scope field - attribute prefix addresses the named scope attribute",
key: telemetrytypes.TelemetryFieldKey{
Name: "attribute.name",
FieldContext: telemetrytypes.FieldContextScope,
},
expectedResult: "scope.attributes.`name`::String",
expectedError: nil,
},
{
// Query like `attribute.attribute_string:string` should resolve to `attributes_string['attribute_string']`.
name: "Attribute key whose name collides with contextual map column resolves as a map lookup",
@@ -304,3 +343,99 @@ func TestColumnExpressionForTimestampAttributeCollision(t *testing.T) {
assert.Contains(t, result, "attributes_number['timestamp']")
})
}
// TestColumnExpressionForScopeDeclaredPath covers select-side resolution of scope names that
// collide with a declared scope path. A short name under scope context (or the bare
// `scope.<x>` spelling that normalizes to it) names both homes and coalesces them when
// metadata knows a same-named scope attribute, and binds to the declared path alone when it
// does not. The full `scope.<x>` name under explicit scope context addresses the declared
// path alone; the explicit `scope.attribute.` prefix addresses the attribute alone.
func TestColumnExpressionForScopeDeclaredPath(t *testing.T) {
ctx := context.Background()
scopeKey := func(name string) *telemetrytypes.TelemetryFieldKey {
return &telemetrytypes.TelemetryFieldKey{
Name: name,
Signal: telemetrytypes.SignalTraces,
FieldContext: telemetrytypes.FieldContextScope,
FieldDataType: telemetrytypes.FieldDataTypeString,
}
}
declaredOnly := map[string][]*telemetrytypes.TelemetryFieldKey{
"scope.name": {scopeKey("scope.name")},
"scope.version": {scopeKey("scope.version")},
}
withAttr := map[string][]*telemetrytypes.TelemetryFieldKey{
"scope.name": {scopeKey("scope.name")},
"scope.version": {scopeKey("scope.version")},
"name": {scopeKey("name")},
"version": {scopeKey("version")},
}
testCases := []struct {
name string
key telemetrytypes.TelemetryFieldKey
keys map[string][]*telemetrytypes.TelemetryFieldKey
expectedResult string
}{
{
name: "short name under scope context binds to the declared path",
key: telemetrytypes.TelemetryFieldKey{Name: "version", FieldContext: telemetrytypes.FieldContextScope},
keys: declaredOnly,
expectedResult: "multiIf(scope.version::String <> '', scope.version::String, NULL)",
},
{
name: "full scope.version name under scope context addresses the declared path alone",
key: telemetrytypes.TelemetryFieldKey{Name: "scope.version", FieldContext: telemetrytypes.FieldContextScope},
keys: withAttr,
expectedResult: "multiIf(scope.version::String <> '', scope.version::String, NULL)",
},
{
name: "full scope.name name under scope context addresses the declared path alone",
key: telemetrytypes.TelemetryFieldKey{Name: "scope.name", FieldContext: telemetrytypes.FieldContextScope},
keys: withAttr,
expectedResult: "multiIf(scope.name::String <> '', scope.name::String, NULL)",
},
{
// `scope.attribute.name` normalizes to {attribute.name, scope}; the `attribute.`
// prefix is dropped so it addresses the scope attribute named `name` — the only
// way to reach it, since `scope.name` is reserved for the declared path.
name: "attribute prefix reaches the named scope attribute",
key: telemetrytypes.TelemetryFieldKey{Name: "attribute.name", FieldContext: telemetrytypes.FieldContextScope},
keys: withAttr,
expectedResult: "multiIf(scope.attributes.`name` IS NOT NULL, scope.attributes.`name`::String, NULL)",
},
{
// the caller supplied the context, so `scope.` is part of the name rather than a
// prefix to strip: this addresses a scope attribute literally named
// `scope.testing.env`, not the attribute `testing.env`
name: "explicit context keeps a scope-prefixed name intact",
key: telemetrytypes.TelemetryFieldKey{Name: "scope.testing.env", FieldContext: telemetrytypes.FieldContextScope},
keys: declaredOnly,
expectedResult: "multiIf(scope.attributes.`scope.testing.env` IS NOT NULL, scope.attributes.`scope.testing.env`::String, NULL)",
},
{
// metadata knows both homes under this name, so the short spelling coalesces
// them instead of being rejected as ambiguous
name: "short name coalesces a known scope attribute with the declared path",
key: telemetrytypes.TelemetryFieldKey{Name: "name", FieldContext: telemetrytypes.FieldContextScope},
keys: withAttr,
expectedResult: "multiIf(scope.attributes.`name` IS NOT NULL, toString(scope.attributes.`name`::String), scope.name::String <> '', toString(scope.name::String), NULL)",
},
{
name: "short version coalesces a known scope attribute with the declared path",
key: telemetrytypes.TelemetryFieldKey{Name: "version", FieldContext: telemetrytypes.FieldContextScope},
keys: withAttr,
expectedResult: "multiIf(scope.attributes.`version` IS NOT NULL, toString(scope.attributes.`version`::String), scope.version::String <> '', toString(scope.version::String), NULL)",
},
}
for _, tc := range testCases {
t.Run(tc.name, func(t *testing.T) {
fm := NewFieldMapper(flaggertest.New(t))
result, err := fm.ColumnExpressionFor(ctx, valuer.UUID{}, 0, 0, &tc.key, telemetrytypes.FieldDataTypeUnspecified, tc.keys)
require.NoError(t, err)
assert.Equal(t, tc.expectedResult, result)
})
}
}

View File

@@ -113,6 +113,20 @@ func BuildCompleteFieldKeyMap(releaseTime time.Time) map[string][]*telemetrytype
FieldDataType: telemetrytypes.FieldDataTypeBool,
},
},
"scope.name": {
{
Name: "scope.name",
FieldContext: telemetrytypes.FieldContextScope,
FieldDataType: telemetrytypes.FieldDataTypeString,
},
},
"scope.version": {
{
Name: "scope.version",
FieldContext: telemetrytypes.FieldContextScope,
FieldDataType: telemetrytypes.FieldDataTypeString,
},
},
// both spellings of an enabled semantic-convention family
"deployment.environment.name": {
{

View File

@@ -15,11 +15,12 @@ type PostableFieldKeysParams struct {
Limit int `query:"limit"`
}
// existingQuery is unsupported until the computed per-trace aggregates it may
// reference can be narrowed on.
// existingQuery may reference the computed per-trace aggregates, which are
// never ingested; those filters are stripped before narrowing values.
type PostableFieldValueParams struct {
PostableFieldKeysParams
Name string `query:"name"`
Name string `query:"name"`
ExistingQuery string `query:"existingQuery"`
}
func NewFieldKeySelectorFromPostableFieldKeysParams(params PostableFieldKeysParams) *telemetrytypes.FieldKeySelector {
@@ -30,6 +31,7 @@ func NewFieldValueSelectorFromPostableFieldValueParams(params PostableFieldValue
return telemetrytypes.NewFieldValueSelectorFromPostableFieldValueParams(telemetrytypes.PostableFieldValueParams{
PostableFieldKeysParams: params.telemetryParams(),
Name: params.Name,
ExistingQuery: params.ExistingQuery,
})
}

View File

@@ -1,5 +1,7 @@
package aiobservabilitytypes
import "strings"
// OpenTelemetry gen_ai semantic-convention attribute keys. Single source of truth
// shared by the AI query builder and the LLM pricing pipeline.
const (
@@ -18,6 +20,20 @@ const (
GenAIOutputMessages = "gen_ai.output.messages"
)
// GenAISpanGateKeys mark a span as gen_ai: an LLM call, a tool call, or an
// agent span. A trace belongs to the AI explorer when any span carries one.
var GenAISpanGateKeys = []string{GenAIRequestModel, GenAIToolName, GenAIAgentName}
// GenAISpanFilterExpression renders the gate as a query-builder filter
// expression: each gate key ORed on EXISTS.
func GenAISpanFilterExpression() string {
exprs := make([]string, 0, len(GenAISpanGateKeys))
for _, key := range GenAISpanGateKeys {
exprs = append(exprs, key+" EXISTS")
}
return strings.Join(exprs, " OR ")
}
// Per-span costs the SigNoz LLM pricing processor attaches; not OTel semconv.
const (
SignozGenAICostInput = "_signoz.gen_ai.cost_input"

View File

@@ -374,7 +374,7 @@ func (q *QueryBuilderQuery[T]) validateAggregations(cfg validationConfig) error
return nil
}
func (m MetricAggregation) ValidateForType() error {
func (m MetricAggregation) ValidateForTypeAndTemporality() error {
if m.SpaceAggregation.IsPercentile() && !m.Type.IsPercentileSpaceAggregationAllowed() {
return errors.Newf(
errors.TypeInvalidInput,
@@ -384,6 +384,17 @@ func (m MetricAggregation) ValidateForType() error {
m.Type.StringValue(),
)
}
// reading a step's distribution out of a cumulative sketch would mean
// subtracting the previous point's sketch, which ClickHouse cannot do
if m.Type == metrictypes.ExpHistogramType && m.Temporality != metrictypes.Delta {
return errors.Newf(
errors.TypeUnsupported,
errors.CodeUnsupported,
"metric `%s` is an exponential histogram recorded with `%s` temporality, which cannot be queried; only `delta` exponential histograms are supported",
m.MetricName,
m.Temporality.StringValue(),
)
}
return nil
}

View File

@@ -1517,10 +1517,11 @@ func TestNonAggregationFieldsSkipped(t *testing.T) {
})
}
func TestMetricAggregationValidateForType(t *testing.T) {
func TestMetricAggregationValidateForTypeAndTemporality(t *testing.T) {
cases := []struct {
name string
metricType metrictypes.Type
temporality metrictypes.Temporality
spaceAggregation metrictypes.SpaceAggregation
comparisonParam *metrictypes.ComparisonSpaceAggregationParam
wantErr bool
@@ -1532,11 +1533,33 @@ func TestMetricAggregationValidateForType(t *testing.T) {
wantErr: false,
},
{
name: "percentile on exponential histogram is allowed",
name: "percentile on delta exponential histogram is allowed",
metricType: metrictypes.ExpHistogramType,
temporality: metrictypes.Delta,
spaceAggregation: metrictypes.SpaceAggregationPercentile99,
wantErr: false,
},
{
name: "cumulative exponential histogram is not allowed",
metricType: metrictypes.ExpHistogramType,
temporality: metrictypes.Cumulative,
spaceAggregation: metrictypes.SpaceAggregationPercentile99,
wantErr: true,
},
{
name: "exponential histogram with unresolved temporality is not allowed",
metricType: metrictypes.ExpHistogramType,
temporality: metrictypes.Unknown,
spaceAggregation: metrictypes.SpaceAggregationPercentile99,
wantErr: true,
},
{
name: "cumulative histogram is unaffected by the exponential histogram rule",
metricType: metrictypes.HistogramType,
temporality: metrictypes.Cumulative,
spaceAggregation: metrictypes.SpaceAggregationPercentile95,
wantErr: false,
},
{
name: "percentile on summary is not allowed",
metricType: metrictypes.SummaryType,
@@ -1562,10 +1585,11 @@ func TestMetricAggregationValidateForType(t *testing.T) {
agg := MetricAggregation{
MetricName: "test_metric",
Type: tc.metricType,
Temporality: tc.temporality,
SpaceAggregation: tc.spaceAggregation,
ComparisonSpaceAggregationParam: tc.comparisonParam,
}
err := agg.ValidateForType()
err := agg.ValidateForTypeAndTemporality()
if tc.wantErr && err == nil {
t.Errorf("expected error, got nil")
}

View File

@@ -23,10 +23,11 @@ var (
const savedViewNameSuffixLen = 8
var (
SourceTraces = Source{valuer.NewString("traces")}
SourceLogs = Source{valuer.NewString("logs")}
SourceMetrics = Source{valuer.NewString("metrics")}
SourceMeter = Source{valuer.NewString("meter")}
SourceTraces = Source{valuer.NewString("traces")}
SourceLogs = Source{valuer.NewString("logs")}
SourceMetrics = Source{valuer.NewString("metrics")}
SourceMeter = Source{valuer.NewString("meter")}
SourceAIObservability = Source{valuer.NewString("ai_observability")}
)
type SavedView struct {
@@ -117,12 +118,13 @@ func (Source) Enum() []any {
SourceLogs,
SourceMetrics,
SourceMeter,
SourceAIObservability,
}
}
func (s Source) Validate() error {
switch s {
case SourceTraces, SourceLogs, SourceMetrics, SourceMeter:
case SourceTraces, SourceLogs, SourceMetrics, SourceMeter, SourceAIObservability:
return nil
default:
return errors.NewInvalidInputf(ErrCodeSavedViewInvalidInput, "invalid source: %s", s.StringValue())

View File

@@ -39,6 +39,7 @@ func TestSourceValidate(t *testing.T) {
{name: "logs", source: SourceLogs},
{name: "metrics", source: SourceMetrics},
{name: "meter", source: SourceMeter},
{name: "ai_observability", source: SourceAIObservability},
{name: "unknown is rejected", source: Source{valuer.NewString("bogus")}, expectError: true},
}

View File

@@ -173,6 +173,21 @@ func TestSavedViewSpecValidate(t *testing.T) {
},
expectError: false,
},
{
name: "builder_ai_query is valid",
spec: SavedViewSpec{
DisplayName: "My View",
PanelType: PanelTypeList,
RequestType: qbtypes.RequestTypeRaw,
Queries: []qbtypes.QueryEnvelope{{
Type: qbtypes.QueryTypeBuilderAI,
Spec: qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Signal: telemetrytypes.SignalTraces,
},
}},
},
expectError: false,
},
{
name: "graph panel query with no aggregation is still rejected",
spec: SavedViewSpec{

View File

@@ -18,7 +18,7 @@ import (
// - Use `scope.` prefix to explicitly indicate and enforce scope context. Example
// - `scope.name`
// - `scope.version`
// - `scope.my.custom.attribute` and `scope.attribute.my.custom.attribute` resolve to same attribute
// - `scope.my.custom.attribute` resolves to the `my.custom.attribute` scope attribute
//
// - Use `attribute.` to explicitly indicate and enforce attribute context. Example
// - `attribute.http.method`
@@ -190,7 +190,7 @@ func (FieldContext) Enum() []any {
FieldContextSpan,
FieldContextTrace,
FieldContextResource,
// FieldContextScope,
FieldContextScope,
FieldContextAttribute,
// FieldContextEvent,
FieldContextBody,

View File

@@ -294,6 +294,17 @@ func TestNormalize(t *testing.T) {
FieldDataType: FieldDataTypeString,
},
},
{
name: "Normalize keeps a prefix that does not match the set context",
input: TelemetryFieldKey{
Name: "scope.name",
FieldContext: FieldContextAttribute,
},
expected: TelemetryFieldKey{
Name: "scope.name",
FieldContext: FieldContextAttribute,
},
},
{
name: "Normalize body field",
input: TelemetryFieldKey{

View File

@@ -19,6 +19,7 @@ pytest_plugins = [
"fixtures.traces",
"fixtures.metrics",
"fixtures.queriercommon",
"fixtures.semconvfamilies",
"fixtures.metadata",
"fixtures.meter",
"fixtures.browser",

View File

@@ -132,7 +132,15 @@ class MetricsSample(ABC):
class MetricsExpHist(ABC):
"""Represents a row in the exp_hist table for exponential histograms."""
"""Represents a row in the exp_hist table for exponential histograms.
`observations` must be non-empty; ClickHouse folds it into the `sketch`
AggregateFunction state on insert.
TODO: take an ExponentialHistogramDataPoint and compute the sketch bytes the
way the collector does, so the fixture exercises the real write path instead
of having ClickHouse build the state.
"""
env: str
temporality: str
@@ -143,7 +151,7 @@ class MetricsExpHist(ABC):
sum: np.float64
min: np.float64
max: np.float64
sketch: bytes
observations: list[int]
flags: np.uint32
def __init__(
@@ -151,11 +159,7 @@ class MetricsExpHist(ABC):
metric_name: str,
fingerprint: np.uint64,
timestamp: datetime.datetime,
count: int,
sum_value: float,
min_value: float,
max_value: float,
sketch: bytes = b"",
observations: list[int],
temporality: str = "Unspecified",
env: str = "default",
flags: int = 0,
@@ -165,28 +169,13 @@ class MetricsExpHist(ABC):
self.metric_name = metric_name
self.fingerprint = fingerprint
self.unix_milli = np.int64(int(timestamp.timestamp() * 1e3))
self.count = np.uint64(count)
self.sum = np.float64(sum_value)
self.min = np.float64(min_value)
self.max = np.float64(max_value)
self.sketch = sketch
self.observations = observations
self.count = np.uint64(len(observations))
self.sum = np.float64(sum(observations))
self.min = np.float64(min(observations))
self.max = np.float64(max(observations))
self.flags = np.uint32(flags)
def to_row(self) -> list:
return [
self.env,
self.temporality,
self.metric_name,
self.fingerprint,
self.unix_milli,
self.count,
self.sum,
self.min,
self.max,
self.sketch,
self.flags,
]
class MetricsMetadata(ABC):
"""Represents a row in the metadata table for metric metadata."""
@@ -429,6 +418,73 @@ class Metrics(ABC):
return metrics
class ExpHistogramMetrics(ABC):
"""High-level exponential histogram representation. Produces both time series
and exp_hist entries."""
metric_name: str
labels: dict[str, str]
temporality: str
timestamp: datetime.datetime
observations: list[int]
@property
def time_series(self) -> MetricsTimeSeries:
return self._time_series
@property
def exp_hist(self) -> MetricsExpHist:
return self._exp_hist
def __init__(
self,
metric_name: str,
observations: list[int],
labels: dict[str, str] = {},
timestamp: datetime.datetime | None = None,
temporality: str = "Delta",
flags: int = 0,
description: str = "",
unit: str = "",
env: str = "default",
resource_attributes: dict[str, str] = {},
scope_attributes: dict[str, str] = {},
) -> None:
if timestamp is None:
timestamp = datetime.datetime.now()
self.metric_name = metric_name
self.labels = labels
self.temporality = temporality
self.timestamp = timestamp
self.observations = observations
self._time_series = MetricsTimeSeries(
metric_name=metric_name,
labels=labels,
timestamp=timestamp,
temporality=temporality,
description=description,
unit=unit,
# the querier resolves the metric type from this column, and only an
# ExponentialHistogram here routes the query to the sketch read
type_="ExponentialHistogram",
is_monotonic=False,
env=env,
resource_attrs=resource_attributes,
scope_attrs=scope_attributes,
)
self._exp_hist = MetricsExpHist(
metric_name=metric_name,
fingerprint=self._time_series.fingerprint,
timestamp=timestamp,
observations=observations,
temporality=temporality,
env=env,
flags=flags,
)
class MetricsReducedTimeSeries(ABC):
"""Represents a row in the time_series_v4_reduced table i.e what
the time_series_v4_reduced_mv materializes for a metric under a
@@ -686,6 +742,47 @@ class MetricsBufferSample(ABC):
]
def insert_time_series_to_clickhouse(conn, time_series: list[MetricsTimeSeries]) -> None:
"""
Insert one distributed_time_series_v4 registration row per (series, hour
bucket), unix_milli floored to the hour — the exporter's exact shape.
Readers floor lookup windows to these buckets: skipping per-bucket
re-registration or keeping raw mid-hour timestamps hides series in ways
production never sees.
"""
time_series_map: dict[tuple[int, int], MetricsTimeSeries] = {}
for ts in time_series:
fp = int(ts.fingerprint)
hour_bucket = int(ts.unix_milli) // 3_600_000
if (fp, hour_bucket) not in time_series_map:
ts.unix_milli = np.int64(hour_bucket * 3_600_000)
time_series_map[(fp, hour_bucket)] = ts
if len(time_series_map) == 0:
return
conn.insert(
database="signoz_metrics",
table="distributed_time_series_v4",
column_names=[
"env",
"temporality",
"metric_name",
"description",
"unit",
"type",
"is_monotonic",
"fingerprint",
"unix_milli",
"labels",
"attrs",
"scope_attrs",
"resource_attrs",
],
data=[ts.to_row() for ts in time_series_map.values()],
)
def insert_metrics_to_clickhouse(conn, metrics: list[Metrics]) -> None:
"""
Insert metrics into ClickHouse tables.
@@ -697,39 +794,7 @@ def insert_metrics_to_clickhouse(conn, metrics: list[Metrics]) -> None:
Pure function so the seeder container can reuse the exact insert path
used by the pytest fixture. `conn` is a clickhouse-connect Client.
"""
# One registration row per (series, hour bucket), unix_milli floored to
# the hour — the exporter's exact shape. Readers floor lookup windows to
# these buckets: skipping per-bucket re-registration or keeping raw
# mid-hour timestamps hides series in ways production never sees.
time_series_map: dict[tuple[int, int], MetricsTimeSeries] = {}
for metric in metrics:
fp = int(metric.time_series.fingerprint)
hour_bucket = int(metric.time_series.unix_milli) // 3_600_000
if (fp, hour_bucket) not in time_series_map:
metric.time_series.unix_milli = np.int64(hour_bucket * 3_600_000)
time_series_map[(fp, hour_bucket)] = metric.time_series
if len(time_series_map) > 0:
conn.insert(
database="signoz_metrics",
table="distributed_time_series_v4",
column_names=[
"env",
"temporality",
"metric_name",
"description",
"unit",
"type",
"is_monotonic",
"fingerprint",
"unix_milli",
"labels",
"attrs",
"scope_attrs",
"resource_attrs",
],
data=[ts.to_row() for ts in time_series_map.values()],
)
insert_time_series_to_clickhouse(conn, [metric.time_series for metric in metrics])
samples = [metric.sample for metric in metrics]
if len(samples) > 0:
@@ -748,6 +813,15 @@ def insert_metrics_to_clickhouse(conn, metrics: list[Metrics]) -> None:
data=[sample.to_row() for sample in samples],
)
insert_metrics_metadata_to_clickhouse(conn, metrics)
def insert_metrics_metadata_to_clickhouse(conn, metrics: list) -> None:
"""
Insert the distributed_metadata rows describing each metric's point, resource
and scope attributes. Accepts anything exposing `time_series`, `labels` and
`timestamp`.
"""
# (metric_name, attr_type, attr_name, attr_value) -> MetricsMetadata
metadata_map: dict[tuple, MetricsMetadata] = {}
for metric in metrics:
@@ -853,6 +927,61 @@ def insert_metrics(
)
def insert_exp_histogram_metrics_to_clickhouse(conn, metrics: list[ExpHistogramMetrics]) -> None:
"""
Insert exponential histograms into ClickHouse tables.
Handles insertion into:
- distributed_time_series_v4 (time series metadata)
- distributed_exp_hist (per-point sketches)
- distributed_metadata (metric attribute metadata)
"""
insert_time_series_to_clickhouse(conn, [metric.time_series for metric in metrics])
# `sketch` is AggregateFunction(quantilesDD(...), UInt64) — the state has to be
# folded server-side, it cannot be sent as a literal. The quantilesDDState
# parameters must match the column's exactly or the INSERT is rejected.
for metric in metrics:
hist = metric.exp_hist
conn.command(
"INSERT INTO signoz_metrics.distributed_exp_hist "
"(env, temporality, metric_name, fingerprint, unix_milli, count, sum, min, max, sketch, flags) "
"SELECT %(env)s, %(temporality)s, %(metric_name)s, %(fingerprint)s, %(unix_milli)s, "
"%(count)s, %(sum)s, %(min)s, %(max)s, "
"quantilesDDState(0.01, 0.5, 0.75, 0.9, 0.95, 0.99)(toUInt64(observation)), %(flags)s "
"FROM (SELECT arrayJoin(%(observations)s) AS observation)",
parameters={
"env": hist.env,
"temporality": hist.temporality,
"metric_name": hist.metric_name,
"fingerprint": int(hist.fingerprint),
"unix_milli": int(hist.unix_milli),
"count": int(hist.count),
"sum": float(hist.sum),
"min": float(hist.min),
"max": float(hist.max),
"observations": hist.observations,
"flags": int(hist.flags),
},
)
insert_metrics_metadata_to_clickhouse(conn, metrics)
@pytest.fixture(name="insert_exp_histogram_metrics", scope="function")
def insert_exp_histogram_metrics(
clickhouse: types.TestContainerClickhouse,
) -> Generator[Callable[[list[ExpHistogramMetrics]], None], Any]:
def _insert_exp_histogram_metrics(metrics: list[ExpHistogramMetrics]) -> None:
insert_exp_histogram_metrics_to_clickhouse(clickhouse.conn, metrics)
yield _insert_exp_histogram_metrics
truncate_metrics_tables(
clickhouse.conn,
clickhouse.env["SIGNOZ_TELEMETRYSTORE_CLICKHOUSE_CLUSTER"],
)
def insert_reduced_metrics_to_clickhouse(
conn,
time_series: list[MetricsReducedTimeSeries],

View File

@@ -999,6 +999,8 @@ def generate_traces_with_corrupt_metadata() -> list[Traces]:
"cloud.provider": "integration",
"cloud.account.id": "000",
"trace_id": "corrupt_data",
"scope_name": "corrupt_data",
"scope.scope.name": "corrupt_data",
},
attributes={
"net.transport": "IP.TCP",
@@ -1007,7 +1009,10 @@ def generate_traces_with_corrupt_metadata() -> list[Traces]:
"http.request.method": "POST",
"http.response.status_code": "200",
"timestamp": "corrupt_data",
"version": "1.0.0",
"scope.scope.version": "1.0.0",
},
scope={"name": "io.signoz.http.server", "version": "2.0.0"},
),
Traces(
timestamp=now - timedelta(seconds=3.5),
@@ -1027,12 +1032,24 @@ def generate_traces_with_corrupt_metadata() -> list[Traces]:
"cloud.provider": "integration",
"cloud.account.id": "000",
"timestamp": "corrupt_data",
"scope.attributes.name": "corrupt_data",
},
attributes={
"db.name": "integration",
"db.operation": "SELECT",
"db.statement": "SELECT * FROM integration",
"trace_d": "corrupt_data",
"scope.attributes.version": "corrupt_data",
},
scope={
"name": "io.opentelemetry.contrib.http",
"version": "1.0.0",
"attributes": {
"telemetry.sdk.language": "cpp",
"name": "not-the-real-name",
"version": "not-the-real-version",
"attributes": "literally-a-key-named-attributes",
},
},
),
Traces(
@@ -1053,12 +1070,15 @@ def generate_traces_with_corrupt_metadata() -> list[Traces]:
"cloud.provider": "integration",
"cloud.account.id": "000",
"duration_nano": "corrupt_data",
"scope.scope.attributes.version": "corrupt_data",
},
attributes={
"http.request.method": "PATCH",
"http.status_code": "404",
"id": "1",
"scope.scope.version": "corrupt_data",
},
scope={"name": "io.signoz.http.client", "version": "2.0.0"},
),
Traces(
timestamp=now - timedelta(seconds=1),
@@ -1077,6 +1097,7 @@ def generate_traces_with_corrupt_metadata() -> list[Traces]:
"host.name": "linux-001",
"cloud.provider": "integration",
"cloud.account.id": "001",
"scope.scope.version": "corrupt_data",
},
attributes={
"message.type": "SENT",
@@ -1084,7 +1105,10 @@ def generate_traces_with_corrupt_metadata() -> list[Traces]:
"messaging.message.id": "001",
"duration_nano": "corrupt_data",
"id": 1,
"scope": "corrupt_data",
"scope.attributes.name": "corrupt_data",
},
scope={"name": "io.signoz.messaging", "version": "3.0.0"},
),
]

View File

@@ -1,5 +1,16 @@
from datetime import datetime, timedelta
from http import HTTPStatus
from fixtures import types
from fixtures.querier import (
Aggregation,
BuilderQuery,
OrderBy,
RequestType,
TelemetryFieldKey,
get_scalar_table_data,
make_query_request,
)
from fixtures.traces import TraceIdGenerator, Traces, TracesKind, TracesStatusCode
@@ -32,10 +43,10 @@ def ai_trace(
*,
now: datetime,
service: str,
user: str,
in_tokens: int | None,
out_tokens: int,
cost: float,
user: str = "user",
cost: float = 0.1,
model: str = "gpt-4o-mini",
environment: str = "production",
) -> list[Traces]:
@@ -74,6 +85,28 @@ def ai_trace(
]
def tool_only_trace(*, now: datetime, service: str) -> list[Traces]:
"""Root + one tool span: passes the gen_ai gate but has NO LLM span."""
trace_id = TraceIdGenerator.trace_id()
root_id = TraceIdGenerator.span_id()
resources = {"service.name": service}
return [
root_span(now=now, trace_id=trace_id, span_id=root_id, resources=resources, duration_s=2),
Traces(
timestamp=now - timedelta(seconds=4),
duration=timedelta(seconds=0.5),
trace_id=trace_id,
span_id=TraceIdGenerator.span_id(),
parent_span_id=root_id,
name="execute_tool",
kind=TracesKind.SPAN_KIND_INTERNAL,
status_code=TracesStatusCode.STATUS_CODE_OK,
resources=resources,
attributes={"gen_ai.tool.name": "get_weather", "gen_ai.tool.type": "function"},
),
]
def ai_trace_mixed_spans(*, now: datetime, service: str, user: str) -> list[Traces]:
"""Root + LLM + tool + agent spans; only the LLM span carries gen_ai.request.model."""
trace_id = TraceIdGenerator.trace_id()
@@ -116,3 +149,48 @@ def ai_trace_mixed_spans(*, now: datetime, service: str, user: str) -> list[Trac
),
child("agent.step", TracesKind.SPAN_KIND_INTERNAL, {"gen_ai.agent.name": "chat-agent"}, 2),
]
def ai_aggregation_query(
service: str,
expression: str,
*,
filter_extra: str = "",
group_by: list[TelemetryFieldKey] | None = None,
alias: str | None = None,
having: str | None = None,
order: list[OrderBy] | None = None,
limit: int | None = None,
step_interval: int | None = None,
) -> dict:
filter_expression = f"service.name = '{service}'"
if filter_extra:
filter_expression += f" AND {filter_extra}"
return BuilderQuery(
signal="traces",
query_type="builder_ai_query",
name="A",
filter_expression=filter_expression,
aggregations=[Aggregation(expression=expression, alias=alias)],
group_by=group_by,
having_expression=having,
order=order,
limit=limit,
step_interval=step_interval,
).to_dict()
def scalar_value(signoz: types.SigNoz, token: str, start_ms: int, end_ms: int, service: str, expression: str, filter_extra: str = "") -> float:
"""The single cell of a one-aggregation, ungrouped scalar query."""
response = make_query_request(
signoz,
token,
start_ms,
end_ms,
[ai_aggregation_query(service, expression, filter_extra=filter_extra)],
request_type=RequestType.SCALAR,
)
assert response.status_code == HTTPStatus.OK, f"{expression}: {response.text}"
data = get_scalar_table_data(response.json())
assert len(data) == 1, f"{expression}: expected one row, got {data}"
return float(data[0][-1])

64
tests/fixtures/semconvfamilies.py vendored Normal file
View File

@@ -0,0 +1,64 @@
from collections.abc import Callable, Generator
from datetime import UTC, datetime, timedelta
import pytest
from fixtures.logs import Logs
from fixtures.traces import TraceIdGenerator, Traces, TracesKind, TracesStatusCode
PREFIX = "semconv-fam"
CURRENT_KEY = "deployment.environment.name"
OLD_KEY = "deployment.environment"
# Row identities. The span name, the log body, and service.name are the identity.
# Tests compare identity sets filtered by PREFIX, so reruns on a reused stack
# with leftover rows stay stable.
OLD = f"{PREFIX}-old" # only the old spelling, value "production"
NEW = f"{PREFIX}-new" # only the current spelling, value "production"
BOTH = f"{PREFIX}-both" # current "staging" and old "production" - the conflict row
NEITHER = f"{PREFIX}-neither" # no member at all
_ROWS = [
(OLD, {OLD_KEY: "production"}, timedelta(seconds=4)),
(NEW, {CURRENT_KEY: "production"}, timedelta(seconds=3)),
(BOTH, {CURRENT_KEY: "staging", OLD_KEY: "production"}, timedelta(seconds=2)),
(NEITHER, {}, timedelta(seconds=1)),
]
@pytest.fixture(name="family_fleet", scope="function")
def family_fleet(
insert_logs: Callable[[list[Logs]], None],
insert_traces: Callable[[list[Traces]], None],
) -> Generator[datetime]:
"""Yields the base timestamp of the inserted rows."""
now = datetime.now(tz=UTC).replace(microsecond=0) - timedelta(minutes=1)
insert_traces(
[
Traces(
timestamp=now - offset,
duration=timedelta(milliseconds=10),
trace_id=TraceIdGenerator.trace_id(),
span_id=TraceIdGenerator.span_id(),
name=identity,
kind=TracesKind.SPAN_KIND_SERVER,
status_code=TracesStatusCode.STATUS_CODE_OK,
resources={"service.name": identity, **family},
attributes=dict(family),
)
for identity, family, offset in _ROWS
]
)
insert_logs(
[
Logs(
timestamp=now - offset,
body=identity,
resources={"service.name": identity, **family},
attributes=dict(family),
)
for identity, family, offset in _ROWS
]
)
yield now

View File

@@ -302,6 +302,7 @@ class Traces(ABC):
db_operation: str
has_error: bool
is_remote: str
scope_json: dict[str, Any]
resource: list[TracesResource]
tag_attributes: list[TracesTagAttributes]
@@ -327,6 +328,7 @@ class Traces(ABC):
links: list[TracesLink] = [],
trace_state: str = "",
flags: np.uint32 = 0,
scope: dict[str, Any] = {},
resource_write_mode: Literal["legacy_only", "dual_write"] = "dual_write",
) -> None:
if timestamp is None:
@@ -408,6 +410,33 @@ class Traces(ABC):
# Calculate resource fingerprint
self.resource_fingerprint = LogsOrTracesFingerprint(self.resources_string).calculate()
# Process scope mirroring the InstrumentationScope on the OTLP span.
scope_name = scope.get("name", "")
scope_version = scope.get("version", "")
scope_string = {k: str(v) for k, v in scope.get("attributes", {}).items()}
self.scope_json = {
"name": scope_name,
"version": scope_version,
"attributes": scope_string,
}
scope_keys = {"scope.name": scope_name, "scope.version": scope_version}
scope_keys.update(scope_string)
for k, v in scope_keys.items():
if v == "":
continue
self.tag_attributes.append(
TracesTagAttributes(
timestamp=timestamp,
tag_key=k,
tag_type="scope",
tag_data_type="string",
string_value=v,
number_value=None,
)
)
self.attribute_keys.append(TracesResourceOrAttributeKeys(name=k, datatype="string", tag_type="scope"))
# Process attributes by type and populate custom fields
self.attribute_string = {}
self.attributes_number = {}
@@ -659,6 +688,7 @@ class Traces(ABC):
self.has_error,
self.is_remote,
self.resource_json,
self.scope_json,
],
dtype=object,
)
@@ -689,6 +719,7 @@ class Traces(ABC):
attributes=data.get("attributes", {}),
trace_state=data.get("trace_state", ""),
flags=data.get("flags", 0),
scope=data.get("scope", {}),
)
@classmethod
@@ -828,6 +859,7 @@ def insert_traces_to_clickhouse(conn, traces: list[Traces]) -> None:
"has_error",
"is_remote",
"resource",
"scope",
],
data=[trace.np_arr() for trace in traces],
)

View File

@@ -72,8 +72,8 @@ def test_ai_list_having_aggregate_filter(
get_token: Callable[[str, str], str],
insert_traces: Callable[[list[Traces]], None],
) -> None:
"""Span + aggregate condition in one filter box splits into WHERE + HAVING; bare
and `trace.` spellings behave identically; an output-only aggregate is rejected."""
"""One filter box splits into WHERE + HAVING; bare and `trace.` spellings behave
identically; an output-only aggregate is rejected."""
now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
service = "ai-it-having"
@@ -322,9 +322,8 @@ def test_ai_list_nested_group_span_or_and_aggregate(
get_token: Callable[[str, str], str],
insert_traces: Callable[[list[Traces]], None],
) -> None:
"""service.name = X AND (has_error = true OR gen_ai.request.model = 'gpt-4o') AND
total_tokens > 100: the nested OR group must not flatten, span predicates go to
WHERE, the aggregate to HAVING."""
"""A nested (span OR span) group ANDed with an aggregate must not flatten: span
predicates go to WHERE, the aggregate to HAVING."""
now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
service = "ai-it-nested"

View File

@@ -0,0 +1,502 @@
from collections.abc import Callable
from datetime import UTC, datetime, timedelta
from http import HTTPStatus
import pytest
from fixtures import types
from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
from fixtures.querier import (
Aggregation,
BuilderQuery,
OrderBy,
RequestType,
TelemetryFieldKey,
get_all_series,
get_scalar_columns,
get_scalar_table_data,
get_series_values,
make_query_request,
)
from fixtures.querierai import ai_aggregation_query, ai_trace, query_window, scalar_value, tool_only_trace
from fixtures.traces import TraceIdGenerator, Traces, TracesKind, TracesStatusCode
def test_ai_scalar_trace_level_aggregations(
signoz: types.SigNoz,
create_user_admin: None, # pylint: disable=unused-argument
get_token: Callable[[str, str], str],
insert_traces: Callable[[list[Traces]], None],
) -> None:
"""Scalars over per-trace values, and the bare-key span domain through the same request type."""
now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
service = "ai-it-agg-scalar"
insert_traces(ai_trace(now=now, service=service, in_tokens=10, out_tokens=100) + ai_trace(now=now, service=service, in_tokens=30, out_tokens=300))
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
start_ms, end_ms = query_window(now)
def value(expression: str) -> float:
return scalar_value(signoz, token, start_ms, end_ms, service, expression)
assert value("avg(trace.output_tokens)") == pytest.approx(200)
assert value("count(trace.trace_id)") == 2
assert value("max(trace.total_tokens)") == pytest.approx(330)
assert value("p50(trace.output_tokens)") == pytest.approx(200) # AggreFuncMap -> quantile(0.50)
# arithmetic inside one function and between functions
assert value("avg(trace.output_tokens + trace.input_tokens)") == pytest.approx(220)
assert value("sum(trace.output_tokens)/count(trace.trace_id)") == pytest.approx(200)
assert value("count()") == 2 # the two LLM spans; roots are not gen_ai
assert value("sum(gen_ai.usage.output_tokens)") == pytest.approx(400)
# multiple trace-level aggregations in one query -> one column per aggregation
multi = BuilderQuery(
signal="traces",
query_type="builder_ai_query",
name="A",
filter_expression=f"service.name = '{service}'",
aggregations=[Aggregation(expression="avg(trace.output_tokens)"), Aggregation(expression="count(trace.trace_id)")],
)
resp = make_query_request(signoz, token, start_ms, end_ms, [multi.to_dict()], request_type=RequestType.SCALAR)
assert resp.status_code == HTTPStatus.OK, resp.text
data = get_scalar_table_data(resp.json())
assert len(data) == 1 and [float(v) for v in data[0]] == [pytest.approx(200), 2], data
def test_ai_scalar_trace_level_filter_qualifies_traces(
signoz: types.SigNoz,
create_user_admin: None, # pylint: disable=unused-argument
get_token: Callable[[str, str], str],
insert_traces: Callable[[list[Traces]], None],
) -> None:
"""A trace-level condition qualifies whole traces before aggregation, on both domains."""
now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
service = "ai-it-agg-qualify"
insert_traces(ai_trace(now=now, service=service, in_tokens=10, out_tokens=100) + ai_trace(now=now, service=service, in_tokens=30, out_tokens=300))
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
start_ms, end_ms = query_window(now)
for expression in (
"sum(trace.output_tokens)", # native trace-domain path
"sum(gen_ai.usage.output_tokens)", # delegated span-domain path (__trace_scope)
):
got = scalar_value(signoz, token, start_ms, end_ms, service, expression, filter_extra="trace.output_tokens > 100")
assert got == pytest.approx(300), expression
# the qualification also constrains delegated (span-domain) time series
resp = make_query_request(
signoz,
token,
start_ms,
end_ms,
[
ai_aggregation_query(
service,
"sum(gen_ai.usage.output_tokens)",
filter_extra="trace.output_tokens > 100",
step_interval=60,
)
],
request_type=RequestType.TIME_SERIES,
)
assert resp.status_code == HTTPStatus.OK, resp.text
assert [v["value"] for v in get_series_values(resp.json(), "A")] == [pytest.approx(300)]
def test_ai_scalar_group_by_model(
signoz: types.SigNoz,
create_user_admin: None, # pylint: disable=unused-argument
get_token: Callable[[str, str], str],
insert_traces: Callable[[list[Traces]], None],
) -> None:
"""Trace-level aggregation grouped by a span attribute."""
now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
service = "ai-it-agg-groupby"
insert_traces(ai_trace(now=now, service=service, in_tokens=10, out_tokens=100, model="gpt-4o") + ai_trace(now=now, service=service, in_tokens=10, out_tokens=300, model="gpt-4o") + ai_trace(now=now, service=service, in_tokens=10, out_tokens=50, model="gpt-4o-mini"))
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
start_ms, end_ms = query_window(now)
resp = make_query_request(
signoz,
token,
start_ms,
end_ms,
[ai_aggregation_query(service, "avg(trace.output_tokens)", group_by=[TelemetryFieldKey(name="gen_ai.request.model")])],
request_type=RequestType.SCALAR,
)
assert resp.status_code == HTTPStatus.OK, resp.text
data = get_scalar_table_data(resp.json())
by_model = {row[0]: float(row[-1]) for row in data}
assert by_model == {"gpt-4o": pytest.approx(200), "gpt-4o-mini": pytest.approx(50)}, data
def test_ai_scalar_group_by_intrinsic_span_column(
signoz: types.SigNoz,
create_user_admin: None, # pylint: disable=unused-argument
get_token: Callable[[str, str], str],
insert_traces: Callable[[list[Traces]], None],
) -> None:
"""Grouping by an intrinsic must not alias the group column to the span column it reads
(`toString(name) AS name` is a cyclic alias ClickHouse rejects)."""
now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
service = "ai-it-agg-groupby-intrinsic"
insert_traces(ai_trace(now=now, service=service, in_tokens=10, out_tokens=100) + ai_trace(now=now, service=service, in_tokens=10, out_tokens=300) + tool_only_trace(now=now, service=service))
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
start_ms, end_ms = query_window(now)
resp = make_query_request(
signoz,
token,
start_ms,
end_ms,
[
ai_aggregation_query(
service,
"count(trace.trace_id)",
group_by=[TelemetryFieldKey(name="name")],
order=[OrderBy(key=TelemetryFieldKey(name="name"), direction="asc")],
)
],
request_type=RequestType.SCALAR,
)
assert resp.status_code == HTTPStatus.OK, resp.text
columns = get_scalar_columns(resp.json())
assert columns[0]["name"] == "name", columns
data = get_scalar_table_data(resp.json())
# the root spans are gated out, so each trace groups under its gen_ai span name
assert [(row[0], int(row[-1])) for row in data] == [("chat gpt-4o-mini", 2), ("execute_tool", 1)], data
def test_ai_timeseries_trace_level_aggregation(
signoz: types.SigNoz,
create_user_admin: None, # pylint: disable=unused-argument
get_token: Callable[[str, str], str],
insert_traces: Callable[[list[Traces]], None],
) -> None:
now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
service = "ai-it-agg-ts"
insert_traces(ai_trace(now=now, service=service, in_tokens=10, out_tokens=100) + ai_trace(now=now, service=service, in_tokens=30, out_tokens=300))
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
start_ms, end_ms = query_window(now)
# all spans fall in one step bucket
resp = make_query_request(
signoz,
token,
start_ms,
end_ms,
[ai_aggregation_query(service, "avg(trace.output_tokens)", step_interval=60)],
request_type=RequestType.TIME_SERIES,
)
assert resp.status_code == HTTPStatus.OK, resp.text
assert [v["value"] for v in get_series_values(resp.json(), "A")] == [pytest.approx(200)]
def test_ai_timeseries_top_n_groups(
signoz: types.SigNoz,
create_user_admin: None, # pylint: disable=unused-argument
get_token: Callable[[str, str], str],
insert_traces: Callable[[list[Traces]], None],
) -> None:
"""A grouped, limited time series ranks groups on whole-window per-trace values in
the requested order."""
now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
service = "ai-it-agg-topn"
insert_traces(ai_trace(now=now, service=service, in_tokens=10, out_tokens=300, model="gpt-4o") + ai_trace(now=now, service=service, in_tokens=10, out_tokens=100, model="gpt-4o") + ai_trace(now=now, service=service, in_tokens=10, out_tokens=50, model="gpt-4o-mini"))
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
start_ms, end_ms = query_window(now)
def top_series(order: list[OrderBy] | None) -> dict:
resp = make_query_request(
signoz,
token,
start_ms,
end_ms,
[
ai_aggregation_query(
service,
"sum(trace.output_tokens)",
group_by=[TelemetryFieldKey(name="gen_ai.request.model")],
alias="total_out",
order=order,
limit=1,
step_interval=60,
)
],
request_type=RequestType.TIME_SERIES,
)
assert resp.status_code == HTTPStatus.OK, resp.text
series = get_all_series(resp.json(), "A")
assert len(series) == 1, f"limit=1 must keep exactly one group, got {len(series)} series"
return series[0]
top = top_series(None) # default ranking: first aggregation desc
assert top["labels"][0]["value"] == "gpt-4o", top["labels"]
assert [v["value"] for v in top["values"]] == [pytest.approx(400)]
bottom = top_series([OrderBy(key=TelemetryFieldKey(name="total_out"), direction="asc")])
assert bottom["labels"][0]["value"] == "gpt-4o-mini", bottom["labels"]
assert [v["value"] for v in bottom["values"]] == [pytest.approx(50)]
def test_ai_scalar_group_order_limit(
signoz: types.SigNoz,
create_user_admin: None, # pylint: disable=unused-argument
get_token: Callable[[str, str], str],
insert_traces: Callable[[list[Traces]], None],
) -> None:
"""Scalar limit is a plain top-N over the grouped rows."""
now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
service = "ai-it-agg-scalar-limit"
insert_traces(
ai_trace(now=now, service=service, in_tokens=10, out_tokens=300, model="gpt-4o")
+ ai_trace(now=now, service=service, in_tokens=10, out_tokens=100, model="gpt-4o")
+ ai_trace(now=now, service=service, in_tokens=10, out_tokens=50, model="gpt-4o-mini")
+ ai_trace(now=now, service=service, in_tokens=10, out_tokens=10, model="gpt-4")
)
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
start_ms, end_ms = query_window(now)
resp = make_query_request(
signoz,
token,
start_ms,
end_ms,
[
ai_aggregation_query(
service,
"sum(trace.output_tokens)",
group_by=[TelemetryFieldKey(name="gen_ai.request.model")],
alias="total_out",
order=[OrderBy(key=TelemetryFieldKey(name="total_out"), direction="desc")],
limit=2,
)
],
request_type=RequestType.SCALAR,
)
assert resp.status_code == HTTPStatus.OK, resp.text
data = get_scalar_table_data(resp.json())
assert [(row[0], float(row[-1])) for row in data] == [("gpt-4o", pytest.approx(400)), ("gpt-4o-mini", pytest.approx(50))], data
def test_ai_timeseries_span_time_bucketing(
signoz: types.SigNoz,
create_user_admin: None, # pylint: disable=unused-argument
get_token: Callable[[str, str], str],
insert_traces: Callable[[list[Traces]], None],
) -> None:
"""Per-trace values are clipped per (bucket, trace), so a trace spanning two buckets
contributes each call's tokens to its own bucket, not the total to both."""
now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
service = "ai-it-agg-buckets"
trace_id = TraceIdGenerator.trace_id()
root_id = TraceIdGenerator.span_id()
resources = {"service.name": service}
def llm(offset_s: float, out_tokens: int) -> Traces:
return Traces(
timestamp=now - timedelta(seconds=offset_s),
duration=timedelta(seconds=1),
trace_id=trace_id,
span_id=TraceIdGenerator.span_id(),
parent_span_id=root_id,
name="chat",
kind=TracesKind.SPAN_KIND_CLIENT,
status_code=TracesStatusCode.STATUS_CODE_OK,
resources=resources,
attributes={"gen_ai.request.model": "gpt-4o-mini", "gen_ai.usage.output_tokens": out_tokens},
)
root = Traces(
timestamp=now - timedelta(seconds=130),
duration=timedelta(seconds=130),
trace_id=trace_id,
span_id=root_id,
parent_span_id="",
name="POST /api/chat",
kind=TracesKind.SPAN_KIND_SERVER,
status_code=TracesStatusCode.STATUS_CODE_OK,
resources=resources,
attributes={"http.request.method": "POST"},
)
# two LLM calls two minutes apart
insert_traces([root, llm(124, 100), llm(4, 300)])
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
start_ms, end_ms = query_window(now)
resp = make_query_request(
signoz,
token,
start_ms,
end_ms,
[ai_aggregation_query(service, "avg(trace.output_tokens)", step_interval=60)],
request_type=RequestType.TIME_SERIES,
)
assert resp.status_code == HTTPStatus.OK, resp.text
series = get_all_series(resp.json(), "A")
assert len(series) == 1, series
assert sorted(v["value"] for v in series[0]["values"]) == [pytest.approx(100), pytest.approx(300)], series
def test_ai_scalar_variables_in_trace_level_filter(
signoz: types.SigNoz,
create_user_admin: None, # pylint: disable=unused-argument
get_token: Callable[[str, str], str],
insert_traces: Callable[[list[Traces]], None],
) -> None:
"""Variables resolve inside trace-level conditions with span-filter semantics."""
now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
service = "ai-it-agg-vars"
insert_traces(ai_trace(now=now, service=service, in_tokens=10, out_tokens=100) + ai_trace(now=now, service=service, in_tokens=30, out_tokens=300))
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
start_ms, end_ms = query_window(now)
query = ai_aggregation_query(service, "sum(trace.output_tokens)", filter_extra="trace.output_tokens > $threshold")
resp = make_query_request(
signoz,
token,
start_ms,
end_ms,
[query],
request_type=RequestType.SCALAR,
variables={"threshold": {"type": "text", "value": 100}},
)
assert resp.status_code == HTTPStatus.OK, resp.text
data = get_scalar_table_data(resp.json())
assert len(data) == 1 and float(data[0][-1]) == pytest.approx(300), data
# an unresolvable $var is a 400 today via aggregate validation
resp = make_query_request(signoz, token, start_ms, end_ms, [query], request_type=RequestType.SCALAR)
assert resp.status_code == HTTPStatus.BAD_REQUEST, resp.text
# quotes in the message are JSON-escaped, so match the halves separately
assert "$threshold" in resp.text and "cannot be used in a trace-level filter" in resp.text, resp.text
# a dynamic variable resolved to __all__ drops the condition (both traces count)
resp = make_query_request(
signoz,
token,
start_ms,
end_ms,
[query],
request_type=RequestType.SCALAR,
variables={"threshold": {"type": "dynamic", "value": "__all__"}},
)
assert resp.status_code == HTTPStatus.OK, resp.text
data = get_scalar_table_data(resp.json())
assert len(data) == 1 and float(data[0][-1]) == pytest.approx(400), data
def test_ai_scalar_tool_only_trace_null_semantics(
signoz: types.SigNoz,
create_user_admin: None, # pylint: disable=unused-argument
get_token: Callable[[str, str], str],
insert_traces: Callable[[list[Traces]], None],
) -> None:
"""A tool-only trace (in the gate, no LLM span) follows plain SQL NULL semantics."""
now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
service = "ai-it-agg-toolonly"
insert_traces(ai_trace(now=now, service=service, in_tokens=10, out_tokens=100) + tool_only_trace(now=now, service=service))
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
start_ms, end_ms = query_window(now)
def value(expression: str, filter_extra: str = "") -> float:
return scalar_value(signoz, token, start_ms, end_ms, service, expression, filter_extra)
assert value("count(trace.trace_id)") == 2, "tool-only trace is an AI trace and must be counted"
assert value("avg(trace.output_tokens)") == pytest.approx(100), "NULL tokens are skipped by avg"
assert value("avg(trace.tool_call_count)") == pytest.approx(0.5), "tool-only trace feeds tool aggregates (1 and 0 calls)"
assert value("count()") == 2, "span-level count sees the LLM and the tool span"
# filtering on LLM activity is explicit, not implicit
assert value("count(trace.trace_id)", filter_extra="trace.llm_call_count > 0") == 1
def test_ai_scalar_having_on_aggregation(
signoz: types.SigNoz,
create_user_admin: None, # pylint: disable=unused-argument
get_token: Callable[[str, str], str],
insert_traces: Callable[[list[Traces]], None],
) -> None:
"""The outer having filters aggregation results per group (by alias)."""
now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
service = "ai-it-agg-having"
insert_traces(ai_trace(now=now, service=service, in_tokens=10, out_tokens=300, model="gpt-4o") + ai_trace(now=now, service=service, in_tokens=10, out_tokens=50, model="gpt-4o-mini"))
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
start_ms, end_ms = query_window(now)
resp = make_query_request(
signoz,
token,
start_ms,
end_ms,
[
ai_aggregation_query(
service,
"avg(trace.output_tokens)",
group_by=[TelemetryFieldKey(name="gen_ai.request.model")],
alias="avg_out",
having="avg_out > 100",
)
],
request_type=RequestType.SCALAR,
)
assert resp.status_code == HTTPStatus.OK, resp.text
data = get_scalar_table_data(resp.json())
assert len(data) == 1 and data[0][0] == "gpt-4o", data
def test_ai_aggregation_rejections(
signoz: types.SigNoz,
create_user_admin: None, # pylint: disable=unused-argument
get_token: Callable[[str, str], str],
insert_traces: Callable[[list[Traces]], None],
) -> None:
now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
service = "ai-it-agg-reject"
insert_traces(ai_trace(now=now, service=service, in_tokens=10, out_tokens=100))
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
start_ms, end_ms = query_window(now)
def expect_bad_request(query: dict, message: str) -> None:
resp = make_query_request(signoz, token, start_ms, end_ms, [query], request_type=RequestType.SCALAR)
assert resp.status_code == HTTPStatus.BAD_REQUEST, resp.text
assert message in resp.text, resp.text
# span-level and trace-level aggregations cannot be mixed in one query
mixed = BuilderQuery(
signal="traces",
query_type="builder_ai_query",
name="A",
filter_expression=f"service.name = '{service}'",
aggregations=[Aggregation(expression="avg(trace.output_tokens)"), Aggregation(expression="count()")],
)
expect_bad_request(mixed.to_dict(), "cannot be mixed")
expect_bad_request(
ai_aggregation_query(service, "avg(trace.output_tokens)", group_by=[TelemetryFieldKey(name="trace.llm_call_count")]),
"grouping by trace-level aggregate",
)
# a bare per-trace column would emit one row per trace instead of one aggregated row
expect_bad_request(ai_aggregation_query(service, "trace.output_tokens"), "must be inside an aggregation function")
# the rate interval divides the whole expression, so it may not carry a second aggregation
expect_bad_request(ai_aggregation_query(service, "rate(trace.trace_id) + avg(trace.output_tokens)"), "combines a rate with another aggregation")
# order-by is stopped earlier, by request validation
expect_bad_request(
ai_aggregation_query(service, "avg(trace.output_tokens)", order=[OrderBy(key=TelemetryFieldKey(name="trace.total_tokens"), direction="desc")]),
"invalid order by key",
)

View File

@@ -2,10 +2,12 @@ from collections.abc import Callable
from datetime import UTC, datetime
from http import HTTPStatus
import pytest
from fixtures import types
from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
from fixtures.metadata import get_field_keys, get_field_values
from fixtures.querierai import ai_trace
from fixtures.metadata import AttributesMetadata, get_field_keys, get_field_values
from fixtures.querierai import ai_trace, ai_trace_mixed_spans
from fixtures.traces import Traces
AI_KEYS_PATH = "/api/v1/ai_observability/fields/keys"
@@ -106,20 +108,94 @@ def test_ai_field_values_suggests_ingested_attribute_values(
assert values["stringValues"] == ["gpt-it-values"], values
def test_ai_field_values_reject_existing_query(
@pytest.mark.parametrize(
"existing_query,search_text,expected",
[
pytest.param(None, "", {"ai-rel-a", "ai-rel-b", "ai-rel-c"}, id="no_query_scopes_to_gen_ai_spans"),
pytest.param("gen_ai.user.id = 'alice'", "", {"ai-rel-a"}, id="span_filter_narrows_under_the_gate"),
pytest.param("llm_call_count > 0", "", {"ai-rel-a", "ai-rel-b", "ai-rel-c"}, id="pure_trace_aggregate_filter_is_stripped"),
pytest.param("llm_call_count > 0 AND gen_ai.user.id = 'alice'", "", {"ai-rel-a"}, id="mixed_filter_keeps_only_the_span_part"),
pytest.param(
"llm_call_count > 0 OR gen_ai.user.id = 'alice'",
"",
{"ai-rel-a", "ai-rel-b", "ai-rel-c"},
id="class_mixing_or_drops_the_filter_not_the_request",
),
pytest.param(
"gen_ai.user.id = ",
"",
{"ai-rel-a", "ai-rel-b", "ai-rel-c"},
id="unparseable_filter_falls_back_to_the_gate",
),
pytest.param(None, "ai-rel-a", {"ai-rel-a"}, id="search_text_narrows_related_values"),
# http.request.method lives on the root span's metadata row, gen_ai.* on
# the LLM/tool/agent rows; rows are per span-shape, so the gate AND a
# cross-span attribute filter can match no single row
pytest.param("http.request.method = 'POST'", "", set(), id="cross_span_attribute_filter_matches_no_row"),
],
)
def test_ai_field_values_related_values(
signoz: types.SigNoz,
create_user_admin: None, # pylint: disable=unused-argument
get_token: Callable[[str, str], str],
insert_traces: Callable[[list[Traces]], None],
insert_attributes_metadata: Callable[[list[AttributesMetadata]], None],
existing_query: str | None,
search_text: str,
expected: set[str],
) -> None:
now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
# existingQuery key resolution reads the trace keys tables, not
# attributes_metadata; a mixed trace registers the gate keys (model/tool/
# agent) plus gen_ai.user.id and http.request.method
insert_traces(ai_trace_mixed_spans(now=now, service="ai-rel-a", user="alice"))
# related values are served from attributes_metadata; one row per gate key,
# the traces row without any gate attribute and the logs row (wrong
# data_source, gate attribute present) must never surface
insert_attributes_metadata(
[
AttributesMetadata(
data_source="traces",
resource_attributes={"service.name": "ai-rel-a"},
attributes={"gen_ai.request.model": "gpt-rel", "gen_ai.user.id": "alice"},
),
AttributesMetadata(
data_source="traces",
resource_attributes={"service.name": "ai-rel-b"},
attributes={"gen_ai.tool.name": "get_weather", "gen_ai.user.id": "bob"},
),
AttributesMetadata(
data_source="traces",
resource_attributes={"service.name": "ai-rel-c"},
attributes={"gen_ai.agent.name": "chat-agent"},
),
AttributesMetadata(
data_source="traces",
resource_attributes={"service.name": "plain-rel"},
attributes={"http.request.method": "POST"},
),
AttributesMetadata(
data_source="logs",
resource_attributes={"service.name": "ai-rel-logs"},
attributes={"gen_ai.request.model": "gpt-rel"},
),
]
)
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
response = get_field_values(
signoz,
token,
{"name": "gen_ai.request.model", "existingQuery": "service.name = 'ai-it-values'"},
AI_VALUES_PATH,
)
assert response.status_code == HTTPStatus.BAD_REQUEST, response.text
params = {"name": "service.name", "searchText": search_text}
if existing_query is not None:
params["existingQuery"] = existing_query
response = get_field_values(signoz, token, params, AI_VALUES_PATH)
assert response.status_code == HTTPStatus.OK, response.text
assert response.json()["status"] == "success"
related = response.json()["data"]["values"].get("relatedValues") or []
assert set(related) == expected, related
def test_ai_field_values_of_computed_aggregate_are_empty(

View File

@@ -0,0 +1,172 @@
from collections.abc import Callable
from datetime import UTC, datetime, timedelta
from http import HTTPStatus
import pytest
from fixtures import types
from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
from fixtures.metrics import ExpHistogramMetrics
from fixtures.querier import (
build_builder_query,
get_all_series,
get_all_warnings,
get_error_message,
get_series_values,
index_series_by_label,
make_query_request,
)
# quantilesDD carries 0.01 relative accuracy and the log-spaced observations put
# neighbouring ranks ~1.25% apart, so a percentile can land a few percent off
PERCENTILE_TOLERANCE = 0.05
@pytest.mark.parametrize(
"space_aggregation, frontend_first, frontend_last, backend_first, backend_last",
[
("p50", 118, 153, 711, 921),
("p95", 1108, 1435, 6651, 8613),
("p99", 1352, 1751, 8113, 10507),
],
)
@pytest.mark.parametrize("time_aggregation", ["", "rate"])
def test_exp_histogram_percentile_delta_grouped(
signoz: types.SigNoz,
create_user_admin: None, # pylint: disable=unused-argument
get_token: Callable[[str, str], str],
insert_exp_histogram_metrics: Callable[[list[ExpHistogramMetrics]], None],
time_aggregation: str,
space_aggregation: str,
frontend_first: float,
frontend_last: float,
backend_first: float,
backend_last: float,
) -> None:
now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
start_ms = int((now - timedelta(minutes=65)).timestamp() * 1000)
end_ms = int(now.timestamp() * 1000)
metric_name = "test_exp_histogram_latency"
insert_exp_histogram_metrics(
[
ExpHistogramMetrics(
metric_name=metric_name,
# log-spaced latencies with a long tail, drifting ~30% higher across
# the hour so each point carries a distinct distribution
observations=[round(base * 1.0125**rank * (1 + minute / 200)) for rank in range(400)],
labels={"service.name": service},
timestamp=now - timedelta(minutes=60 - minute),
temporality="Delta",
)
for service, base in (("frontend", 10), ("backend", 60))
for minute in range(60)
]
)
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
query = build_builder_query(
"A",
metric_name,
time_aggregation,
space_aggregation,
temporality="delta",
group_by=["service.name"],
)
response = make_query_request(signoz, token, start_ms, end_ms, [query])
assert response.status_code == HTTPStatus.OK, response.text
data = response.json()
assert get_all_warnings(data) == [], f"unexpected warnings: {get_all_warnings(data)}"
series_by_service = index_series_by_label(get_all_series(data, "A"), "service.name")
assert set(series_by_service.keys()) == {"frontend", "backend"}, f"got series {set(series_by_service.keys())}"
for service, first, last in (
("frontend", frontend_first, frontend_last),
("backend", backend_first, backend_last),
):
values = [point["value"] for point in sorted(series_by_service[service]["values"], key=lambda point: point["timestamp"])]
assert len(values) == 60, f"{service}: expected a point per minute, got {len(values)}"
assert values[0] == pytest.approx(first, rel=PERCENTILE_TOLERANCE), f"{service} {space_aggregation} at the oldest point: got {values[0]}, want ~{first}"
assert values[-1] == pytest.approx(last, rel=PERCENTILE_TOLERANCE), f"{service} {space_aggregation} at the newest point: got {values[-1]}, want ~{last}"
# every observation drifts up minute over minute, so the sketch must too
assert values == sorted(values), f"{service} {space_aggregation} is not non-decreasing: {values}"
def test_exp_histogram_percentile_delta_merges_across_series(
signoz: types.SigNoz,
create_user_admin: None, # pylint: disable=unused-argument
get_token: Callable[[str, str], str],
insert_exp_histogram_metrics: Callable[[list[ExpHistogramMetrics]], None],
) -> None:
now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
start_ms = int((now - timedelta(minutes=65)).timestamp() * 1000)
end_ms = int(now.timestamp() * 1000)
metric_name = "test_exp_histogram_latency_merged"
insert_exp_histogram_metrics(
[
ExpHistogramMetrics(
metric_name=metric_name,
observations=[round(base * 1.0125**rank * (1 + minute / 200)) for rank in range(400)],
labels={"service.name": service},
timestamp=now - timedelta(minutes=60 - minute),
temporality="Delta",
)
for service, base in (("frontend", 10), ("backend", 60))
for minute in range(60)
]
)
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
query = build_builder_query("A", metric_name, "", "p95", temporality="delta")
response = make_query_request(signoz, token, start_ms, end_ms, [query])
assert response.status_code == HTTPStatus.OK, response.text
data = response.json()
assert get_all_warnings(data) == [], f"unexpected warnings: {get_all_warnings(data)}"
# both services' sketches merge into one, so p95 sits well above the frontend's
# own p95 (~1108) and below the backend's (~6651)
values = [point["value"] for point in sorted(get_series_values(data, "A"), key=lambda point: point["timestamp"])]
assert len(values) == 60, f"expected a point per minute, got {len(values)}"
assert values[0] == pytest.approx(5188, rel=PERCENTILE_TOLERANCE), f"oldest point: got {values[0]}, want ~5188"
assert values[-1] == pytest.approx(6718, rel=PERCENTILE_TOLERANCE), f"newest point: got {values[-1]}, want ~6718"
def test_exp_histogram_percentile_cumulative_is_rejected(
signoz: types.SigNoz,
create_user_admin: None, # pylint: disable=unused-argument
get_token: Callable[[str, str], str],
insert_exp_histogram_metrics: Callable[[list[ExpHistogramMetrics]], None],
) -> None:
now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
start_ms = int((now - timedelta(minutes=65)).timestamp() * 1000)
end_ms = int(now.timestamp() * 1000)
metric_name = "test_exp_histogram_latency_cumulative"
insert_exp_histogram_metrics(
[
ExpHistogramMetrics(
metric_name=metric_name,
observations=[round(10 * 1.0125**rank * (1 + minute / 200)) for rank in range(400)],
labels={"service.name": "frontend"},
timestamp=now - timedelta(minutes=60 - minute),
temporality="Cumulative",
)
for minute in range(60)
]
)
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
query = build_builder_query("A", metric_name, "", "p95", temporality="cumulative")
response = make_query_request(signoz, token, start_ms, end_ms, [query])
# the request is well formed; it is the stored metric that cannot be read, so
# this is reported as unsupported rather than as bad input
assert response.status_code == HTTPStatus.NOT_IMPLEMENTED, response.text
assert "only `delta` exponential histograms are supported" in get_error_message(response.json()), response.text

View File

@@ -1240,6 +1240,13 @@ def test_traces_list_span_scope(
lambda x: {"duration_nano": int(x[1].duration_nano), "span_id": x[1].span_id, "timestamp": format_timestamp(x[1].timestamp), "trace_id": x[1].trace_id},
id="select_attribute_duration_order_intrinsic",
),
# Case 9: filter on the intrinsic scope.version. Only x[1] should match.
pytest.param(
BuilderQuery(signal="traces", name="A", select_fields=[TelemetryFieldKey("timestamp")], filter_expression="scope.version = '1.0.0'", limit=1),
HTTPStatus.OK,
lambda x: {"span_id": x[1].span_id, "timestamp": format_timestamp(x[1].timestamp), "trace_id": x[1].trace_id},
id="filter_scope_version",
),
],
)
def test_traces_list_with_corrupt_data(
@@ -1283,6 +1290,168 @@ def test_traces_list_with_corrupt_data(
assert get_rows(response)[0]["data"] == expected(traces)
@pytest.mark.parametrize(
"filter_expression,expected_indices",
[
# Intrinsic scope.name / scope.version resolve to the JSON sub-columns.
pytest.param("scope.name = 'io.signoz.payment'", [1], id="intrinsic_scope_name"),
pytest.param("scope.version = '2.3.1'", [0], id="intrinsic_scope_version"),
# A scope attribute resolves against the scope JSON column's attributes.
pytest.param("scope.telemetry.sdk.language = 'python'", [1], id="scope_attribute"),
# A scope attribute whose own name carries a `scope.` prefix. `scope.prefixed`
# normalizes to {prefixed, scope} and must still resolve to the attribute.
pytest.param("scope.prefixed = 'prefixed-val'", [0], id="scope_prefixed_attribute"),
# `env.tier` is a span attribute on span 0 and a scope attribute on
# span 1. Unprefixed -> no explicit context, so it is checked in every
# applicable context (attribute OR scope) and both spans match.
pytest.param("env.tier = 'gold'", [0, 1], id="bare_cross_context"),
# The explicit `scope.` prefix forces scope context only, so span 0's
# span attribute is ignored — only span 1 matches.
pytest.param("scope.env.tier = 'gold'", [1], id="scope_prefixed_cross_context"),
# `scope.name` names both homes it can resolve to: the declared scope.name field
# (span 0) and a same-named `name` scope attribute (span 1), the same way any
# other name colliding across contexts unions. `scope.attribute.name` addresses
# the attribute alone.
pytest.param("scope.name = 'io.signoz.checkout'", [0, 1], id="scope_name_unions_attribute"),
# The `scope.name` spelling is also a real stored key: span 2 carries a span
# attribute literally named `scope.name`, so it matches too.
pytest.param("scope.name = 'attr-scope-name'", [2], id="scope_name_matches_stored_spelling"),
# The explicit `scope.attribute.` prefix addresses the scope attribute alone, without
# the declared path. Span 1 has a `name` scope attribute = 'io.signoz.checkout'.
pytest.param("scope.attribute.name = 'io.signoz.checkout'", [1], id="scope_attribute_name"),
# `version` as a scope attribute: no span carries one (span 1's 4.5.6 is the declared
# scope.version, not a scope attribute), so this matches nothing.
pytest.param("scope.attribute.version = '4.5.6'", [], id="scope_attribute_version_none"),
# An unprefixed `name` is checked in every applicable context: the span `name`
# column (span 2) and a `name` scope attribute (span 1). It does not reach the
# declared scope.name field (span 0), which only the `scope.` prefix addresses.
pytest.param("name = 'io.signoz.checkout'", [1, 2], id="bare_name_unions_scope_attribute"),
# A value that no resolvable key holds (scope.name/scope.version field,
# a `name`/`version` scope attribute, or a same-named attribute/resource)
# returns nothing.
pytest.param("scope.version = 'corrupt_data'", [], id="scope_version_no_match"),
pytest.param("scope.name = 'corrupt_data'", [], id="scope_name_no_match"),
],
)
def test_traces_list_with_scope_filter(
signoz: types.SigNoz,
create_user_admin: None, # pylint: disable=unused-argument
get_token: Callable[[str, str], str],
insert_traces: Callable[[list[Traces]], None],
filter_expression: str,
expected_indices: list[int],
) -> None:
"""
Setup three spans with different scope key resolution:
- x[0]: scope.name/version 'io.signoz.checkout'/'2.3.1'; span attribute
env.tier='gold'.
- x[1]: scope.name/version 'io.signoz.payment'/'4.5.6'; scope attributes
telemetry.sdk.language='python', env.tier='gold', and a `name` scope
attribute colliding with x[0]'s scope.name value.
- x[2]: span name 'io.signoz.checkout' (colliding with x[0]'s scope.name
value) and a span attribute literally named `scope.name`.
Tests:
- Filtering on scope.name / scope.version / a scope attribute.
- An unprefixed key is resolved across contexts (scope checked alongside
attribute / intrinsic), while a `scope.`-prefixed key is scope-only.
- `scope.name`/`scope.version` name every home they resolve to: the declared JSON
sub-column and a same-named `name`/`version` scope attribute. The explicit
`scope.attribute.` prefix addresses the attribute alone.
- a bare `name` reaches the span `name` column and a `name` scope attribute, but
never the declared scope.name field.
"""
now = datetime.now(tz=UTC).replace(microsecond=0)
trace_id = TraceIdGenerator.trace_id()
span_ids = [TraceIdGenerator.span_id() for _ in range(3)]
traces = [
Traces(
timestamp=now - timedelta(seconds=4),
duration=timedelta(seconds=2),
trace_id=trace_id,
span_id=span_ids[0],
parent_span_id="",
name="GET /checkout",
kind=TracesKind.SPAN_KIND_SERVER,
status_code=TracesStatusCode.STATUS_CODE_OK,
resources={"service.name": "checkout"},
attributes={"http.request.method": "GET", "env.tier": "gold"},
scope={
"name": "io.signoz.checkout",
"version": "2.3.1",
# a scope attribute whose own name carries a `scope.` prefix
"attributes": {"telemetry.sdk.language": "go", "scope.prefixed": "prefixed-val"},
},
),
Traces(
timestamp=now - timedelta(seconds=2),
duration=timedelta(seconds=1),
trace_id=trace_id,
span_id=span_ids[1],
parent_span_id="",
name="POST /pay",
kind=TracesKind.SPAN_KIND_SERVER,
status_code=TracesStatusCode.STATUS_CODE_OK,
resources={"service.name": "payment"},
attributes={"http.request.method": "POST"},
# env.tier is a scope attribute here (cross-context with span 0);
# `name` is a scope attribute colliding with span 0's scope.name.
scope={
"name": "io.signoz.payment",
"version": "4.5.6",
"attributes": {
"telemetry.sdk.language": "python",
"env.tier": "gold",
"name": "io.signoz.checkout",
},
},
),
Traces(
timestamp=now - timedelta(seconds=1),
duration=timedelta(seconds=1),
trace_id=trace_id,
span_id=span_ids[2],
parent_span_id="",
# span name collides with span 0's scope.name value
name="io.signoz.checkout",
kind=TracesKind.SPAN_KIND_SERVER,
status_code=TracesStatusCode.STATUS_CODE_OK,
resources={"service.name": "probe"},
# a span attribute named `scope.name`
attributes={"scope.name": "attr-scope-name"},
scope={"name": "span-gamma", "version": "9.9.9"},
),
]
insert_traces(traces)
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
start_ms = int((now - timedelta(minutes=1)).timestamp() * 1000)
end_ms = int((now + timedelta(seconds=1)).timestamp() * 1000)
response = make_query_request(
signoz,
token,
start_ms=start_ms,
end_ms=end_ms,
request_type=RequestType.RAW,
queries=[
BuilderQuery(
signal="traces",
name="A",
select_fields=[TelemetryFieldKey("timestamp")],
filter_expression=filter_expression,
limit=10,
).to_dict()
],
)
assert response.status_code == HTTPStatus.OK, response.text
got_span_ids = {row["data"]["span_id"] for row in get_rows(response)}
expected_span_ids = {traces[i].span_id for i in expected_indices}
assert got_span_ids == expected_span_ids
@pytest.mark.parametrize("surface", ["filter", "select", "order"])
def test_traces_list_unknown_span_context_synthesizes(
signoz: types.SigNoz,

View File

@@ -592,6 +592,66 @@ def test_saved_view_lifecycle(
assert response.status_code == HTTPStatus.NOT_FOUND
def test_ai_observability_view_with_builder_ai_query_roundtrip(
signoz: SigNoz,
create_user_admin: Operation, # pylint: disable=unused-argument
get_token: Callable[[str, str], str],
):
"""builder_ai_query implies the traces signal -- the spec is sent without
one and must read back with signal pinned to "traces"."""
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
headers = {"Authorization": f"Bearer {token}"}
response = requests.post(
signoz.self.host_configs["8080"].get(BASE_URL),
json={
"name": "ai-observability-overview",
"generateName": False,
"source": "ai_observability",
"schemaVersion": "v2",
"spec": {
"displayName": "ai-observability-overview",
"requestType": "scalar",
"queries": [{"type": "builder_ai_query", "spec": {"name": "A", "aggregations": [{"expression": "count()"}], "disabled": False, "legend": ""}}],
"selectedFields": [],
"panelType": "table",
"display": {"maxLines": 0, "fontSize": "", "format": "", "color": ""},
},
},
headers=headers,
timeout=5,
)
assert response.status_code == HTTPStatus.CREATED, response.text
view_id = response.json()["data"]["id"]
try:
response = requests.get(
signoz.self.host_configs["8080"].get(f"{BASE_URL}/{view_id}"),
headers=headers,
timeout=5,
)
assert response.status_code == HTTPStatus.OK, response.text
got = response.json()["data"]
assert got["source"] == "ai_observability"
assert got["spec"]["queries"][0]["type"] == "builder_ai_query"
assert got["spec"]["queries"][0]["spec"]["signal"] == "traces"
response = requests.get(
signoz.self.host_configs["8080"].get(BASE_URL),
params={"source": "ai_observability"},
headers=headers,
timeout=5,
)
assert response.status_code == HTTPStatus.OK, response.text
assert {v["name"] for v in response.json()["data"]} == {"ai-observability-overview"}
finally:
requests.delete(
signoz.self.host_configs["8080"].get(f"{BASE_URL}/{view_id}"),
headers=headers,
timeout=5,
)
def test_empty_name_derives_a_slug_from_display_name(
signoz: SigNoz,
create_user_admin: Operation, # pylint: disable=unused-argument

View File

@@ -0,0 +1,220 @@
from collections.abc import Callable
from datetime import datetime, timedelta
from http import HTTPStatus
import pytest
from fixtures import types
from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
from fixtures.querier import (
RequestType,
build_aggregation,
build_group_by_field,
build_order_by,
build_raw_query,
build_traces_scalar_query,
get_column_data_from_response,
make_query_request,
)
from fixtures.semconvfamilies import (
BOTH,
CURRENT_KEY,
NEITHER,
NEW,
OLD,
OLD_KEY,
PREFIX,
)
FILTER_MATRIX = [
pytest.param("{key} = 'production'", {OLD, NEW}, id="eq_matches_either_spelling"),
pytest.param("{key} = 'staging'", {BOTH}, id="eq_current_wins_on_conflict"),
pytest.param("{key} != 'production'", {BOTH, NEITHER}, id="neq_keeps_keyless_and_conflict"),
pytest.param("{key} IN ['production', 'staging']", {OLD, NEW, BOTH}, id="in_matches_merged_value"),
pytest.param("{key} NOT IN ['production']", {BOTH, NEITHER}, id="not_in_keeps_keyless"),
pytest.param("{key} LIKE '%prod%'", {OLD, NEW}, id="like_matches_merged_value"),
pytest.param("{key} EXISTS", {OLD, NEW, BOTH}, id="exists_is_any_member"),
pytest.param("{key} NOT EXISTS", {NEITHER}, id="not_exists_is_no_member"),
pytest.param("{key} != 'production' AND {key} EXISTS", {BOTH}, id="neq_composed_with_exists"),
]
LITERAL_MATRIX = [
pytest.param("{key} = 'production'", {NEW}, id="literal_eq_reads_one_spelling"),
pytest.param("{key} != 'production'", {OLD, BOTH, NEITHER}, id="literal_neq_reads_one_spelling"),
]
@pytest.mark.parametrize("expression_template,expected", FILTER_MATRIX)
@pytest.mark.parametrize("requested_key", [CURRENT_KEY, OLD_KEY], ids=["current", "old"])
@pytest.mark.parametrize("context", ["resource", "attribute"])
def test_family_filters(
signoz: types.SigNoz,
create_user_admin: None, # pylint: disable=unused-argument
get_token: Callable[[str, str], str],
family_fleet: datetime,
context: str,
requested_key: str,
expression_template: str,
expected: set[str],
) -> None:
"""The result set is a property of the family, not of the requested spelling."""
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
expression = expression_template.format(key=f"{context}.{requested_key}")
response = make_query_request(
signoz,
token,
start_ms=int((family_fleet - timedelta(minutes=2)).timestamp() * 1000),
end_ms=int((family_fleet + timedelta(minutes=1)).timestamp() * 1000),
request_type=RequestType.RAW,
queries=[
build_raw_query(
"A",
"traces",
limit=100,
filter_expression=expression,
order=[build_order_by("timestamp", "asc")],
select_fields=[{"name": "span.name"}],
)
],
)
assert response.status_code == HTTPStatus.OK, response.text
matched = {name for name in get_column_data_from_response(response.json(), "name") if name.startswith(PREFIX)}
assert matched == expected, expression
@pytest.mark.parametrize("expression_template,expected", LITERAL_MATRIX)
def test_flag_off_stays_literal(
signoz_families_off: types.SigNoz,
create_user_admin: None, # pylint: disable=unused-argument
get_token: Callable[[str, str], str],
family_fleet: datetime,
expression_template: str,
expected: set[str],
) -> None:
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
expression = expression_template.format(key=f"resource.{CURRENT_KEY}")
response = make_query_request(
signoz_families_off,
token,
start_ms=int((family_fleet - timedelta(minutes=2)).timestamp() * 1000),
end_ms=int((family_fleet + timedelta(minutes=1)).timestamp() * 1000),
request_type=RequestType.RAW,
queries=[
build_raw_query(
"A",
"traces",
limit=100,
filter_expression=expression,
order=[build_order_by("timestamp", "asc")],
select_fields=[{"name": "span.name"}],
)
],
)
assert response.status_code == HTTPStatus.OK, response.text
matched = {name for name in get_column_data_from_response(response.json(), "name") if name.startswith(PREFIX)}
assert matched == expected, expression
@pytest.mark.parametrize("expression_template,expected", LITERAL_MATRIX)
def test_logs_stay_literal_with_flag_on(
signoz: types.SigNoz,
create_user_admin: None, # pylint: disable=unused-argument
get_token: Callable[[str, str], str],
family_fleet: datetime,
expression_template: str,
expected: set[str],
) -> None:
"""Only traces have family support today."""
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
expression = expression_template.format(key=f"resource.{CURRENT_KEY}")
response = make_query_request(
signoz,
token,
start_ms=int((family_fleet - timedelta(minutes=2)).timestamp() * 1000),
end_ms=int((family_fleet + timedelta(minutes=1)).timestamp() * 1000),
request_type=RequestType.RAW,
queries=[
build_raw_query(
"A",
"logs",
limit=100,
filter_expression=expression,
order=[build_order_by("timestamp", "asc")],
select_fields=[{"name": "body"}],
)
],
)
assert response.status_code == HTTPStatus.OK, response.text
matched = {body for body in get_column_data_from_response(response.json(), "body") if body.startswith(PREFIX)}
assert matched == expected, expression
@pytest.mark.parametrize("requested_key", [CURRENT_KEY, OLD_KEY], ids=["current", "old"])
def test_group_by_merges_and_echoes_requested_spelling(
signoz: types.SigNoz,
create_user_admin: None, # pylint: disable=unused-argument
get_token: Callable[[str, str], str],
family_fleet: datetime,
requested_key: str,
) -> None:
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
response = make_query_request(
signoz,
token,
start_ms=int((family_fleet - timedelta(minutes=2)).timestamp() * 1000),
end_ms=int((family_fleet + timedelta(minutes=1)).timestamp() * 1000),
request_type=RequestType.SCALAR,
queries=[
build_traces_scalar_query(
[build_aggregation("count()")],
filter_expression=f"service.name LIKE '{PREFIX}%'",
group_by=[build_group_by_field(requested_key, "string", "resource")],
)
],
)
assert response.status_code == HTTPStatus.OK, response.text
result = response.json()["data"]["data"]["results"][0]
group_column = result["columns"][0]
assert group_column["name"] == requested_key, group_column
assert group_column["columnType"] == "group", group_column
groups = {row[0] for row in result["data"]}
assert {"production", "staging"}.issubset(groups), groups
assert None in groups, groups
def test_bare_name_prefers_resource_and_warns(
signoz: types.SigNoz,
create_user_admin: None, # pylint: disable=unused-argument
get_token: Callable[[str, str], str],
family_fleet: datetime,
) -> None:
"""Both contexts carry the family, so a bare name is ambiguous: resolution
warns and keeps the resource side."""
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
response = make_query_request(
signoz,
token,
start_ms=int((family_fleet - timedelta(minutes=2)).timestamp() * 1000),
end_ms=int((family_fleet + timedelta(minutes=1)).timestamp() * 1000),
request_type=RequestType.RAW,
queries=[
build_raw_query(
"A",
"traces",
limit=100,
filter_expression=f"{CURRENT_KEY} = 'production'",
order=[build_order_by("timestamp", "asc")],
select_fields=[{"name": "span.name"}],
)
],
)
assert response.status_code == HTTPStatus.OK, response.text
matched = {name for name in get_column_data_from_response(response.json(), "name") if name.startswith(PREFIX)}
assert matched == {OLD, NEW}
warning = response.json()["data"].get("warning") or {}
messages = " ".join(entry.get("message", "") for entry in warning.get("warnings", []))
assert "ambiguous" in messages.lower(), messages

View File

@@ -0,0 +1,55 @@
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_semconv_families(
network: Network,
zeus: types.TestContainerDocker,
gateway: types.TestContainerDocker,
sqlstore: types.TestContainerSQL,
clickhouse: types.TestContainerClickhouse,
request: pytest.FixtureRequest,
pytestconfig: pytest.Config,
) -> types.SigNoz:
return create_signoz(
network=network,
zeus=zeus,
gateway=gateway,
sqlstore=sqlstore,
clickhouse=clickhouse,
request=request,
pytestconfig=pytestconfig,
cache_key="signoz-semconv-families",
env_overrides={
"SIGNOZ_FLAGGER_CONFIG_BOOLEAN_RESOLVE__SEMCONV__FAMILIES": True,
},
)
@pytest.fixture(name="signoz_families_off", scope="package")
def signoz_families_off(
network: Network,
zeus: types.TestContainerDocker,
gateway: types.TestContainerDocker,
sqlstore: types.TestContainerSQL,
clickhouse: types.TestContainerClickhouse,
request: pytest.FixtureRequest,
pytestconfig: pytest.Config,
) -> types.SigNoz:
"""Shares the sqlstore and clickhouse with the flag-on instance, so the
same admin token and seeded rows work."""
return create_signoz(
network=network,
zeus=zeus,
gateway=gateway,
sqlstore=sqlstore,
clickhouse=clickhouse,
request=request,
pytestconfig=pytestconfig,
cache_key="signoz-semconv-families-off",
env_overrides={},
)