Compare commits

..

1 Commits

Author SHA1 Message Date
Ashwin Bhatkal
06dd0ed3d3 feat(alert-channel-integrations): google chat final leg (#12335)
Some checks are pending
build-staging / prepare (push) Waiting to run
build-staging / js-build (push) Blocked by required conditions
build-staging / go-build (push) Blocked by required conditions
build-staging / staging (push) Blocked by required conditions
cacheci / tests (push) Waiting to run
Release Drafter / update_release_draft (push) Waiting to run
* feat(alert-channels): add google chat channel type, defaults and url validation

Prefills the title and description templates verbatim from the backend's
DefaultGoogleChatReceiverConfig, and extracts the slack templates into
SlackInitialConfig so the form can restore them when the type changes.

* feat(alert-channels): add google chat option and form fields

* feat(alert-channels): wire google chat create, edit and test calls

* test(alert-channels): cover the google chat channel form

* fix(googlechat): use standard markdown in default templates for cardsV2
2026-08-05 06:13:25 +00:00
28 changed files with 696 additions and 2412 deletions

View File

@@ -24,6 +24,8 @@
"tooltip_opsgenie_api_key": "Learn how to obtain the API key from your OpsGenie account [here](https://support.atlassian.com/opsgenie/docs/integrate-opsgenie-with-prometheus/).",
"tooltip_email_to": "Enter email addresses separated by commas.",
"tooltip_ms_teams_url": "The URL of the Microsoft Teams [webhook](https://support.microsoft.com/en-us/office/create-incoming-webhooks-with-workflows-for-microsoft-teams-8ae491c7-0394-4861-ba59-055e33f75498) to send alerts to. Learn more about Microsoft Teams integration in the docs [here](https://signoz.io/docs/alerts-management/notification-channel/ms-teams/).",
"tooltip_google_chat_url": "The URL of the Google Chat space [incoming webhook](https://developers.google.com/workspace/chat/quickstart/webhooks) to send alerts to. It must be an https URL on chat.googleapis.com.",
"google_chat_webhook_url_invalid": "Webhook URL must be an https URL on chat.googleapis.com",
"field_slack_recipient": "Recipient",
"field_slack_title": "Title",

View File

@@ -24,6 +24,8 @@
"tooltip_opsgenie_api_key": "Learn how to obtain the API key from your OpsGenie account [here](https://support.atlassian.com/opsgenie/docs/integrate-opsgenie-with-prometheus/).",
"tooltip_email_to": "Enter email addresses separated by commas.",
"tooltip_ms_teams_url": "The URL of the Microsoft Teams [webhook](https://support.microsoft.com/en-us/office/create-incoming-webhooks-with-workflows-for-microsoft-teams-8ae491c7-0394-4861-ba59-055e33f75498) to send alerts to. Learn more about Microsoft Teams integration in the docs [here](https://signoz.io/docs/alerts-management/notification-channel/ms-teams/).",
"tooltip_google_chat_url": "The URL of the Google Chat space [incoming webhook](https://developers.google.com/workspace/chat/quickstart/webhooks) to send alerts to. It must be an https URL on chat.googleapis.com.",
"google_chat_webhook_url_invalid": "Webhook URL must be an https URL on chat.googleapis.com",
"field_slack_recipient": "Recipient",
"field_slack_title": "Title",
"field_slack_description": "Description",

View File

@@ -1,18 +1,28 @@
import CreateAlertChannels from 'container/CreateAlertChannels';
import { ChannelType } from 'container/CreateAlertChannels/config';
import { GoogleChatInitialConfig } from 'container/CreateAlertChannels/defaults';
import {
googleChatDescriptionDefaultValue,
googleChatTitleDefaultValue,
opsGenieDescriptionDefaultValue,
opsGenieMessageDefaultValue,
opsGeniePriorityDefaultValue,
pagerDutyAdditionalDetailsDefaultValue,
pagerDutyDescriptionDefaultVaule,
pagerDutyDescriptionDefaultValue,
pagerDutySeverityTextDefaultValue,
slackDescriptionDefaultValue,
slackTitleDefaultValue,
} from 'mocks-server/__mockdata__/alerts';
import { server } from 'mocks-server/server';
import { rest } from 'msw';
import { act, fireEvent, render, screen, waitFor } from 'tests/test-utils';
import {
act,
fireEvent,
render,
screen,
userEvent,
waitFor,
} from 'tests/test-utils';
import { testLabelInputAndHelpValue } from './testUtils';
@@ -225,7 +235,7 @@ describe('Create Alert Channel', () => {
);
expect(descriptionTextArea).toHaveTextContent(
pagerDutyDescriptionDefaultVaule,
pagerDutyDescriptionDefaultValue,
);
});
it('Should check if Severity label, info (help_pager_severity), and textbox are displayed properly', () => {
@@ -419,5 +429,150 @@ describe('Create Alert Channel', () => {
expect(descriptionTextArea).toHaveTextContent(slackDescriptionDefaultValue);
});
});
describe('Google Chat', () => {
const validWebhookUrl =
'https://chat.googleapis.com/v1/spaces/AAAA/messages?key=dummy_key&token=dummy_token';
beforeEach(() => {
render(<CreateAlertChannels preType={ChannelType.GoogleChat} />);
});
it('Should check if the selected item in the type dropdown has text "Google Chat"', () => {
expect(screen.getByText('Google Chat')).toBeInTheDocument();
});
it('Should check if Webhook URL label and input are displayed properly', () => {
testLabelInputAndHelpValue({
labelText: 'field_webhook_url',
testId: 'webhook-url-textbox',
});
});
it('Should check if Title contains the google chat template', () => {
expect(screen.getByTestId('title-textarea')).toHaveTextContent(
googleChatTitleDefaultValue,
);
});
it('Should check if Description contains the google chat template', () => {
expect(screen.getByTestId('description-textarea')).toHaveTextContent(
googleChatDescriptionDefaultValue,
);
});
it('Should check if saving with a webhook url outside chat.googleapis.com displays error notification', async () => {
const user = userEvent.setup();
await user.type(
screen.getByTestId('channel-name-textbox'),
'gchat-channel',
);
await user.type(
screen.getByTestId('webhook-url-textbox'),
'https://example.com/webhook',
);
await user.click(screen.getByTestId('save-channel-button'));
await waitFor(() =>
expect(errorNotification).toHaveBeenCalledWith({
message: 'Error',
description: 'google_chat_webhook_url_invalid',
}),
);
});
it('Should check if saving sends a googlechat_configs payload', async () => {
let requestBody: unknown;
server.use(
rest.post('http://localhost/api/v1/channels', async (req, res, ctx) => {
requestBody = await req.json();
return res(
ctx.status(201),
ctx.json({ status: 'success', data: 'channel created' }),
);
}),
);
const user = userEvent.setup();
await user.type(
screen.getByTestId('channel-name-textbox'),
'gchat-channel',
);
await user.type(screen.getByTestId('webhook-url-textbox'), validWebhookUrl);
await user.click(screen.getByTestId('save-channel-button'));
await waitFor(() =>
expect(successNotification).toHaveBeenCalledWith({
message: 'Success',
description: 'channel_creation_done',
}),
);
expect(requestBody).toStrictEqual({
name: 'gchat-channel',
googlechat_configs: [
{
webhook_url: validWebhookUrl,
title: GoogleChatInitialConfig.title,
text: GoogleChatInitialConfig.text,
send_resolved: true,
},
],
});
});
});
describe('Changing the channel type', () => {
async function selectType(
user: ReturnType<typeof userEvent.setup>,
optionText: string,
): Promise<void> {
// the type dropdown opens on the inner search input of the antd select
await user.click(screen.getByRole('combobox'));
await user.click(await screen.findByTitle(optionText));
}
it('Should check if switching to Google Chat and back swaps the prefilled templates', async () => {
const user = userEvent.setup();
render(<CreateAlertChannels preType={ChannelType.Slack} />);
await selectType(user, 'Google Chat');
await waitFor(() =>
expect(screen.getByTestId('title-textarea')).toHaveTextContent(
googleChatTitleDefaultValue,
),
);
expect(screen.getByTestId('description-textarea')).toHaveTextContent(
googleChatDescriptionDefaultValue,
);
await selectType(user, 'Slack');
await waitFor(() =>
expect(screen.getByTestId('title-textarea')).toHaveTextContent(
slackTitleDefaultValue,
),
);
expect(screen.getByTestId('description-textarea')).toHaveTextContent(
slackDescriptionDefaultValue,
);
});
it('Should check if switching to Pagerduty prefills the pagerduty description and not the opsgenie one', async () => {
const user = userEvent.setup();
render(<CreateAlertChannels preType={ChannelType.Opsgenie} />);
await selectType(user, 'Pagerduty');
await waitFor(() =>
expect(screen.getByTestId('pager-description-textarea')).toHaveTextContent(
pagerDutyDescriptionDefaultValue,
),
);
});
});
});
});

View File

