mirror of
https://github.com/SigNoz/signoz.git
synced 2026-08-18 18:50:48 +01:00
Compare commits
29 Commits
feat/googl
...
issue_5602
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
9098e683e0 | ||
|
|
a53b937854 | ||
|
|
606511df34 | ||
|
|
640072a570 | ||
|
|
6c7c4a333c | ||
|
|
15fb851ce0 | ||
|
|
3e73614f34 | ||
|
|
07979db754 | ||
|
|
6e8659006c | ||
|
|
c138aa7da6 | ||
|
|
0d3f7ed51d | ||
|
|
28f0e06c55 | ||
|
|
19c5cb9984 | ||
|
|
f6a823c000 | ||
|
|
23f03973c5 | ||
|
|
0c0e969cfc | ||
|
|
c870efa12d | ||
|
|
d250f190a7 | ||
|
|
97c49c870b | ||
|
|
31efe177a4 | ||
|
|
d502d12ac3 | ||
|
|
bd9f15a716 | ||
|
|
813ef988c9 | ||
|
|
40e6799285 | ||
|
|
1caa60a3cd | ||
|
|
3f781f0083 | ||
|
|
6aec05cf7a | ||
|
|
683a52f35a | ||
|
|
e924fa1e62 |
@@ -8,19 +8,12 @@ import {
|
||||
|
||||
import ChangelogRenderer from '../components/ChangelogRenderer';
|
||||
|
||||
// Mock react-markdown to render children as plain text and a sample
|
||||
// anchor through the `components.a` override
|
||||
// Mock react-markdown to just render children as plain text
|
||||
jest.mock(
|
||||
'react-markdown',
|
||||
() =>
|
||||
function ReactMarkdown({ children, components }: any) {
|
||||
const Anchor = components?.a;
|
||||
return (
|
||||
<div>
|
||||
{children}
|
||||
{Anchor && <Anchor href="https://signoz.io/docs">docs</Anchor>}
|
||||
</div>
|
||||
);
|
||||
function ReactMarkdown({ children }: any) {
|
||||
return <div>{children}</div>;
|
||||
},
|
||||
);
|
||||
|
||||
@@ -69,14 +62,4 @@ describe('ChangelogRenderer', () => {
|
||||
expect(screen.getByAltText('Media')).toBeInTheDocument();
|
||||
expect(screen.getByText('Description for feature 1')).toBeInTheDocument();
|
||||
});
|
||||
|
||||
it('renders markdown links that open in a new tab', () => {
|
||||
render(<ChangelogRenderer changelog={mockChangelog} />);
|
||||
const links = screen.getAllByRole('link', { name: 'docs' });
|
||||
expect(links.length).toBeGreaterThan(0);
|
||||
links.forEach((link) => {
|
||||
expect(link).toHaveAttribute('target', '_blank');
|
||||
expect(link).toHaveAttribute('rel', 'noopener noreferrer');
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
@@ -13,19 +13,6 @@ interface Props {
|
||||
changelog: ChangelogSchema;
|
||||
}
|
||||
|
||||
interface LinkProps {
|
||||
href?: string;
|
||||
children?: React.ReactNode;
|
||||
}
|
||||
|
||||
function Link({ href, children }: LinkProps): JSX.Element {
|
||||
return (
|
||||
<a href={href} target="_blank" rel="noopener noreferrer">
|
||||
{children}
|
||||
</a>
|
||||
);
|
||||
}
|
||||
|
||||
function renderMedia(media: Media): JSX.Element | null {
|
||||
if (SupportedImageTypes.includes(media.ext)) {
|
||||
return (
|
||||
@@ -75,9 +62,7 @@ function ChangelogRenderer({ changelog }: Props): JSX.Element {
|
||||
<div key={feature.id}>
|
||||
<div className="changelog-renderer-section-title">{feature.title}</div>
|
||||
{feature.media && renderMedia(feature.media)}
|
||||
<ReactMarkdown components={{ a: Link }}>
|
||||
{feature.description}
|
||||
</ReactMarkdown>
|
||||
<ReactMarkdown>{feature.description}</ReactMarkdown>
|
||||
</div>
|
||||
))}
|
||||
</div>
|
||||
@@ -86,9 +71,7 @@ function ChangelogRenderer({ changelog }: Props): JSX.Element {
|
||||
<div className="changelog-renderer-bug-fixes">
|
||||
<div className="changelog-renderer-section-title">Bug Fixes</div>
|
||||
{changelog.bug_fixes && (
|
||||
<ReactMarkdown components={{ a: Link }}>
|
||||
{changelog.bug_fixes}
|
||||
</ReactMarkdown>
|
||||
<ReactMarkdown>{changelog.bug_fixes}</ReactMarkdown>
|
||||
)}
|
||||
</div>
|
||||
)}
|
||||
@@ -96,9 +79,7 @@ function ChangelogRenderer({ changelog }: Props): JSX.Element {
|
||||
<div className="changelog-renderer-maintenance">
|
||||
<div className="changelog-renderer-section-title">Maintenance</div>
|
||||
{changelog.maintenance && (
|
||||
<ReactMarkdown components={{ a: Link }}>
|
||||
{changelog.maintenance}
|
||||
</ReactMarkdown>
|
||||
<ReactMarkdown>{changelog.maintenance}</ReactMarkdown>
|
||||
)}
|
||||
</div>
|
||||
)}
|
||||
|
||||
@@ -1,5 +1,9 @@
|
||||
// temporary flag to be removed with old log details code.
|
||||
export const isLogDetailsV2 = true;
|
||||
import getLocalStorage from 'api/browser/localstorage/get';
|
||||
import { LOCALSTORAGE } from 'constants/localStorage';
|
||||
|
||||
// Temp feature flag before actual roll-out
|
||||
export const isLogDetailsV2 =
|
||||
getLocalStorage(LOCALSTORAGE.LOG_DETAILS_V2) === 'true';
|
||||
|
||||
export const VIEW_TYPES = {
|
||||
OVERVIEW: 'OVERVIEW',
|
||||
|
||||
@@ -100,7 +100,6 @@ function LogDetailInner({
|
||||
// Don't close if clicking on drawer content, overlays, or portal elements
|
||||
if (
|
||||
target.closest('[data-log-detail-ignore="true"]') ||
|
||||
target.closest('.log-detail-drawer') ||
|
||||
target.closest('.cm-tooltip-autocomplete') ||
|
||||
target.closest('.drawer-popover') ||
|
||||
target.closest('.query-status-popover') ||
|
||||
|
||||
@@ -13,6 +13,7 @@ export enum LOCALSTORAGE {
|
||||
TRACES_LIST_COLUMNS = 'TRACES_LIST_COLUMNS',
|
||||
LOGS_LIST_COLUMNS = 'LOGS_LIST_COLUMNS',
|
||||
LOGS_LIST_COLUMN_SIZING = 'LOGS_LIST_COLUMN_SIZING',
|
||||
LOG_DETAILS_V2 = 'LOG_DETAILS_V2',
|
||||
LOGGED_IN_USER_NAME = 'LOGGED_IN_USER_NAME',
|
||||
LOGGED_IN_USER_EMAIL = 'LOGGED_IN_USER_EMAIL',
|
||||
CHAT_SUPPORT = 'CHAT_SUPPORT',
|
||||
|
||||
@@ -37,7 +37,7 @@ import { useInfraMonitoringFontSize } from './useInfraMonitoringTablePreferences
|
||||
import styles from './K8sExpandedRow.module.scss';
|
||||
import { buildExpressionFromGroupMeta } from './utils';
|
||||
import { logInfraColumnSortedEvent } from 'container/InfraMonitoringK8sV2/Base/events';
|
||||
import { getUnstableCurrentSearchParams } from 'utils/getUnstableCurrentSearchParams';
|
||||
import { getUnstableCurrentSearchParams } from 'container/TopNav/DateTimeSelectionV2/utils/getUnstableCurrentSearchParams';
|
||||
import { QueryParams } from 'constants/query';
|
||||
|
||||
const EXPANDED_ROW_LIMIT = 10;
|
||||
|
||||
@@ -9,11 +9,7 @@ function Overview(): JSX.Element {
|
||||
|
||||
return (
|
||||
<div className={styles.overview} data-testid="llm-observability-overview">
|
||||
<DashboardContainer
|
||||
dashboard={dashboard}
|
||||
refetch={refetch}
|
||||
canEditDashboardOverride={false}
|
||||
/>
|
||||
<DashboardContainer dashboard={dashboard} refetch={refetch} />
|
||||
</div>
|
||||
);
|
||||
}
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
{
|
||||
"id": "llm-observability-overview",
|
||||
"orgId": "",
|
||||
"locked": false,
|
||||
"locked": true,
|
||||
"name": "AI Observability Overview",
|
||||
"schemaVersion": "v6",
|
||||
"source": "system",
|
||||
@@ -1146,4 +1146,4 @@
|
||||
}
|
||||
]
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -23,9 +23,9 @@ import { useLogAttributeActions } from './hooks/useLogAttributeActions';
|
||||
import TableView from './TableView';
|
||||
import {
|
||||
aggregateAttributesResourcesToObject,
|
||||
buildPrettyViewData,
|
||||
getBodyDisplayString,
|
||||
getSanitizedLogBody,
|
||||
parseJsonStringBody,
|
||||
removeEscapeCharacters,
|
||||
} from './utils';
|
||||
|
||||
@@ -71,7 +71,11 @@ function Overview({
|
||||
|
||||
if (isLogDetailsV2) {
|
||||
const raw = aggregateAttributesResourcesToObject(logData);
|
||||
const prettyData = buildPrettyViewData(raw);
|
||||
const prettyData = Object.fromEntries(
|
||||
Object.entries({ ...raw, body: parseJsonStringBody(raw.body) }).filter(
|
||||
([, value]) => value !== undefined,
|
||||
),
|
||||
);
|
||||
return (
|
||||
<div className="overview-container">
|
||||
<DataViewer
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
export enum LogAttributeBucket {
|
||||
ATTRIBUTES = 'attributes',
|
||||
RESOURCES = 'resource',
|
||||
RESOURCES = 'resources',
|
||||
SCOPE = 'scope',
|
||||
}
|
||||
|
||||
|
||||
@@ -33,7 +33,7 @@ describe('buildLogFilterTarget', () => {
|
||||
|
||||
it('maps `resources` with Resource type', () => {
|
||||
expect(
|
||||
buildLogFilterTarget(['resource', 'service.name'], 'api', true),
|
||||
buildLogFilterTarget(['resources', 'service.name'], 'api', true),
|
||||
).toMatchObject({
|
||||
fieldKey: 'service.name',
|
||||
metricsType: MetricsType.Resource,
|
||||
@@ -53,30 +53,6 @@ describe('buildLogFilterTarget', () => {
|
||||
});
|
||||
});
|
||||
|
||||
describe('nested attribute values (parsed JSON)', () => {
|
||||
it('marks a sub-field of a parsed attribute copy-only (restricted, no group-by)', () => {
|
||||
const t = buildLogFilterTarget(['attributes', 'payload', 'x'], 1, true);
|
||||
expect(t.isRestricted).toBe(true);
|
||||
expect(t.groupBySupported).toBe(false);
|
||||
expect(t.groupByKey).toBeUndefined();
|
||||
});
|
||||
|
||||
it('leaves a top-level attribute (depth 2) filterable', () => {
|
||||
const t = buildLogFilterTarget(['attributes', 'payload'], 'v', true);
|
||||
expect(t.isRestricted).toBe(false);
|
||||
expect(t.groupBySupported).toBe(true);
|
||||
});
|
||||
|
||||
it('does not restrict nested resource/scope values', () => {
|
||||
expect(
|
||||
buildLogFilterTarget(['resource', 'k8s', 'pod'], 'p', true).isRestricted,
|
||||
).toBe(false);
|
||||
expect(
|
||||
buildLogFilterTarget(['scope', 'a', 'b'], 'v', true).isRestricted,
|
||||
).toBe(false);
|
||||
});
|
||||
});
|
||||
|
||||
describe('restricted fields (timestamp / id)', () => {
|
||||
it.each(['timestamp', 'id'])(
|
||||
'marks %s restricted with no group-by',
|
||||
@@ -89,30 +65,6 @@ describe('buildLogFilterTarget', () => {
|
||||
);
|
||||
});
|
||||
|
||||
describe('group-by-restricted fields (trace_id)', () => {
|
||||
it('allows filtering but not group-by on top-level trace_id', () => {
|
||||
const t = buildLogFilterTarget(['trace_id'], 'abc123', true);
|
||||
expect(t.isRestricted).toBe(false);
|
||||
expect(t.filterInOperator).toBe('=');
|
||||
expect(t.groupBySupported).toBe(false);
|
||||
expect(t.groupByKey).toBeUndefined();
|
||||
});
|
||||
|
||||
it.each([
|
||||
['resource', ['resource', 'trace_id']],
|
||||
['attributes', ['attributes', 'trace_id']],
|
||||
])(
|
||||
'blocks group-by on a %s field named trace_id, keeping filter',
|
||||
(_bucket, path) => {
|
||||
const t = buildLogFilterTarget(path as string[], 'abc123', true);
|
||||
expect(t.isRestricted).toBe(false);
|
||||
expect(t.filterInOperator).toBe('=');
|
||||
expect(t.groupBySupported).toBe(false);
|
||||
expect(t.groupByKey).toBeUndefined();
|
||||
},
|
||||
);
|
||||
});
|
||||
|
||||
describe('body scalars', () => {
|
||||
it('maps a top-level body scalar to body.<key> with =/!=, groupable when json body on', () => {
|
||||
const t = buildLogFilterTarget(['body', 'message'], 'hello', true);
|
||||
|
||||
@@ -5,10 +5,7 @@ import {
|
||||
QUERY_BUILDER_FUNCTIONS,
|
||||
} from 'constants/antlrQueryConstants';
|
||||
import { OPERATORS as QUERY_BUILDER_OPERATORS } from 'constants/queryBuilder';
|
||||
import {
|
||||
RESTRICTED_GROUP_BY_FIELDS,
|
||||
RESTRICTED_SELECTED_FIELDS,
|
||||
} from 'container/LogsFilters/config';
|
||||
import { RESTRICTED_SELECTED_FIELDS } from 'container/LogsFilters/config';
|
||||
import { MetricsType } from 'container/MetricsApplication/constant';
|
||||
import { getOperatorValue } from 'container/QueryBuilder/filters/QueryBuilderSearch/utils';
|
||||
import { chooseAutocompleteFromCustomValue } from 'lib/newQueryBuilder/chooseAutocompleteFromCustomValue';
|
||||
@@ -86,24 +83,15 @@ export const buildLogFilterTarget = (
|
||||
if (root !== 'body') {
|
||||
const fieldKey =
|
||||
fieldKeyPath.length > 1 ? fieldKeyPath.slice(1).join('.') : String(root);
|
||||
// Temporarily removing filter/group-by support for nested attributes.
|
||||
// This will be removed once backend starts to support these actions.
|
||||
const isNestedAttributeValue =
|
||||
root === LogAttributeBucket.ATTRIBUTES && fieldKeyPath.length > 2;
|
||||
|
||||
const isRestricted =
|
||||
RESTRICTED_SELECTED_FIELDS.includes(fieldKey) || isNestedAttributeValue;
|
||||
|
||||
const groupBySupported =
|
||||
!isRestricted && !RESTRICTED_GROUP_BY_FIELDS.includes(fieldKey);
|
||||
const isRestricted = RESTRICTED_SELECTED_FIELDS.includes(fieldKey);
|
||||
return {
|
||||
fieldKey,
|
||||
filterInOperator: OPERATORS['='],
|
||||
filterOutOperator: OPERATORS['!='],
|
||||
dataType: getDataTypes(value),
|
||||
metricsType: metricsTypeForRoot(root),
|
||||
groupBySupported,
|
||||
groupByKey: groupBySupported ? fieldKey : undefined,
|
||||
groupBySupported: !isRestricted,
|
||||
groupByKey: isRestricted ? undefined : fieldKey,
|
||||
isRestricted,
|
||||
};
|
||||
}
|
||||
|
||||
@@ -3,79 +3,45 @@ import { DataTypes } from 'types/api/queryBuilder/queryAutocompleteResponse';
|
||||
|
||||
import {
|
||||
aggregateAttributesResourcesToObject,
|
||||
buildPrettyViewData,
|
||||
flattenObject,
|
||||
getDataTypes,
|
||||
getSanitizedLogBody,
|
||||
parseJsonStringValue,
|
||||
parseJsonStringBody,
|
||||
recursiveParseJSON,
|
||||
} from './utils';
|
||||
|
||||
describe('parseJsonStringValue', () => {
|
||||
describe('parseJsonStringBody', () => {
|
||||
it('parses a JSON-object string into an object', () => {
|
||||
expect(parseJsonStringValue('{"a":1,"b":{"c":2}}')).toStrictEqual({
|
||||
expect(parseJsonStringBody('{"a":1,"b":{"c":2}}')).toStrictEqual({
|
||||
a: 1,
|
||||
b: { c: 2 },
|
||||
});
|
||||
});
|
||||
|
||||
it('parses a JSON-array string into an array', () => {
|
||||
expect(parseJsonStringValue('[1,2,3]')).toStrictEqual([1, 2, 3]);
|
||||
expect(parseJsonStringBody('[1,2,3]')).toStrictEqual([1, 2, 3]);
|
||||
});
|
||||
|
||||
it('returns a plain (non-JSON) string unchanged', () => {
|
||||
expect(parseJsonStringValue('plain log line')).toBe('plain log line');
|
||||
expect(parseJsonStringBody('plain log line')).toBe('plain log line');
|
||||
});
|
||||
|
||||
it('returns a string that is not object/array-looking unchanged', () => {
|
||||
expect(parseJsonStringValue('42')).toBe('42');
|
||||
expect(parseJsonStringBody('42')).toBe('42');
|
||||
});
|
||||
|
||||
it('returns an invalid JSON string unchanged', () => {
|
||||
expect(parseJsonStringValue('{not valid}')).toBe('{not valid}');
|
||||
expect(parseJsonStringBody('{not valid}')).toBe('{not valid}');
|
||||
});
|
||||
|
||||
it('returns an already-object value unchanged (same reference)', () => {
|
||||
const value = { message: 'hi', a: 1 };
|
||||
expect(parseJsonStringValue(value)).toBe(value);
|
||||
it('returns an already-object body unchanged (same reference)', () => {
|
||||
const body = { message: 'hi', a: 1 };
|
||||
expect(parseJsonStringBody(body)).toBe(body);
|
||||
});
|
||||
|
||||
it('leaves a value larger than the 128KB parse guard as a string', () => {
|
||||
it('leaves a body larger than the 128KB parse guard as a string', () => {
|
||||
const huge = `{"x":"${'a'.repeat(130 * 1024)}"}`;
|
||||
expect(parseJsonStringValue(huge)).toBe(huge);
|
||||
});
|
||||
});
|
||||
|
||||
describe('buildPrettyViewData', () => {
|
||||
const baseRaw = {
|
||||
id: 'log-1',
|
||||
timestamp: 1234,
|
||||
body: 'hello',
|
||||
attributes: {},
|
||||
resource: {},
|
||||
scope: {},
|
||||
} as any;
|
||||
|
||||
it('parses a JSON-string body into a tree', () => {
|
||||
const result = buildPrettyViewData({ ...baseRaw, body: '{"a":1}' });
|
||||
expect(result.body).toStrictEqual({ a: 1 });
|
||||
});
|
||||
|
||||
it('parses attribute values that are JSON strings, leaves others as-is', () => {
|
||||
const result = buildPrettyViewData({
|
||||
...baseRaw,
|
||||
attributes: { payload: '{"x":1}', name: 'cart', count: 3 },
|
||||
});
|
||||
expect(result.attributes).toStrictEqual({
|
||||
payload: { x: 1 },
|
||||
name: 'cart',
|
||||
count: 3,
|
||||
});
|
||||
});
|
||||
|
||||
it('drops undefined fields so they do not render as empty rows', () => {
|
||||
const result = buildPrettyViewData({ ...baseRaw, trace_id: undefined });
|
||||
expect('trace_id' in result).toBe(false);
|
||||
expect(parseJsonStringBody(huge)).toBe(huge);
|
||||
});
|
||||
});
|
||||
|
||||
@@ -99,7 +65,7 @@ describe('aggregateAttributesResourcesToObject', () => {
|
||||
'http.method': 'GET',
|
||||
retries: 3,
|
||||
});
|
||||
expect(result.resource).toStrictEqual({ 'service.name': 'cart' });
|
||||
expect(result.resources).toStrictEqual({ 'service.name': 'cart' });
|
||||
expect(result.scope).toStrictEqual({ lib: 'otel' });
|
||||
expect(result.body).toBe('hello');
|
||||
expect(result.id).toBe('log-1');
|
||||
|
||||
@@ -276,7 +276,7 @@ export const aggregateAttributesResourcesToObject = (
|
||||
traceFlags: logData.traceFlags,
|
||||
traceId: logData.traceId,
|
||||
attributes: {},
|
||||
resource: {},
|
||||
resources: {},
|
||||
scope: {},
|
||||
severity_text: logData.severity_text,
|
||||
severity_number: logData.severity_number,
|
||||
@@ -290,8 +290,8 @@ export const aggregateAttributesResourcesToObject = (
|
||||
outputJson.attributes = outputJson.attributes || {};
|
||||
Object.assign(outputJson.attributes, logData[key as keyof ILog]);
|
||||
} else if (key.startsWith('resources_')) {
|
||||
outputJson.resource = outputJson.resource || {};
|
||||
Object.assign(outputJson.resource, logData[key as keyof ILog]);
|
||||
outputJson.resources = outputJson.resources || {};
|
||||
Object.assign(outputJson.resources, logData[key as keyof ILog]);
|
||||
} else if (key.startsWith('scope_string')) {
|
||||
outputJson.scope = outputJson.scope || {};
|
||||
Object.assign(outputJson.scope, logData[key as keyof ILog]);
|
||||
@@ -315,57 +315,30 @@ export const aggregateAttributesResourcesToString = (logData: ILog): string => {
|
||||
}
|
||||
};
|
||||
|
||||
const MAX_JSON_PARSE_BYTES = 128 * 1024;
|
||||
const MAX_JSON_BODY_PARSE_BYTES = 128 * 1024;
|
||||
|
||||
// A JSON-encoded object/array string is parsed so DataViewer renders it as a tree
|
||||
// instead of one escaped string; non-JSON / plain-text values are returned unchanged.
|
||||
// A JSON-encoded object/array `body` is parsed so DataViewer renders it as a
|
||||
// tree instead of one escaped string; plain-text bodies are returned unchanged.
|
||||
// Guarded against very large payloads.
|
||||
export const parseJsonStringValue = (value: unknown): unknown => {
|
||||
if (typeof value !== 'string') {
|
||||
return value;
|
||||
export const parseJsonStringBody = (body: ILog['body']): ILog['body'] => {
|
||||
if (typeof body !== 'string') {
|
||||
return body;
|
||||
}
|
||||
const trimmed = value.trim();
|
||||
const trimmed = body.trim();
|
||||
const looksLikeJson = trimmed.startsWith('{') || trimmed.startsWith('[');
|
||||
if (!looksLikeJson || trimmed.length > MAX_JSON_PARSE_BYTES) {
|
||||
return value;
|
||||
if (!looksLikeJson || trimmed.length > MAX_JSON_BODY_PARSE_BYTES) {
|
||||
return body;
|
||||
}
|
||||
try {
|
||||
const parsed = JSON.parse(trimmed);
|
||||
return parsed !== null && typeof parsed === 'object' ? parsed : value;
|
||||
return parsed !== null && typeof parsed === 'object'
|
||||
? (parsed as ILogBody)
|
||||
: body;
|
||||
} catch {
|
||||
return value;
|
||||
return body;
|
||||
}
|
||||
};
|
||||
|
||||
// Parse each attribute value that's a stringified JSON string into an object
|
||||
// Non-JSON values are left unchanged.
|
||||
const parseAttributeJsonValues = (
|
||||
attributes: Record<string, unknown>,
|
||||
): Record<string, unknown> => {
|
||||
const parsed: Record<string, unknown> = {};
|
||||
Object.keys(attributes).forEach((key) => {
|
||||
parsed[key] = parseJsonStringValue(attributes[key]);
|
||||
});
|
||||
return parsed;
|
||||
};
|
||||
|
||||
export const buildPrettyViewData = (
|
||||
raw: ILogAggregateAttributesResources,
|
||||
): Record<string, unknown> => {
|
||||
const prettyData: Record<string, unknown> = { ...raw };
|
||||
prettyData.body = parseJsonStringValue(raw.body);
|
||||
prettyData.attributes = parseAttributeJsonValues(raw.attributes);
|
||||
|
||||
// drop undefined fields so they don't render as empty rows
|
||||
Object.keys(prettyData).forEach((key) => {
|
||||
if (prettyData[key] === undefined) {
|
||||
delete prettyData[key];
|
||||
}
|
||||
});
|
||||
|
||||
return prettyData;
|
||||
};
|
||||
|
||||
const isFloat = (num: number): boolean => num % 1 !== 0;
|
||||
|
||||
const isBooleanString = (str: string): boolean =>
|
||||
|
||||
@@ -2,9 +2,6 @@ import { blue, red } from '@ant-design/colors';
|
||||
|
||||
export const RESTRICTED_SELECTED_FIELDS = ['timestamp', 'id'];
|
||||
|
||||
// Fields that can be filtered on but not grouped by in the log details view.
|
||||
export const RESTRICTED_GROUP_BY_FIELDS = ['body', 'trace_id'];
|
||||
|
||||
export const ICON_STYLE = {
|
||||
PLUS: { color: blue[5] },
|
||||
CLOSE: { color: red[5] },
|
||||
|
||||
@@ -10,7 +10,7 @@ import DateTimeSelection from '../index';
|
||||
import {
|
||||
__resetSearchParamsGetter,
|
||||
__setSearchParamsGetterForTest,
|
||||
} from 'utils/getUnstableCurrentSearchParams';
|
||||
} from '../utils/getUnstableCurrentSearchParams';
|
||||
import { queryClient, TestWrapper } from './testUtils';
|
||||
|
||||
const mockSafeNavigate = jest.fn();
|
||||
|
||||
@@ -10,7 +10,7 @@ import DateTimeSelection from '../index';
|
||||
import {
|
||||
__resetSearchParamsGetter,
|
||||
__setSearchParamsGetterForTest,
|
||||
} from 'utils/getUnstableCurrentSearchParams';
|
||||
} from '../utils/getUnstableCurrentSearchParams';
|
||||
import { queryClient, TestWrapper } from './testUtils';
|
||||
|
||||
const mockSafeNavigate = jest.fn();
|
||||
|
||||
@@ -24,7 +24,7 @@ import DateTimeSelection from '../index';
|
||||
import {
|
||||
__resetSearchParamsGetter,
|
||||
__setSearchParamsGetterForTest,
|
||||
} from 'utils/getUnstableCurrentSearchParams';
|
||||
} from '../utils/getUnstableCurrentSearchParams';
|
||||
|
||||
const queryClient = new QueryClient({
|
||||
defaultOptions: {
|
||||
|
||||
@@ -5,7 +5,7 @@ import DateTimeSelection from '../index';
|
||||
import {
|
||||
__resetSearchParamsGetter,
|
||||
__setSearchParamsGetterForTest,
|
||||
} from 'utils/getUnstableCurrentSearchParams';
|
||||
} from '../utils/getUnstableCurrentSearchParams';
|
||||
import { queryClient, TestWrapper } from './testUtils';
|
||||
|
||||
const mockSafeNavigate = jest.fn();
|
||||
|
||||
@@ -10,7 +10,7 @@ import DateTimeSelection from '../index';
|
||||
import {
|
||||
__resetSearchParamsGetter,
|
||||
__setSearchParamsGetterForTest,
|
||||
} from 'utils/getUnstableCurrentSearchParams';
|
||||
} from '../utils/getUnstableCurrentSearchParams';
|
||||
import { queryClient, TestWrapper, createMockMoment } from './testUtils';
|
||||
|
||||
const mockSafeNavigate = jest.fn();
|
||||
|
||||
@@ -54,7 +54,7 @@ import {
|
||||
Time,
|
||||
TimeRange,
|
||||
} from './types';
|
||||
import { getUnstableCurrentSearchParams } from 'utils/getUnstableCurrentSearchParams';
|
||||
import { getUnstableCurrentSearchParams } from './utils/getUnstableCurrentSearchParams';
|
||||
|
||||
import './DateTimeSelectionV2.styles.scss';
|
||||
|
||||
|
||||
@@ -189,8 +189,7 @@ function DashboardActions({
|
||||
onClick: (): void => void handleClone(),
|
||||
});
|
||||
}
|
||||
|
||||
if (canEditDashboard && (isAuthor || user.role === USER_ROLES.ADMIN)) {
|
||||
if (isAuthor || user.role === USER_ROLES.ADMIN) {
|
||||
dashboardGroup.push({
|
||||
key: 'lock',
|
||||
label: isDashboardLocked ? 'Unlock dashboard' : 'Lock dashboard',
|
||||
|
||||
@@ -46,11 +46,23 @@ beforeAll(() => {
|
||||
});
|
||||
|
||||
// jest.config maps the real hook to a no-op mock; this suite needs real navigation.
|
||||
jest.mock('hooks/useSafeNavigate', () =>
|
||||
jest
|
||||
.requireActual('tests/browser-history-safe-navigate')
|
||||
.createBrowserHistorySafeNavigateMock(),
|
||||
);
|
||||
jest.mock('hooks/useSafeNavigate', () => {
|
||||
const { useHistory } = jest.requireActual('react-router-dom');
|
||||
return {
|
||||
useSafeNavigate: (): unknown => {
|
||||
const history = useHistory();
|
||||
return {
|
||||
safeNavigate: (to: string, opts?: { replace?: boolean }): void => {
|
||||
if (opts?.replace) {
|
||||
history.replace(to);
|
||||
} else {
|
||||
history.push(to);
|
||||
}
|
||||
},
|
||||
};
|
||||
},
|
||||
};
|
||||
});
|
||||
|
||||
jest.mock('api/querySuggestions/getKeySuggestions', () => ({
|
||||
getKeySuggestions: jest
|
||||
@@ -192,12 +204,9 @@ function Harness(): JSX.Element {
|
||||
);
|
||||
}
|
||||
|
||||
const INITIAL_ROUTE = '/dashboard/dash-1';
|
||||
|
||||
const renderHarness = (): void => {
|
||||
window.history.replaceState(null, '', INITIAL_ROUTE);
|
||||
render(
|
||||
<MemoryRouter initialEntries={[INITIAL_ROUTE]}>
|
||||
<MemoryRouter initialEntries={['/dashboard/dash-1']}>
|
||||
<CompatRouter>
|
||||
<QueryClientProvider client={new QueryClient()}>
|
||||
<ReduxProvider store={configureStore([])(appStore.getState())}>
|
||||
|
||||
@@ -16,11 +16,23 @@ import ViewPanelModal from '../ViewPanelModal/ViewPanelModal';
|
||||
import { useViewPanel } from '../hooks/useViewPanel';
|
||||
|
||||
// jest.config maps the real hook to a no-op mock; this suite needs real navigation.
|
||||
jest.mock('hooks/useSafeNavigate', () =>
|
||||
jest
|
||||
.requireActual('tests/browser-history-safe-navigate')
|
||||
.createBrowserHistorySafeNavigateMock(),
|
||||
);
|
||||
jest.mock('hooks/useSafeNavigate', () => {
|
||||
const { useHistory } = jest.requireActual('react-router-dom');
|
||||
return {
|
||||
useSafeNavigate: (): unknown => {
|
||||
const history = useHistory();
|
||||
return {
|
||||
safeNavigate: (to: string, opts?: { replace?: boolean }): void => {
|
||||
if (opts?.replace) {
|
||||
history.replace(to);
|
||||
} else {
|
||||
history.push(to);
|
||||
}
|
||||
},
|
||||
};
|
||||
},
|
||||
};
|
||||
});
|
||||
|
||||
jest.mock(
|
||||
'pages/DashboardPageV2/DashboardContainer/hooks/usePanelQuery',
|
||||
@@ -138,12 +150,9 @@ function Harness(): JSX.Element {
|
||||
);
|
||||
}
|
||||
|
||||
const INITIAL_ROUTE = '/dashboard/dash-1';
|
||||
|
||||
const renderHarness = (): void => {
|
||||
window.history.replaceState(null, '', INITIAL_ROUTE);
|
||||
render(
|
||||
<MemoryRouter initialEntries={[INITIAL_ROUTE]}>
|
||||
<MemoryRouter initialEntries={['/dashboard/dash-1']}>
|
||||
<CompatRouter>
|
||||
<QueryClientProvider client={new QueryClient()}>
|
||||
<ReduxProvider store={configureStore([])(appStore.getState())}>
|
||||
|
||||
@@ -14,11 +14,23 @@ import { useViewPanelMode } from '../ViewPanelModal/useViewPanelMode';
|
||||
import { useViewPanel } from '../hooks/useViewPanel';
|
||||
|
||||
// jest.config maps the real hook to a no-op mock; this suite needs real navigation.
|
||||
jest.mock('hooks/useSafeNavigate', () =>
|
||||
jest
|
||||
.requireActual('tests/browser-history-safe-navigate')
|
||||
.createBrowserHistorySafeNavigateMock(),
|
||||
);
|
||||
jest.mock('hooks/useSafeNavigate', () => {
|
||||
const { useHistory } = jest.requireActual('react-router-dom');
|
||||
return {
|
||||
useSafeNavigate: (): unknown => {
|
||||
const history = useHistory();
|
||||
return {
|
||||
safeNavigate: (to: string, opts?: { replace?: boolean }): void => {
|
||||
if (opts?.replace) {
|
||||
history.replace(to);
|
||||
} else {
|
||||
history.push(to);
|
||||
}
|
||||
},
|
||||
};
|
||||
},
|
||||
};
|
||||
});
|
||||
|
||||
jest.mock(
|
||||
'pages/DashboardPageV2/DashboardContainer/hooks/usePanelQuery',
|
||||
@@ -172,12 +184,9 @@ function Harness(): JSX.Element {
|
||||
);
|
||||
}
|
||||
|
||||
const INITIAL_ROUTE = '/dashboard/dash-1';
|
||||
|
||||
const renderHarness = (): void => {
|
||||
window.history.replaceState(null, '', INITIAL_ROUTE);
|
||||
render(
|
||||
<MemoryRouter initialEntries={[INITIAL_ROUTE]}>
|
||||
<MemoryRouter initialEntries={['/dashboard/dash-1']}>
|
||||
<CompatRouter>
|
||||
<QueryBuilderProvider>
|
||||
<Harness />
|
||||
|
||||
@@ -19,20 +19,11 @@ import { resolveDashboardImage } from 'pages/DashboardPageV2/DashboardContainer/
|
||||
interface DashboardContainerProps {
|
||||
dashboard: DashboardtypesGettableDashboardV2DTO;
|
||||
refetch: () => void;
|
||||
/**
|
||||
* @deprecated
|
||||
* `canEditDashboardOverride` is a temporary solution to allow the dashboard to be view only.
|
||||
* This is only used for LLM Observability.
|
||||
* It will be removed in the future.
|
||||
* TODO: @Ashwin / @Abhi — remove when the final solution is implemented.
|
||||
*/
|
||||
canEditDashboardOverride?: boolean;
|
||||
}
|
||||
|
||||
function DashboardContainer({
|
||||
dashboard,
|
||||
refetch,
|
||||
canEditDashboardOverride,
|
||||
}: DashboardContainerProps): JSX.Element {
|
||||
const spec = dashboard.spec;
|
||||
const image = resolveDashboardImage(dashboard.image);
|
||||
@@ -54,11 +45,10 @@ function DashboardContainer({
|
||||
// Seed during render (not an effect) so the first Panel render already sees the id —
|
||||
// useDashboardFetchRequired throws on a missing id. setEditContext self-guards.
|
||||
const setEditContext = useDashboardStore((s) => s.setEditContext);
|
||||
|
||||
setEditContext({
|
||||
dashboardId: dashboard.id,
|
||||
isLocked,
|
||||
canEditDashboard: canEditDashboardOverride ?? canEditDashboard,
|
||||
canEditDashboard,
|
||||
refetch,
|
||||
});
|
||||
|
||||
|
||||
@@ -37,6 +37,7 @@ import { OptionsQuery } from 'container/OptionsMenu/types';
|
||||
import { useGetCompositeQueryParam } from 'hooks/queryBuilder/useGetCompositeQueryParam';
|
||||
import { updateStepInterval } from 'hooks/queryBuilder/useStepInterval';
|
||||
import { useSafeNavigate } from 'hooks/useSafeNavigate';
|
||||
import useUrlQuery from 'hooks/useUrlQuery';
|
||||
import { createIdFromObjectFields } from 'lib/createIdFromObjectFields';
|
||||
import { createNewBuilderItemName } from 'lib/newQueryBuilder/createNewBuilderItemName';
|
||||
import { getOperatorsBySourceAndPanelType } from 'lib/newQueryBuilder/getOperatorsBySourceAndPanelType';
|
||||
@@ -65,7 +66,6 @@ import {
|
||||
} from 'types/common/queryBuilder';
|
||||
import { sanitizeOrderByForExplorer } from 'utils/sanitizeOrderBy';
|
||||
import { v4 as uuid } from 'uuid';
|
||||
import { getUnstableCurrentSearchParams } from 'utils/getUnstableCurrentSearchParams';
|
||||
|
||||
export const QueryBuilderContext = createContext<QueryBuilderContextType>({
|
||||
currentQuery: initialQueriesMap.metrics,
|
||||
@@ -105,6 +105,7 @@ export const QueryBuilderContext = createContext<QueryBuilderContextType>({
|
||||
export function QueryBuilderProvider({
|
||||
children,
|
||||
}: PropsWithChildren): JSX.Element {
|
||||
const urlQuery = useUrlQuery();
|
||||
const location = useLocation();
|
||||
|
||||
const currentPathnameRef = useRef<string | null>(location.pathname);
|
||||
@@ -121,7 +122,7 @@ export function QueryBuilderProvider({
|
||||
null,
|
||||
);
|
||||
|
||||
const panelTypeQueryParams = getUnstableCurrentSearchParams().get(
|
||||
const panelTypeQueryParams = urlQuery.get(
|
||||
QueryParams.panelTypes,
|
||||
) as PANEL_TYPES | null;
|
||||
|
||||
@@ -975,7 +976,6 @@ export function QueryBuilderProvider({
|
||||
unit: query.unit || initialQueryState.unit,
|
||||
};
|
||||
|
||||
const urlQuery = getUnstableCurrentSearchParams();
|
||||
const pagination = urlQuery.get(QueryParams.pagination);
|
||||
|
||||
if (pagination) {
|
||||
@@ -1014,7 +1014,7 @@ export function QueryBuilderProvider({
|
||||
|
||||
safeNavigate(generatedUrl, { newTab });
|
||||
},
|
||||
[location.pathname, safeNavigate],
|
||||
[location.pathname, safeNavigate, urlQuery],
|
||||
);
|
||||
|
||||
const handleSetConfig = useCallback(
|
||||
|
||||
@@ -1,54 +0,0 @@
|
||||
// Mock factory for suites that need `useSafeNavigate` to navigate for real.
|
||||
//
|
||||
// `jest.config.ts` maps every `hooks/useSafeNavigate` import to the no-op
|
||||
// `__tests__/safeNavigateMock.ts`, so a suite that drives navigation has to opt
|
||||
// out with its own `jest.mock`.
|
||||
//
|
||||
// In production `safeNavigate` goes through `createBrowserHistory`, which writes
|
||||
// `window.location` as well as notifying the router. `MemoryRouter` never touches
|
||||
// `window`, so anything reading `getUnstableCurrentSearchParams()` sees an empty
|
||||
// search and drops the params the test just navigated with. This mock writes both.
|
||||
//
|
||||
// The `jest.mock` factory is hoisted above imports, so require it inside:
|
||||
//
|
||||
// jest.mock('hooks/useSafeNavigate', () =>
|
||||
// jest
|
||||
// .requireActual('tests/browser-history-safe-navigate')
|
||||
// .createBrowserHistorySafeNavigateMock(),
|
||||
// );
|
||||
|
||||
import type { History } from 'history';
|
||||
|
||||
interface SafeNavigateOptions {
|
||||
replace?: boolean;
|
||||
}
|
||||
|
||||
interface UseSafeNavigateModule {
|
||||
useSafeNavigate: () => {
|
||||
safeNavigate: (to: string, options?: SafeNavigateOptions) => void;
|
||||
};
|
||||
}
|
||||
|
||||
export function createBrowserHistorySafeNavigateMock(): UseSafeNavigateModule {
|
||||
const { useHistory } = jest.requireActual<{ useHistory: () => History }>(
|
||||
'react-router-dom',
|
||||
);
|
||||
|
||||
return {
|
||||
useSafeNavigate: () => {
|
||||
const history = useHistory();
|
||||
|
||||
return {
|
||||
safeNavigate: (to: string, options?: SafeNavigateOptions): void => {
|
||||
if (options?.replace) {
|
||||
window.history.replaceState(null, '', to);
|
||||
history.replace(to);
|
||||
} else {
|
||||
window.history.pushState(null, '', to);
|
||||
history.push(to);
|
||||
}
|
||||
},
|
||||
};
|
||||
},
|
||||
};
|
||||
}
|
||||
@@ -41,6 +41,6 @@ type OmitAttributesResources = Pick<
|
||||
|
||||
export type ILogAggregateAttributesResources = OmitAttributesResources & {
|
||||
attributes: Record<string, never>;
|
||||
resource: Record<string, never>;
|
||||
resources: Record<string, never>;
|
||||
scope: Record<string, never>;
|
||||
};
|
||||
|
||||
@@ -7,6 +7,7 @@ import (
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"net/http"
|
||||
"net/url"
|
||||
"unicode/utf8"
|
||||
|
||||
"github.com/SigNoz/signoz/pkg/alertmanager/alertmanagertemplate"
|
||||
@@ -185,7 +186,18 @@ func (n *Notifier) Notify(ctx context.Context, alerts ...*types.Alert) (bool, er
|
||||
}
|
||||
}
|
||||
|
||||
resp, err := notify.PostJSON(ctx, n.client, n.conf.WebhookURL.String(), buf) //nolint:bodyclose
|
||||
// Thread same-rule alerts together: threadKey is a stable hash of the
|
||||
// alert group key. Changing a rule's grouping starts a new thread.
|
||||
u, err := url.Parse(n.conf.WebhookURL.String())
|
||||
if err != nil {
|
||||
return false, errors.WrapInternalf(err, errors.CodeInternal, "parse google chat webhook url")
|
||||
}
|
||||
q := u.Query()
|
||||
q.Set("threadKey", key.Hash())
|
||||
q.Set("messageReplyOption", "REPLY_MESSAGE_FALLBACK_TO_NEW_THREAD")
|
||||
u.RawQuery = q.Encode()
|
||||
|
||||
resp, err := notify.PostJSON(ctx, n.client, u.String(), buf) //nolint:bodyclose
|
||||
if err != nil {
|
||||
return true, notify.RedactURL(err)
|
||||
}
|
||||
|
||||
@@ -245,7 +245,7 @@ func TestGoogleChatMessageSizeLimit(t *testing.T) {
|
||||
assert.LessOrEqual(t, bodyLen, maxMessageBytes, "posted body must be within the size limit")
|
||||
}
|
||||
|
||||
func TestGoogleChatWebhookURLVerbatim(t *testing.T) {
|
||||
func TestGoogleChatThreading(t *testing.T) {
|
||||
var query url.Values
|
||||
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
query = r.URL.Query()
|
||||
@@ -253,11 +253,25 @@ func TestGoogleChatWebhookURLVerbatim(t *testing.T) {
|
||||
}))
|
||||
defer server.Close()
|
||||
|
||||
n := newTestNotifier(t, server.URL+"?key=abc&token=xyz", "T", "")
|
||||
_, err := n.Notify(newTestContext(), newTestAlerts("X")...)
|
||||
require.NoError(t, err)
|
||||
cases := []struct{ name, groupKey string }{
|
||||
{"rule a", "{ruleId=\"aaa\"}"},
|
||||
{"rule b", "{ruleId=\"bbb\"}"},
|
||||
}
|
||||
seen := map[string]string{}
|
||||
for _, c := range cases {
|
||||
t.Run(c.name, func(t *testing.T) {
|
||||
n := newTestNotifier(t, server.URL, "T", "")
|
||||
ctx := notify.WithGroupKey(context.Background(), c.groupKey)
|
||||
_, err := n.Notify(ctx, newTestAlerts("X")...)
|
||||
require.NoError(t, err)
|
||||
|
||||
assert.Equal(t, url.Values{"key": {"abc"}, "token": {"xyz"}}, query, "configured webhook URL must be posted verbatim, with no params added")
|
||||
assert.Equal(t, "REPLY_MESSAGE_FALLBACK_TO_NEW_THREAD", query.Get("messageReplyOption"))
|
||||
threadKey := query.Get("threadKey")
|
||||
assert.Equal(t, notify.Key(c.groupKey).Hash(), threadKey, "threadKey must be the group key hash")
|
||||
seen[c.name] = threadKey
|
||||
})
|
||||
}
|
||||
assert.NotEqual(t, seen["rule a"], seen["rule b"], "distinct group keys must yield distinct threadKeys")
|
||||
}
|
||||
|
||||
func TestGoogleChatCustomTemplateMarkdown(t *testing.T) {
|
||||
|
||||
@@ -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()}
|
||||
|
||||
@@ -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) {
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
@@ -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")
|
||||
}
|
||||
@@ -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, ".$`") {
|
||||
|
||||
@@ -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
|
||||
}
|
||||
@@ -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))
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
@@ -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,
|
||||
@@ -521,6 +562,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 +727,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))
|
||||
|
||||
13
tests/e2e/pnpm-lock.yaml
generated
13
tests/e2e/pnpm-lock.yaml
generated
@@ -4,9 +4,6 @@ settings:
|
||||
autoInstallPeers: true
|
||||
excludeLinksFromLockfile: false
|
||||
|
||||
overrides:
|
||||
brace-expansion@>=5.0.0 <5.0.9: '>=5.0.9 <6'
|
||||
|
||||
importers:
|
||||
|
||||
.:
|
||||
@@ -380,9 +377,9 @@ packages:
|
||||
resolution: {integrity: sha512-BLrgEcRTwX2o6gGxGOCNyMvGSp35YofuYzw9h1IMTRmKqttAZZVU67bdb9Pr2vUHA8+j3i2tJfjO6C6+4myGTA==}
|
||||
engines: {node: 18 || 20 || >=22}
|
||||
|
||||
brace-expansion@5.0.9:
|
||||
resolution: {integrity: sha512-ScQ4IuvIEF1TMlP7Zt+vjJ//9zlPb2SDcxWxM3bk8s6t6GGdJ7KO1dCcTidOPJKePW30LE/2cT7wCyPho9/Wxg==}
|
||||
engines: {node: 20 || >=22}
|
||||
brace-expansion@5.0.5:
|
||||
resolution: {integrity: sha512-VZznLgtwhn+Mact9tfiwx64fA9erHH/MCXEUfB/0bX/6Fz6ny5EGTXYltMocqg4xFAQZtnO3DHWWXi8RiuN7cQ==}
|
||||
engines: {node: 18 || 20 || >=22}
|
||||
|
||||
cross-spawn@7.0.6:
|
||||
resolution: {integrity: sha512-uV2QOWP2nWzsy2aMp8aRibhi9dlzF5Hgh5SHaB9OiTGEyDTiJJyx0uy51QXdyWbtAHNua4XJzUKca3OzKUd3vA==}
|
||||
@@ -848,7 +845,7 @@ snapshots:
|
||||
|
||||
balanced-match@4.0.4: {}
|
||||
|
||||
brace-expansion@5.0.9:
|
||||
brace-expansion@5.0.5:
|
||||
dependencies:
|
||||
balanced-match: 4.0.4
|
||||
|
||||
@@ -1001,7 +998,7 @@ snapshots:
|
||||
|
||||
minimatch@10.2.5:
|
||||
dependencies:
|
||||
brace-expansion: 5.0.9
|
||||
brace-expansion: 5.0.5
|
||||
|
||||
ms@2.1.3: {}
|
||||
|
||||
|
||||
@@ -1,6 +0,0 @@
|
||||
# Security floors for vulnerable transitive deps. Where possible, targets are
|
||||
# capped to avoid crossing breaking versions (major; and minor for 0.x).
|
||||
overrides:
|
||||
# via: eslint-plugin-playwright > eslint@10 > minimatch@10.2.5 (brace-expansion ^5.0.5)
|
||||
# remove: blocked — minimatch@10.2.6 (latest) only widens to ^5.0.8, still vulnerable
|
||||
'brace-expansion@>=5.0.0 <5.0.9': '>=5.0.9 <6'
|
||||
82
tests/fixtures/querierai.py
vendored
82
tests/fixtures/querierai.py
vendored
@@ -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])
|
||||
|
||||
@@ -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"
|
||||
|
||||
|
||||
502
tests/integration/tests/querierai/02_ai_aggregations.py
Normal file
502
tests/integration/tests/querierai/02_ai_aggregations.py
Normal 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",
|
||||
)
|
||||
Reference in New Issue
Block a user