@@ -5,7 +5,7 @@ import {
opsGenieMessageDefaultValue,
opsGeniePriorityDefaultValue,
pagerDutyAdditionalDetailsDefaultValue,
pagerDutyDescriptionDefaultVaule,
pagerDutyDescriptionDefaultValue,
pagerDutySeverityTextDefaultValue,
slackDescriptionDefaultValue,
slackTitleDefaultValue,
@@ -150,7 +150,7 @@ describe('Create Alert Channel (Normal User)', () => {
);
expect(descriptionTextArea).toHaveTextContent(
pagerDutyDescriptionDefaultVaule,
pagerDutyDescriptionDefaultValue,
);
});
it('Should check if Severity label, info (help_pager_severity), and textbox are displayed properly', () => {

View File

@@ -104,6 +104,7 @@ export enum ChannelType {
Pagerduty = 'pagerduty',
Opsgenie = 'opsgenie',
MsTeams = 'msteams',
GoogleChat = 'googlechat',
}
// LabelFilterStatement will be used for preparing filter conditions / matchers
@@ -125,3 +126,11 @@ export interface MsTeamsChannel extends Channel {
title?: string;
text?: string;
}
export interface GoogleChatChannel extends Channel {
// incoming webhook url of the google chat space, must be an
// https url on chat.googleapis.com
webhook_url?: string;
title?: string;
text?: string;
}

View File

@@ -1,4 +1,51 @@
import { EmailChannel, OpsgenieChannel, PagerChannel } from './config';
import {
ChannelType,
EmailChannel,
GoogleChatChannel,
MsTeamsChannel,
OpsgenieChannel,
PagerChannel,
SlackChannel,
WebhookChannel,
} from './config';
// shared by slack and ms teams, both render the same title / description boxes
export const SlackInitialConfig: Partial<SlackChannel> = {
text: `{{ range .Alerts -}}
*Alert:* {{ .Labels.alertname }}{{ if .Labels.severity }} - {{ .Labels.severity }}{{ end }}
*Summary:* {{ .Annotations.summary }}
*Description:* {{ .Annotations.description }}
*RelatedLogs:* {{ if gt (len .Annotations.related_logs) 0 -}} View in <{{ .Annotations.related_logs }}|logs explorer> {{- end}}
*RelatedTraces:* {{ if gt (len .Annotations.related_traces) 0 -}} View in <{{ .Annotations.related_traces }}|traces explorer> {{- end}}
*Details:*
{{ range .Labels.SortedPairs }} • *{{ .Name }}:* {{ .Value }}
{{ end }}
{{ end }}`,
title: `[{{ .Status | toUpper }}{{ if eq .Status "firing" }}:{{ .Alerts.Firing | len }}{{ end }}] {{ .CommonLabels.alertname }} for {{ .CommonLabels.job }}
{{- if gt (len .CommonLabels) (len .GroupLabels) -}}
{{" "}}(
{{- with .CommonLabels.Remove .GroupLabels.Names }}
{{- range $index, $label := .SortedPairs -}}
{{ if $index }}, {{ end }}
{{- $label.Name }}="{{ $label.Value -}}"
{{- end }}
{{- end -}}
)
{{- end }}`,
};
// mirrors DefaultGoogleChatReceiverConfig in pkg/types/alertmanagertypes/googlechat.go,
// which the backend applies when title / text are left empty
export const GoogleChatInitialConfig: Partial<GoogleChatChannel> = {
title: `[{{ .Status | toUpper }}{{ if eq .Status "firing" }}:{{ .Alerts.Firing | len }}{{ end }}] {{ .CommonLabels.alertname }}`,
text: `{{ range .Alerts -}}
**Alert:** {{ .Labels.alertname }}{{ if .Labels.severity }} ({{ .Labels.severity }}){{ end }}{{ if .Annotations.summary }}
**Summary:** {{ .Annotations.summary }}{{ end }}{{ if .Annotations.description }}
**Description:** {{ .Annotations.description }}{{ end }}
{{ end }}`,
};
export const PagerInitialConfig: Partial<PagerChannel> = {
description: `[{{ .Status | toUpper }}{{ if eq .Status "firing" }}:{{ .Alerts.Firing | len }}{{ end }}] {{ .CommonLabels.alertname }} for {{ .CommonLabels.job }}
@@ -446,3 +493,26 @@ export const EmailInitialConfig: Partial<EmailChannel> = {
</body>
</html>`,
};
// prefilled values of every channel type, keyed by type so the form can apply
// exactly one set of defaults and swap it when the type changes
export const ChannelInitialConfig: Record<
ChannelType,
Partial<
SlackChannel &
WebhookChannel &
PagerChannel &
MsTeamsChannel &
OpsgenieChannel &
EmailChannel &
GoogleChatChannel
>
> = {
[ChannelType.Slack]: SlackInitialConfig,
[ChannelType.MsTeams]: SlackInitialConfig,
[ChannelType.GoogleChat]: GoogleChatInitialConfig,
[ChannelType.Pagerduty]: PagerInitialConfig,
[ChannelType.Opsgenie]: OpsgenieInitialConfig,
[ChannelType.Email]: EmailInitialConfig,
[ChannelType.Webhook]: {},
};

View File

@@ -14,16 +14,24 @@ import testPagerApi from 'api/channels/testPager';
import testSlackApi from 'api/channels/testSlack';
import testWebhookApi from 'api/channels/testWebhook';
import logEvent from 'api/common/logEvent';
import {
useCreateChannel,
useTestChannel,
} from 'api/generated/services/channels';
import { RenderErrorResponseDTO } from 'api/generated/services/sigNoz.schemas';
import { ErrorType } from 'api/generatedAPIInstance';
import ROUTES from 'constants/routes';
import FormAlertChannels from 'container/FormAlertChannels';
import { useNotifications } from 'hooks/useNotifications';
import history from 'lib/history';
import { useErrorModal } from 'providers/ErrorModalProvider';
import APIError from 'types/api/error';
import { toAPIError } from 'utils/errorUtils';
import {
ChannelType,
EmailChannel,
GoogleChatChannel,
MsTeamsChannel,
OpsgenieChannel,
PagerChannel,
@@ -31,12 +39,12 @@ import {
ValidatePagerChannel,
WebhookChannel,
} from './config';
import { ChannelInitialConfig } from './defaults';
import {
EmailInitialConfig,
OpsgenieInitialConfig,
PagerInitialConfig,
} from './defaults';
import { isChannelType } from './utils';
isChannelType,
isValidGoogleChatWebhookURL,
prepareGoogleChatRequest,
} from './utils';
import './CreateAlertChannels.styles.scss';
@@ -60,69 +68,38 @@ function CreateAlertChannels({
PagerChannel &
MsTeamsChannel &
OpsgenieChannel &
EmailChannel
EmailChannel &
GoogleChatChannel
>
>({
>(() => ({
send_resolved: true,
text: `{{ range .Alerts -}}
*Alert:* {{ .Labels.alertname }}{{ if .Labels.severity }} - {{ .Labels.severity }}{{ end }}
*Summary:* {{ .Annotations.summary }}
*Description:* {{ .Annotations.description }}
*RelatedLogs:* {{ if gt (len .Annotations.related_logs) 0 -}} View in <{{ .Annotations.related_logs }}|logs explorer> {{- end}}
*RelatedTraces:* {{ if gt (len .Annotations.related_traces) 0 -}} View in <{{ .Annotations.related_traces }}|traces explorer> {{- end}}
*Details:*
{{ range .Labels.SortedPairs }} • *{{ .Name }}:* {{ .Value }}
{{ end }}
{{ end }}`,
title: `[{{ .Status | toUpper }}{{ if eq .Status "firing" }}:{{ .Alerts.Firing | len }}{{ end }}] {{ .CommonLabels.alertname }} for {{ .CommonLabels.job }}
{{- if gt (len .CommonLabels) (len .GroupLabels) -}}
{{" "}}(
{{- with .CommonLabels.Remove .GroupLabels.Names }}
{{- range $index, $label := .SortedPairs -}}
{{ if $index }}, {{ end }}
{{- $label.Name }}="{{ $label.Value -}}"
{{- end }}
{{- end -}}
)
{{- end }}`,
});
...ChannelInitialConfig[preType],
}));
const [savingState, setSavingState] = useState<boolean>(false);
const [testingState, setTestingState] = useState<boolean>(false);
const { notifications } = useNotifications();
const { mutateAsync: createChannel } = useCreateChannel();
const { mutateAsync: testChannel } = useTestChannel();
const [type, setType] = useState<ChannelType>(preType);
const onTypeChangeHandler = useCallback(
(value: string) => {
const currentType = type;
setType(value as ChannelType);
if (value === ChannelType.Pagerduty && currentType !== value) {
// reset config to pager defaults
setSelectedConfig({
name: selectedConfig?.name,
send_resolved: selectedConfig.send_resolved,
...PagerInitialConfig,
});
const nextType = value as ChannelType;
if (nextType === type) {
return;
}
if (value === ChannelType.Opsgenie && currentType !== value) {
setSelectedConfig((selectedConfig) => ({
...selectedConfig,
...OpsgenieInitialConfig,
}));
}
setType(nextType);
// reset config to email defaults
if (value === ChannelType.Email && currentType !== value) {
setSelectedConfig((selectedConfig) => ({
...selectedConfig,
...EmailInitialConfig,
}));
}
// the fields the types share (title, text, description) keep the value of
// the type that was selected before, so the new type's defaults have to be
// written to both the config and the form
const defaults = ChannelInitialConfig[nextType];
setSelectedConfig((selectedConfig) => ({ ...selectedConfig, ...defaults }));
formInstance.setFieldsValue(defaults);
},
[type, selectedConfig],
[type, formInstance],
);
const prepareSlackRequest = useCallback(
@@ -407,6 +384,56 @@ function CreateAlertChannels({
showErrorModal,
]);
const validateGoogleChatConfig = useCallback((): boolean => {
if (!selectedConfig.webhook_url) {
notifications.error({
message: 'Error',
description: t('webhook_url_required'),
});
return false;
}
if (!isValidGoogleChatWebhookURL(selectedConfig.webhook_url)) {
notifications.error({
message: 'Error',
description: t('google_chat_webhook_url_invalid'),
});
return false;
}
return true;
}, [selectedConfig.webhook_url, notifications, t]);
const onGoogleChatHandler = useCallback(async () => {
if (!validateGoogleChatConfig()) {
return { status: 'failed', statusMessage: t('channel_creation_failed') };
}
setSavingState(true);
try {
await createChannel({ data: prepareGoogleChatRequest(selectedConfig) });
notifications.success({
message: 'Success',
description: t('channel_creation_done'),
});
history.replace(ROUTES.ALL_CHANNELS);
return { status: 'success', statusMessage: t('channel_creation_done') };
} catch (error) {
showErrorModal(toAPIError(error as ErrorType<RenderErrorResponseDTO>));
return { status: 'failed', statusMessage: t('channel_creation_failed') };
} finally {
setSavingState(false);
}
}, [
validateGoogleChatConfig,
createChannel,
selectedConfig,
notifications,
t,
showErrorModal,
]);
const onSaveHandler = useCallback(
async (value: ChannelType) => {
if (!selectedConfig.name) {
@@ -424,6 +451,7 @@ function CreateAlertChannels({
[ChannelType.Opsgenie]: onOpsgenieHandler,
[ChannelType.MsTeams]: onMsTeamsHandler,
[ChannelType.Email]: onEmailHandler,
[ChannelType.GoogleChat]: onGoogleChatHandler,
};
if (isChannelType(value)) {
@@ -455,6 +483,7 @@ function CreateAlertChannels({
onOpsgenieHandler,
onMsTeamsHandler,
onEmailHandler,
onGoogleChatHandler,
notifications,
t,
],
@@ -492,6 +521,13 @@ function CreateAlertChannels({
request = prepareEmailRequest();
await testEmail(request);
break;
case ChannelType.GoogleChat:
if (!validateGoogleChatConfig()) {
setTestingState(false);
return;
}
await testChannel({ data: prepareGoogleChatRequest(selectedConfig) });
break;
default:
notifications.error({
message: 'Error',
@@ -513,7 +549,11 @@ function CreateAlertChannels({
status: 'Test success',
});
} catch (error) {
showErrorModal(error as APIError);
showErrorModal(
error instanceof APIError
? error
: toAPIError(error as ErrorType<RenderErrorResponseDTO>),
);
logEvent('Alert Channel: Test notification', {
type: channelType,
@@ -535,6 +575,8 @@ function CreateAlertChannels({
prepareSlackRequest,
prepareMsTeamsRequest,
prepareEmailRequest,
validateGoogleChatConfig,
testChannel,
notifications,
],
);
@@ -562,9 +604,6 @@ function CreateAlertChannels({
initialValue: {
type,
...selectedConfig,
...PagerInitialConfig,
...OpsgenieInitialConfig,
...EmailInitialConfig,
},
}}
/>

View File

@@ -1,4 +1,39 @@
import { ChannelType } from './config';
import {
AlertmanagertypesPostableChannelDTO,
ConfigSecretURLDTO,
} from 'api/generated/services/sigNoz.schemas';
import { ChannelType, GoogleChatChannel } from './config';
export const isChannelType = (type: string): type is ChannelType =>
Object.values(ChannelType).includes(type as ChannelType);
const GOOGLE_CHAT_WEBHOOK_HOST = 'chat.googleapis.com';
// the backend enforces the same two rules, this is only for a nicer error experience
export const isValidGoogleChatWebhookURL = (url: string): boolean => {
try {
const { protocol, hostname } = new URL(url);
return (
protocol === 'https:' && hostname.toLowerCase() === GOOGLE_CHAT_WEBHOOK_HOST
);
} catch {
return false;
}
};
// create, update and test all send the same body shape
export const prepareGoogleChatRequest = (
config: Partial<GoogleChatChannel>,
): AlertmanagertypesPostableChannelDTO => ({
name: config.name || '',
googlechat_configs: [
{
// the generated type models go's config.SecretURL as an object, the api takes a string
webhook_url: (config.webhook_url || '') as unknown as ConfigSecretURLDTO,
title: config.title || '',
text: config.text || '',
send_resolved: config.send_resolved || false,
},
],
});

View File

@@ -14,10 +14,17 @@ import testPagerApi from 'api/channels/testPager';
import testSlackApi from 'api/channels/testSlack';
import testWebhookApi from 'api/channels/testWebhook';
import logEvent from 'api/common/logEvent';
import {
useTestChannel,
useUpdateChannelByID,
} from 'api/generated/services/channels';
import { RenderErrorResponseDTO } from 'api/generated/services/sigNoz.schemas';
import { ErrorType } from 'api/generatedAPIInstance';
import ROUTES from 'constants/routes';
import {
ChannelType,
EmailChannel,
GoogleChatChannel,
MsTeamsChannel,
OpsgenieChannel,
PagerChannel,
@@ -25,10 +32,15 @@ import {
ValidatePagerChannel,
WebhookChannel,
} from 'container/CreateAlertChannels/config';
import {
isValidGoogleChatWebhookURL,
prepareGoogleChatRequest,
} from 'container/CreateAlertChannels/utils';
import FormAlertChannels from 'container/FormAlertChannels';
import { useNotifications } from 'hooks/useNotifications';
import history from 'lib/history';
import APIError from 'types/api/error';
import { toAPIError } from 'utils/errorUtils';
function EditAlertChannels({
initialValue,
@@ -45,7 +57,8 @@ function EditAlertChannels({
PagerChannel &
MsTeamsChannel &
OpsgenieChannel &
EmailChannel
EmailChannel &
GoogleChatChannel
>
>({
...initialValue,
@@ -54,6 +67,26 @@ function EditAlertChannels({
const [testingState, setTestingState] = useState<boolean>(false);
const { notifications } = useNotifications();
const { mutateAsync: updateChannel } = useUpdateChannelByID();
const { mutateAsync: testChannel } = useTestChannel();
const notifyError = useCallback(
(error: unknown): APIError => {
const apiError =
error instanceof APIError
? error
: toAPIError(error as ErrorType<RenderErrorResponseDTO>);
notifications.error({
message: apiError.getErrorCode(),
description: apiError.getErrorMessage(),
});
return apiError;
},
[notifications],
);
const [type, setType] = useState<ChannelType>(
initialValue?.type ? (initialValue.type as ChannelType) : ChannelType.Slack,
);
@@ -364,6 +397,61 @@ function EditAlertChannels({
}
}, [prepareMsTeamsRequest, t, notifications, selectedConfig]);
const validateGoogleChatConfig = useCallback((): string => {
if (!selectedConfig?.webhook_url) {
return t('webhook_url_required');
}
if (!isValidGoogleChatWebhookURL(selectedConfig.webhook_url)) {
return t('google_chat_webhook_url_invalid');
}
return '';
}, [selectedConfig, t]);
const onGoogleChatEditHandler = useCallback(async () => {
const validationError = validateGoogleChatConfig();
if (validationError !== '') {
notifications.error({
message: 'Error',
description: validationError,
});
return { status: 'failed', statusMessage: validationError };
}
setSavingState(true);
try {
await updateChannel({
pathParams: { id },
data: prepareGoogleChatRequest(selectedConfig),
});
notifications.success({
message: 'Success',
description: t('channel_edit_done'),
});
history.replace(ROUTES.ALL_CHANNELS);
return { status: 'success', statusMessage: t('channel_edit_done') };
} catch (error) {
const apiError = notifyError(error);
return {
status: 'failed',
statusMessage: apiError.getErrorMessage() || t('channel_edit_failed'),
};
} finally {
setSavingState(false);
}
}, [
validateGoogleChatConfig,
updateChannel,
id,
selectedConfig,
notifications,
notifyError,
t,
]);
const onSaveHandler = useCallback(
async (value: ChannelType) => {
let result;
@@ -379,6 +467,8 @@ function EditAlertChannels({
result = await onOpsgenieEditHandler();
} else if (value === ChannelType.Email) {
result = await onEmailEditHandler();
} else if (value === ChannelType.GoogleChat) {
result = await onGoogleChatEditHandler();
}
logEvent('Alert Channel: Save channel', {
type: value,
@@ -397,6 +487,7 @@ function EditAlertChannels({
onMsTeamsEditHandler,
onOpsgenieEditHandler,
onEmailEditHandler,
onGoogleChatEditHandler,
],
);
@@ -438,6 +529,19 @@ function EditAlertChannels({
await testEmail(request);
}
break;
case ChannelType.GoogleChat: {
const validationError = validateGoogleChatConfig();
if (validationError !== '') {
notifications.error({
message: 'Error',
description: validationError,
});
setTestingState(false);
return;
}
await testChannel({ data: prepareGoogleChatRequest(selectedConfig) });
break;
}
default:
notifications.error({
message: 'Error',
@@ -459,10 +563,7 @@ function EditAlertChannels({
status: 'Test success',
});
} catch (error) {
notifications.error({
message: (error as APIError).getErrorCode(),
description: (error as APIError).getErrorMessage(),
});
notifyError(error);
logEvent('Alert Channel: Test notification', {
type: channelType,
sendResolvedAlert: selectedConfig?.send_resolved,
@@ -476,6 +577,9 @@ function EditAlertChannels({
// eslint-disable-next-line react-hooks/exhaustive-deps
[
t,
notifyError,
validateGoogleChatConfig,
testChannel,
prepareWebhookRequest,
preparePagerRequest,
prepareSlackRequest,

View File

@@ -0,0 +1,82 @@
import { Dispatch, SetStateAction } from 'react';
import { useTranslation } from 'react-i18next';
import { Form, Input } from 'antd';
import { MarkdownRenderer } from 'components/MarkdownRenderer/MarkdownRenderer';
import { GoogleChatChannel } from '../../CreateAlertChannels/config';
import { isValidGoogleChatWebhookURL } from '../../CreateAlertChannels/utils';
function GoogleChat({ setSelectedConfig }: GoogleChatProps): JSX.Element {
const { t } = useTranslation('channels');
return (
<>
<Form.Item
name="webhook_url"
label={t('field_webhook_url')}
required
rules={[
{
validator: (_, value: string): Promise<void> =>
!value || isValidGoogleChatWebhookURL(value)
? Promise.resolve()
: Promise.reject(new Error(t('google_chat_webhook_url_invalid'))),
},
]}
tooltip={{
title: (
<MarkdownRenderer
markdownContent={t('tooltip_google_chat_url')}
variables={{}}
/>
),
overlayInnerStyle: { maxWidth: 400 },
placement: 'right',
}}
>
<Input
onChange={(event): void => {
setSelectedConfig((value) => ({
...value,
webhook_url: event.target.value,
}));
}}
data-testid="webhook-url-textbox"
/>
</Form.Item>
<Form.Item name="title" label={t('field_slack_title')}>
<Input.TextArea
rows={4}
onChange={(event): void =>
setSelectedConfig((value) => ({
...value,
title: event.target.value,
}))
}
data-testid="title-textarea"
/>
</Form.Item>
<Form.Item name="text" label={t('field_slack_description')}>
<Input.TextArea
rows={4}
onChange={(event): void =>
setSelectedConfig((value) => ({
...value,
text: event.target.value,
}))
}
data-testid="description-textarea"
placeholder={t('placeholder_slack_description')}
/>
</Form.Item>
</>
);
}
interface GoogleChatProps {
setSelectedConfig: Dispatch<SetStateAction<Partial<GoogleChatChannel>>>;
}
export default GoogleChat;

View File

@@ -9,6 +9,7 @@ import ROUTES from 'constants/routes';
import {
ChannelType,
EmailChannel,
GoogleChatChannel,
OpsgenieChannel,
PagerChannel,
SlackChannel,
@@ -17,6 +18,7 @@ import {
import history from 'lib/history';
import EmailSettings from './Settings/Email';
import GoogleChatSettings from './Settings/GoogleChat';
import MsTeamsSettings from './Settings/MsTeams';
import OpsgenieSettings from './Settings/Opsgenie';
import PagerSettings from './Settings/Pager';
@@ -49,6 +51,8 @@ function FormAlertChannels({
return <PagerSettings setSelectedConfig={setSelectedConfig} />;
case ChannelType.MsTeams:
return <MsTeamsSettings setSelectedConfig={setSelectedConfig} />;
case ChannelType.GoogleChat:
return <GoogleChatSettings setSelectedConfig={setSelectedConfig} />;
case ChannelType.Opsgenie:
return <OpsgenieSettings setSelectedConfig={setSelectedConfig} />;
case ChannelType.Email:
@@ -129,6 +133,14 @@ function FormAlertChannels({
<Select.Option value="msteams" key="msteams" data-testid="select-option">
Microsoft Teams
</Select.Option>
<Select.Option
value="googlechat"
key="googlechat"
data-testid="select-option"
>
Google Chat
</Select.Option>
</Select>
</Form.Item>
@@ -176,7 +188,8 @@ interface FormAlertChannelsProps {
WebhookChannel &
PagerChannel &
OpsgenieChannel &
EmailChannel
EmailChannel &
GoogleChatChannel
>
>
>;

View File

@@ -27,9 +27,13 @@ export const slackTitleDefaultValue = `[{{ .Status | toUpper }}{{ if eq .Status
export const slackDescriptionDefaultValue = `{{ range .Alerts -}} *Alert:* {{ .Labels.alertname }}{{ if .Labels.severity }} - {{ .Labels.severity }}{{ end }} *Summary:* {{ .Annotations.summary }} *Description:* {{ .Annotations.description }} *RelatedLogs:* {{ if gt (len .Annotations.related_logs) 0 -}} View in <{{ .Annotations.related_logs }}|logs explorer> {{- end}} *RelatedTraces:* {{ if gt (len .Annotations.related_traces) 0 -}} View in <{{ .Annotations.related_traces }}|traces explorer> {{- end}} *Details:* {{ range .Labels.SortedPairs }} • *{{ .Name }}:* {{ .Value }} {{ end }} {{ end }}`;
export const googleChatTitleDefaultValue = `[{{ .Status | toUpper }}{{ if eq .Status "firing" }}:{{ .Alerts.Firing | len }}{{ end }}] {{ .CommonLabels.alertname }}`;
export const googleChatDescriptionDefaultValue = `{{ range .Alerts -}} **Alert:** {{ .Labels.alertname }}{{ if .Labels.severity }} ({{ .Labels.severity }}){{ end }}{{ if .Annotations.summary }} **Summary:** {{ .Annotations.summary }}{{ end }}{{ if .Annotations.description }} **Description:** {{ .Annotations.description }}{{ end }} {{ end }}`;
export const editSlackDescriptionDefaultValue = `{{ range .Alerts -}} *Alert:* {{ .Labels.alertname }}{{ if .Labels.severity }} - {{ .Labels.severity }}{{ end }} dummy_summary *Summary:* {{ .Annotations.summary }} *Description:* {{ .Annotations.description }} *Details:* {{ range .Labels.SortedPairs }} • *{{ .Name }}:* {{ .Value }} {{ end }} {{ end }}`;
export const pagerDutyDescriptionDefaultVaule = `{{ if gt (len .Alerts.Firing) 0 -}} Alerts Firing: {{ range .Alerts.Firing }} - Message: {{ .Annotations.description }} Labels: {{ range .Labels.SortedPairs }} - {{ .Name }} = {{ .Value }} {{ end }} Annotations: {{ range .Annotations.SortedPairs }} - {{ .Name }} = {{ .Value }} {{ end }} Source: {{ .GeneratorURL }} {{ end }} {{- end }} {{ if gt (len .Alerts.Resolved) 0 -}} Alerts Resolved: {{ range .Alerts.Resolved }} - Message: {{ .Annotations.description }} Labels: {{ range .Labels.SortedPairs }} - {{ .Name }} = {{ .Value }} {{ end }} Annotations: {{ range .Annotations.SortedPairs }} - {{ .Name }} = {{ .Value }} {{ end }} Source: {{ .GeneratorURL }} {{ end }} {{- end }}`;
export const pagerDutyDescriptionDefaultValue = `[{{ .Status | toUpper }}{{ if eq .Status "firing" }}:{{ .Alerts.Firing | len }}{{ end }}] {{ .CommonLabels.alertname }} for {{ .CommonLabels.job }} {{- if gt (len .CommonLabels) (len .GroupLabels) -}} {{" "}}( {{- with .CommonLabels.Remove .GroupLabels.Names }} {{- range $index, $label := .SortedPairs -}} {{ if $index }}, {{ end }} {{- $label.Name }}="{{ $label.Value -}}" {{- end }} {{- end -}} ) {{- end }}`;
export const pagerDutyAdditionalDetailsDefaultValue = JSON.stringify({
firing: `{{ .Alerts.Firing | toJson }}`,

View File

@@ -10,6 +10,7 @@ import Spinner from 'components/Spinner';
import ROUTES from 'constants/routes';
import {
ChannelType,
GoogleChatChannel,
MsTeamsChannel,
PagerChannel,
SlackChannel,
@@ -59,11 +60,20 @@ function ChannelsEdit(): JSX.Element {
const prepChannelConfig = (): {
type: string;
channel: SlackChannel & WebhookChannel & PagerChannel & MsTeamsChannel;
channel: SlackChannel &
WebhookChannel &
PagerChannel &
MsTeamsChannel &
GoogleChatChannel;
} => {
let channel: SlackChannel & WebhookChannel & PagerChannel & MsTeamsChannel = {
let channel: SlackChannel &
WebhookChannel &
PagerChannel &
MsTeamsChannel &
GoogleChatChannel = {
name: '',
};
if (value && 'slack_configs' in value) {
const slackConfig = value.slack_configs[0];
channel = slackConfig;
@@ -81,6 +91,16 @@ function ChannelsEdit(): JSX.Element {
channel,
};
}
if (value && 'googlechat_configs' in value) {
const [googleChatConfig] = value.googlechat_configs;
channel = googleChatConfig;
return {
type: ChannelType.GoogleChat,
channel,
};
}
if (value && 'pagerduty_configs' in value) {
const pagerConfig = value.pagerduty_configs[0];
channel = pagerConfig;

View File

@@ -1,79 +0,0 @@
package querybuilder
import (
"strings"
"github.com/SigNoz/signoz/pkg/errors"
grammar "github.com/SigNoz/signoz/pkg/parser/filterquery/grammar"
qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
"github.com/antlr4-go/antlr/v4"
)
// ExprKeys returns the field keys referenced in *key positions* of a filter
// expression. Unlike QueryStringToKeysSelectors (which scans raw KEY tokens and so
// also picks up unquoted values — in `x > $threshold` it reports `$threshold`), this
// walks the parse tree and collects only KeyContext nodes.
func ExprKeys(query string) []*telemetrytypes.TelemetryFieldKey {
var keys []*telemetrytypes.TelemetryFieldKey
var walk func(node antlr.Tree)
walk = func(node antlr.Tree) {
if kc, ok := node.(*grammar.KeyContext); ok {
key := telemetrytypes.GetFieldKeyFromKeyText(kc.GetText())
keys = append(keys, &key)
return
}
for i := 0; i < node.GetChildCount(); i++ {
walk(node.GetChild(i))
}
}
// syntax errors are ignored here; downstream re-parsing surfaces them
tree, _ := parseFilterQuery(query)
walk(tree)
return keys
}
// ValidateVariablesInExpr checks the variable references in an expression's value
// positions upfront, so a broken reference fails with a targeted error instead of
// the where-clause visitor's combined "Found N errors" (whose details ride in the
// error's additionals). Lookup mirrors the visitor: verbatim, then with a leading
// `$` stripped. A `$`-prefixed token that resolves to nothing is an error — it can
// never be a valid literal; a bare token that resolves to nothing is left to mean
// itself.
func ValidateVariablesInExpr(query string, variables map[string]qbtypes.VariableItem) error {
var err error
var walk func(node antlr.Tree)
walk = func(node antlr.Tree) {
if err != nil {
return
}
if vc, ok := node.(*grammar.ValueContext); ok {
// only unquoted textual values can be variable references
if vc.KEY() == nil {
return
}
text := vc.GetText()
item, ok := variables[text]
if !ok {
item, ok = variables[strings.TrimPrefix(text, "$")]
}
if !ok {
if strings.HasPrefix(text, "$") {
err = errors.NewInvalidInputf(errors.CodeInvalidInput, "unknown variable %q", text)
}
return
}
if values, isList := item.Value.([]any); isList && len(values) == 0 {
err = errors.NewInvalidInputf(errors.CodeInvalidInput,
"variable %q used in expression has an empty list value", strings.TrimPrefix(text, "$"))
}
return
}
for i := 0; i < node.GetChildCount(); i++ {
walk(node.GetChild(i))
}
}
tree, _ := parseFilterQuery(query)
walk(tree)
return err
}

View File

@@ -1,40 +0,0 @@
package querybuilder
import (
"testing"
qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
"github.com/stretchr/testify/require"
)
func TestExprKeys(t *testing.T) {
names := func(expr string) []string {
var out []string
for _, k := range ExprKeys(expr) {
out = append(out, k.Name)
}
return out
}
// value-position tokens are not keys, unlike QueryStringToKeysSelectors
require.Equal(t, []string{"output_tokens"}, names("output_tokens > $threshold"))
// the trace. prefix parses into FieldContext, leaving the bare name
require.Equal(t, []string{"output_tokens"}, names("trace.output_tokens > 1000"))
require.Equal(t, []string{"a", "b"}, names("a > 1 AND b IN ('x', 'y')"))
}
func TestValidateVariablesInExpr(t *testing.T) {
vars := map[string]qbtypes.VariableItem{
"threshold": {Type: qbtypes.TextBoxVariableType, Value: float64(1000)},
"empty": {Type: qbtypes.QueryVariableType, Value: []any{}},
"all": {Type: qbtypes.DynamicVariableType, Value: "__all__"},
}
require.NoError(t, ValidateVariablesInExpr("x > $threshold", vars))
require.NoError(t, ValidateVariablesInExpr("x > threshold", vars))
require.NoError(t, ValidateVariablesInExpr("x IN $all", vars))
require.NoError(t, ValidateVariablesInExpr("m = 'cost$usd'", vars)) // quoted literals are not references
require.NoError(t, ValidateVariablesInExpr("x > bare_word", vars)) // bare non-variable means itself
require.ErrorContains(t, ValidateVariablesInExpr("x > $bogus", vars), `unknown variable "$bogus"`)
require.ErrorContains(t, ValidateVariablesInExpr("x IN $empty", vars), "empty list")
}

View File

@@ -74,8 +74,5 @@ func Scope() scopedtraces.TraceScope {
FieldKeys: gateKeys,
Columns: columns,
DefaultOrderAlias: "last_activity_time",
// a tool/agent-only window slice contributes NULL tokens but would still count
// as a trace, making count(trace.trace_id) and avg(trace.output_tokens) disagree
ActivityGateAlias: "llm_call_count",
}
}

View File

@@ -993,8 +993,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.` and `tracefield.` are equivalent spellings of a trace-level aggregate:
// both resolve through the same synthetic key.
// `trace.` marks a trace-level aggregate; `tracefield.` routes trace-level too but is
// not a rewritable alias, so the HAVING rewriter rejects it.
func TestBuild_TraceList_TraceContextPrefix(t *testing.T) {
b := newTestBuilder(t)
build := func(q qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]) (*qbtypes.Statement, error) {
@@ -1002,19 +1002,19 @@ func TestBuild_TraceList_TraceContextPrefix(t *testing.T) {
return b.Build(context.Background(), valuer.UUID{}, testStartMs, testEndMs, qbtypes.RequestTypeTrace, q, nil)
}
viaTrace, err := build(qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
_, err := build(qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Filter: &qbtypes.Filter{Expression: "trace.output_tokens > 1000"}})
require.NoError(t, err)
viaTracefield, err := build(qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
_, err = build(qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Filter: &qbtypes.Filter{Expression: "tracefield.output_tokens > 1000"}})
require.NoError(t, err)
assert.Equal(t, viaTrace.Query, viaTracefield.Query)
require.Error(t, err)
assert.Contains(t, err.Error(), "Invalid references in `Having` expression: [tracefield.output_tokens]")
viaHaving, err := build(qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
_, err = build(qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Having: &qbtypes.Having{Expression: "tracefield.output_tokens > 1000"}})
require.NoError(t, err)
assert.Equal(t, viaTrace.Query, viaHaving.Query)
require.Error(t, err)
assert.Contains(t, err.Error(), "Invalid references in `Having` expression: [tracefield.output_tokens]")
_, err = build(qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Filter: &qbtypes.Filter{Expression: "trace.span_count > 3"}})
@@ -1022,9 +1022,7 @@ func TestBuild_TraceList_TraceContextPrefix(t *testing.T) {
assert.Contains(t, err.Error(), "cannot be used")
}
// Query variables in a trace-level condition resolve through the standard filter
// pipeline, exactly like span-level filters: bound args, list/IN handling, dynamic
// __all__ dropping the condition.
// Query variables in a trace-level condition are substituted into the HAVING.
func TestBuild_TraceList_VariableInAggregateFilter(t *testing.T) {
b := newTestBuilder(t)
build := func(expr string, vars map[string]qbtypes.VariableItem) (*qbtypes.Statement, error) {
@@ -1036,19 +1034,17 @@ func TestBuild_TraceList_VariableInAggregateFilter(t *testing.T) {
}, vars)
}
// scalar variable -> replaced to a literal (canonical pkg/variables semantics),
// then parsed and bound as an arg by the filter pipeline
// scalar variable -> literal in HAVING
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 > ?")
assert.Contains(t, stmt.Args, float64(700))
assert.Contains(t, stmt.Query, "HAVING output_tokens > 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",

View File

@@ -1,255 +0,0 @@
package aistatementbuilder
import (
"context"
"strings"
"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"
)
// Scalar / time-series (trace-level aggregation) Build tests. The
// rewriteTraceAggregation unit tests live in scopedtracesstatementbuilder; these
// exercise the full builder through the gen_ai scope.
// Mixing domains across separate aggregations of 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")
}
// A trace-level filter over an output-only aggregate is rejected on the
// aggregation paths too (it is not computable in the mask-pruned scan).
func TestBuild_Aggregation_OutputOnlyFilterRejected(t *testing.T) {
b := newTestBuilder(t)
for _, rt := range []qbtypes.RequestType{qbtypes.RequestTypeScalar, qbtypes.RequestTypeRaw} {
_, err := b.Build(context.Background(), valuer.UUID{}, testStartMs, testEndMs, rt,
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 per-trace columns are rejected as group-by / order keys with a
// targeted error (not the field mapper's generic "field not found").
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`)
_, err = b.Build(ctx, valuer.UUID{}, testStartMs, testEndMs, qbtypes.RequestTypeRaw,
qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Signal: telemetrytypes.SignalTraces,
Order: []qbtypes.OrderBy{{Key: qbtypes.OrderByKey{TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{Name: "trace.output_tokens"}}}},
}, nil)
require.ErrorContains(t, err, `ordering the span list by trace-level aggregate "trace.output_tokens" is not supported`)
_, 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)"}},
Order: []qbtypes.OrderBy{{Key: qbtypes.OrderByKey{TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{Name: "trace.total_tokens"}}}},
}, nil)
require.ErrorContains(t, err, `ordering by trace-level aggregate "trace.total_tokens" is not supported`)
// ordering by the aggregation itself (expression or alias) stays valid
_, 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)
}
// Query variables resolve inside trace-level filter conditions on every request type,
// as bound args via the standard filter pipeline; unknown $vars fail with a variable
// error, not an "unknown aggregate" one.
func TestBuild_Aggregation_VariablesInTraceFilter(t *testing.T) {
b := newTestBuilder(t)
ctx := context.Background()
vars := map[string]qbtypes.VariableItem{
"threshold": {Type: qbtypes.TextBoxVariableType, Value: float64(1000)},
}
for _, rt := range []qbtypes.RequestType{qbtypes.RequestTypeScalar, qbtypes.RequestTypeRaw, qbtypes.RequestTypeTrace} {
q := qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Signal: telemetrytypes.SignalTraces,
Filter: &qbtypes.Filter{Expression: "trace.output_tokens > $threshold"},
}
if rt == qbtypes.RequestTypeScalar {
q.Aggregations = []qbtypes.TraceAggregation{{Expression: "avg(trace.output_tokens)"}}
}
stmt, err := b.Build(ctx, valuer.UUID{}, testStartMs, testEndMs, rt, q, vars)
require.NoError(t, err, rt.StringValue())
assert.Contains(t, stmt.Query, "HAVING output_tokens > ?", rt.StringValue())
assert.Contains(t, stmt.Args, float64(1000), rt.StringValue())
_, err = b.Build(ctx, valuer.UUID{}, testStartMs, testEndMs, rt, q, nil)
require.ErrorContains(t, err, `unknown variable "$threshold"`, rt.StringValue())
}
// a dynamic variable resolved to __all__ skips the trace-level condition, exactly
// like span filters — no qualification CTE is built
stmt, 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)"}},
Filter: &qbtypes.Filter{Expression: "trace.output_tokens IN $models"},
}, map[string]qbtypes.VariableItem{
"models": {Type: qbtypes.DynamicVariableType, Value: "__all__"},
})
require.NoError(t, err)
assert.NotContains(t, stmt.Query, "__qualified")
// list variables render as IN with bound args
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)
assert.Contains(t, stmt.Query, "HAVING llm_call_count IN (?, ?)")
}
// A resource-attribute condition prunes the qualification scan the same way it prunes
// the trace list's matched pass: __qualified references the __resource_filter CTE, the
// delegated __trace_scope inlines the fingerprint subquery.
func TestBuild_Aggregation_QualificationResourcePruned(t *testing.T) {
b := newTestBuilder(t)
ctx := context.Background()
filter := &qbtypes.Filter{Expression: "service.name = 'api' AND trace.output_tokens > 1000"}
stmt, 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)"}},
Filter: filter,
}, nil)
require.NoError(t, err)
qualified := stmt.Query[strings.Index(stmt.Query, "__qualified"):strings.Index(stmt.Query, "__ai_traces")]
assert.Contains(t, qualified, "resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter)")
stmt, err = b.Build(ctx, valuer.UUID{}, testStartMs, testEndMs, qbtypes.RequestTypeRaw,
qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Signal: telemetrytypes.SignalTraces,
Filter: filter,
Limit: 10,
}, nil)
require.NoError(t, err)
scope := stmt.Query[strings.Index(stmt.Query, "__trace_scope"):]
assert.Contains(t, scope, "resource_fingerprint GLOBAL IN (SELECT fingerprint FROM (SELECT")
}
// ---------------------------------------------------------------------------
// Full-query goldens — native trace-domain pipeline
// ---------------------------------------------------------------------------
// Scalar over per-trace values, no filter: one window-clipped per-trace scan, outer
// avg across traces.
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 __ai_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
HAVING (countIf(mapContains(attributes_string, 'gen_ai.request.model'))) > 0
)
SELECT avg(output_tokens) AS __result_0
FROM __ai_traces
ORDER BY __result_0 DESC
SETTINGS distributed_product_mode='allow', max_memory_usage=10000000000
`, stmt)
}
// Time series over per-trace values: the per-trace scan buckets by span time
// (per-bucket clipping), the outer aggregation is 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 __ai_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
HAVING (countIf(mapContains(attributes_string, 'gen_ai.request.model'))) > 0
)
SELECT ts, avg(output_tokens) AS __result_0
FROM __ai_traces
GROUP BY ts
SETTINGS distributed_product_mode='allow', max_memory_usage=10000000000
`, stmt)
}
// Span-level scalar with a trace-level filter: delegated to the trace builder with
// the gate ANDed, constrained by the __trace_scope qualification.
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)
got := renderSQL(t, stmt)
assert.Contains(t, got, "__trace_scope AS (")
assert.Contains(t, got, "HAVING output_tokens > 1000")
assert.Contains(t, got, "trace_id GLOBAL IN (SELECT trace_id FROM __trace_scope)")
assert.Contains(t, got, "AS __result_0")
}

View File

@@ -1,112 +0,0 @@
package aistatementbuilder
import (
"context"
"testing"
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"
)
// Span list with a mixed filter: gen_ai spans matching the span-level part, in
// traces whose window-clipped aggregates satisfy the trace-level part (the
// __trace_scope qualification on the delegated path).
func TestBuild_SpanList_TraceScoped(t *testing.T) {
b := newTestBuilder(t)
stmt, err := b.Build(context.Background(), valuer.UUID{}, testStartMs, testEndMs, qbtypes.RequestTypeRaw,
qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Signal: telemetrytypes.SignalTraces,
Filter: &qbtypes.Filter{Expression: "gen_ai.request.model = 'gpt-4o-mini' AND trace.output_tokens > 1000"},
Limit: 10,
}, nil)
require.NoError(t, err)
got := renderSQL(t, stmt)
assert.Contains(t, got, "__trace_scope AS (")
assert.Contains(t, got, "HAVING output_tokens > 1000")
assert.Contains(t, got, "trace_id GLOBAL IN (SELECT trace_id FROM __trace_scope)")
assert.Contains(t, got, "gpt-4o-mini")
}
// Without a trace-level condition nothing changes: the span list stays a single
// gated span scan (no __trace_scope CTE).
func TestBuild_SpanList_NoTraceFilter_NoScope(t *testing.T) {
b := newTestBuilder(t)
stmt, err := b.Build(context.Background(), valuer.UUID{}, testStartMs, testEndMs, qbtypes.RequestTypeRaw,
qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Signal: telemetrytypes.SignalTraces,
Filter: &qbtypes.Filter{Expression: "gen_ai.request.model = 'gpt-4o-mini'"},
Limit: 10,
}, nil)
require.NoError(t, err)
assert.NotContains(t, stmt.Query, "__trace_scope")
}
// The span-list trace-level filter shares the trace list's rules: output-only
// aggregates are rejected, OR-mixing the two classes is rejected, and explicitly
// trace-level order keys get a targeted error — while bare span columns that happen
// to share a name with an aggregate alias (duration_nano) stay orderable.
func TestBuild_SpanList_TraceFilter_Validation(t *testing.T) {
b := newTestBuilder(t)
build := func(q qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]) error {
q.Signal = telemetrytypes.SignalTraces
_, err := b.Build(context.Background(), valuer.UUID{}, testStartMs, testEndMs, qbtypes.RequestTypeRaw, q, nil)
return err
}
err := build(qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Filter: &qbtypes.Filter{Expression: "trace.span_count > 3"},
})
require.ErrorContains(t, err, `aggregate "span_count" cannot be used`)
err = build(qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Filter: &qbtypes.Filter{Expression: "trace.output_tokens > 1000 OR kind_string = 'Client'"},
})
require.ErrorContains(t, err, "cannot be combined")
err = build(qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Order: []qbtypes.OrderBy{{Key: qbtypes.OrderByKey{TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{Name: "trace.output_tokens"}}}},
})
require.ErrorContains(t, err, `ordering the span list by trace-level aggregate "trace.output_tokens" is not supported`)
err = build(qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Order: []qbtypes.OrderBy{{Key: qbtypes.OrderByKey{TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{Name: "duration_nano"}}, Direction: qbtypes.OrderDirectionDesc}},
Limit: 10,
})
require.NoError(t, err, "bare duration_nano is a span column, not a trace-level key")
}
// Variables in a trace-level condition on the span list get the trace list's
// treatment: resolved and bound as args, __all__ drops the condition (no scope CTE),
// tracefield. spelling behaves like trace..
func TestBuild_SpanList_TraceFilter_Variables(t *testing.T) {
b := newTestBuilder(t)
build := func(expr string, vars map[string]qbtypes.VariableItem) (*qbtypes.Statement, error) {
return b.Build(context.Background(), valuer.UUID{}, testStartMs, testEndMs, qbtypes.RequestTypeRaw,
qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
Signal: telemetrytypes.SignalTraces,
Filter: &qbtypes.Filter{Expression: expr},
Limit: 10,
}, vars)
}
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 > ?")
assert.Contains(t, stmt.Args, float64(700))
stmt, err = build("trace.output_tokens > $threshold",
map[string]qbtypes.VariableItem{"threshold": {Type: qbtypes.DynamicVariableType, Value: "__all__"}})
require.NoError(t, err)
assert.NotContains(t, stmt.Query, "__trace_scope")
viaTrace, err := build("trace.output_tokens > 1000", nil)
require.NoError(t, err)
viaTracefield, err := build("tracefield.output_tokens > 1000", nil)
require.NoError(t, err)
assert.Equal(t, viaTrace.Query, viaTracefield.Query)
}

View File

@@ -16,9 +16,6 @@ type TraceScope struct {
Columns []TraceColumn
// DefaultOrderAlias is sorted by (desc) when the query gives no order.
DefaultOrderAlias string
// ActivityGateAlias names the column that must be > 0 for a per-trace row to feed
// trace-level (trace.) aggregations; empty disables the gate.
ActivityGateAlias string
}
// TraceColumn is one per-trace output column.

View File

@@ -4,6 +4,7 @@ import (
"context"
"fmt"
"log/slog"
"sort"
"strings"
"github.com/SigNoz/signoz/pkg/errors"
@@ -18,6 +19,7 @@ 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"
)
@@ -112,27 +114,14 @@ func (b *scopedTraceStatementBuilder) Build(
case qbtypes.RequestTypeTrace:
return b.buildTraceListQuery(ctx, orgID, querybuilder.ToNanoSecs(start), querybuilder.ToNanoSecs(end), query, variables)
case qbtypes.RequestTypeRaw:
if err := b.validateGroupByAndOrder(requestType, query); err != nil {
return nil, err
}
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
}
}
// traceScopedStatementBuilder is the delegate's optional capability of constraining a
// query to a set of trace ids (implemented by the traces statement builder).
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 *qbtypes.Statement) (*qbtypes.Statement, error)
}
// buildDelegated splits the user filter, ANDs the base gate into its span-level part,
// and delegates to the standard trace builder. A trace-level part (trace.output_tokens
// > 1000) becomes a window-clipped qualification the delegate constrains trace_id by.
// Serves the span list (raw) and span-level scalar/time-series.
// buildDelegated ANDs the base gate into the user filter and delegates to the
// standard trace builder (the span-list / raw path).
func (b *scopedTraceStatementBuilder) buildDelegated(
ctx context.Context,
orgID valuer.UUID,
@@ -141,42 +130,17 @@ func (b *scopedTraceStatementBuilder) buildDelegated(
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)
if query.Filter != nil && strings.TrimSpace(query.Filter.Expression) != "" {
expr = fmt.Sprintf("(%s) AND (%s)", gate, query.Filter.Expression)
}
// 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, 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)
return b.traceStmtBuilder.Build(ctx, orgID, start, end, requestType, gated, variables)
}
// buildTraceListQuery wires the CTE pipeline (start/end are nanoseconds):
@@ -220,17 +184,19 @@ func (b *scopedTraceStatementBuilder) buildTraceListQuery(
if err != nil {
return nil, err
}
orderableSet := orderableAliasSet(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(), start, end, variables, matchedSB)
fp, err := b.splitFilter(ctx, orgID, query, b.aggregateAliasSet(), orderableSet, start, end, variables, matchedSB)
if err != nil {
return nil, err
}
matchedFrag, matchedArgs, err := b.buildMatchedCTE(matchedSB, start, end, startBucket, endBucket, resolved, orders, maskExpr, fp, resourcePred, limit, query.Offset)
matchedFrag, matchedArgs, err := b.buildMatchedCTE(matchedSB, start, end, startBucket, endBucket, resolved, orders, orderableSet, maskExpr, fp, resourcePred, limit, query.Offset)
if err != nil {
return nil, err
}
@@ -414,28 +380,27 @@ func (b *scopedTraceStatementBuilder) resolveListOrders(order []qbtypes.OrderBy,
return orders, nil
}
// filterParts is the user filter split into a span-level predicate (widens the
// matched WHERE prune and becomes a countIf existence check in HAVING) and the
// resolved trace-level HAVING (nil when there is none).
// filterParts is the user filter split into a span-level predicate and a trace-level
// HAVING expression.
type filterParts struct {
spanPred string
hasSpanFilter bool
having *traceHaving
havingExpr string
warnings []string
warningsURL string
}
// 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.
func (b *scopedTraceStatementBuilder) splitFilter(ctx context.Context, orgID valuer.UUID, query qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation], classifySet 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 (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, orderableSet map[string]struct{}, 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
}
havingExpr = traceExpr
fp.havingExpr = traceExpr
if strings.TrimSpace(spanExpr) != "" {
pred, warnings, url, err := b.resolveSpanPredicate(ctx, orgID, start, end, spanExpr, variables, sb)
if err != nil {
@@ -450,17 +415,23 @@ func (b *scopedTraceStatementBuilder) splitFilter(ctx context.Context, orgID val
}
}
if query.Having != nil && strings.TrimSpace(query.Having.Expression) != "" {
if havingExpr != "" {
havingExpr = fmt.Sprintf("(%s) AND (%s)", havingExpr, query.Having.Expression)
if fp.havingExpr != "" {
fp.havingExpr = fmt.Sprintf("(%s) AND (%s)", fp.havingExpr, query.Having.Expression)
} else {
havingExpr = query.Having.Expression
fp.havingExpr = query.Having.Expression
}
}
having, err := b.resolveTraceHaving(ctx, havingExpr, variables, sb)
if err != nil {
// 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, orderableSet); err != nil {
return fp, err
}
fp.having = having
return fp, nil
}
@@ -502,8 +473,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, maskExpr string, fp filterParts, resourcePred string, limit, offset int) (string, []any, error) {
needed := neededMatchedAliases(orders, fp.having)
func (b *scopedTraceStatementBuilder) buildMatchedCTE(sb *sqlbuilder.SelectBuilder, start, end, startBucket, endBucket uint64, resolved []resolvedColumn, orders []listOrder, orderableSet map[string]struct{}, maskExpr string, fp filterParts, resourcePred string, limit, offset int) (string, []any, error) {
needed := neededMatchedAliases(orders, fp.havingExpr, orderableSet)
selects := []string{"trace_id"}
for _, rc := range resolved {
if _, ok := needed[rc.alias]; !ok {
@@ -540,8 +511,22 @@ func (b *scopedTraceStatementBuilder) buildMatchedCTE(sb *sqlbuilder.SelectBuild
having = append(having, "countIf("+maskExpr+") > 0")
having = append(having, "countIf("+fp.spanPred+") > 0")
}
if fp.having != nil {
having = append(having, fp.having.pred)
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(orderableSet)*2)
for a := range orderableSet {
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 len(having) > 0 {
sb.Having(strings.Join(having, " AND "))
@@ -619,33 +604,47 @@ func orderableAliasSet(resolved []resolvedColumn) map[string]struct{} {
}
// neededMatchedAliases is the minimal alias set the matched pass must select: those
// in ORDER BY plus those the resolved trace-level HAVING touches.
func neededMatchedAliases(orders []listOrder, having *traceHaving) map[string]struct{} {
// in ORDER BY plus those in the aggregate HAVING.
func neededMatchedAliases(orders []listOrder, havingExpr string, orderableSet map[string]struct{}) map[string]struct{} {
needed := make(map[string]struct{})
for _, o := range orders {
needed[o.alias] = struct{}{}
}
if having != nil {
for name := range having.used {
for _, name := range traceAggregateNames(havingExpr) {
if _, ok := orderableSet[name]; ok {
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 not
// computable in the matched pass (e.g. span_count, duration_nano) with a targeted
// top-level error — the same check inside the where-clause visitor would surface only
// as a detail of a combined error. Key positions only: `x > $threshold` references x.
// computable in the matched pass.
func validateAggregateFilter(havingExpr string, orderableSet map[string]struct{}) error {
if strings.TrimSpace(havingExpr) == "" {
return nil
}
for _, key := range querybuilder.ExprKeys(havingExpr) {
name := strings.TrimPrefix(strings.TrimPrefix(key.Name, "tracefield."), "trace.")
allowed := make([]string, 0, len(orderableSet))
for a := range orderableSet {
allowed = append(allowed, a)
}
sort.Strings(allowed)
for _, name := range traceAggregateNames(havingExpr) {
if _, ok := orderableSet[name]; !ok {
return errors.NewInvalidInputf(errors.CodeInvalidInput,
"aggregate %q cannot be used in the trace-list filter; filterable aggregates: %s", name, strings.Join(sortedAliases(orderableSet), ", "))
"aggregate %q cannot be used in the trace-list filter; filterable aggregates: %s", name, strings.Join(allowed, ", "))
}
}
return nil

View File

@@ -1,825 +0,0 @@
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"
)
// This file implements scalar / time-series for builder_ai_query.
//
// Aggregations come in two domains, chosen per expression by the `trace.` prefix:
// - span-level (bare keys): aggregate over individual gen_ai spans. Delegated to the
// standard trace builder with the gate ANDed in; a trace-level filter part becomes
// a __trace_scope qualification (see buildDelegated).
// - trace-level (`trace.` prefix): aggregate over window-clipped per-trace values
// (avg(trace.output_tokens) = average per trace). Runs the native pipeline below.
//
// Native pipeline (buildTraceAggregationQuery):
//
// __qualified traces whose window-clipped aggregates satisfy the trace-level
// │ filter — whole-window values, so a trace qualifies once. Only
// ▼ present when the filter has a trace-level part.
// __ai_traces per-trace values: windowed, mask-pruned GROUP BY trace_id
// │ (+ time bucket for time series → per-bucket clipping, + group-by
// ▼ columns), spans filtered by gate AND span-level filter; rows with
// main no LLM activity are dropped (activity gate). Outer aggregation over
// the per-trace rows → __result_i.
// 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 scalar/time-series requests by aggregation domain.
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.validateGroupByAndOrder(requestType, query); err != nil {
return nil, err
}
if len(traceAggs) == 0 {
return b.buildDelegated(ctx, orgID, start, end, requestType, query, variables)
}
return b.buildTraceAggregationQuery(ctx, orgID, querybuilder.ToNanoSecs(start), querybuilder.ToNanoSecs(end), requestType, query, variables, traceAggs)
}
// classifyAggregations splits the aggregations into span-domain (delegated) vs
// trace-domain (over per-trace values). 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) {
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 the scope's per-trace column set (static) usable in
// trace-level aggregations and filters.
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
}
// validateGroupByAndOrder rejects trace-level (trace.) per-trace columns used as a
// group-by key or an order key with a targeted error, instead of the generic "field
// not found" the field mapper would raise. An order key that names an aggregation
// (alias / expression / index) is exempt — that is the way to order by a trace-level
// aggregation's result.
func (b *scopedTraceStatementBuilder) validateGroupByAndOrder(requestType qbtypes.RequestType, query qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]) error {
aliases := b.aggregateAliasSet()
for _, gb := range query.GroupBy {
if isTraceLevelKey(gb.Name, gb.FieldContext, aliases) {
return errors.NewInvalidInputf(errors.CodeInvalidInput,
"grouping by trace-level aggregate %q is not supported; group by span attributes instead (e.g. gen_ai.request.model, service.name)", gb.Name)
}
}
for _, o := range query.Order {
if _, isAgg := traceAggOrderIndex(o, query); isAgg {
continue
}
if !isTraceLevelKey(o.Key.Name, o.Key.FieldContext, aliases) {
continue
}
if requestType == qbtypes.RequestTypeRaw {
return errors.NewInvalidInputf(errors.CodeInvalidInput,
"ordering the span list by trace-level aggregate %q is not supported; order by span columns instead (e.g. timestamp, duration_nano)", o.Key.Name)
}
return errors.NewInvalidInputf(errors.CodeInvalidInput,
"ordering by trace-level aggregate %q is not supported; order by the aggregation itself (its alias or expression) or a group-by key", o.Key.Name)
}
return nil
}
// isTraceLevelKey reports whether a group-by / order key explicitly names a
// trace-level per-trace aggregate (trace./tracefield. prefix or trace field context).
// Bare names pass through: they may legitimately be span columns that share a name
// with an aggregate alias (duration_nano, timestamp).
func isTraceLevelKey(name string, fieldContext telemetrytypes.FieldContext, aliases map[string]struct{}) bool {
stripped := strings.TrimPrefix(strings.TrimPrefix(name, "tracefield."), "trace.")
if _, ok := aliases[stripped]; !ok {
return false
}
return stripped != name || fieldContext == telemetrytypes.FieldContextTrace
}
// rewriteTraceAggregation parses one aggregation expression. When it references
// trace.-prefixed per-trace columns it returns the expression rewritten to run over
// the per-trace scan (trace.output_tokens → output_tokens, arithmetic between
// trace. columns allowed, function names mapped via AggreFuncMap) with isTrace=true;
// a pure span-level expression returns isTrace=false and is left 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)
}
return &traceAggregation{expr: chparser.Format(sel.SelectItems[0]), used: v.used, isRate: v.isRate}, true, nil
}
// traceAggVisitor walks the aggregation AST, classifying column references and
// rewriting trace.-prefixed ones (bare paths, backquoted identifiers, and either
// nested in arithmetic) to the per-trace column aliases in place. It keeps an
// ancestor stack (Enter/Leave) to tell a column identifier from a path segment,
// function name, or alias, and to reject trace. columns inside *If combinators.
type traceAggVisitor struct {
chparser.DefaultASTVisitor
traceCols map[string]struct{}
used map[string]struct{}
stack []chparser.Expr
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]
}
// enclosingCombinator returns the name of a surrounding *If-combinator function, if any.
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
}
// 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 {
ref := chparser.Format(p)
col, isTrace := traceColumnRef(ref)
if !isTrace {
v.hasSpan = true
return nil
}
if err := v.acceptTraceColumn(ref, 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 a
// trace-level reference (rewritten in place); any other column identifier is
// span-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
}
}
col, isTrace := traceColumnRef(i.Name)
if !isTrace {
v.hasSpan = true
return nil
}
if err := v.acceptTraceColumn(i.Name, col); err != nil {
return err
}
i.Name = col
return nil
}
// acceptTraceColumn validates one trace-level column reference and records it.
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{}{}
}
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
if aggFunc.Rate {
v.isRate = true
}
return nil
}
// traceColumnRef reports whether text is a pure trace.-prefixed column reference
// (trace.output_tokens / tracefield.output_tokens) and returns the bare column name.
func traceColumnRef(text string) (string, bool) {
text = strings.TrimSpace(text)
var rest string
if r, ok := strings.CutPrefix(text, "trace."); ok {
rest = r
} else if r, ok := strings.CutPrefix(text, "tracefield."); ok {
rest = r
} else {
return "", false
}
if rest == "" || strings.ContainsAny(rest, " ()'\"`,+-*/<>=!") {
return "", false
}
return rest, 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 builds the qualification statement — trace ids whose
// window-clipped per-trace aggregates satisfy the trace-level filter — used as the
// delegate's __trace_scope. When the query's filter references resource attributes,
// the scan is pruned to matching resource fingerprints (inlined, since the caller
// embeds this statement standalone). start/end are ns. Returns nil when every
// trace-level condition was dropped by variable resolution.
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, error) {
keys, err := b.fetchKeys(ctx, orgID)
if err != nil {
return nil, err
}
sb := sqlbuilder.NewSelectBuilder()
maskExpr, resolved, err := b.resolveFor(ctx, orgID, start, end, keys, sb)
if err != nil {
return nil, err
}
having, err := b.resolveTraceHaving(ctx, traceExpr, variables, sb)
if err != nil {
return nil, err
}
if having == nil {
return nil, nil
}
var resourcePred string
if stmt, err := b.resourceFilterStmt(ctx, orgID, query, start, end, variables); err != nil {
return nil, err
} else if stmt != nil {
inlined, err := embedExpr(sb, stmt.Query, stmt.Args)
if err != nil {
return nil, err
}
resourcePred = fmt.Sprintf("resource_fingerprint GLOBAL IN (SELECT fingerprint FROM (%s))", inlined)
}
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}, nil
}
// resourceFilterStmt resolves the fingerprint statement for the resource attributes
// in the query's filter; nil when there are none.
func (b *scopedTraceStatementBuilder) resourceFilterStmt(
ctx context.Context,
orgID valuer.UUID,
query qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation],
start, end uint64,
variables map[string]qbtypes.VariableItem,
) (*qbtypes.Statement, error) {
return b.resourceFilterStmtBuilder.Build(ctx, orgID, start, end, qbtypes.RequestTypeRaw, query, variables)
}
// embedExpr inlines a pre-built statement into sb, replacing each `?` placeholder
// with a builder Var so the args are tracked in appearance order. A count mismatch
// would silently shift args into the wrong slots — error out instead.
func embedExpr(sb *sqlbuilder.SelectBuilder, expr string, args []any) (string, error) {
if n := strings.Count(expr, "?"); n != len(args) {
return "", errors.NewInternalf(errors.CodeInternal,
"scoped trace builder: %d placeholders != %d args embedding %q", n, len(args), expr)
}
var out strings.Builder
ai := 0
for i := 0; i < len(expr); i++ {
if expr[i] == '?' {
out.WriteString(sb.Var(args[ai]))
ai++
continue
}
out.WriteByte(expr[i])
}
return out.String(), nil
}
// groupColumn is a resolved span-attribute group-by column (arg-free expression).
type groupColumn struct {
name string
expr string
}
// 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
havingPred string // resolved HAVING predicate over the selected aliases
activityExpr string // aggregate expr that must be > 0 for a row to survive
}
// buildPerTraceScan renders the scan: window + gate mask (+ span filter, resource
// prune, qualification), grouped by trace_id (+ ts bucket, group-by columns).
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.name))
}
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)")
}
sb.Where(where...)
groupBy := []string{"trace_id"}
if o.stepSeconds > 0 {
groupBy = append(groupBy, "ts")
}
for _, gc := range o.groupCols {
groupBy = append(groupBy, "`"+gc.name+"`")
}
sb.GroupBy(groupBy...)
var having []string
if strings.TrimSpace(o.activityExpr) != "" {
having = append(having, "("+o.activityExpr+") > 0")
}
if strings.TrimSpace(o.havingPred) != "" {
having = append(having, o.havingPred)
}
if len(having) > 0 {
sb.Having(strings.Join(having, " AND "))
}
return sb.BuildWithFlavor(sqlbuilder.ClickHouse)
}
// resolveGroupColumns resolves span-attribute group-by keys through the field mapper
// (metadata-aware), for selection inside the per-trace scan.
func (b *scopedTraceStatementBuilder) resolveGroupColumns(ctx context.Context, orgID valuer.UUID, start, end uint64, groupBy []qbtypes.GroupByKey) ([]groupColumn, error) {
if len(groupBy) == 0 {
return nil, nil
}
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,
})
}
keys, _, err := b.metadataStore.GetKeysMulti(ctx, orgID, selectors)
if err != nil {
return nil, err
}
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{name: 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
}
// newScanContext resolves everything a per-trace scan embeds against a fresh builder.
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, 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 builds the native pipeline (see the file comment).
// 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) {
keys, err := b.fetchKeys(ctx, orgID)
if err != nil {
return nil, err
}
var spanExpr, traceExpr string
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
}
}
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)
if err != nil {
return nil, err
}
groupNames := make([]string, 0, len(groupCols))
for _, gc := range groupCols {
groupNames = append(groupNames, "`"+gc.name+"`")
}
needed := make(map[string]struct{})
for _, ta := range traceAggs {
for a := range ta.used {
needed[a] = struct{}{}
}
}
stepSeconds := int64(0)
rateInterval := (end - start) / querybuilder.NsToSeconds
if requestType == qbtypes.RequestTypeTimeSeries {
stepSeconds = int64(query.StepInterval.Seconds())
rateInterval = uint64(stepSeconds)
}
// 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("__ai_traces")
// grouped, limited time series → rank groups on whole-window per-trace values
// (exact for non-composable aggregates) and constrain the main query to the top-N.
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,
activityExpr: activityGate(b.scope, tsc.resolved),
})
cteFragments = append(cteFragments, fmt.Sprintf("__ai_traces_total AS (%s)", totalSQL))
cteArgs = append(cteArgs, totalArgs)
limitSQL, limitArgs := outerLimitSQL(query, traceAggs, groupNames, (end-start)/querybuilder.NsToSeconds)
cteFragments = append(cteFragments, fmt.Sprintf("__limit_cte AS (%s)", limitSQL))
cteArgs = append(cteArgs, limitArgs)
tuple := "(" + strings.Join(groupNames, ", ") + ")"
sb.Where(fmt.Sprintf("%s IN (SELECT %s FROM __limit_cte)", tuple, 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,
// LLM-activity gate: per-trace rows with no in-scope activity in their
// window/bucket slice are dropped, so e.g. count(trace.trace_id) and
// avg(trace.output_tokens) agree on the set of traces they see.
activityExpr: activityGate(b.scope, msc.resolved),
})
cteFragments = append(cteFragments, fmt.Sprintf("__ai_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", orderBy.Key.Name, 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", orderBy.Key.Name, 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
}
// activityGate resolves the scope's activity-gate column to its aggregate expression;
// empty when the scope declares none.
func activityGate(scope TraceScope, resolved []resolvedColumn) string {
if scope.ActivityGateAlias == "" {
return ""
}
for _, rc := range resolved {
if rc.alias == scope.ActivityGateAlias {
return rc.expr
}
}
return ""
}
// rendered returns the outer aggregation SQL, dividing rate aggregations by the
// interval (step for time series, window length for scalar).
func (ta traceAggregation) rendered(rateInterval uint64) string {
if ta.isRate {
return fmt.Sprintf("%s/%d", ta.expr, rateInterval)
}
return ta.expr
}
// outerLimitSQL renders the top-N group selection for a grouped, limited time
// series: the outer aggregations over whole-window per-trace values, ranked and
// limited.
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("__ai_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", orderBy.Key.Name, orderBy.Direction.StringValue()))
}
}
if len(query.Order) == 0 {
sb.OrderBy("__result_0 DESC")
}
sb.Limit(query.Limit)
return sb.BuildWithFlavor(sqlbuilder.ClickHouse)
}
// traceAggOrderIndex reports whether an order key refers to the i-th aggregation
// (by alias, expression, or index), mirroring the trace builder.
func traceAggOrderIndex(k qbtypes.OrderBy, q qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]) (int, bool) {
for i, agg := range q.Aggregations {
if k.Key.Name == agg.Alias ||
k.Key.Name == agg.Expression ||
k.Key.Name == fmt.Sprintf("%d", i) {
return i, true
}
}
return 0, false
}

View File

@@ -1,60 +0,0 @@
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: "tracefield prefix", expr: "sum(tracefield.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"},
{name: "countIf over trace col rejected", expr: "countIf(trace.output_tokens > 1000)", wantErr: "not supported"},
}
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)
assert.Equal(t, tc.isTrace, isTrace)
if !tc.isTrace {
return
}
assert.Equal(t, tc.want, ta.expr)
for _, u := range tc.used {
assert.Contains(t, ta.used, u)
}
assert.Len(t, ta.used, len(tc.used))
})
}
}

View File

@@ -1,154 +0,0 @@
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 column aliases (args bound into the scan's builder) plus the aliases it
// references (so scans select only what the predicate needs).
type traceHaving struct {
pred string
used map[string]struct{}
}
// resolveTraceHaving resolves a trace-level filter expression through the standard
// filter pipeline (PrepareWhereClause) against the per-trace column aliases, so
// operators and bound args behave exactly as in span-level filters. Query variables
// are resolved by the canonical replacement (pkg/variables) first — a dynamic
// variable set to __all__ drops its condition for any operator. Returns nil when the
// expression is empty or every condition was dropped. 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
}
allowed := b.orderableColumnSet()
// upfront targeted errors: the visitor folds condition errors into a combined
// "Found N errors" whose details are not part of the error message
if err := validateAggregateFilter(expr, allowed); err != nil {
return nil, err
}
if err := querybuilder.ValidateVariablesInExpr(expr, variables); err != nil {
return nil, err
}
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
}
}
// every user-facing spelling of an alias resolves to the same synthetic key:
// bare + trace.-prefixed here; tracefield. parses to FieldContextTrace, which
// matches the bare entry's context
fieldKeys := make(map[string][]*telemetrytypes.TelemetryFieldKey, len(allowed)*2)
for alias := range allowed {
key := &telemetrytypes.TelemetryFieldKey{Name: alias, FieldContext: telemetrytypes.FieldContextTrace}
fieldKeys[alias] = []*telemetrytypes.TelemetryFieldKey{key}
fieldKeys["trace."+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
}
return &traceHaving{pred: prepared.Expr, used: cb.used}, nil
}
// aliasConditionBuilder renders filter conditions directly against the per-trace
// column aliases. It records the aliases it touches; a key that resolves to no alias
// is an unknown/unfilterable aggregate.
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 {
name := strings.TrimPrefix(strings.TrimPrefix(key.Name, "tracefield."), "trace.")
return nil, nil, errors.NewInvalidInputf(errors.CodeInvalidInput,
"aggregate %q cannot be used in an AI trace-list filter; filterable aggregates: %s",
name, strings.Join(sortedAliases(c.allowed), ", "))
}
alias := matching[0].Name
c.used[alias] = struct{}{}
col := quoteAlias(alias)
var cond string
switch op {
case qbtypes.FilterOperatorEqual:
cond = sb.E(col, value)
case qbtypes.FilterOperatorNotEqual:
cond = sb.NE(col, value)
case qbtypes.FilterOperatorGreaterThan:
cond = sb.G(col, value)
case qbtypes.FilterOperatorGreaterThanOrEq:
cond = sb.GE(col, value)
case qbtypes.FilterOperatorLessThan:
cond = sb.L(col, value)
case qbtypes.FilterOperatorLessThanOrEq:
cond = sb.LE(col, value)
case qbtypes.FilterOperatorIn, qbtypes.FilterOperatorNotIn:
values, ok := value.([]any)
if !ok {
values = []any{value}
}
if op == qbtypes.FilterOperatorIn {
cond = sb.In(col, values...)
} else {
cond = sb.NotIn(col, values...)
}
case qbtypes.FilterOperatorBetween, qbtypes.FilterOperatorNotBetween:
values, ok := value.([]any)
if !ok || len(values) != 2 {
return nil, nil, errors.NewInvalidInputf(errors.CodeInvalidInput,
"between on trace-level aggregate %q requires exactly two values", alias)
}
if op == qbtypes.FilterOperatorBetween {
cond = sb.Between(col, values[0], values[1])
} else {
cond = sb.NotBetween(col, values[0], values[1])
}
default:
return nil, nil, errors.NewInvalidInputf(errors.CodeInvalidInput,
"trace-level aggregate %q supports only comparison operators (=, !=, <, <=, >, >=, in, between)", alias)
}
return []string{cond}, nil, nil
}

View File

@@ -32,10 +32,6 @@ type traceQueryStatementBuilder struct {
resourceFilterResolver *resourcefilter.ResourceFingerprintResolver[qbtypes.TraceAggregation]
aggExprRewriter qbtypes.AggExprRewriter
skipResourceFingerprintEnabled bool
// traceScope, when set (only on the per-call copy made by BuildTraceScoped),
// constrains raw/scalar/time-series queries to spans whose trace_id is in the
// scope statement, attached as a __trace_scope CTE.
traceScope *qbtypes.Statement
}
var _ qbtypes.StatementBuilder[qbtypes.TraceAggregation] = (*traceQueryStatementBuilder)(nil)
@@ -99,34 +95,6 @@ func NewTraceQueryStatementBuilder(
}
}
// BuildTraceScoped is Build with the query additionally constrained to spans whose
// trace_id is 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 *qbtypes.Statement,
) (*qbtypes.Statement, error) {
scoped := *b
scoped.traceScope = traceScope
return scoped.Build(ctx, orgID, start, end, requestType, query, variables)
}
// attachTraceScope adds the trace-scope condition to sb and returns the CTE fragment
// + args to prepend; both empty when no scope is set.
func (b *traceQueryStatementBuilder) attachTraceScope(sb *sqlbuilder.SelectBuilder) (string, []any) {
if b.traceScope == nil {
return "", nil
}
sb.Where("trace_id GLOBAL IN (SELECT trace_id FROM __trace_scope)")
return fmt.Sprintf("__trace_scope AS (%s)", b.traceScope.Query), b.traceScope.Args
}
// Build builds a SQL query for traces based on the given parameters.
func (b *traceQueryStatementBuilder) Build(
ctx context.Context,
@@ -350,11 +318,6 @@ func (b *traceQueryStatementBuilder) buildListQuery(
cteArgs = append(cteArgs, args)
}
if scopeFrag, scopeArgs := b.attachTraceScope(sb); scopeFrag != "" {
cteFragments = append(cteFragments, scopeFrag)
cteArgs = append(cteArgs, scopeArgs)
}
for i, field := range query.SelectFields {
expr, err := b.fm.ColumnExpressionFor(ctx, orgID, start, end, &field, telemetrytypes.FieldDataTypeUnspecified, keys)
if err != nil {
@@ -556,11 +519,6 @@ func (b *traceQueryStatementBuilder) buildTimeSeriesQuery(
cteArgs = append(cteArgs, args)
}
if scopeFrag, scopeArgs := b.attachTraceScope(sb); scopeFrag != "" {
cteFragments = append(cteFragments, scopeFrag)
cteArgs = append(cteArgs, scopeArgs)
}
sb.SelectMore(fmt.Sprintf(
"toStartOfInterval(timestamp, INTERVAL %d SECOND) AS ts",
int64(query.StepInterval.Seconds()),
@@ -721,13 +679,6 @@ 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 scopeFrag, scopeArgs := b.attachTraceScope(sb); scopeFrag != "" && !skipResourceCTE {
cteFragments = append(cteFragments, scopeFrag)
cteArgs = append(cteArgs, scopeArgs)
}
allAggChArgs := []any{}
fieldNames := make([]string, 0, len(query.GroupBy))

View File

@@ -214,40 +214,6 @@ def test_ai_span_list_excludes_non_gen_ai_spans(
assert "POST /api/chat" not in names # root span excluded
def test_ai_span_list_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:
"""Span list (raw) with a trace-level condition qualifies whole traces first:
out-tokens 100 / 300 and `trace.output_tokens > 100` keeps only the large trace's span."""
now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
service = "ai-it-spanlist-tracefilter"
small = ai_trace(now=now, service=service, user="a", in_tokens=10, out_tokens=100, cost=0.1)
large = ai_trace(now=now, service=service, user="b", in_tokens=30, out_tokens=300, cost=0.2)
insert_traces(small + large)
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
start_ms, end_ms = query_window(now)
query = BuilderQuery(
signal="traces",
query_type="builder_ai_query",
name="A",
filter_expression=f"service.name = '{service}' AND trace.output_tokens > 100",
limit=10,
)
resp = make_query_request(signoz, token, start_ms, end_ms, [query.to_dict()], request_type=RequestType.RAW)
assert resp.status_code == HTTPStatus.OK, resp.text
rows = resp.json()["data"]["data"]["results"][0]["rows"]
assert len(rows) == 1, f"expected only the large trace's LLM span, got {len(rows)} rows"
body = json.dumps(rows)
assert large[0].trace_id in body
assert small[0].trace_id not in body
def test_ai_list_having_or_aggregates(
signoz: types.SigNoz,
create_user_admin: None, # pylint: disable=unused-argument

View File

@@ -1,594 +0,0 @@
"""
Integration tests for query_type="builder_ai_query" scalar / time-series aggregations.
Aggregations come in two domains, chosen per expression by the `trace.` prefix:
- span-level (bare keys): over individual gen_ai spans (count(), sum(gen_ai.*))
- trace-level (trace.*): over window-clipped per-trace values (avg(trace.output_tokens))
A trace-level condition in the filter (trace.output_tokens > N) qualifies traces by
their window-clipped per-trace values in every request type — the span-list variant
of this is covered in 01_ai_traces.py (test_ai_span_list_trace_level_filter).
Each test tags its spans with a unique service.name and filters on it, so tests do
not interfere with each other's data.
"""
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_scalar_table_data,
make_query_request,
)
from fixtures.traces import TraceIdGenerator, Traces, TracesKind, TracesStatusCode
def _ai_trace(
*,
now: datetime,
service: str,
in_tokens: int,
out_tokens: int,
model: str = "gpt-4o-mini",
) -> list[Traces]:
"""A minimal AI trace: root span + one LLM span with gen_ai attributes."""
trace_id = TraceIdGenerator.trace_id()
root_id = TraceIdGenerator.span_id()
resources = {"service.name": service}
root = Traces(
timestamp=now - timedelta(seconds=5),
duration=timedelta(seconds=2),
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"},
)
llm = Traces(
timestamp=now - timedelta(seconds=4),
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": model,
"gen_ai.usage.input_tokens": in_tokens,
"gen_ai.usage.output_tokens": out_tokens,
},
)
return [root, llm]
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 [
Traces(
timestamp=now - timedelta(seconds=5),
duration=timedelta(seconds=2),
trace_id=trace_id,
span_id=root_id,
parent_span_id="",
name="POST /api/tool",
kind=TracesKind.SPAN_KIND_SERVER,
status_code=TracesStatusCode.STATUS_CODE_OK,
resources=resources,
attributes={"http.request.method": "POST"},
),
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 _window_ms(now: datetime) -> tuple[int, int]:
start_ms = int((now - timedelta(minutes=10)).timestamp() * 1000)
end_ms = int((now + timedelta(minutes=1)).timestamp() * 1000)
return start_ms, end_ms
def _scalar_query(
service: str,
expression: str,
*,
filter_extra: str = "",
group_by: list[TelemetryFieldKey] | None = None,
alias: str | None = None,
having: str | None = None,
limit: 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,
limit=limit,
).to_dict()
def _scalar_value(signoz: types.SigNoz, token: str, start_ms: int, end_ms: int, service: str, expression: str) -> float:
"""Run one single-aggregation scalar query and return its value."""
resp = make_query_request(
signoz,
token,
start_ms,
end_ms,
[_scalar_query(service, expression)],
request_type=RequestType.SCALAR,
)
assert resp.status_code == HTTPStatus.OK, f"{expression}: {resp.text}"
data = get_scalar_table_data(resp.json())
assert len(data) == 1, f"{expression}: expected one row, got {data}"
return float(data[0][-1])
def _series_values(response_json: dict) -> list[list[float]]:
"""Per-series lists of bucket values (bucket order as returned)."""
series = response_json["data"]["data"]["results"][0]["aggregations"][0]["series"]
return [[v["value"] for v in ser["values"]] for ser in series]
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:
"""
Trace-level (trace.) scalar aggregations over per-trace values: two traces with
out-tokens 100 / 300 give avg(trace.output_tokens)=200 and count(trace.trace_id)=2,
while the span-level count() sees the two LLM spans (root spans are gated out).
"""
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 = _window_ms(now)
def scalar_value(expression: str) -> float:
return _scalar_value(signoz, token, start_ms, end_ms, service, expression)
assert scalar_value("avg(trace.output_tokens)") == pytest.approx(200)
assert scalar_value("count(trace.trace_id)") == 2
assert scalar_value("max(trace.total_tokens)") == pytest.approx(330)
assert scalar_value("p50(trace.output_tokens)") == pytest.approx(200) # AggreFuncMap -> quantile(0.50)
# arithmetic inside one function and between functions
assert scalar_value("avg(trace.output_tokens + trace.input_tokens)") == pytest.approx(220)
assert scalar_value("sum(trace.output_tokens)/count(trace.trace_id)") == pytest.approx(200)
# span-level domain still works through the same request type
assert scalar_value("count()") == 2 # the two LLM spans; roots are not gen_ai
assert scalar_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 in the filter qualifies whole traces before aggregation:
with out-tokens 100 / 300, `trace.output_tokens > 100` keeps only the 300 trace
for both trace-level and span-level aggregations.
"""
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 = _window_ms(now)
for expression, expected in (
("sum(trace.output_tokens)", 300), # native trace-domain path
("sum(gen_ai.usage.output_tokens)", 300), # delegated span-domain path (__trace_scope)
):
resp = make_query_request(
signoz,
token,
start_ms,
end_ms,
[_scalar_query(service, expression, filter_extra="trace.output_tokens > 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 float(data[0][-1]) == pytest.approx(expected), f"{expression}: {data}"
# the qualification also constrains delegated (span-domain) time series
ts = BuilderQuery(
signal="traces",
query_type="builder_ai_query",
name="A",
filter_expression=f"service.name = '{service}' AND trace.output_tokens > 100",
aggregations=[Aggregation(expression="sum(gen_ai.usage.output_tokens)")],
step_interval=60,
)
resp = make_query_request(signoz, token, start_ms, end_ms, [ts.to_dict()], request_type=RequestType.TIME_SERIES)
assert resp.status_code == HTTPStatus.OK, resp.text
values = _series_values(resp.json())
assert values == [[pytest.approx(300)]], values
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: per-model avg of per-trace tokens."""
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 = _window_ms(now)
resp = make_query_request(
signoz,
token,
start_ms,
end_ms,
[_scalar_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_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:
"""Time-series over per-trace values: all spans fall in one step bucket, avg=200."""
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 = _window_ms(now)
query = BuilderQuery(
signal="traces",
query_type="builder_ai_query",
name="A",
filter_expression=f"service.name = '{service}'",
aggregations=[Aggregation(expression="avg(trace.output_tokens)")],
step_interval=60,
)
resp = make_query_request(
signoz,
token,
start_ms,
end_ms,
[query.to_dict()],
request_type=RequestType.TIME_SERIES,
)
assert resp.status_code == HTTPStatus.OK, resp.text
values = _series_values(resp.json())
assert values == [[pytest.approx(200)]], values
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:
"""
Grouped, limited time series ranks groups on whole-window per-trace values
(__ai_traces_total -> __limit_cte) and returns only the top-N: gpt-4o sums to
400 across two traces vs gpt-4o-mini's 50, so limit=1 keeps only gpt-4o.
"""
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 = _window_ms(now)
query = BuilderQuery(
signal="traces",
query_type="builder_ai_query",
name="A",
filter_expression=f"service.name = '{service}'",
aggregations=[Aggregation(expression="sum(trace.output_tokens)")],
group_by=[TelemetryFieldKey(name="gen_ai.request.model")],
step_interval=60,
limit=1,
)
resp = make_query_request(signoz, token, start_ms, end_ms, [query.to_dict()], request_type=RequestType.TIME_SERIES)
assert resp.status_code == HTTPStatus.OK, resp.text
series = resp.json()["data"]["data"]["results"][0]["aggregations"][0]["series"]
assert len(series) == 1, f"limit=1 must keep only the top group, got {len(series)} series"
assert series[0]["labels"][0]["value"] == "gpt-4o", series[0]["labels"]
assert [v["value"] for v in series[0]["values"]] == [pytest.approx(400)]
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): one trace with two LLM calls
two minutes apart contributes each call's tokens to its own bucket, not the
whole-trace 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"},
)
insert_traces([root, _llm(124, 100), _llm(4, 300)])
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
start_ms, end_ms = _window_ms(now)
query = BuilderQuery(
signal="traces",
query_type="builder_ai_query",
name="A",
filter_expression=f"service.name = '{service}'",
aggregations=[Aggregation(expression="avg(trace.output_tokens)")],
step_interval=60,
)
resp = make_query_request(signoz, token, start_ms, end_ms, [query.to_dict()], request_type=RequestType.TIME_SERIES)
assert resp.status_code == HTTPStatus.OK, resp.text
values = _series_values(resp.json())
assert len(values) == 1, values
assert sorted(values[0]) == [pytest.approx(100), pytest.approx(300)], f"each call's tokens in its own bucket: {values}"
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:
"""
Query variables resolve inside trace-level conditions with span-filter semantics;
an unresolvable $var is a 400, not a silent literal comparison.
"""
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 = _window_ms(now)
query = _scalar_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
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 "unknown variable" in 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_activity_gate_excludes_tool_only_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 tool-only trace (in the gen_ai gate, no LLM span) must not feed trace-level
aggregations: count(trace.trace_id) sees only the LLM trace, while the span-level
count() still sees both gen_ai spans (LLM + tool).
"""
now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
service = "ai-it-agg-gate"
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 = _window_ms(now)
def scalar_value(expression: str) -> float:
return _scalar_value(signoz, token, start_ms, end_ms, service, expression)
# count and avg agree on the trace set — the gate's purpose
assert scalar_value("count(trace.trace_id)") == 1, "tool-only trace must be dropped by the LLM-activity gate"
assert scalar_value("avg(trace.output_tokens)") == pytest.approx(100), "avg over the same gated trace set"
assert scalar_value("count()") == 2, "span-level count still sees the tool span"
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 = _window_ms(now)
resp = make_query_request(
signoz,
token,
start_ms,
end_ms,
[
_scalar_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:
"""Targeted 400s: mixed domains, group-by on a trace column, raw order by a trace column."""
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 = _window_ms(now)
# 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()")],
)
resp = make_query_request(signoz, token, start_ms, end_ms, [mixed.to_dict()], request_type=RequestType.SCALAR)
assert resp.status_code == HTTPStatus.BAD_REQUEST, resp.text
assert "cannot be mixed" in resp.text
# grouping by a trace-level per-trace column is rejected with a targeted error
bad_group = BuilderQuery(
signal="traces",
query_type="builder_ai_query",
name="A",
filter_expression=f"service.name = '{service}'",
aggregations=[Aggregation(expression="avg(trace.output_tokens)")],
group_by=[TelemetryFieldKey(name="trace.llm_call_count")],
)
resp = make_query_request(signoz, token, start_ms, end_ms, [bad_group.to_dict()], request_type=RequestType.SCALAR)
assert resp.status_code == HTTPStatus.BAD_REQUEST, resp.text
assert "grouping by trace-level aggregate" in resp.text
# ordering the span list by a trace-level column is rejected with a targeted error
bad_order = BuilderQuery(
signal="traces",
query_type="builder_ai_query",
name="A",
filter_expression=f"service.name = '{service}'",
order=[OrderBy(key=TelemetryFieldKey(name="trace.output_tokens"), direction="desc")],
limit=10,
)
resp = make_query_request(signoz, token, start_ms, end_ms, [bad_order.to_dict()], request_type=RequestType.RAW)
assert resp.status_code == HTTPStatus.BAD_REQUEST, resp.text
assert "ordering the span list by trace-level aggregate" in resp.text

View File

@@ -1,38 +0,0 @@
import pytest
from testcontainers.core.container import Network
from fixtures import types
from fixtures.signoz import create_signoz
@pytest.fixture(name="signoz", scope="package")
def signoz_ai_observability(
network: Network,
migrator: types.Operation, # pylint: disable=unused-argument
zeus: types.TestContainerDocker,
gateway: types.TestContainerDocker,
sqlstore: types.TestContainerSQL,
clickhouse: types.TestContainerClickhouse,
request: pytest.FixtureRequest,
pytestconfig: pytest.Config,
) -> types.SigNoz:
"""
Package-scoped SigNoz instance with AI observability enabled. source=ai
queries rely on the metadata store surfacing the static gen_ai key
definitions (enrichWithGenAIKeys), which is gated on this flag — without it
the gate keys (gen_ai.tool.name, gen_ai.agent.name, ...) only resolve once
a span carrying them has been ingested.
"""
return create_signoz(
network=network,
zeus=zeus,
gateway=gateway,
sqlstore=sqlstore,
clickhouse=clickhouse,
request=request,
pytestconfig=pytestconfig,
cache_key="signoz-ai-observability",
env_overrides={
"SIGNOZ_FLAGGER_CONFIG_BOOLEAN_ENABLE__AI__OBSERVABILITY": True,
},
)