feat: ai trace view cloumn

Hardcode a Trace View column whitelist (AI O11y TDD shape), own visibility
in a separate localStorage key, and wire Options → Edit columns via
FieldsSelector.availableFields so toggling is instant with no selectFields
refetch. Demoed on existing Trace View.

Co-authored-by: Gaurav Tewari <gauravtewari111@gmail.com>

.github/workflows/integrationci.yaml

@@ -39,6 +39,8 @@ jobs:
       matrix:
         suite:
           - alerts
+          - alertmanager
+          - alertmanagerrotation
           - basepath
           - callbackauthn
           - cloudintegrations

docs/api/openapi.yml

@@ -15477,72 +15477,6 @@ paths:
       summary: Lock dashboard (v2)
       tags:
       - dashboard
-  /api/v2/dashboards/{id}/migrate:
-    post:
-      deprecated: false
-      description: 'This endpoint retries the v1→v2 (Perses) migration on a dashboard
-        still stored in the v1 schema and returns the v2-shape result. It is idempotent:
-        a dashboard already in the v2 schema is returned unchanged.'
-      operationId: MigrateDashboardV2
-      parameters:
-      - in: path
-        name: id
-        required: true
-        schema:
-          type: string
-      responses:
-        "200":
-          content:
-            application/json:
-              schema:
-                properties:
-                  data:
-                    $ref: '#/components/schemas/DashboardtypesGettableDashboardV2'
-                  status:
-                    type: string
-                required:
-                - status
-                - data
-                type: object
-          description: OK
-        "400":
-          content:
-            application/json:
-              schema:
-                $ref: '#/components/schemas/RenderErrorResponse'
-          description: Bad Request
-        "401":
-          content:
-            application/json:
-              schema:
-                $ref: '#/components/schemas/RenderErrorResponse'
-          description: Unauthorized
-        "403":
-          content:
-            application/json:
-              schema:
-                $ref: '#/components/schemas/RenderErrorResponse'
-          description: Forbidden
-        "404":
-          content:
-            application/json:
-              schema:
-                $ref: '#/components/schemas/RenderErrorResponse'
-          description: Not Found
-        "500":
-          content:
-            application/json:
-              schema:
-                $ref: '#/components/schemas/RenderErrorResponse'
-          description: Internal Server Error
-      security:
-      - api_key:
-        - EDITOR
-      - tokenizer:
-        - EDITOR
-      summary: Migrate dashboard to v2
-      tags:
-      - dashboard
   /api/v2/factor_password/forgot:
     post:
       deprecated: false

ee/modules/dashboard/impldashboard/module.go

@@ -276,10 +276,6 @@ func (module *module) GetV2(ctx context.Context, orgID valuer.UUID, id valuer.UU
 	return module.pkgDashboardModule.GetV2(ctx, orgID, id)
 }
 
-func (module *module) MigrateV2(ctx context.Context, orgID valuer.UUID, id valuer.UUID) (*dashboardtypes.DashboardV2, error) {
-	return module.pkgDashboardModule.MigrateV2(ctx, orgID, id)
-}
-
 func (module *module) UpdateV2(ctx context.Context, orgID valuer.UUID, id valuer.UUID, updatedBy string, updatable dashboardtypes.UpdatableDashboardV2) (*dashboardtypes.DashboardV2, error) {
 	return module.pkgDashboardModule.UpdateV2(ctx, orgID, id, updatedBy, updatable)
 }

frontend/src/api/generated/services/dashboard/index.ts

@@ -52,8 +52,6 @@ import type {
 	ListDashboardsV2200,
 	ListDashboardsV2Params,
 	LockDashboardV2PathParameters,
-	MigrateDashboardV2200,
-	MigrateDashboardV2PathParameters,
 	PatchDashboardV2200,
 	PatchDashboardV2PathParameters,
 	PinDashboardV2PathParameters,
@@ -1806,85 +1804,6 @@ export const useLockDashboardV2 = <
 > => {
 	return useMutation(getLockDashboardV2MutationOptions(options));
 };
-/**
- * This endpoint retries the v1→v2 (Perses) migration on a dashboard still stored in the v1 schema and returns the v2-shape result. It is idempotent: a dashboard already in the v2 schema is returned unchanged.
- * @summary Migrate dashboard to v2
- */
-export const migrateDashboardV2 = (
-	{ id }: MigrateDashboardV2PathParameters,
-	signal?: AbortSignal,
-) => {
-	return GeneratedAPIInstance<MigrateDashboardV2200>({
-		url: `/api/v2/dashboards/${id}/migrate`,
-		method: 'POST',
-		signal,
-	});
-};
-
-export const getMigrateDashboardV2MutationOptions = <
-	TError = ErrorType<RenderErrorResponseDTO>,
-	TContext = unknown,
->(options?: {
-	mutation?: UseMutationOptions<
-		Awaited<ReturnType<typeof migrateDashboardV2>>,
-		TError,
-		{ pathParams: MigrateDashboardV2PathParameters },
-		TContext
-	>;
-}): UseMutationOptions<
-	Awaited<ReturnType<typeof migrateDashboardV2>>,
-	TError,
-	{ pathParams: MigrateDashboardV2PathParameters },
-	TContext
-> => {
-	const mutationKey = ['migrateDashboardV2'];
-	const { mutation: mutationOptions } = options
-		? options.mutation &&
-			'mutationKey' in options.mutation &&
-			options.mutation.mutationKey
-			? options
-			: { ...options, mutation: { ...options.mutation, mutationKey } }
-		: { mutation: { mutationKey } };
-
-	const mutationFn: MutationFunction<
-		Awaited<ReturnType<typeof migrateDashboardV2>>,
-		{ pathParams: MigrateDashboardV2PathParameters }
-	> = (props) => {
-		const { pathParams } = props ?? {};
-
-		return migrateDashboardV2(pathParams);
-	};
-
-	return { mutationFn, ...mutationOptions };
-};
-
-export type MigrateDashboardV2MutationResult = NonNullable<
-	Awaited<ReturnType<typeof migrateDashboardV2>>
->;
-
-export type MigrateDashboardV2MutationError = ErrorType<RenderErrorResponseDTO>;
-
-/**
- * @summary Migrate dashboard to v2
- */
-export const useMigrateDashboardV2 = <
-	TError = ErrorType<RenderErrorResponseDTO>,
-	TContext = unknown,
->(options?: {
-	mutation?: UseMutationOptions<
-		Awaited<ReturnType<typeof migrateDashboardV2>>,
-		TError,
-		{ pathParams: MigrateDashboardV2PathParameters },
-		TContext
-	>;
-}): UseMutationResult<
-	Awaited<ReturnType<typeof migrateDashboardV2>>,
-	TError,
-	{ pathParams: MigrateDashboardV2PathParameters },
-	TContext
-> => {
-	return useMutation(getMigrateDashboardV2MutationOptions(options));
-};
 /**
  * This endpoint returns the sanitized v2-shape dashboard data for public access. Each panel query is reduced to a safe field subset, so filters and raw query strings are not exposed.
  * @summary Get public dashboard data (v2)

frontend/src/api/generated/services/sigNoz.schemas.ts

@@ -11164,17 +11164,6 @@ export type UnlockDashboardV2PathParameters = {
 export type LockDashboardV2PathParameters = {
 	id: string;
 };
-export type MigrateDashboardV2PathParameters = {
-	id: string;
-};
-export type MigrateDashboardV2200 = {
-	data: DashboardtypesGettableDashboardV2DTO;
-	/**
-	 * @type string
-	 */
-	status: string;
-};
-
 export type GetFeatures200 = {
 	/**
 	 * @type array

frontend/src/components/FieldsSelector/FieldsSelector.tsx

@@ -31,6 +31,13 @@ interface FieldsSelectorProps {
 	width?: number;
 	height?: number;
 	defaultPosition?: { x: number; y: number };
+	/**
+	 * Caller-supplied field list. When provided, Other Fields offers exactly
+	 * these and key discovery is skipped — use it when the set of selectable
+	 * fields is known up front rather than fetched. Omit to discover keys from
+	 * the API for `signal`.
+	 */
+	availableFields?: TelemetryFieldKey[];
 }
 
 type FieldsSelectorContentProps = Omit<FieldsSelectorProps, 'isOpen'>;
@@ -49,6 +56,7 @@ function FieldsSelectorContent({
 	width = DEFAULT_PANEL_WIDTH,
 	height,
 	defaultPosition,
+	availableFields,
 }: FieldsSelectorContentProps): JSX.Element {
 	const resolvedHeight =
 		height ?? window.innerHeight - DEFAULT_PANEL_HEIGHT_OFFSET;
@@ -153,6 +161,7 @@ function FieldsSelectorContent({
 					addedFields={draftFields}
 					onAdd={handleAdd}
 					isAtLimit={isAtLimit}
+					availableFields={availableFields}
 				/>
 
 				{hasUnsavedChanges && (

frontend/src/components/FieldsSelector/OtherFields.tsx

@@ -21,6 +21,11 @@ interface OtherFieldsProps {
 	addedFields: TelemetryFieldKey[];
 	onAdd: (field: TelemetryFieldKey) => void;
 	isAtLimit: boolean;
+	/**
+	 * Caller-supplied field list. When provided, key discovery is skipped and
+	 * these are filtered locally by the search term instead.
+	 */
+	availableFields?: TelemetryFieldKey[];
 }
 
 function OtherFields({
@@ -29,7 +34,10 @@ function OtherFields({
 	addedFields,
 	onAdd,
 	isAtLimit,
+	availableFields,
 }: OtherFieldsProps): JSX.Element {
+	const useRegistry = Boolean(availableFields);
+
 	const { data, isFetching } = useGetQueryKeySuggestions(
 		{
 			signal,
@@ -41,11 +49,34 @@ function OtherFields({
 				signal,
 				debouncedInputValue,
 			],
-			enabled: true,
+			enabled: !useRegistry,
 		},
 	);
 
 	const otherFields: TelemetryFieldKey[] = useMemo(() => {
+		const addedIds = new Set(
+			addedFields.map((f) => f.key ?? buildCompositeKey(f.name, f.fieldContext)),
+		);
+
+		if (useRegistry && availableFields) {
+			const search = debouncedInputValue.trim().toLowerCase();
+			return availableFields
+				.filter((attr) => {
+					const id = attr.key ?? buildCompositeKey(attr.name, attr.fieldContext);
+					if (addedIds.has(id)) {
+						return false;
+					}
+					if (!search) {
+						return true;
+					}
+					return attr.name.toLowerCase().includes(search);
+				})
+				.map((attr) => ({
+					...attr,
+					key: attr.key ?? buildCompositeKey(attr.name, attr.fieldContext),
+				}));
+		}
+
 		const suggestions = Object.values(data?.data.data.keys || {}).flat();
 		// Normalize: synthesize `key` once so downstream reads can trust it.
 		const normalizedSuggestions: TelemetryFieldKey[] = suggestions.map(
@@ -57,15 +88,12 @@ function OtherFields({
 				fieldDataType: attr.fieldDataType,
 			}),
 		);
-		const addedIds = new Set(
-			addedFields.map((f) => f.key ?? buildCompositeKey(f.name, f.fieldContext)),
-		);
 		return normalizedSuggestions.filter(
 			(attr) => !addedIds.has(attr.key as string),
 		);
-	}, [data, addedFields]);
+	}, [data, addedFields, availableFields, debouncedInputValue, useRegistry]);
 
-	if (isFetching) {
+	if (!useRegistry && isFetching) {
 		return (
 			<div className={cx(styles.section, styles.sectionOther)}>
 				<div className={styles.sectionHeader}>OTHER FIELDS</div>

frontend/src/constants/localStorage.ts

@@ -11,6 +11,8 @@ export enum LOCALSTORAGE {
 	TRACES_LIST_OPTIONS = 'TRACES_LIST_OPTIONS',
 	GRAPH_VISIBILITY_STATES = 'GRAPH_VISIBILITY_STATES',
 	TRACES_LIST_COLUMNS = 'TRACES_LIST_COLUMNS',
+	/** AI Observability Trace View column visibility. */
+	AI_TRACE_VIEW_COLUMNS = 'AI_TRACE_VIEW_COLUMNS',
 	LOGS_LIST_COLUMNS = 'LOGS_LIST_COLUMNS',
 	LOGS_LIST_COLUMN_SIZING = 'LOGS_LIST_COLUMN_SIZING',
 	LOGGED_IN_USER_NAME = 'LOGGED_IN_USER_NAME',

frontend/src/container/AIObservability/TraceView/aiTraceViewColumns.tsx

@@ -0,0 +1,178 @@
+import { Typography } from '@signozhq/ui/typography';
+import { LOCALSTORAGE } from 'constants/localStorage';
+import {
+	BASE_TRACE_VIEW_COLUMNS,
+	renderTraceDurationCell,
+	TraceViewColumn,
+} from 'container/TracesExplorer/TracesView/configs';
+import { TraceViewColumnSelection } from 'container/TracesExplorer/TracesView/useTraceViewColumns';
+
+function isBlank(value: unknown): boolean {
+	return value === undefined || value === null || value === '';
+}
+
+function renderCountCell(value: unknown): JSX.Element {
+	if (isBlank(value)) {
+		return <Typography>—</Typography>;
+	}
+	const count = Number(value);
+	return (
+		<Typography>
+			{Number.isFinite(count) ? count.toLocaleString() : String(value)}
+		</Typography>
+	);
+}
+
+function renderCostCell(value: unknown): JSX.Element {
+	if (isBlank(value)) {
+		return <Typography>—</Typography>;
+	}
+	const cost = Number(value);
+	return (
+		<Typography>
+			{Number.isFinite(cost) ? `$${cost.toFixed(4)}` : String(value)}
+		</Typography>
+	);
+}
+
+/**
+ * Trace-level gen_ai aggregates. The query-range response does not carry these
+ * yet, so they render as em dashes until the AI trace API lands — visible but
+ * empty is intentional, it lets the column set be reviewed ahead of the data.
+ */
+const AI_ONLY_COLUMNS: TraceViewColumn[] = [
+	{
+		field: {
+			name: 'input_tokens',
+			fieldContext: 'trace',
+			fieldDataType: 'int64',
+		},
+		title: 'Input Tokens',
+		render: renderCountCell,
+	},
+	{
+		field: {
+			name: 'output_tokens',
+			fieldContext: 'trace',
+			fieldDataType: 'int64',
+		},
+		title: 'Output Tokens',
+		render: renderCountCell,
+	},
+	{
+		field: {
+			name: 'total_tokens',
+			fieldContext: 'trace',
+			fieldDataType: 'int64',
+		},
+		title: 'Total Tokens',
+		render: renderCountCell,
+	},
+	{
+		field: {
+			name: 'llm_call_count',
+			fieldContext: 'trace',
+			fieldDataType: 'int64',
+		},
+		title: 'LLM Calls',
+		render: renderCountCell,
+	},
+	{
+		field: {
+			name: 'tool_call_count',
+			fieldContext: 'trace',
+			fieldDataType: 'int64',
+		},
+		title: 'Tool Calls',
+		render: renderCountCell,
+	},
+	{
+		field: {
+			name: 'distinct_tool_count',
+			fieldContext: 'trace',
+			fieldDataType: 'int64',
+		},
+		title: 'Distinct Tools',
+		render: renderCountCell,
+	},
+	{
+		field: {
+			name: 'estimated_cost_usd',
+			fieldContext: 'trace',
+			fieldDataType: 'float64',
+		},
+		title: 'Est. Cost (USD)',
+		render: renderCostCell,
+	},
+	{
+		field: {
+			name: 'max_llm_latency_ns',
+			fieldContext: 'trace',
+			fieldDataType: 'int64',
+		},
+		title: 'Max LLM Latency',
+		render: renderTraceDurationCell,
+	},
+	{
+		field: {
+			name: 'last_activity_time',
+			fieldContext: 'trace',
+			fieldDataType: 'int64',
+		},
+		title: 'Last Activity',
+	},
+	{
+		field: { name: 'start_time', fieldContext: 'trace', fieldDataType: 'int64' },
+		title: 'Start Time',
+	},
+	{
+		field: { name: 'end_time', fieldContext: 'trace', fieldDataType: 'int64' },
+		title: 'End Time',
+	},
+	{
+		field: {
+			name: 'trace_duration_nano',
+			fieldContext: 'trace',
+			fieldDataType: 'int64',
+		},
+		title: 'Trace Duration',
+		render: renderTraceDurationCell,
+	},
+	{
+		field: { name: 'error_count', fieldContext: 'trace', fieldDataType: 'int64' },
+		title: 'Errors',
+		render: renderCountCell,
+	},
+	{
+		field: {
+			name: 'root_span_name',
+			fieldContext: 'trace',
+			fieldDataType: 'string',
+		},
+		title: 'Root Span Name',
+	},
+	{
+		field: { name: 'input', fieldContext: 'trace', fieldDataType: 'string' },
+		title: 'Input',
+	},
+	{
+		field: { name: 'output', fieldContext: 'trace', fieldDataType: 'string' },
+		title: 'Output',
+	},
+];
+
+export const AI_TRACE_VIEW_COLUMNS: TraceViewColumn[] = [
+	...BASE_TRACE_VIEW_COLUMNS,
+	...AI_ONLY_COLUMNS,
+];
+
+/**
+ * Hand this to `<TracesView columnSelection={…} />` to get the AI column set
+ * plus the Options → Edit columns picker. Module-level so its identity is
+ * stable across renders.
+ */
+export const AI_TRACE_VIEW_COLUMN_SELECTION: TraceViewColumnSelection = {
+	columns: AI_TRACE_VIEW_COLUMNS,
+	storageKey: LOCALSTORAGE.AI_TRACE_VIEW_COLUMNS,
+	defaultVisible: BASE_TRACE_VIEW_COLUMNS.map((column) => column.field.name),
+};

frontend/src/container/TracesExplorer/Controls/Controls.module.scss

@@ -11,4 +11,12 @@
 	align-items: center;
 	gap: 4px;
 	cursor: pointer;
+
+	// Rendered as a <button> for keyboard access — strip the native chrome so it
+	// still reads as the inline text trigger it was.
+	padding: 0;
+	border: none;
+	background: none;
+	color: inherit;
+	font: inherit;
 }

frontend/src/container/TracesExplorer/Controls/index.tsx

@@ -5,6 +5,7 @@ import FieldsSelector from 'components/FieldsSelector';
 import Controls, { ControlsProps } from 'container/Controls';
 import { OptionsMenuConfig } from 'container/OptionsMenu/types';
 import useQueryPagination from 'hooks/queryPagination/useQueryPagination';
+import { TelemetryFieldKey } from 'types/api/v5/queryRange';
 import { DataSource } from 'types/common/queryBuilder';
 
 import styles from './Controls.module.scss';
@@ -15,6 +16,7 @@ function TraceExplorerControls({
 	perPageOptions,
 	config,
 	showSizeChanger = true,
+	availableFields,
 }: TraceExplorerControlsProps): JSX.Element | null {
 	const { t } = useTranslation(['trace']);
 	const [isFieldsSelectorOpen, setIsFieldsSelectorOpen] = useState(false);
@@ -30,13 +32,15 @@ function TraceExplorerControls({
 		<div className={styles.container}>
 			{config?.fieldsSelector && (
 				<>
-					<div
+					<button
+						type="button"
 						className={styles.optionsTrigger}
 						onClick={(): void => setIsFieldsSelectorOpen(true)}
+						data-testid="trace-view-options-trigger"
 					>
 						{t('options_menu.options')}
 						<Settings size="md" />
-					</div>
+					</button>
 					<FieldsSelector
 						isOpen={isFieldsSelectorOpen}
 						title="Edit columns"
@@ -44,6 +48,7 @@ function TraceExplorerControls({
 						onFieldsChange={config.fieldsSelector.onFieldsChange}
 						onClose={(): void => setIsFieldsSelectorOpen(false)}
 						signal={DataSource.TRACES}
+						availableFields={availableFields}
 					/>
 				</>
 			)}
@@ -63,20 +68,20 @@ function TraceExplorerControls({
 	);
 }
 
-TraceExplorerControls.defaultProps = {
-	config: null,
-};
-
 type TraceExplorerControlsProps = Pick<
 	ControlsProps,
 	'isLoading' | 'totalCount' | 'perPageOptions'
 > & {
 	config?: OptionsMenuConfig | null;
 	showSizeChanger?: boolean;
+	/** Forwarded to FieldsSelector — see `availableFields` there. */
+	availableFields?: TelemetryFieldKey[];
 };
 
 TraceExplorerControls.defaultProps = {
+	config: null,
 	showSizeChanger: true,
+	availableFields: undefined,
 };
 
 export default memo(TraceExplorerControls);

frontend/src/container/TracesExplorer/TracesView/configs.tsx

@@ -4,47 +4,105 @@ import { Typography } from '@signozhq/ui/typography';
 import ROUTES from 'constants/routes';
 import { getMs } from 'container/Trace/Filters/Panel/PanelBody/Duration/util';
 import { DEFAULT_PER_PAGE_OPTIONS } from 'hooks/queryPagination';
+import { TelemetryFieldKey } from 'types/api/v5/queryRange';
 import { ListItem } from 'types/api/widgets/getQuery';
 
 export const PER_PAGE_OPTIONS: number[] = [10, ...DEFAULT_PER_PAGE_OPTIONS];
 
-export const columns: ColumnsType<ListItem['data']> = [
+/**
+ * One Trace View column: the telemetry field it reads, its header, and how the
+ * cell renders. Callers own their column sets — Trace View has no knowledge of
+ * any particular product's columns.
+ */
+export interface TraceViewColumn {
+	field: TelemetryFieldKey;
+	title: string;
+	/** Defaults to `renderTraceCellValue`. */
+	render?: (value: unknown) => JSX.Element;
+}
+
+function isBlank(value: unknown): boolean {
+	return value === undefined || value === null || value === '';
+}
+
+/** Fallback cell: em dash when empty, otherwise stringified. */
+export function renderTraceCellValue(value: unknown): JSX.Element {
+	if (isBlank(value)) {
+		return <Typography>—</Typography>;
+	}
+	return <Typography>{String(value)}</Typography>;
+}
+
+/** Nanosecond duration rendered as milliseconds. */
+export function renderTraceDurationCell(value: unknown): JSX.Element {
+	if (isBlank(value)) {
+		return <Typography>—</Typography>;
+	}
+	return <Typography>{getMs(String(value))}ms</Typography>;
+}
+
+function renderTraceIdCell(value: unknown): JSX.Element {
+	if (isBlank(value)) {
+		return <Typography>—</Typography>;
+	}
+	return (
+		<Link
+			to={generatePath(ROUTES.TRACE_DETAIL, {
+				id: String(value),
+			})}
+			data-testid="trace-id"
+		>
+			{String(value)}
+		</Link>
+	);
+}
+
+/**
+ * The root-span columns Trace View renders when the caller configures no
+ * column selection. Matches the pre-selection behaviour exactly.
+ */
+export const BASE_TRACE_VIEW_COLUMNS: TraceViewColumn[] = [
 	{
+		field: {
+			name: 'service.name',
+			fieldContext: 'resource',
+			fieldDataType: 'string',
+		},
 		title: 'Root Service Name',
-		dataIndex: 'service.name',
-		key: 'serviceName',
 	},
 	{
+		field: { name: 'name', fieldContext: 'span', fieldDataType: 'string' },
 		title: 'Root Operation Name',
-		dataIndex: 'name',
-		key: 'name',
 	},
 	{
+		field: {
+			name: 'duration_nano',
+			fieldContext: 'span',
+			fieldDataType: 'int64',
+		},
 		title: 'Root Duration (in ms)',
-		dataIndex: 'duration_nano',
-		key: 'durationNano',
-		render: (duration: number): JSX.Element => (
-			<Typography>{getMs(String(duration))}ms</Typography>
-		),
+		render: renderTraceDurationCell,
 	},
 	{
+		field: { name: 'span_count', fieldContext: 'trace', fieldDataType: 'int64' },
 		title: 'No of Spans',
-		dataIndex: 'span_count',
-		key: 'span_count',
 	},
 	{
+		field: { name: 'trace_id', fieldContext: 'span', fieldDataType: 'string' },
 		title: 'TraceID',
-		dataIndex: 'trace_id',
-		key: 'traceID',
-		render: (traceID: string): JSX.Element => (
-			<Link
-				to={generatePath(ROUTES.TRACE_DETAIL, {
-					id: traceID,
-				})}
-				data-testid="trace-id"
-			>
-				{traceID}
-			</Link>
-		),
+		render: renderTraceIdCell,
 	},
 ];
+
+/** Build antd columns, preserving the order given. */
+export function buildTraceViewColumns(
+	columns: TraceViewColumn[],
+): ColumnsType<ListItem['data']> {
+	return columns.map(({ field, title, render }) => ({
+		title,
+		dataIndex: field.name,
+		key: field.name,
+		render: (value: unknown): JSX.Element =>
+			(render ?? renderTraceCellValue)(value),
+	}));
+}

frontend/src/container/TracesExplorer/TracesView/index.tsx

@@ -34,14 +34,26 @@ import DOCLINKS from 'utils/docLinks';
 
 import TraceExplorerControls from '../Controls';
 import { TracesLoading } from '../TraceLoading/TraceLoading';
-import { columns, PER_PAGE_OPTIONS } from './configs';
+import {
+	BASE_TRACE_VIEW_COLUMNS,
+	buildTraceViewColumns,
+	PER_PAGE_OPTIONS,
+} from './configs';
 import { ActionsContainer, Container } from './styles';
+import useTraceViewColumns, {
+	TraceViewColumnSelection,
+} from './useTraceViewColumns';
 
 interface TracesViewProps {
 	isFilterApplied: boolean;
 	setWarning: Dispatch<SetStateAction<Warning | undefined>>;
 	setIsLoadingQueries: Dispatch<SetStateAction<boolean>>;
 	queryKeyRef?: MutableRefObject<any>;
+	/**
+	 * Opt in to user-editable columns. Omit for the base root-span columns with
+	 * no Options → Edit columns picker.
+	 */
+	columnSelection?: TraceViewColumnSelection;
 }
 
 function TracesView({
@@ -49,6 +61,7 @@ function TracesView({
 	setWarning,
 	setIsLoadingQueries,
 	queryKeyRef,
+	columnSelection,
 }: TracesViewProps): JSX.Element {
 	const { stagedQuery, panelType } = useQueryBuilder();
 
@@ -62,6 +75,32 @@ function TracesView({
 		QueryParams.pagination,
 	);
 
+	// Column visibility is owned here rather than by useOptionsMenu, whose
+	// TRACES_LIST_OPTIONS storage is already claimed by List View.
+	const { visibleColumns, selectedFields, availableFields, onFieldsChange } =
+		useTraceViewColumns(columnSelection);
+
+	const fieldsSelectorConfig = useMemo(
+		() =>
+			columnSelection
+				? {
+						fieldsSelector: {
+							value: selectedFields,
+							onFieldsChange,
+						},
+					}
+				: null,
+		[columnSelection, selectedFields, onFieldsChange],
+	);
+
+	const tableColumns = useMemo(
+		() =>
+			buildTraceViewColumns(
+				columnSelection ? visibleColumns : BASE_TRACE_VIEW_COLUMNS,
+			),
+		[columnSelection, visibleColumns],
+	);
+
 	const transformedQuery = useMemo(
 		() => getListViewQuery(stagedQuery || initialQueriesMap.traces),
 		[stagedQuery],
@@ -76,6 +115,8 @@ function TracesView({
 			stagedQuery,
 			panelType,
 			paginationQueryData,
+			// Column visibility is deliberately absent: it is client-side only, and
+			// this array doubles as the parent's cancelQueries handle.
 		],
 		[
 			globalSelectedTime,
@@ -100,6 +141,8 @@ function TracesView({
 			params: {
 				dataSource: 'traces',
 			},
+			// No selectColumns: the backend returns all columns and visibility is
+			// resolved client-side, so toggling a column is refetch-free.
 			tableParams: {
 				pagination: paginationQueryData,
 			},
@@ -162,6 +205,8 @@ function TracesView({
 							isLoading={isLoading}
 							totalCount={responseData?.length || 0}
 							perPageOptions={PER_PAGE_OPTIONS}
+							config={fieldsSelectorConfig}
+							availableFields={columnSelection ? availableFields : undefined}
 						/>
 					</div>
 				</ActionsContainer>
@@ -190,7 +235,7 @@ function TracesView({
 			{(tableData || []).length !== 0 && (
 				<ResizeTable
 					loading={isLoading}
-					columns={columns}
+					columns={tableColumns}
 					tableLayout="fixed"
 					dataSource={tableData}
 					scroll={{ x: true }}
@@ -203,6 +248,7 @@ function TracesView({
 
 TracesView.defaultProps = {
 	queryKeyRef: undefined,
+	columnSelection: undefined,
 };
 
 export default memo(TracesView);

frontend/src/container/TracesExplorer/TracesView/useTraceViewColumns.ts

@@ -0,0 +1,123 @@
+import { useCallback, useMemo, useState } from 'react';
+import getLocalStorageKey from 'api/browser/localstorage/get';
+import setLocalStorageKey from 'api/browser/localstorage/set';
+import { LOCALSTORAGE } from 'constants/localStorage';
+import { TelemetryFieldKey } from 'types/api/v5/queryRange';
+
+import { TraceViewColumn } from './configs';
+
+/**
+ * Opt-in column selection for Trace View. Passing one enables the Options →
+ * Edit columns picker; omitting it leaves Trace View on its base columns with
+ * no picker at all.
+ *
+ * Pass a module-level constant, not an inline literal — a fresh object each
+ * render invalidates the memoised column set on every pass.
+ */
+export interface TraceViewColumnSelection {
+	/** Every column offered, in default display order. */
+	columns: TraceViewColumn[];
+	/** Where this caller persists visibility. Must be unique per view. */
+	storageKey: LOCALSTORAGE;
+	/** Field names visible before the user customises anything. */
+	defaultVisible: string[];
+}
+
+interface UseTraceViewColumnsReturn {
+	/** Columns to render, in the user's persisted order. */
+	visibleColumns: TraceViewColumn[];
+	/** Visible fields, for the picker's "added" list. */
+	selectedFields: TelemetryFieldKey[];
+	/** Every offered field, for the picker's "other" list. */
+	availableFields: TelemetryFieldKey[];
+	onFieldsChange: (fields: TelemetryFieldKey[]) => void;
+}
+
+function readStoredKeys(
+	selection: TraceViewColumnSelection | undefined,
+): string[] {
+	if (!selection) {
+		return [];
+	}
+
+	const { storageKey, columns, defaultVisible } = selection;
+	const raw = getLocalStorageKey(storageKey);
+	if (!raw) {
+		return defaultVisible;
+	}
+
+	try {
+		const parsed = JSON.parse(raw) as string[];
+		if (!Array.isArray(parsed) || parsed.length === 0) {
+			return defaultVisible;
+		}
+		// Drop anything the caller no longer offers, so a stale localStorage
+		// entry from an earlier column set self-heals instead of rendering blank.
+		const allowed = new Set(columns.map((column) => column.field.name));
+		const filtered = parsed.filter((key) => allowed.has(key));
+		return filtered.length > 0 ? filtered : defaultVisible;
+	} catch {
+		return defaultVisible;
+	}
+}
+
+/**
+ * Owns Trace View column visibility client-side: toggling reorders or hides
+ * columns without touching the query, so it never triggers a refetch.
+ */
+function useTraceViewColumns(
+	selection?: TraceViewColumnSelection,
+): UseTraceViewColumnsReturn {
+	const [visibleKeys, setVisibleKeys] = useState<string[]>(() =>
+		readStoredKeys(selection),
+	);
+
+	const offeredColumns = selection?.columns;
+
+	const columnsByName = useMemo(
+		() =>
+			new Map((offeredColumns ?? []).map((column) => [column.field.name, column])),
+		[offeredColumns],
+	);
+
+	// Ordered by visibleKeys, so reordering in the picker moves the column.
+	const visibleColumns = useMemo(
+		() =>
+			visibleKeys
+				.map((key) => columnsByName.get(key))
+				.filter((column): column is TraceViewColumn => Boolean(column)),
+		[columnsByName, visibleKeys],
+	);
+
+	const selectedFields = useMemo(
+		() => visibleColumns.map((column) => column.field),
+		[visibleColumns],
+	);
+
+	const availableFields = useMemo(
+		() => (offeredColumns ?? []).map((column) => column.field),
+		[offeredColumns],
+	);
+
+	const onFieldsChange = useCallback(
+		(fields: TelemetryFieldKey[]): void => {
+			if (!selection) {
+				return;
+			}
+			const nextKeys = fields.map((field) => field.name);
+			const keys = nextKeys.length > 0 ? nextKeys : selection.defaultVisible;
+			setVisibleKeys(keys);
+			setLocalStorageKey(selection.storageKey, JSON.stringify(keys));
+		},
+		[selection],
+	);
+
+	return {
+		visibleColumns,
+		selectedFields,
+		availableFields,
+		onFieldsChange,
+	};
+}
+
+export default useTraceViewColumns;

frontend/src/pages/TracesExplorer/index.tsx

@@ -17,6 +17,7 @@ import { usePageActions } from 'container/AIAssistant/pageActions/usePageActions
 import ExplorerOptionWrapper from 'container/ExplorerOptions/ExplorerOptionWrapper';
 import { useOptionsMenu } from 'container/OptionsMenu';
 import LeftToolbarActions from 'container/QueryBuilder/components/ToolbarActions/LeftToolbarActions';
+import { AI_TRACE_VIEW_COLUMN_SELECTION } from 'container/AIObservability/TraceView/aiTraceViewColumns';
 import RightToolbarActions from 'container/QueryBuilder/components/ToolbarActions/RightToolbarActions';
 import Toolbar from 'container/Toolbar/Toolbar';
 import {
@@ -330,6 +331,10 @@ function TracesExplorer(): JSX.Element {
 									setWarning={setWarning}
 									setIsLoadingQueries={setIsLoadingQueries}
 									queryKeyRef={listQueryKeyRef}
+									// DEMO ONLY — surfaces the AI Observability column set here because
+									// the AI Explorer does not exist yet. Drop this prop (and the import)
+									// to return Trace View to its five base columns.
+									columnSelection={AI_TRACE_VIEW_COLUMN_SELECTION}
 								/>
 							</div>
 						)}

grammar/FilterQuery.g4

@@ -112,15 +112,12 @@ functionCall
     ;
 
 /*
- * Full-text search call: search('needle')
- *
- * Uses the shared functionParamList so future scoped forms like
- * search(body, 'abc') / search(attribute, 'abc') need no grammar change. Today
- * only a single needle is supported. Unlike bare/quoted free text (`fullText`),
- * which only targets the body column, search() fans out across every field.
+ * Full-text search: search('term') or scoped search('term', body, ...).
+ * First param is the search term; the rest are field-context scopes (body/attribute/
+ * resource/log), quoted or bare. Handled in the visitor — no grammar change.
  */
 searchCall
-    : SEARCH LPAREN functionParamList RPAREN
+    : SEARCH LPAREN valueList RPAREN
     ;
 
 // Function parameters can be keys, single scalar values, or arrays

pkg/alertmanager/alertmanagerserver/server.go

@@ -241,9 +241,12 @@ func (server *Server) PutAlerts(ctx context.Context, postableAlerts alertmanager
 }
 
 func (server *Server) SetConfig(ctx context.Context, alertmanagerConfig *alertmanagertypes.Config) error {
-	config := alertmanagerConfig.AlertmanagerConfig()
+	resolved, err := alertmanagerConfig.Resolved()
+	if err != nil {
+		return err
+	}
+	config := resolved.AlertmanagerConfig()
 
-	var err error
 	// Load SigNoz's alertmanager notification templates from the configured
 	// globs. The upstream default templates (default.tmpl, email.tmpl) are
 	// always loaded from the embedded alertmanager assets inside FromGlobs, so
@@ -275,7 +278,7 @@ func (server *Server) SetConfig(ctx context.Context, alertmanagerConfig *alertma
 			server.logger.InfoContext(ctx, "skipping creation of receiver not referenced by any route", slog.String("receiver", rcv.Name))
 			continue
 		}
-		extendedRcv, err := alertmanagerConfig.GetReceiver(rcv.Name)
+		extendedRcv, err := resolved.GetReceiver(rcv.Name)
 		if err != nil {
 			return err
 		}
@@ -350,7 +353,7 @@ func (server *Server) SetConfig(ctx context.Context, alertmanagerConfig *alertma
 	go server.dispatcher.Run()
 	go server.inhibitor.Run()
 
-	server.alertmanagerConfig = alertmanagerConfig
+	server.alertmanagerConfig = resolved
 	return nil
 }
 

pkg/alertmanager/config.go

@@ -1,7 +1,6 @@
 package alertmanager
 
 import (
-	"net/url"
 	"strings"
 	"time"
 
@@ -26,11 +25,6 @@ type Signoz struct {
 	alertmanagerserver.Config `mapstructure:",squash" yaml:",squash"`
 }
 
-type Legacy struct {
-	// ApiURL is the URL of the legacy signoz alertmanager.
-	ApiURL *url.URL `mapstructure:"api_url"`
-}
-
 func NewConfigFactory() factory.ConfigFactory {
 	return factory.NewConfigFactory(factory.MustNewName("alertmanager"), newConfig)
 }

pkg/alertmanager/signozalertmanager/provider.go

@@ -167,6 +167,10 @@ func (provider *provider) UpdateChannelByReceiverAndID(ctx context.Context, orgI
 		return err
 	}
 
+	if err := config.SetGlobalConfig(provider.config.Signoz.Global); err != nil {
+		return err
+	}
+
 	if err := config.UpdateReceiver(receiver); err != nil {
 		return err
 	}
@@ -217,6 +221,10 @@ func (provider *provider) CreateChannel(ctx context.Context, orgID string, recei
 		return nil, err
 	}
 
+	if err := config.SetGlobalConfig(provider.config.Signoz.Global); err != nil {
+		return nil, err
+	}
+
 	if err := config.CreateReceiver(receiver); err != nil {
 		return nil, err
 	}

pkg/apiserver/signozapiserver/dashboard.go

@@ -85,23 +85,6 @@ func (provider *provider) addDashboardRoutes(router *mux.Router) error {
 		return err
 	}
 
-	if err := router.Handle("/api/v2/dashboards/{id}/migrate", handler.New(provider.authzMiddleware.EditAccess(provider.dashboardHandler.MigrateV2), handler.OpenAPIDef{
-		ID:                  "MigrateDashboardV2",
-		Tags:                []string{"dashboard"},
-		Summary:             "Migrate dashboard to v2",
-		Description:         "This endpoint retries the v1→v2 (Perses) migration on a dashboard still stored in the v1 schema and returns the v2-shape result. It is idempotent: a dashboard already in the v2 schema is returned unchanged.",
-		Request:             nil,
-		RequestContentType:  "",
-		Response:            new(dashboardtypes.GettableDashboardV2),
-		ResponseContentType: "application/json",
-		SuccessStatusCode:   http.StatusOK,
-		ErrorStatusCodes:    []int{http.StatusBadRequest, http.StatusNotFound},
-		Deprecated:          false,
-		SecuritySchemes:     newSecuritySchemes(types.RoleEditor),
-	})).Methods(http.MethodPost).GetError(); err != nil {
-		return err
-	}
-
 	if err := router.Handle("/api/v2/dashboards/{id}", handler.New(provider.authzMiddleware.ViewAccess(provider.dashboardHandler.GetV2), handler.OpenAPIDef{
 		ID:                  "GetDashboardV2",
 		Tags:                []string{"dashboard"},

pkg/flagger/flaggertest/flaggertest.go

@@ -29,13 +29,18 @@ func New(t *testing.T) flagger.Flagger {
 
 // WithUseJSONBody returns a Flagger with use_json_body set to the given value.
 func WithUseJSONBody(t *testing.T, enabled bool) flagger.Flagger {
+	return WithBooleanFlags(t, map[string]bool{
+		flagger.FeatureUseJSONBody.String(): enabled,
+	})
+}
+
+// WithBooleanFlags returns a Flagger with the given boolean flags, keyed by feature name.
+func WithBooleanFlags(t *testing.T, flags map[string]bool) flagger.Flagger {
 	t.Helper()
 	registry := flagger.MustNewRegistry()
 	cfg := flagger.Config{}
-	if enabled {
-		cfg.Config.Boolean = map[string]bool{
-			flagger.FeatureUseJSONBody.String(): true,
-		}
+	if len(flags) > 0 {
+		cfg.Config.Boolean = flags
 	}
 	fl, err := flagger.New(
 		context.Background(),

pkg/modules/dashboard/dashboard.go

@@ -63,9 +63,6 @@ type Module interface {
 
 	GetV2(ctx context.Context, orgID valuer.UUID, id valuer.UUID) (*dashboardtypes.DashboardV2, error)
 
-	// MigrateV2 retries the v1→v2 migration on a dashboard still stored in the v1 schema.
-	MigrateV2(ctx context.Context, orgID valuer.UUID, id valuer.UUID) (*dashboardtypes.DashboardV2, error)
-
 	ListV2(ctx context.Context, orgID valuer.UUID, params *dashboardtypes.ListDashboardsV2Params) (*dashboardtypes.ListableDashboardV2, error)
 
 	ListForUserV2(ctx context.Context, orgID valuer.UUID, userID valuer.UUID, params *dashboardtypes.ListDashboardsV2Params) (*dashboardtypes.ListableDashboardForUserV2, error)
@@ -135,8 +132,6 @@ type Handler interface {
 
 	GetV2(http.ResponseWriter, *http.Request)
 
-	MigrateV2(http.ResponseWriter, *http.Request)
-
 	ListV2(http.ResponseWriter, *http.Request)
 
 	ListForUserV2(http.ResponseWriter, *http.Request)

pkg/modules/dashboard/impldashboard/v2_handler.go

@@ -207,38 +207,6 @@ func (handler *handler) GetV2(rw http.ResponseWriter, r *http.Request) {
 	render.Success(rw, http.StatusOK, dashboard.ToGettableDashboardV2())
 }
 
-func (handler *handler) MigrateV2(rw http.ResponseWriter, r *http.Request) {
-	ctx, cancel := context.WithTimeout(r.Context(), 10*time.Second)
-	defer cancel()
-
-	claims, err := authtypes.ClaimsFromContext(ctx)
-	if err != nil {
-		render.Error(rw, err)
-		return
-	}
-
-	orgID := valuer.MustNewUUID(claims.OrgID)
-
-	id := mux.Vars(r)["id"]
-	if id == "" {
-		render.Error(rw, errors.Newf(errors.TypeInvalidInput, errors.CodeInvalidInput, "id is missing in the path"))
-		return
-	}
-	dashboardID, err := valuer.NewUUID(id)
-	if err != nil {
-		render.Error(rw, err)
-		return
-	}
-
-	dashboard, err := handler.module.MigrateV2(ctx, orgID, dashboardID)
-	if err != nil {
-		render.Error(rw, err)
-		return
-	}
-
-	render.Success(rw, http.StatusOK, dashboard.ToGettableDashboardV2())
-}
-
 func (handler *handler) LockV2(rw http.ResponseWriter, r *http.Request) {
 	handler.lockUnlockV2(rw, r, true)
 }

pkg/modules/dashboard/impldashboard/v2_module.go

@@ -4,7 +4,6 @@ import (
 	"context"
 
 	"github.com/SigNoz/signoz/pkg/errors"
-	"github.com/SigNoz/signoz/pkg/transition"
 	"github.com/SigNoz/signoz/pkg/types/coretypes"
 	"github.com/SigNoz/signoz/pkg/types/dashboardtypes"
 	"github.com/SigNoz/signoz/pkg/types/tagtypes"
@@ -122,51 +121,6 @@ func (module *module) GetV2(ctx context.Context, orgID valuer.UUID, id valuer.UU
 	return storable.ToDashboardV2(tags)
 }
 
-// MigrateV2 retries the v1→v2 migration on a dashboard still stored as v1 (one the
-// bulk 103 migration skipped or failed). Idempotent: an already-v2 one is unchanged.
-func (module *module) MigrateV2(ctx context.Context, orgID valuer.UUID, id valuer.UUID) (*dashboardtypes.DashboardV2, error) {
-	storable, err := module.store.Get(ctx, orgID, id)
-	if err != nil {
-		return nil, err
-	}
-
-	// Already migrated: return as-is.
-	if storable.IsV2() {
-		tags, err := module.tagModule.ListForResource(ctx, orgID, coretypes.KindDashboard, id)
-		if err != nil {
-			return nil, err
-		}
-		return storable.ToDashboardV2(tags)
-	}
-
-	// v1→v2 needs v5-shaped queries; run v4→v5 in place first.
-	transition.NewDashboardMigrateV5(module.settings.Logger(), nil, nil).Migrate(ctx, storable.Data)
-
-	v2, err := storable.ConvertV1ToV2()
-	if err != nil {
-		return nil, err
-	}
-
-	err = module.store.RunInTx(ctx, func(ctx context.Context) error {
-		resolvedTags, err := module.tagModule.SyncTags(ctx, orgID, coretypes.KindDashboard, v2.ID, tagtypes.NewPostableTagsFromTags(v2.Tags))
-		if err != nil {
-			return err
-		}
-		v2.Tags = resolvedTags
-
-		storableV2, err := v2.ToStorableDashboard()
-		if err != nil {
-			return err
-		}
-		return module.store.Update(ctx, orgID, storableV2)
-	})
-	if err != nil {
-		return nil, err
-	}
-
-	return v2, nil
-}
-
 func (module *module) UpdateV2(ctx context.Context, orgID valuer.UUID, id valuer.UUID, updatedBy string, updatable dashboardtypes.UpdatableDashboardV2) (*dashboardtypes.DashboardV2, error) {
 	if err := updatable.Validate(); err != nil {
 		return nil, err

pkg/modules/rulestatehistory/implrulestatehistory/condition_builder.go

@@ -35,7 +35,7 @@ func (c *conditionBuilder) ConditionFor(
 	sb *sqlbuilder.SelectBuilder,
 ) ([]string, []string, error) {
 
-	// has/hasAny/hasAll/hasToken are logs-body-only; reject for rule state history.
+	// has/hasAny/hasAll/hasToken/search are logs-only functions; reject for rule state history.
 	if err := querybuilder.NewFunctionUnsupportedError(operator); err != nil {
 		return nil, nil, err
 	}

pkg/parser/filterquery/grammar/filterquery_parser.go

@@ -137,8 +137,8 @@ func filterqueryParserInit() {
 		0, 0, 191, 19, 1, 0, 0, 0, 192, 190, 1, 0, 0, 0, 193, 194, 7, 2, 0, 0,
 		194, 21, 1, 0, 0, 0, 195, 196, 7, 3, 0, 0, 196, 197, 5, 1, 0, 0, 197, 198,
 		3, 26, 13, 0, 198, 199, 5, 2, 0, 0, 199, 23, 1, 0, 0, 0, 200, 201, 5, 27,
-		0, 0, 201, 202, 5, 1, 0, 0, 202, 203, 3, 26, 13, 0, 203, 204, 5, 2, 0,
-		0, 204, 25, 1, 0, 0, 0, 205, 210, 3, 28, 14, 0, 206, 207, 5, 5, 0, 0, 207,
+		0, 0, 201, 202, 5, 1, 0, 0, 202, 203, 3, 18, 9, 0, 203, 204, 5, 2, 0, 0,
+		204, 25, 1, 0, 0, 0, 205, 210, 3, 28, 14, 0, 206, 207, 5, 5, 0, 0, 207,
 		209, 3, 28, 14, 0, 208, 206, 1, 0, 0, 0, 209, 212, 1, 0, 0, 0, 210, 208,
 		1, 0, 0, 0, 210, 211, 1, 0, 0, 0, 211, 27, 1, 0, 0, 0, 212, 210, 1, 0,
 		0, 0, 213, 217, 3, 34, 17, 0, 214, 217, 3, 32, 16, 0, 215, 217, 3, 30,
@@ -2945,7 +2945,7 @@ type ISearchCallContext interface {
 	// Getter signatures
 	SEARCH() antlr.TerminalNode
 	LPAREN() antlr.TerminalNode
-	FunctionParamList() IFunctionParamListContext
+	ValueList() IValueListContext
 	RPAREN() antlr.TerminalNode
 
 	// IsSearchCallContext differentiates from other interfaces.
@@ -2992,10 +2992,10 @@ func (s *SearchCallContext) LPAREN() antlr.TerminalNode {
 	return s.GetToken(FilterQueryParserLPAREN, 0)
 }
 
-func (s *SearchCallContext) FunctionParamList() IFunctionParamListContext {
+func (s *SearchCallContext) ValueList() IValueListContext {
 	var t antlr.RuleContext
 	for _, ctx := range s.GetChildren() {
-		if _, ok := ctx.(IFunctionParamListContext); ok {
+		if _, ok := ctx.(IValueListContext); ok {
 			t = ctx.(antlr.RuleContext)
 			break
 		}
@@ -3005,7 +3005,7 @@ func (s *SearchCallContext) FunctionParamList() IFunctionParamListContext {
 		return nil
 	}
 
-	return t.(IFunctionParamListContext)
+	return t.(IValueListContext)
 }
 
 func (s *SearchCallContext) RPAREN() antlr.TerminalNode {
@@ -3064,7 +3064,7 @@ func (p *FilterQueryParser) SearchCall() (localctx ISearchCallContext) {
 	}
 	{
 		p.SetState(202)
-		p.FunctionParamList()
+		p.ValueList()
 	}
 	{
 		p.SetState(203)

pkg/querier/builder_query.go

@@ -20,6 +20,8 @@ import (
 	"github.com/SigNoz/signoz/pkg/valuer"
 )
 
+const estimateTimeout = 5 * time.Second
+
 const traceOutsideRangeWarn = "Query %s references a trace_id that exists between %s and %s (UTC) but lies outside the selected time range; adjust the time range to see results"
 
 type builderQuery[T any] struct {
@@ -247,6 +249,10 @@ func (q *builderQuery[T]) Execute(ctx context.Context) (*qbtypes.Result, error)
 		return nil, err
 	}
 
+	if err := q.enforceEstimate(ctx, stmt); err != nil {
+		return nil, err
+	}
+
 	// Execute the query with proper context for partial value detection
 	result, err := q.executeWithContext(ctx, stmt.Query, stmt.Args)
 	if err != nil {
@@ -258,6 +264,65 @@ func (q *builderQuery[T]) Execute(ctx context.Context) (*qbtypes.Result, error)
 	return result, nil
 }
 
+// estimateRows returns the per-shard EXPLAIN ESTIMATE scan rows for a cost-guarded
+// statement. guarded=false means nothing to enforce; a non-nil error means reject.
+// Callers own the budget comparison (per-statement or cumulative).
+func (q *builderQuery[T]) estimateRows(ctx context.Context, stmt *qbtypes.Statement) (int64, bool, error) {
+	if stmt.CostGuard == nil || stmt.CostGuard.MaxScanRows <= 0 {
+		return 0, false, nil
+	}
+
+	estCtx, cancel := context.WithTimeout(ctx, estimateTimeout)
+	defer cancel()
+
+	entries, err := q.telemetryStore.Estimate(estCtx, stmt.Query, stmt.Args...)
+	if err != nil {
+		if ctx.Err() != nil {
+			return 0, true, ctx.Err()
+		}
+		if errors.Is(estCtx.Err(), context.DeadlineExceeded) {
+			return 0, true, errors.NewInvalidInputf(errors.CodeInvalidInput,
+				"This query is too broad to plan within %s", estimateTimeout).
+				WithSuggestions(costGuardSuggestions(stmt.CostGuard.Warning)...)
+		}
+		return 0, true, err
+	}
+
+	var rows int64
+	for _, e := range entries {
+		rows += e.Rows
+	}
+	return rows, true, nil
+}
+
+// enforceEstimate rejects a scan-heavy statement whose estimate exceeds its own
+// budget, before executing. Budget 0 disables.
+func (q *builderQuery[T]) enforceEstimate(ctx context.Context, stmt *qbtypes.Statement) error {
+	rows, guarded, err := q.estimateRows(ctx, stmt)
+	if err != nil {
+		return err
+	}
+	if !guarded {
+		return nil
+	}
+	if budget := stmt.CostGuard.MaxScanRows; rows > budget {
+		return errors.NewInvalidInputf(errors.CodeInvalidInput,
+			"This query would scan about %d rows per shard in this range, over the per-shard limit of %d", rows, budget).
+			WithSuggestions(costGuardSuggestions(stmt.CostGuard.Warning)...)
+	}
+	return nil
+}
+
+// costGuardSuggestions leads with the requirement's advisory (e.g. the search() hint),
+// then how to get under budget.
+func costGuardSuggestions(advisory string) []string {
+	suggestions := make([]string, 0, 2)
+	if advisory != "" {
+		suggestions = append(suggestions, advisory)
+	}
+	return append(suggestions, "Narrow the time range or add a more selective filter.")
+}
+
 // narrowWindowByTraceID inspects the filter for trace_id predicates and clamps
 // [fromMS,toMS] to the time range stored in signoz_traces.distributed_trace_summary.
 // Returns the (possibly narrowed) window, overlap=false when the trace lies
@@ -491,6 +556,10 @@ func (q *builderQuery[T]) executeWindowList(ctx context.Context) (*qbtypes.Resul
 	var warnings []string
 	var warningsDocURL string
 
+	// Cumulative across visited buckets: the budget bounds the whole query's per-shard
+	// scan, not each bucket independently.
+	var estimatedScan int64
+
 	for _, r := range buckets {
 		q.spec.Offset = 0
 		q.spec.Limit = need
@@ -501,6 +570,18 @@ func (q *builderQuery[T]) executeWindowList(ctx context.Context) (*qbtypes.Resul
 		}
 		warnings = stmt.Warnings
 		warningsDocURL = stmt.WarningsDocURL
+		rowsEst, guarded, err := q.estimateRows(ctx, stmt)
+		if err != nil {
+			return nil, err
+		}
+		if guarded {
+			estimatedScan += rowsEst
+			if budget := stmt.CostGuard.MaxScanRows; estimatedScan > budget {
+				return nil, errors.NewInvalidInputf(errors.CodeInvalidInput,
+					"This query would scan about %d rows per shard across the time range, over the per-shard limit of %d", estimatedScan, budget).
+					WithSuggestions(costGuardSuggestions(stmt.CostGuard.Warning)...)
+			}
+		}
 		// Execute with proper context for partial value detection
 		res, err := q.executeWithContext(ctx, stmt.Query, stmt.Args)
 		if err != nil {

pkg/querybuilder/constants.go

@@ -8,6 +8,9 @@ const (
 	// BodyFullTextSearchDefaultWarning is emitted when a full-text search or "body" searches are hit
 	// with New JSON Body enhancements.
 	BodyFullTextSearchDefaultWarning = "Full text searches default to `body.message:string`. Use `body.<key>` to search a different field inside body"
+
+	// SearchWarning is emitted on every search() call — it scans all fields.
+	SearchWarning = "search() runs across all fields and can be slow and expensive. Prefer a specific field, e.g. `<context>.<field_key>:<type>`"
 )
 
 var (

pkg/querybuilder/key_resolution.go

@@ -161,14 +161,16 @@ func inferDataTypesFromList(values []any) []telemetrytypes.FieldDataType {
 	return out
 }
 
-// NewFunctionUnsupportedError returns the error for a has/hasAny/hasAll/hasToken operator
-// on a builder that doesn't support it (logs body only), or nil for other operators.
+// NewFunctionUnsupportedError returns the error for a has/hasAny/hasAll/hasToken/search
+// operator on a builder that doesn't support it (logs only), or nil for other operators.
 func NewFunctionUnsupportedError(operator qbtypes.FilterOperator) error {
 	switch operator {
 	case qbtypes.FilterOperatorHasToken:
 		return errors.NewInvalidInputf(errors.CodeInvalidInput, "function `hasToken` only supports body field as first parameter").WithUrl(hasTokenFunctionDocURL)
 	case qbtypes.FilterOperatorHas, qbtypes.FilterOperatorHasAny, qbtypes.FilterOperatorHasAll:
 		return errors.NewInvalidInputf(errors.CodeInvalidInput, "function `%s` supports only body JSON search", operator.FunctionName()).WithUrl(functionBodyJSONSearchDocURL)
+	case qbtypes.FilterOperatorSearch:
+		return errors.NewInvalidInputf(errors.CodeInvalidInput, "function `search` is only supported for logs")
 	default:
 		return nil
 	}

pkg/querybuilder/where_clause_visitor.go

@@ -43,6 +43,8 @@ type filterExpressionVisitor struct {
 	keysWithWarnings map[string]bool
 	startNs          uint64
 	endNs            uint64
+
+	requiresCostGuard bool
 }
 
 type FilterExprVisitorOpts struct {
@@ -81,9 +83,10 @@ func newFilterExpressionVisitor(opts FilterExprVisitorOpts) *filterExpressionVis
 }
 
 type PreparedWhereClause struct {
-	WhereClause    *sqlbuilder.WhereClause
-	Warnings       []string
-	WarningsDocURL string
+	WhereClause       *sqlbuilder.WhereClause
+	Warnings          []string
+	WarningsDocURL    string
+	RequiresCostGuard bool
 }
 
 func (p PreparedWhereClause) IsEmpty() bool {
@@ -165,12 +168,12 @@ func PrepareWhereClause(query string, opts FilterExprVisitorOpts) (PreparedWhere
 
 	// Return empty where clause so callers can skip the WHERE clause
 	if cond == "" || cond == SkipConditionLiteral {
-		return PreparedWhereClause{WhereClause: nil, Warnings: visitor.warnings, WarningsDocURL: visitor.mainWarnURL}, nil
+		return PreparedWhereClause{WhereClause: nil, Warnings: visitor.warnings, WarningsDocURL: visitor.mainWarnURL, RequiresCostGuard: visitor.requiresCostGuard}, nil
 	}
 
 	whereClause := sqlbuilder.NewWhereClause().AddWhereExpr(visitor.builder.Args, cond)
 
-	return PreparedWhereClause{WhereClause: whereClause, Warnings: visitor.warnings, WarningsDocURL: visitor.mainWarnURL}, nil
+	return PreparedWhereClause{WhereClause: whereClause, Warnings: visitor.warnings, WarningsDocURL: visitor.mainWarnURL, RequiresCostGuard: visitor.requiresCostGuard}, nil
 }
 
 // Visit dispatches to the specific visit method based on node type.
@@ -776,11 +779,77 @@ func normalizeFunctionValue(operator qbtypes.FilterOperator, functionName string
 	return valueParams, nil
 }
 
-// VisitSearchCall handles search('needle'). The search() function is parsed but
-// not yet implemented; reject it with a clear invalid-input error.
+// VisitSearchCall handles search('term'[, body, resource, …]): a case-insensitive
+// search term plus optional field-context scopes, ORing one FilterOperatorSearch per
+// scope (no scope = keyless, covering every field).
 func (v *filterExpressionVisitor) VisitSearchCall(ctx *grammar.SearchCallContext) any {
-	v.errors = append(v.errors, "function `search` is not yet supported")
-	return ErrorConditionLiteral
+	// Flag scan-heavy so the statement builder attaches the cost guard.
+	v.requiresCostGuard = true
+
+	valueList := ctx.ValueList()
+	if valueList == nil {
+		v.errors = append(v.errors, "function `search` expects a search term, e.g. search('error')")
+		return ErrorConditionLiteral
+	}
+	params := valueList.AllValue()
+	if len(params) == 0 {
+		v.errors = append(v.errors, "function `search` expects a search term, e.g. search('error')")
+		return ErrorConditionLiteral
+	}
+
+	searchText, ok := searchParamText(params[0])
+	if !ok {
+		v.errors = append(v.errors, "function `search` expects a search term as its first argument, e.g. search('error')")
+		return ErrorConditionLiteral
+	}
+
+	var fieldContexts []telemetrytypes.FieldContext
+	if len(params) == 1 {
+		fieldContexts = []telemetrytypes.FieldContext{telemetrytypes.FieldContextUnspecified}
+	} else {
+		for _, p := range params[1:] {
+			scopeText, sok := searchParamText(p)
+			if !sok {
+				v.errors = append(v.errors, "function `search` expects each scope to be a context, e.g. search('error', body, resource)")
+				return ErrorConditionLiteral
+			}
+			fc, fok := telemetrytypes.FieldContextFromText(scopeText)
+			if !fok {
+				v.errors = append(v.errors, fmt.Sprintf("invalid search scope %q; expected a field context: body, attribute, resource, or log", scopeText))
+				return ErrorConditionLiteral
+			}
+			fieldContexts = append(fieldContexts, fc)
+		}
+	}
+
+	var conds []string
+	for _, fieldContext := range fieldContexts {
+		key := telemetrytypes.NewTelemetryFieldKey("", fieldContext, telemetrytypes.FieldDataTypeUnspecified)
+		scoped, cok := v.buildConditions(key, nil, qbtypes.FilterOperatorSearch, searchText)
+		if !cok {
+			return ErrorConditionLiteral
+		}
+		conds = append(conds, scoped...)
+	}
+	if len(conds) == 0 {
+		return SkipConditionLiteral
+	}
+	if len(conds) == 1 {
+		return conds[0]
+	}
+	return v.builder.Or(conds...)
+}
+
+// searchParamText returns an argument's raw token text (quoted or bare) rather than its
+// visited value, so a bare word stays literal and search(1000000) isn't "1e+06".
+func searchParamText(val grammar.IValueContext) (string, bool) {
+	if val == nil {
+		return "", false
+	}
+	if val.QUOTED_TEXT() != nil {
+		return trimQuotes(val.QUOTED_TEXT().GetText()), true
+	}
+	return val.GetText(), true
 }
 
 // VisitFunctionParamList handles the parameter list for function calls.

pkg/signoz/provider.go

@@ -233,6 +233,7 @@ func NewSQLMigrationProviderFactories(
 		sqlmigration.NewFillDashboardMeterSourceFactory(sqlstore, dashboardStore),
 		sqlmigration.NewUpdateRoleTransactionGroupsFactory(),
 		sqlmigration.NewFillDashboardSpecCollectionsFactory(sqlstore, dashboardStore),
+		sqlmigration.NewScrubEmailChannelTransportFactory(sqlstore),
 	)
 }
 

pkg/sqlmigration/107_scrub_email_channel_transport.go

@@ -0,0 +1,274 @@
+package sqlmigration
+
+import (
+	"context"
+	"crypto/md5"
+	"encoding/json"
+	"fmt"
+	"log/slog"
+
+	"github.com/SigNoz/signoz/pkg/errors"
+	"github.com/SigNoz/signoz/pkg/factory"
+	"github.com/SigNoz/signoz/pkg/sqlstore"
+	"github.com/uptrace/bun"
+	"github.com/uptrace/bun/migrate"
+)
+
+type scrubEmailChannelTransport struct {
+	sqlstore sqlstore.SQLStore
+	logger   *slog.Logger
+}
+
+type alertmanagerConfigScrubRow struct {
+	bun.BaseModel `bun:"table:alertmanager_config"`
+
+	ID     string `bun:"id"`
+	Config string `bun:"config"`
+}
+
+type notificationChannelScrubRow struct {
+	bun.BaseModel `bun:"table:notification_channel"`
+
+	ID   string `bun:"id"`
+	Data string `bun:"data"`
+}
+
+var emailTransportKeys = []string{
+	"from",
+	"hello",
+	"smarthost",
+	"auth_username",
+	"auth_password",
+	"auth_password_file",
+	"auth_secret",
+	"auth_secret_file",
+	"auth_identity",
+	"require_tls",
+	"tls_config",
+	"force_implicit_tls",
+}
+
+var globalSMTPKeys = []string{
+	"smtp_from",
+	"smtp_hello",
+	"smtp_smarthost",
+	"smtp_auth_username",
+	"smtp_auth_password",
+	"smtp_auth_password_file",
+	"smtp_auth_secret",
+	"smtp_auth_secret_file",
+	"smtp_auth_identity",
+	"smtp_require_tls",
+	"smtp_tls_config",
+	"smtp_force_implicit_tls",
+}
+
+func NewScrubEmailChannelTransportFactory(sqlstore sqlstore.SQLStore) factory.ProviderFactory[SQLMigration, Config] {
+	return factory.NewProviderFactory(
+		factory.MustNewName("scrub_email_channel_transport"),
+		func(ctx context.Context, ps factory.ProviderSettings, c Config) (SQLMigration, error) {
+			return &scrubEmailChannelTransport{sqlstore: sqlstore, logger: ps.Logger}, nil
+		},
+	)
+}
+
+func (migration *scrubEmailChannelTransport) Register(migrations *migrate.Migrations) error {
+	if err := migrations.Register(migration.Up, migration.Down); err != nil {
+		return err
+	}
+	return nil
+}
+
+func (migration *scrubEmailChannelTransport) Up(ctx context.Context, db *bun.DB) error {
+	tx, err := db.BeginTx(ctx, nil)
+	if err != nil {
+		return err
+	}
+
+	defer func() {
+		_ = tx.Rollback()
+	}()
+
+	if err := migration.scrubConfigs(ctx, tx); err != nil {
+		return err
+	}
+
+	if err := migration.scrubChannels(ctx, tx); err != nil {
+		return err
+	}
+
+	return tx.Commit()
+}
+
+func (migration *scrubEmailChannelTransport) scrubConfigs(ctx context.Context, tx bun.Tx) error {
+	rows := make([]*alertmanagerConfigScrubRow, 0)
+	if err := tx.NewSelect().Model(&rows).Scan(ctx); err != nil {
+		return err
+	}
+
+	for _, row := range rows {
+		cfg := make(map[string]json.RawMessage)
+		if err := json.Unmarshal([]byte(row.Config), &cfg); err != nil {
+			migration.logger.WarnContext(ctx, "skipping alertmanager config with unreadable config", slog.String("config_id", row.ID), errors.Attr(err))
+			continue
+		}
+
+		changed := false
+
+		if globalRaw, ok := cfg["global"]; ok && string(globalRaw) != "null" {
+			global := make(map[string]json.RawMessage)
+			if err := json.Unmarshal(globalRaw, &global); err != nil {
+				migration.logger.WarnContext(ctx, "skipping alertmanager config with unreadable global", slog.String("config_id", row.ID), errors.Attr(err))
+				continue
+			}
+			if deleteKeys(global, globalSMTPKeys) {
+				newGlobal, err := json.Marshal(global)
+				if err != nil {
+					migration.logger.WarnContext(ctx, "skipping alertmanager config, cannot marshal global", slog.String("config_id", row.ID), errors.Attr(err))
+					continue
+				}
+				cfg["global"] = newGlobal
+				changed = true
+			}
+		}
+
+		if receiversRaw, ok := cfg["receivers"]; ok && string(receiversRaw) != "null" {
+			receivers := make([]map[string]json.RawMessage, 0)
+			if err := json.Unmarshal(receiversRaw, &receivers); err != nil {
+				migration.logger.WarnContext(ctx, "skipping alertmanager config with unreadable receivers", slog.String("config_id", row.ID), errors.Attr(err))
+				continue
+			}
+
+			receiversChanged := false
+			unreadable := false
+			for _, receiver := range receivers {
+				scrubbed, err := scrubEmailConfigs(receiver)
+				if err != nil {
+					migration.logger.WarnContext(ctx, "skipping alertmanager config with unreadable email configs", slog.String("config_id", row.ID), errors.Attr(err))
+					unreadable = true
+					break
+				}
+				receiversChanged = receiversChanged || scrubbed
+			}
+			if unreadable {
+				continue
+			}
+
+			if receiversChanged {
+				newReceivers, err := json.Marshal(receivers)
+				if err != nil {
+					migration.logger.WarnContext(ctx, "skipping alertmanager config, cannot marshal receivers", slog.String("config_id", row.ID), errors.Attr(err))
+					continue
+				}
+				cfg["receivers"] = newReceivers
+				changed = true
+			}
+		}
+
+		if !changed {
+			continue
+		}
+
+		newConfig, err := json.Marshal(cfg)
+		if err != nil {
+			migration.logger.WarnContext(ctx, "skipping alertmanager config, cannot marshal config", slog.String("config_id", row.ID), errors.Attr(err))
+			continue
+		}
+
+		if _, err := tx.NewUpdate().
+			Model((*alertmanagerConfigScrubRow)(nil)).
+			Set("config = ?", string(newConfig)).
+			Set("hash = ?", fmt.Sprintf("%x", md5.Sum(newConfig))).
+			Where("id = ?", row.ID).
+			Exec(ctx); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+func (migration *scrubEmailChannelTransport) scrubChannels(ctx context.Context, tx bun.Tx) error {
+	rows := make([]*notificationChannelScrubRow, 0)
+	if err := tx.NewSelect().Model(&rows).Where("type = ?", "email").Scan(ctx); err != nil {
+		return err
+	}
+
+	for _, row := range rows {
+		receiver := make(map[string]json.RawMessage)
+		if err := json.Unmarshal([]byte(row.Data), &receiver); err != nil {
+			migration.logger.WarnContext(ctx, "skipping notification channel with unreadable data", slog.String("channel_id", row.ID), errors.Attr(err))
+			continue
+		}
+
+		scrubbed, err := scrubEmailConfigs(receiver)
+		if err != nil {
+			migration.logger.WarnContext(ctx, "skipping notification channel with unreadable email configs", slog.String("channel_id", row.ID), errors.Attr(err))
+			continue
+		}
+		if !scrubbed {
+			continue
+		}
+
+		newData, err := json.Marshal(receiver)
+		if err != nil {
+			migration.logger.WarnContext(ctx, "skipping notification channel, cannot marshal data", slog.String("channel_id", row.ID), errors.Attr(err))
+			continue
+		}
+
+		if _, err := tx.NewUpdate().
+			Model((*notificationChannelScrubRow)(nil)).
+			Set("data = ?", string(newData)).
+			Where("id = ?", row.ID).
+			Exec(ctx); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+func scrubEmailConfigs(receiver map[string]json.RawMessage) (bool, error) {
+	emailConfigsRaw, ok := receiver["email_configs"]
+	if !ok || string(emailConfigsRaw) == "null" {
+		return false, nil
+	}
+
+	emailConfigs := make([]map[string]json.RawMessage, 0)
+	if err := json.Unmarshal(emailConfigsRaw, &emailConfigs); err != nil {
+		return false, err
+	}
+
+	changed := false
+	for _, emailConfig := range emailConfigs {
+		changed = deleteKeys(emailConfig, emailTransportKeys) || changed
+	}
+
+	if !changed {
+		return false, nil
+	}
+
+	newEmailConfigs, err := json.Marshal(emailConfigs)
+	if err != nil {
+		return false, err
+	}
+	receiver["email_configs"] = newEmailConfigs
+
+	return true, nil
+}
+
+func deleteKeys(m map[string]json.RawMessage, keys []string) bool {
+	deleted := false
+	for _, key := range keys {
+		if _, ok := m[key]; ok {
+			delete(m, key)
+			deleted = true
+		}
+	}
+
+	return deleted
+}
+
+func (migration *scrubEmailChannelTransport) Down(context.Context, *bun.DB) error {
+	return nil
+}

pkg/statementbuilder/config.go

@@ -15,6 +15,12 @@ type SkipResourceFingerprint struct {
 type Config struct {
 	// SkipResourceFingerprint configures when the resource fingerprint subquery is skipped in favor of main-table filtering.
 	SkipResourceFingerprint SkipResourceFingerprint `yaml:"skip_resource_fingerprint" mapstructure:"skip_resource_fingerprint"`
+	// SearchMaxScanRows caps per-shard rows a search() may scan, enforced by the querier
+	// via EXPLAIN ESTIMATE (0 disables).
+	SearchMaxScanRows int64 `yaml:"search_max_scan_rows" mapstructure:"search_max_scan_rows"`
+	// SearchMaxScanRowsJSONBody is the same budget for body_v2, where each row costs far
+	// more: toString() rebuilds every document and no skip index prunes (0 disables).
+	SearchMaxScanRowsJSONBody int64 `yaml:"search_max_scan_rows_json_body" mapstructure:"search_max_scan_rows_json_body"`
 }
 
 func NewConfig() Config {
@@ -23,6 +29,8 @@ func NewConfig() Config {
 			Enabled:   false,
 			Threshold: 100000,
 		},
+		SearchMaxScanRows:         60_000_000,
+		SearchMaxScanRowsJSONBody: 6_000_000,
 	}
 }
 
@@ -31,5 +39,11 @@ func (c Config) Validate() error {
 	if c.SkipResourceFingerprint.Enabled && c.SkipResourceFingerprint.Threshold == 0 {
 		return errors.NewInvalidInputf(errors.CodeInvalidInput, "skip_resource_fingerprint.threshold must be > 0 when enabled")
 	}
+	if c.SearchMaxScanRows < 0 {
+		return errors.NewInvalidInputf(errors.CodeInvalidInput, "search_max_scan_rows must not be negative, got %v", c.SearchMaxScanRows)
+	}
+	if c.SearchMaxScanRowsJSONBody < 0 {
+		return errors.NewInvalidInputf(errors.CodeInvalidInput, "search_max_scan_rows_json_body must not be negative, got %v", c.SearchMaxScanRowsJSONBody)
+	}
 	return nil
 }

pkg/statementbuilder/logsstatementbuilder/search_cost_guard_test.go

@@ -0,0 +1,85 @@
+package logsstatementbuilder
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/SigNoz/signoz/pkg/flagger/flaggertest"
+	"github.com/SigNoz/signoz/pkg/instrumentation/instrumentationtest"
+	"github.com/SigNoz/signoz/pkg/querybuilder"
+	"github.com/SigNoz/signoz/pkg/statementbuilder"
+	"github.com/SigNoz/signoz/pkg/telemetryschema/logstelemetryschema"
+	qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
+	"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
+	"github.com/SigNoz/signoz/pkg/types/telemetrytypes/telemetrytypestest"
+	"github.com/SigNoz/signoz/pkg/valuer"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// TestSearchCostGuard asserts Build attaches the CostGuard budget and its advisory.
+func TestSearchCostGuard(t *testing.T) {
+	releaseTime := time.Date(2024, 1, 15, 10, 0, 0, 0, time.UTC)
+	ctx := context.Background()
+	start := uint64(releaseTime.Add(-5 * time.Minute).UnixMilli())
+	end := uint64(releaseTime.UnixMilli())
+
+	fl := flaggertest.WithBooleanFlags(t, map[string]bool{})
+	fm := logstelemetryschema.NewFieldMapper(fl)
+	cb := logstelemetryschema.NewConditionBuilder(fm, fl)
+	store := telemetrytypestest.NewMockMetadataStore()
+	store.KeysMap = logstelemetryschema.BuildCompleteFieldKeyMap(releaseTime)
+	rewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, fm, cb, fl)
+	sb := NewLogQueryStatementBuilder(
+		instrumentationtest.New().ToProviderSettings(),
+		store, fm, cb, rewriter, logstelemetryschema.DefaultFullTextColumn, fl, nil,
+		statementbuilder.Config{SearchMaxScanRows: 100000, SkipResourceFingerprint: statementbuilder.SkipResourceFingerprint{Enabled: false, Threshold: 100000}},
+	)
+	query := qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]{
+		Signal: telemetrytypes.SignalLogs,
+		Filter: &qbtypes.Filter{Expression: "search('error')"},
+		Limit:  1,
+	}
+
+	stmt, err := sb.Build(ctx, valuer.UUID{}, start, end, qbtypes.RequestTypeRaw, query, nil)
+	require.NoError(t, err)
+	require.NotNil(t, stmt.CostGuard)
+	assert.Equal(t, int64(100000), stmt.CostGuard.MaxScanRows)
+	assert.Contains(t, stmt.Warnings, querybuilder.SearchWarning)
+}
+
+// TestSearchCostGuardJSONBody asserts body_v2 gets its own, lower budget.
+func TestSearchCostGuardJSONBody(t *testing.T) {
+	releaseTime := time.Date(2024, 1, 15, 10, 0, 0, 0, time.UTC)
+	ctx := context.Background()
+	start := uint64(releaseTime.Add(-5 * time.Minute).UnixMilli())
+	end := uint64(releaseTime.UnixMilli())
+
+	fl := flaggertest.WithUseJSONBody(t, true)
+	fm := logstelemetryschema.NewFieldMapper(fl)
+	cb := logstelemetryschema.NewConditionBuilder(fm, fl)
+	store := telemetrytypestest.NewMockMetadataStore()
+	store.KeysMap = logstelemetryschema.BuildCompleteFieldKeyMap(releaseTime)
+	rewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, fm, cb, fl)
+	sb := NewLogQueryStatementBuilder(
+		instrumentationtest.New().ToProviderSettings(),
+		store, fm, cb, rewriter, logstelemetryschema.DefaultFullTextColumn, fl, nil,
+		statementbuilder.Config{
+			SearchMaxScanRows:         100000,
+			SearchMaxScanRowsJSONBody: 10000,
+			SkipResourceFingerprint:   statementbuilder.SkipResourceFingerprint{Enabled: false, Threshold: 100000},
+		},
+	)
+	query := qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]{
+		Signal: telemetrytypes.SignalLogs,
+		Filter: &qbtypes.Filter{Expression: "search('error')"},
+		Limit:  1,
+	}
+
+	stmt, err := sb.Build(ctx, valuer.UUID{}, start, end, qbtypes.RequestTypeRaw, query, nil)
+	require.NoError(t, err)
+	require.NotNil(t, stmt.CostGuard)
+	assert.Equal(t, int64(10000), stmt.CostGuard.MaxScanRows)
+	assert.Contains(t, stmt.Warnings, querybuilder.SearchWarning)
+}

pkg/statementbuilder/logsstatementbuilder/statement_builder.go

@@ -41,14 +41,16 @@ type logQueryStatementBuilder struct {
 	fl                             flagger.Flagger
 	skipResourceFingerprintEnabled bool
 
-	fullTextColumn *telemetrytypes.TelemetryFieldKey
+	fullTextColumn            *telemetrytypes.TelemetryFieldKey
+	searchMaxScanRows         int64
+	searchMaxScanRowsJSONBody int64
 }
 
 var _ qbtypes.StatementBuilder[qbtypes.LogAggregation] = (*logQueryStatementBuilder)(nil)
 
 // NewFactory returns a provider factory for the logs statement builder. Its New
 // internalizes the FieldMapper, ConditionBuilder, and AggExprRewriter, and reads
-// SkipResourceFingerprint from the config.
+// SkipResourceFingerprint and the search() scan budgets from the config.
 func NewFactory(
 	telemetryStore telemetrystore.TelemetryStore,
 	metadataStore telemetrytypes.MetadataStore,
@@ -94,7 +96,7 @@ func NewLogQueryStatementBuilder(
 		cfg.SkipResourceFingerprint.Threshold,
 	)
 
-	return &logQueryStatementBuilder{
+	b := &logQueryStatementBuilder{
 		logger:                         logsSettings.Logger(),
 		metadataStore:                  metadataStore,
 		fm:                             fieldMapper,
@@ -103,8 +105,11 @@ func NewLogQueryStatementBuilder(
 		aggExprRewriter:                aggExprRewriter,
 		fl:                             fl,
 		skipResourceFingerprintEnabled: cfg.SkipResourceFingerprint.Enabled,
+		searchMaxScanRows:              cfg.SearchMaxScanRows,
+		searchMaxScanRowsJSONBody:      cfg.SearchMaxScanRowsJSONBody,
 		fullTextColumn:                 fullTextColumn,
 	}
+	return b
 }
 
 // Build builds a SQL query for logs based on the given parameters.
@@ -150,9 +155,26 @@ func (b *logQueryStatementBuilder) Build(
 	}
 
 	stmt.Warnings = append(stmt.Warnings, warnings...)
+	// Surface the guard's advisory to the user alongside the other warnings.
+	if stmt.CostGuard != nil && stmt.CostGuard.Warning != "" {
+		stmt.Warnings = append(stmt.Warnings, stmt.CostGuard.Warning)
+	}
 	return stmt, nil
 }
 
+// costGuardFor pairs the search() advisory with the budget for the body path taken —
+// body_v2 has its own, lower one. nil when the statement needs no guard.
+func (b *logQueryStatementBuilder) costGuardFor(ctx context.Context, orgID valuer.UUID, required bool) *qbtypes.CostGuard {
+	if !required {
+		return nil
+	}
+	maxScanRows := b.searchMaxScanRows
+	if b.fl.BooleanOrEmpty(ctx, flagger.FeatureUseJSONBody, featuretypes.NewFlaggerEvaluationContext(orgID)) {
+		maxScanRows = b.searchMaxScanRowsJSONBody
+	}
+	return &qbtypes.CostGuard{Warning: querybuilder.SearchWarning, MaxScanRows: maxScanRows}
+}
+
 func getKeySelectors(query qbtypes.QueryBuilderQuery[qbtypes.LogAggregation], bodyJSONEnabled bool) ([]*telemetrytypes.FieldKeySelector, []string) {
 	var keySelectors []*telemetrytypes.FieldKeySelector
 	var warnings []string
@@ -389,6 +411,7 @@ func (b *logQueryStatementBuilder) buildListQuery(
 		Args:           finalArgs,
 		Warnings:       preparedWhereClause.Warnings,
 		WarningsDocURL: preparedWhereClause.WarningsDocURL,
+		CostGuard:      b.costGuardFor(ctx, orgID, preparedWhereClause.RequiresCostGuard),
 	}
 
 	return stmt, nil
@@ -555,6 +578,7 @@ func (b *logQueryStatementBuilder) buildTimeSeriesQuery(
 		Args:           finalArgs,
 		Warnings:       preparedWhereClause.Warnings,
 		WarningsDocURL: preparedWhereClause.WarningsDocURL,
+		CostGuard:      b.costGuardFor(ctx, orgID, preparedWhereClause.RequiresCostGuard),
 	}
 
 	return stmt, nil
@@ -682,6 +706,7 @@ func (b *logQueryStatementBuilder) buildScalarQuery(
 		Args:           finalArgs,
 		Warnings:       preparedWhereClause.Warnings,
 		WarningsDocURL: preparedWhereClause.WarningsDocURL,
+		CostGuard:      b.costGuardFor(ctx, orgID, preparedWhereClause.RequiresCostGuard),
 	}
 
 	return stmt, nil

pkg/telemetryschema/audittelemetryschema/condition_builder.go

@@ -134,7 +134,7 @@ func (c *conditionBuilder) ConditionFor(
 	sb *sqlbuilder.SelectBuilder,
 ) ([]string, []string, error) {
 
-	// has/hasAny/hasAll/hasToken are logs-body-only; reject for audit.
+	// has/hasAny/hasAll/hasToken/search are logs-only functions; reject for audit.
 	if err := querybuilder.NewFunctionUnsupportedError(operator); err != nil {
 		return nil, nil, err
 	}

pkg/telemetryschema/logstelemetryschema/condition_builder.go

@@ -3,6 +3,7 @@ package logstelemetryschema
 import (
 	"context"
 	"fmt"
+	"regexp"
 
 	schema "github.com/SigNoz/signoz-otel-collector/cmd/signozschemamigrator/schema_migrator"
 	"github.com/SigNoz/signoz/pkg/errors"
@@ -27,6 +28,52 @@ func NewConditionBuilder(fm qbtypes.FieldMapper, fl flagger.Flagger) *conditionB
 	return &conditionBuilder{fm: fm, fl: fl}
 }
 
+// conditionForSearch ORs a case-insensitive match of the search term across the key
+// context's searchable columns (unspecified context = every column).
+func (c *conditionBuilder) conditionForSearch(
+	ctx context.Context,
+	orgID valuer.UUID,
+	key *telemetrytypes.TelemetryFieldKey,
+	value any,
+	sb *sqlbuilder.SelectBuilder,
+) ([]string, []string, error) {
+	// QuoteMeta + LOWER on both sides, not (?i): a literal match that can still use the
+	// LOWER(toString(body_v2)) skip index.
+	term := regexp.QuoteMeta(fmt.Sprintf("%v", value))
+
+	useJSONBody := c.fl.BooleanOrEmpty(ctx, flagger.FeatureUseJSONBody, featuretypes.NewFlaggerEvaluationContext(orgID))
+
+	var conditions []string
+
+	for _, col := range searchColumns(key.FieldContext, useJSONBody) {
+		switch col.Type.GetType() {
+		case schema.ColumnTypeEnumMap:
+			keysExpr := fmt.Sprintf("mapKeys(%s)", col.Name)
+			valsExpr := fmt.Sprintf("mapValues(%s)", col.Name)
+			// match() needs a String array; cast non-string map values first.
+			if mc, ok := col.Type.(schema.MapColumnType); ok && mc.ValueType.GetType() != schema.ColumnTypeEnumString {
+				valsExpr = fmt.Sprintf("arrayMap(x -> toString(x), mapValues(%s))", col.Name)
+			}
+			conditions = append(conditions, sb.Or(
+				fmt.Sprintf("arrayExists(x -> match(LOWER(x), LOWER(%s)), %s)", sb.Var(term), keysExpr),
+				fmt.Sprintf("arrayExists(x -> match(LOWER(x), LOWER(%s)), %s)", sb.Var(term), valsExpr),
+			))
+		case schema.ColumnTypeEnumJSON:
+			conditions = append(conditions, fmt.Sprintf("match(LOWER(toString(%s)), LOWER(%s))", col.Name, sb.Var(term)))
+		case schema.ColumnTypeEnumString, schema.ColumnTypeEnumLowCardinality:
+			conditions = append(conditions, fmt.Sprintf("match(LOWER(%s), LOWER(%s))", col.Name, sb.Var(term)))
+		default:
+			return nil, nil, errors.NewInternalf(errors.CodeInternal, "search does not support the column type of %q", col.Name)
+		}
+	}
+
+	if len(conditions) == 0 {
+		return nil, nil, nil
+	}
+	// The advisory rides on CostGuard (set by the visitor), not warnings.
+	return []string{sb.Or(conditions...)}, nil, nil
+}
+
 // isBodyJSONSearch reports whether a key addresses a path within the body JSON. Only
 // an explicit Body context qualifies; a bare, context-less `body` (e.g. full-text
 // `count_distinct(body)` or `body EXISTS`) is a full-text match, not a `$.body` path.
@@ -408,6 +455,11 @@ func (c *conditionBuilder) ConditionFor(
 	matches := querybuilder.MatchingFieldKeys(key, fieldKeys)
 	skipResourceFilter := options.SkipResourceFilter
 
+	// search() resolves its own (optional) scope; handle it before key resolution.
+	if operator == qbtypes.FilterOperatorSearch {
+		return c.conditionForSearch(ctx, orgID, key, value, sb)
+	}
+
 	keys, warning := querybuilder.ResolveKeys(key, matches)
 	var warnings []string
 	if warning != "" {

pkg/telemetryschema/logstelemetryschema/field_mapper.go

@@ -635,3 +635,37 @@ func (m *fieldMapper) existsExpressionFor(
 	}
 	return querybuilder.ExistsExpression(columns, key, tsStart, tsEnd, fieldExpression, exists)
 }
+
+// searchColumns is the single source of truth for the columns search() fans out across,
+// by context; body is body_v2 when useJSONBody, else the legacy body string.
+func searchColumns(fieldContext telemetrytypes.FieldContext, useJSONBody bool) []*schema.Column {
+	switch fieldContext {
+	case telemetrytypes.FieldContextLog:
+		return []*schema.Column{
+			logsV2Columns[LogsV2SeverityTextColumn],
+			logsV2Columns[LogsV2TraceIDColumn],
+			logsV2Columns[LogsV2SpanIDColumn],
+		}
+	case telemetrytypes.FieldContextBody:
+		if useJSONBody {
+			return []*schema.Column{logsV2Columns[LogsV2BodyV2Column]}
+		}
+		return []*schema.Column{logsV2Columns[LogsV2BodyColumn]}
+	case telemetrytypes.FieldContextAttribute:
+		return []*schema.Column{
+			logsV2Columns[LogsV2AttributesStringColumn],
+			logsV2Columns[LogsV2AttributesNumberColumn],
+			logsV2Columns[LogsV2AttributesBoolColumn],
+		}
+	case telemetrytypes.FieldContextResource:
+		return []*schema.Column{
+			logsV2Columns[LogsV2ResourcesStringColumn],
+		}
+	default:
+		columns := searchColumns(telemetrytypes.FieldContextLog, useJSONBody)
+		columns = append(columns, searchColumns(telemetrytypes.FieldContextBody, useJSONBody)...)
+		columns = append(columns, searchColumns(telemetrytypes.FieldContextAttribute, useJSONBody)...)
+		columns = append(columns, searchColumns(telemetrytypes.FieldContextResource, useJSONBody)...)
+		return columns
+	}
+}

pkg/telemetryschema/logstelemetryschema/filter_expr_search_test.go

@@ -0,0 +1,284 @@
+package logstelemetryschema
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/SigNoz/signoz/pkg/flagger"
+	"github.com/SigNoz/signoz/pkg/flagger/flaggertest"
+	"github.com/SigNoz/signoz/pkg/instrumentation/instrumentationtest"
+	"github.com/SigNoz/signoz/pkg/querybuilder"
+	"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
+	"github.com/huandu/go-sqlbuilder"
+	"github.com/stretchr/testify/require"
+)
+
+// searchFanOut returns the WHERE fragment search() fans out to; bodyExpr differs
+// between the legacy string body and the body_v2 JSON column.
+func searchFanOut(bodyExpr string) string {
+	return "(match(LOWER(severity_text), LOWER(?)) OR match(LOWER(trace_id), LOWER(?)) OR match(LOWER(span_id), LOWER(?)) OR " +
+		bodyExpr + " OR " +
+		"(arrayExists(x -> match(LOWER(x), LOWER(?)), mapKeys(attributes_string)) OR arrayExists(x -> match(LOWER(x), LOWER(?)), mapValues(attributes_string))) OR " +
+		"(arrayExists(x -> match(LOWER(x), LOWER(?)), mapKeys(attributes_number)) OR arrayExists(x -> match(LOWER(x), LOWER(?)), arrayMap(x -> toString(x), mapValues(attributes_number)))) OR " +
+		"(arrayExists(x -> match(LOWER(x), LOWER(?)), mapKeys(attributes_bool)) OR arrayExists(x -> match(LOWER(x), LOWER(?)), arrayMap(x -> toString(x), mapValues(attributes_bool)))) OR " +
+		"(arrayExists(x -> match(LOWER(x), LOWER(?)), mapKeys(resources_string)) OR arrayExists(x -> match(LOWER(x), LOWER(?)), mapValues(resources_string))))"
+}
+
+// searchArgs returns v once per bound parameter search() emits — one per searchable
+// column expression (currently 12).
+func searchArgs(v any) []any {
+	const searchColumnParams = 12
+	args := make([]any, searchColumnParams)
+	for i := range args {
+		args[i] = v
+	}
+	return args
+}
+
+// TestFilterExprSearch covers search('term') fanning out across every searchable
+// column via FilterOperatorSearch.
+func TestFilterExprSearch(t *testing.T) {
+	releaseTime := time.Date(2024, 1, 15, 10, 0, 0, 0, time.UTC)
+	inWindowStart := uint64(releaseTime.Add(-5 * time.Minute).UnixNano())
+	inWindowEnd := uint64(releaseTime.Add(5 * time.Minute).UnixNano())
+
+	legacyBody := "match(LOWER(body), LOWER(?))"
+	jsonBody := "match(LOWER(toString(body_v2)), LOWER(?))"
+
+	// Single-context scope fragments (the fan-out narrowed to one context).
+	logScope := "(match(LOWER(severity_text), LOWER(?)) OR match(LOWER(trace_id), LOWER(?)) OR match(LOWER(span_id), LOWER(?)))"
+	resourceScope := "(arrayExists(x -> match(LOWER(x), LOWER(?)), mapKeys(resources_string)) OR arrayExists(x -> match(LOWER(x), LOWER(?)), mapValues(resources_string)))"
+
+	serviceNameEq := "(multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, mapContains(resources_string, 'service.name'), resources_string['service.name'], NULL) = ? " +
+		"AND multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, mapContains(resources_string, 'service.name'), resources_string['service.name'], NULL) IS NOT NULL)"
+
+	testCases := []struct {
+		name                  string
+		query                 string
+		jsonBodyEnabled       bool
+		fullTextColumn        *telemetrytypes.TelemetryFieldKey
+		startNs               uint64
+		endNs                 uint64
+		shouldPass            bool
+		expectedQuery         string
+		expectedArgs          []any
+		expectWarning         bool
+		expectedErrorContains string
+	}{
+		{
+			name:           "quoted, legacy body",
+			query:          "search('error')",
+			fullTextColumn: DefaultFullTextColumn,
+			startNs:        inWindowStart,
+			endNs:          inWindowEnd,
+			shouldPass:     true,
+			expectedQuery:  "WHERE " + searchFanOut(legacyBody),
+			expectedArgs:   searchArgs("error"),
+			expectWarning:  true,
+		},
+		{
+			name:            "quoted, json body",
+			query:           "search('error')",
+			jsonBodyEnabled: true,
+			fullTextColumn:  DefaultFullTextColumn,
+			startNs:         inWindowStart,
+			endNs:           inWindowEnd,
+			shouldPass:      true,
+			expectedQuery:   "WHERE " + searchFanOut(jsonBody),
+			expectedArgs:    searchArgs("error"),
+			expectWarning:   true,
+		},
+		{
+			name:           "bare word",
+			query:          "search(timeout)",
+			fullTextColumn: DefaultFullTextColumn,
+			startNs:        inWindowStart,
+			endNs:          inWindowEnd,
+			shouldPass:     true,
+			expectedQuery:  "WHERE " + searchFanOut(legacyBody),
+			expectedArgs:   searchArgs("timeout"),
+			expectWarning:  true,
+		},
+		{
+			name:           "negated",
+			query:          "NOT search('error')",
+			fullTextColumn: DefaultFullTextColumn,
+			startNs:        inWindowStart,
+			endNs:          inWindowEnd,
+			shouldPass:     true,
+			expectedQuery:  "WHERE NOT (" + searchFanOut(legacyBody) + ")",
+			expectedArgs:   searchArgs("error"),
+			expectWarning:  true,
+		},
+		{
+			name:           "combined with field filter",
+			query:          "search('error') AND service.name=\"api\"",
+			fullTextColumn: DefaultFullTextColumn,
+			startNs:        inWindowStart,
+			endNs:          inWindowEnd,
+			shouldPass:     true,
+			expectedQuery:  "WHERE (" + searchFanOut(legacyBody) + " AND " + serviceNameEq + ")",
+			expectedArgs:   append(searchArgs("error"), "api"),
+			expectWarning:  true,
+		},
+		{
+			// The builder caps no window; the querier's estimate gate bounds scan cost.
+			name:           "wide window builds (estimate gate lives in querier)",
+			query:          "search('error')",
+			fullTextColumn: DefaultFullTextColumn,
+			startNs:        uint64(releaseTime.Add(-10 * time.Hour).UnixNano()),
+			endNs:          inWindowEnd,
+			shouldPass:     true,
+			expectedQuery:  "WHERE " + searchFanOut(legacyBody),
+			expectedArgs:   searchArgs("error"),
+			expectWarning:  true,
+		},
+		{
+			// fullTextColumn governs only bare/quoted free text, so search() must
+			// work when it is unset.
+			name:           "independent of full text column",
+			query:          "search('error')",
+			fullTextColumn: nil,
+			startNs:        inWindowStart,
+			endNs:          inWindowEnd,
+			shouldPass:     true,
+			expectedQuery:  "WHERE " + searchFanOut(legacyBody),
+			expectedArgs:   searchArgs("error"),
+			expectWarning:  true,
+		},
+		{
+			// The bare word is the literal search term; Normalize would strip "resource.".
+			name:           "bare word with context prefix is not normalized",
+			query:          "search(resource.deployment)",
+			fullTextColumn: DefaultFullTextColumn,
+			startNs:        inWindowStart,
+			endNs:          inWindowEnd,
+			shouldPass:     true,
+			expectedQuery:  "WHERE " + searchFanOut(legacyBody),
+			expectedArgs:   searchArgs("resource\\.deployment"),
+			expectWarning:  true,
+		},
+		{
+			// Literal digits, not %v of a parsed float64 (which would scan "1e+06").
+			name:           "numeric search term is not scientific notation",
+			query:          "search(1000000)",
+			fullTextColumn: DefaultFullTextColumn,
+			startNs:        inWindowStart,
+			endNs:          inWindowEnd,
+			shouldPass:     true,
+			expectedQuery:  "WHERE " + searchFanOut(legacyBody),
+			expectedArgs:   searchArgs("1000000"),
+			expectWarning:  true,
+		},
+		{
+			name:           "scoped to body, legacy",
+			query:          "search('error', body)",
+			fullTextColumn: DefaultFullTextColumn,
+			startNs:        inWindowStart,
+			endNs:          inWindowEnd,
+			shouldPass:     true,
+			expectedQuery:  "WHERE (" + legacyBody + ")",
+			expectedArgs:   []any{"error"},
+			expectWarning:  true,
+		},
+		{
+			name:            "scoped to body, json",
+			query:           "search('error', body)",
+			jsonBodyEnabled: true,
+			fullTextColumn:  DefaultFullTextColumn,
+			startNs:         inWindowStart,
+			endNs:           inWindowEnd,
+			shouldPass:      true,
+			expectedQuery:   "WHERE (" + jsonBody + ")",
+			expectedArgs:    []any{"error"},
+			expectWarning:   true,
+		},
+		{
+			name:           "scoped to resource (quoted scope)",
+			query:          "search('error', 'resource')",
+			fullTextColumn: DefaultFullTextColumn,
+			startNs:        inWindowStart,
+			endNs:          inWindowEnd,
+			shouldPass:     true,
+			expectedQuery:  "WHERE (" + resourceScope + ")",
+			expectedArgs:   []any{"error", "error"},
+			expectWarning:  true,
+		},
+		{
+			name:           "scoped to log fields",
+			query:          "search('error', log)",
+			fullTextColumn: DefaultFullTextColumn,
+			startNs:        inWindowStart,
+			endNs:          inWindowEnd,
+			shouldPass:     true,
+			expectedQuery:  "WHERE " + logScope,
+			expectedArgs:   []any{"error", "error", "error"},
+			expectWarning:  true,
+		},
+		{
+			name:           "scoped to multiple contexts",
+			query:          "search('error', body, resource)",
+			fullTextColumn: DefaultFullTextColumn,
+			startNs:        inWindowStart,
+			endNs:          inWindowEnd,
+			shouldPass:     true,
+			expectedQuery:  "WHERE ((" + legacyBody + ") OR (" + resourceScope + "))",
+			expectedArgs:   []any{"error", "error", "error"},
+			expectWarning:  true,
+		},
+		{
+			name:                  "invalid scope",
+			query:                 "search('error', 'timeout')",
+			fullTextColumn:        DefaultFullTextColumn,
+			startNs:               inWindowStart,
+			endNs:                 inWindowEnd,
+			shouldPass:            false,
+			expectedErrorContains: "invalid search scope",
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			fl := flaggertest.WithBooleanFlags(t, map[string]bool{
+				flagger.FeatureUseJSONBody.String(): tc.jsonBodyEnabled,
+			})
+			fm := NewFieldMapper(fl)
+			cb := NewConditionBuilder(fm, fl)
+			keys := BuildCompleteFieldKeyMap(releaseTime)
+
+			opts := querybuilder.FilterExprVisitorOpts{
+				Context:          context.Background(),
+				Logger:           instrumentationtest.New().Logger(),
+				FieldMapper:      fm,
+				ConditionBuilder: cb,
+				FieldKeys:        keys,
+				FullTextColumn:   tc.fullTextColumn,
+				StartNs:          tc.startNs,
+				EndNs:            tc.endNs,
+			}
+
+			clause, err := querybuilder.PrepareWhereClause(tc.query, opts)
+
+			if !tc.shouldPass {
+				require.Error(t, err)
+				require.True(t, detailContains(err, tc.expectedErrorContains),
+					"error %v should contain %q", err, tc.expectedErrorContains)
+				return
+			}
+
+			require.NoError(t, err)
+			require.False(t, clause.IsEmpty())
+
+			sql, args := clause.WhereClause.BuildWithFlavor(sqlbuilder.ClickHouse)
+			require.Equal(t, tc.expectedQuery, sql)
+			require.Equal(t, tc.expectedArgs, args)
+
+			if tc.expectWarning {
+				// The visitor only flags the guard; the statement builder
+				// materializes it from config.
+				require.True(t, clause.RequiresCostGuard)
+			}
+		})
+	}
+}

pkg/telemetryschema/metricstelemetryschema/condition_builder.go

@@ -157,7 +157,7 @@ func (c *conditionBuilder) ConditionFor(
 	sb *sqlbuilder.SelectBuilder,
 ) ([]string, []string, error) {
 
-	// has/hasAny/hasAll/hasToken are logs-body-only; reject for metrics.
+	// has/hasAny/hasAll/hasToken/search are logs-only functions; reject for metrics.
 	if err := querybuilder.NewFunctionUnsupportedError(operator); err != nil {
 		return nil, nil, err
 	}

pkg/telemetryschema/tracestelemetryschema/condition_builder.go

@@ -205,7 +205,7 @@ func (c *conditionBuilder) ConditionFor(
 	sb *sqlbuilder.SelectBuilder,
 ) ([]string, []string, error) {
 
-	// has/hasAny/hasAll/hasToken are logs-body-only; reject for traces.
+	// has/hasAny/hasAll/hasToken/search are logs-only functions; reject for traces.
 	if err := querybuilder.NewFunctionUnsupportedError(operator); err != nil {
 		return nil, nil, err
 	}

pkg/types/alertmanagertypes/channel_test.go

@@ -35,12 +35,8 @@ func TestNewConfigFromChannels(t *testing.T) {
 					"email_configs": []any{map[string]any{
 						"send_resolved": false,
 						"to":            "test@example.com",
-						"from":          "alerts@example.com",
-						"hello":         "localhost",
-						"smarthost":     "smtp.example.com:587",
-						"require_tls":   true,
+						"smarthost":     "",
 						"html":          "{{ template \"email.default.html\" . }}",
-						"tls_config":    map[string]any{"insecure_skip_verify": false},
 						"threading":     map[string]any{},
 					}},
 				},
@@ -63,7 +59,6 @@ func TestNewConfigFromChannels(t *testing.T) {
 					"slack_configs": []any{map[string]any{
 						"send_resolved": true,
 						"api_url":       "https://slack.com/api/test",
-						"app_url":       "https://slack.com/api/chat.postMessage",
 						"channel":       "#alerts",
 						"callback_id":   "{{ template \"slack.default.callbackid\" . }}",
 						"color":         "{{ if eq .Status \"firing\" }}danger{{ else }}good{{ end }}",
@@ -77,12 +72,6 @@ func TestNewConfigFromChannels(t *testing.T) {
 						"title":         "{{ template \"slack.default.title\" . }}",
 						"title_link":    "{{ template \"slack.default.titlelink\" . }}",
 						"username":      "{{ template \"slack.default.username\" . }}",
-						"http_config": map[string]any{
-							"tls_config":       map[string]any{"insecure_skip_verify": false},
-							"follow_redirects": true,
-							"enable_http2":     true,
-							"proxy_url":        nil,
-						},
 					}},
 				},
 			},
@@ -104,7 +93,6 @@ func TestNewConfigFromChannels(t *testing.T) {
 					"pagerduty_configs": []any{map[string]any{
 						"send_resolved": false,
 						"service_key":   "test",
-						"url":           "https://events.pagerduty.com/v2/enqueue",
 						"client":        "{{ template \"pagerduty.default.client\" . }}",
 						"client_url":    "{{ template \"pagerduty.default.clientURL\" . }}",
 						"description":   "{{ template \"pagerduty.default.description\" .}}",
@@ -116,12 +104,6 @@ func TestNewConfigFromChannels(t *testing.T) {
 							"num_resolved": "{{ .Alerts.Resolved | len }}",
 							"resolved":     "{{ .Alerts.Resolved | toJson }}",
 						},
-						"http_config": map[string]any{
-							"tls_config":       map[string]any{"insecure_skip_verify": false},
-							"follow_redirects": true,
-							"enable_http2":     true,
-							"proxy_url":        nil,
-						},
 					}},
 				},
 			},
@@ -148,7 +130,6 @@ func TestNewConfigFromChannels(t *testing.T) {
 					"pagerduty_configs": []any{map[string]any{
 						"send_resolved": false,
 						"service_key":   "test",
-						"url":           "https://events.pagerduty.com/v2/enqueue",
 						"client":        "{{ template \"pagerduty.default.client\" . }}",
 						"client_url":    "{{ template \"pagerduty.default.clientURL\" . }}",
 						"description":   "{{ template \"pagerduty.default.description\" .}}",
@@ -160,12 +141,6 @@ func TestNewConfigFromChannels(t *testing.T) {
 							"num_resolved": "{{ .Alerts.Resolved | len }}",
 							"resolved":     "{{ .Alerts.Resolved | toJson }}",
 						},
-						"http_config": map[string]any{
-							"tls_config":       map[string]any{"insecure_skip_verify": false},
-							"follow_redirects": true,
-							"enable_http2":     true,
-							"proxy_url":        nil,
-						},
 					}},
 				},
 				{
@@ -173,7 +148,6 @@ func TestNewConfigFromChannels(t *testing.T) {
 					"slack_configs": []any{map[string]any{
 						"send_resolved": true,
 						"api_url":       "https://slack.com/api/test",
-						"app_url":       "https://slack.com/api/chat.postMessage",
 						"channel":       "#alerts",
 						"callback_id":   "{{ template \"slack.default.callbackid\" . }}",
 						"color":         "{{ if eq .Status \"firing\" }}danger{{ else }}good{{ end }}",
@@ -187,12 +161,6 @@ func TestNewConfigFromChannels(t *testing.T) {
 						"title":         "{{ template \"slack.default.title\" . }}",
 						"title_link":    "{{ template \"slack.default.titlelink\" . }}",
 						"username":      "{{ template \"slack.default.username\" . }}",
-						"http_config": map[string]any{
-							"tls_config":       map[string]any{"insecure_skip_verify": false},
-							"follow_redirects": true,
-							"enable_http2":     true,
-							"proxy_url":        nil,
-						},
 					}},
 				},
 			},

pkg/types/alertmanagertypes/config.go

@@ -117,6 +117,11 @@ func NewConfigFromStoreableConfig(sc *StoreableConfig) (*Config, error) {
 		return nil, err
 	}
 
+	// It must be replaced with an empty, non-nil global, upstream swaps nil for
+	// DefaultGlobalConfig, which would let a path that skips SetGlobalConfig pass
+	// validation and fail silently at delivery instead of failing fast here.
+	alertmanagerConfig.Global = &config.GlobalConfig{}
+
 	return &Config{
 		alertmanagerConfig: alertmanagerConfig,
 		customConfigs:      customConfigs,
@@ -174,7 +179,7 @@ func newConfigFromString(s string) (*config.Config, map[string]customReceiverCon
 	return amConfig, customConfigs, nil
 }
 
-func newRawFromConfig(c *config.Config, customConfigs map[string]customReceiverConfigs) []byte {
+func extendedReceivers(c *config.Config, customConfigs map[string]customReceiverConfigs) []*Receiver {
 	receivers := make([]*Receiver, len(c.Receivers))
 	for i := range c.Receivers {
 		base := c.Receivers[i]
@@ -185,7 +190,14 @@ func newRawFromConfig(c *config.Config, customConfigs map[string]customReceiverC
 		}
 	}
 
-	b, err := json.Marshal(storedConfig{Config: c, Receivers: receivers})
+	return receivers
+}
+
+func newRawFromConfig(c *config.Config, customConfigs map[string]customReceiverConfigs) []byte {
+	persistable := *c
+	persistable.Global = nil
+
+	b, err := json.Marshal(storedConfig{Config: &persistable, Receivers: extendedReceivers(c, customConfigs)})
 	if err != nil {
 		// Taking inspiration from the upstream. This is never expected to happen.
 		return []byte(fmt.Sprintf("<error creating config string: %s>", err))
@@ -206,6 +218,37 @@ func (c *Config) flush() {
 	c.storeableConfig.UpdatedAt = time.Now()
 }
 
+func (c *Config) Resolved() (*Config, error) {
+	raw, err := json.Marshal(storedConfig{Config: c.alertmanagerConfig, Receivers: extendedReceivers(c.alertmanagerConfig, c.customConfigs)})
+	if err != nil {
+		return nil, err
+	}
+
+	alertmanagerConfig, customConfigs, err := newConfigFromString(string(raw))
+	if err != nil {
+		return nil, err
+	}
+
+	if err := alertmanagerConfig.UnmarshalYAML(func(i interface{}) error { return nil }); err != nil {
+		return nil, err
+	}
+
+	storeableConfig := *c.storeableConfig
+	resolved := &Config{
+		alertmanagerConfig: alertmanagerConfig,
+		customConfigs:      customConfigs,
+		storeableConfig:    &storeableConfig,
+	}
+	resolved.applyNativeDefaults()
+
+	return resolved, nil
+}
+
+func (c *Config) validate() error {
+	_, err := c.Resolved()
+	return err
+}
+
 func (c *Config) CopyWithReset() (*Config, error) {
 	newConfig, err := NewDefaultConfig(
 		*c.alertmanagerConfig.Global,
@@ -271,6 +314,15 @@ func (c *Config) StoreableConfig() *StoreableConfig {
 	return c.storeableConfig
 }
 
+func cloneReceiver(receiver *Receiver) (*Receiver, error) {
+	raw, err := json.Marshal(receiver)
+	if err != nil {
+		return nil, err
+	}
+
+	return NewReceiver(string(raw))
+}
+
 func (c *Config) CreateReceiver(receiver *Receiver) error {
 	// check that receiver name is not already used
 	for _, existingReceiver := range c.alertmanagerConfig.Receivers {
@@ -279,16 +331,21 @@ func (c *Config) CreateReceiver(receiver *Receiver) error {
 		}
 	}
 
-	route, err := NewRouteFromReceiver(receiver)
+	owned, err := cloneReceiver(receiver)
+	if err != nil {
+		return err
+	}
+
+	route, err := NewRouteFromReceiver(owned)
 	if err != nil {
 		return err
 	}
 
 	c.alertmanagerConfig.Route.Routes = append(c.alertmanagerConfig.Route.Routes, route)
-	c.alertmanagerConfig.Receivers = append(c.alertmanagerConfig.Receivers, *receiver.Receiver)
-	c.setCustomConfigs(receiver)
+	c.alertmanagerConfig.Receivers = append(c.alertmanagerConfig.Receivers, *owned.Receiver)
+	c.setCustomConfigs(owned)
 
-	if err := c.alertmanagerConfig.UnmarshalYAML(func(i interface{}) error { return nil }); err != nil {
+	if err := c.validate(); err != nil {
 		return err
 	}
 	c.applyNativeDefaults()
@@ -313,16 +370,21 @@ func (c *Config) GetReceiver(name string) (*Receiver, error) {
 }
 
 func (c *Config) UpdateReceiver(receiver *Receiver) error {
+	owned, err := cloneReceiver(receiver)
+	if err != nil {
+		return err
+	}
+
 	// find and update receiver
 	for i, existingReceiver := range c.alertmanagerConfig.Receivers {
-		if existingReceiver.Name == receiver.Name {
-			c.alertmanagerConfig.Receivers[i] = *receiver.Receiver
-			c.setCustomConfigs(receiver)
+		if existingReceiver.Name == owned.Name {
+			c.alertmanagerConfig.Receivers[i] = *owned.Receiver
+			c.setCustomConfigs(owned)
 			break
 		}
 	}
 
-	if err := c.alertmanagerConfig.UnmarshalYAML(func(i interface{}) error { return nil }); err != nil {
+	if err := c.validate(); err != nil {
 		return err
 	}
 	c.applyNativeDefaults()

pkg/types/alertmanagertypes/config_test.go

@@ -330,6 +330,150 @@ func TestSetGlobalConfigPreservesSMTPRequireTLS(t *testing.T) {
 	}
 }
 
+func newSMTPGlobalConfig() GlobalConfig {
+	return GlobalConfig{
+		SMTPFrom:         "alerts@example.com",
+		SMTPHello:        "example.com",
+		SMTPSmarthost:    config.HostPort{Host: "smtp.sendgrid.net", Port: "587"},
+		SMTPAuthUsername: "apikey",
+		SMTPAuthPassword: "operator-secret",
+		SMTPRequireTLS:   true,
+	}
+}
+
+func newEmailTestConfig(t *testing.T) *Config {
+	t.Helper()
+
+	cfg, err := NewDefaultConfig(
+		newSMTPGlobalConfig(),
+		RouteConfig{GroupInterval: time.Minute, GroupWait: time.Minute, RepeatInterval: time.Minute},
+		"1",
+	)
+	require.NoError(t, err)
+
+	receiver, err := NewReceiver(`{"name":"email-receiver","email_configs":[{"to":"team@example.com"}]}`)
+	require.NoError(t, err)
+	require.NoError(t, cfg.CreateReceiver(receiver))
+
+	return cfg
+}
+
+func TestStoreableConfigCarriesNoSMTPSettings(t *testing.T) {
+	cfg := newEmailTestConfig(t)
+
+	raw := cfg.StoreableConfig().Config
+	assert.NotContains(t, raw, "operator-secret")
+	assert.NotContains(t, raw, "smtp.sendgrid.net")
+	assert.NotContains(t, raw, "apikey")
+	assert.NotContains(t, raw, "alerts@example.com")
+
+	assert.Equal(t, "operator-secret", string(cfg.alertmanagerConfig.Global.SMTPAuthPassword))
+}
+
+func TestStoreableConfigCarriesNoGlobal(t *testing.T) {
+	cfg := newEmailTestConfig(t)
+
+	stored := map[string]json.RawMessage{}
+	require.NoError(t, json.Unmarshal([]byte(cfg.StoreableConfig().Config), &stored))
+	assert.NotContains(t, stored, "global")
+}
+
+func TestSetGlobalConfigDoesNotChangeStoreableHash(t *testing.T) {
+	cfg := newEmailTestConfig(t)
+	hash := cfg.StoreableConfig().Hash
+
+	require.NoError(t, cfg.SetGlobalConfig(GlobalConfig{SMTPSmarthost: config.HostPort{Host: "smtp.other.net", Port: "2525"}, SMTPAuthPassword: "rotated-secret"}))
+
+	assert.Equal(t, hash, cfg.StoreableConfig().Hash)
+	assert.NotContains(t, cfg.StoreableConfig().Config, "rotated-secret")
+}
+
+func TestNewConfigFromStoreableConfigDiscardsStoredGlobal(t *testing.T) {
+	stored := &StoreableConfig{
+		Config: `{"global":{"resolve_timeout":"5m","smtp_smarthost":"email-smtp.us-east-1.amazonaws.com:587","smtp_auth_password":"old-secret","slack_api_url":"https://hooks.slack.com/services/T/B/X"},"route":{"receiver":"default-receiver"},"receivers":[{"name":"default-receiver"}]}`,
+		OrgID:  "1",
+	}
+
+	cfg, err := NewConfigFromStoreableConfig(stored)
+	require.NoError(t, err)
+
+	assert.Equal(t, &config.GlobalConfig{}, cfg.alertmanagerConfig.Global)
+}
+
+func TestResolvedFillsEmailTransportFromGlobal(t *testing.T) {
+	cfg := newEmailTestConfig(t)
+
+	resolved, err := cfg.Resolved()
+	require.NoError(t, err)
+
+	receiver, err := resolved.GetReceiver("email-receiver")
+	require.NoError(t, err)
+	require.Len(t, receiver.EmailConfigs, 1)
+
+	got := receiver.EmailConfigs[0]
+	assert.Equal(t, "team@example.com", got.To)
+	assert.Equal(t, "smtp.sendgrid.net:587", got.Smarthost.String())
+	assert.Equal(t, "alerts@example.com", got.From)
+	assert.Equal(t, "apikey", got.AuthUsername)
+	assert.Equal(t, "operator-secret", string(got.AuthPassword))
+	require.NotNil(t, got.RequireTLS)
+	assert.True(t, *got.RequireTLS)
+
+	stored, err := cfg.GetReceiver("email-receiver")
+	require.NoError(t, err)
+	assert.Empty(t, stored.EmailConfigs[0].Smarthost.String())
+	assert.NotContains(t, cfg.StoreableConfig().Config, "operator-secret")
+}
+
+func TestStaleStoredSMTPSettingsAreReplacedOnLoad(t *testing.T) {
+	stored := &StoreableConfig{
+		Config: `{"global":{"resolve_timeout":"5m","smtp_from":"old@example.com","smtp_hello":"localhost","smtp_smarthost":"email-smtp.us-east-1.amazonaws.com:587","smtp_auth_username":"old-user","smtp_auth_password":"old-secret","smtp_require_tls":true},"route":{"receiver":"default-receiver","group_by":["ruleId"],"routes":[{"receiver":"email-receiver","continue":true,"matchers":["ruleId=~\"-1\""]}],"group_wait":"30s","group_interval":"5m","repeat_interval":"4h"},"receivers":[{"name":"default-receiver"},{"name":"email-receiver","email_configs":[{"send_resolved":false,"to":"team@example.com","from":"old@example.com","hello":"localhost","smarthost":"email-smtp.us-east-1.amazonaws.com:587","auth_username":"old-user","auth_password":"old-secret","require_tls":true}]}]}`,
+		OrgID:  "1",
+	}
+
+	cfg, err := NewConfigFromStoreableConfig(stored)
+	require.NoError(t, err)
+
+	loaded, err := cfg.GetReceiver("email-receiver")
+	require.NoError(t, err)
+	require.Len(t, loaded.EmailConfigs, 1)
+	assert.Empty(t, loaded.EmailConfigs[0].Smarthost.String())
+	assert.Empty(t, string(loaded.EmailConfigs[0].AuthPassword))
+
+	require.NoError(t, cfg.SetGlobalConfig(newSMTPGlobalConfig()))
+
+	resolved, err := cfg.Resolved()
+	require.NoError(t, err)
+
+	receiver, err := resolved.GetReceiver("email-receiver")
+	require.NoError(t, err)
+	got := receiver.EmailConfigs[0]
+	assert.Equal(t, "smtp.sendgrid.net:587", got.Smarthost.String())
+	assert.Equal(t, "operator-secret", string(got.AuthPassword))
+	assert.Equal(t, "alerts@example.com", got.From)
+
+	assert.NotContains(t, cfg.StoreableConfig().Config, "old-secret")
+	assert.NotContains(t, cfg.StoreableConfig().Config, "amazonaws.com")
+	assert.NotContains(t, cfg.StoreableConfig().Config, "operator-secret")
+}
+
+func TestCreateReceiverDoesNotMutateCaller(t *testing.T) {
+	cfg := newEmailTestConfig(t)
+
+	resolved, err := cfg.Resolved()
+	require.NoError(t, err)
+	receiver, err := resolved.GetReceiver("email-receiver")
+	require.NoError(t, err)
+	require.Equal(t, "smtp.sendgrid.net:587", receiver.EmailConfigs[0].Smarthost.String())
+
+	throwaway, err := cfg.CopyWithReset()
+	require.NoError(t, err)
+	require.NoError(t, throwaway.CreateReceiver(receiver))
+
+	assert.Equal(t, "smtp.sendgrid.net:587", receiver.EmailConfigs[0].Smarthost.String())
+	assert.Equal(t, "operator-secret", string(receiver.EmailConfigs[0].AuthPassword))
+}
+
 // Round-trip: create → serialize → reload → GetReceiver still has the configs.
 func TestConfigPreservesGoogleChatConfigs(t *testing.T) {
 	webhookURL, err := url.Parse("https://chat.googleapis.com/v1/spaces/test/messages")

pkg/types/alertmanagertypes/receiver.go

@@ -37,6 +37,7 @@ func NewReceiver(input string) (*Receiver, error) {
 	if err != nil {
 		return nil, err
 	}
+	stripEmailTransport(withDefaults)
 	receiver.Receiver = withDefaults
 
 	// Extend this block when adding another native notifier type.
@@ -53,6 +54,23 @@ func NewReceiver(input string) (*Receiver, error) {
 	return receiver, nil
 }
 
+func stripEmailTransport(base *config.Receiver) {
+	for _, ec := range base.EmailConfigs {
+		ec.From = ""
+		ec.Hello = ""
+		ec.Smarthost = config.HostPort{}
+		ec.AuthUsername = ""
+		ec.AuthPassword = ""
+		ec.AuthPasswordFile = ""
+		ec.AuthSecret = ""
+		ec.AuthSecretFile = ""
+		ec.AuthIdentity = ""
+		ec.RequireTLS = nil
+		ec.TLSConfig = nil
+		ec.ForceImplicitTLS = nil
+	}
+}
+
 func defaultedBaseReceiver(base *config.Receiver) (*config.Receiver, error) {
 	bytes, err := yaml.Marshal(base)
 	if err != nil {
@@ -102,7 +120,12 @@ func TestReceiver(ctx context.Context, receiver *Receiver, receiverIntegrationsF
 		return err
 	}
 
-	defaultedReceiver, err := testConfig.GetReceiver(receiver.Name)
+	resolvedConfig, err := testConfig.Resolved()
+	if err != nil {
+		return err
+	}
+
+	defaultedReceiver, err := resolvedConfig.GetReceiver(receiver.Name)
 	if err != nil {
 		return err
 	}

pkg/types/alertmanagertypes/receiver_test.go

@@ -46,6 +46,31 @@ func TestNewReceiver(t *testing.T) {
 	}
 }
 
+func TestNewReceiverStripsEmailTransport(t *testing.T) {
+	receiver, err := NewReceiver(`{"name":"email","email_configs":[{"to":"team@example.com","from":"attacker@example.com","hello":"example.com","smarthost":"smtp.example.com:587","auth_username":"user","auth_password":"supersecret","auth_secret":"alsosecret","auth_identity":"id","require_tls":false,"headers":{"Subject":"custom"}}]}`)
+	require.NoError(t, err)
+	require.Len(t, receiver.EmailConfigs, 1)
+
+	got := receiver.EmailConfigs[0]
+	assert.Equal(t, "team@example.com", got.To)
+	assert.Equal(t, map[string]string{"Subject": "custom"}, got.Headers)
+
+	assert.Empty(t, got.From)
+	assert.Empty(t, got.Hello)
+	assert.Empty(t, got.Smarthost.String())
+	assert.Empty(t, got.AuthUsername)
+	assert.Empty(t, string(got.AuthPassword))
+	assert.Empty(t, string(got.AuthSecret))
+	assert.Empty(t, got.AuthIdentity)
+	assert.Nil(t, got.RequireTLS)
+	assert.Nil(t, got.TLSConfig)
+
+	bytes, err := json.Marshal(receiver)
+	require.NoError(t, err)
+	assert.NotContains(t, string(bytes), "supersecret")
+	assert.NotContains(t, string(bytes), "smtp.example.com")
+}
+
 // Omitted fields fall back to DefaultGoogleChatReceiverConfig.
 func TestNewReceiverGoogleChatAppliesDefaults(t *testing.T) {
 	receiver, err := NewReceiver(`{"name":"googlechat","googlechat_configs":[{"webhook_url":"https://chat.googleapis.com/v1/spaces/test/messages"}]}`)

pkg/types/dashboardtypes/perses_v1_to_v2_queries.go

@@ -179,7 +179,6 @@ func (d *v1Decoder) collectV1QueryEnvelopes(widget map[string]any, panelKind Pan
 			normalizePreV5GroupBy(q)
 			normalizePreV5PageSize(q, rowLimitPanel)
 			normalizeQueryLimit(q)
-			normalizeQueryOffset(q)
 			if needsAggregation {
 				ensureDefaultAggregation(q)
 			}
@@ -199,7 +198,6 @@ func (d *v1Decoder) collectV1QueryEnvelopes(widget map[string]any, panelKind Pan
 		assignMissingFormulaNames(formulas)
 		for _, f := range formulas {
 			normalizePreV5QueryData(f, widgetType, panelKind)
-			normalizeQueryLimit(f)
 			name := d.readString(f, "queryName")
 			env := qb.WrapInV5Envelope(name, f, string(qb.QueryTypeFormula.StringValue()))
 			backfillFormulaFields(env, f)
@@ -221,8 +219,6 @@ func (d *v1Decoder) collectV1QueryEnvelopes(widget map[string]any, panelKind Pan
 			normalizePreV5QueryData(op, widgetType, panelKind)
 			normalizePreV5GroupBy(op)
 			normalizeOrderByKeys(op)
-			normalizeQueryLimit(op)
-			normalizeQueryOffset(op)
 			name := d.readString(op, "queryName")
 			out = append(out, traceOperatorEnvelope(name, expression, op))
 		}

pkg/types/dashboardtypes/perses_v1_to_v2_queries_malformed.go

@@ -618,32 +618,15 @@ func normalizePreV5PageSize(query map[string]any, rowLimitPanel bool) {
 	}
 }
 
-// normalizeQueryLimit coerces limit to the int the v5 decode expects: v1 stored it
-// as a string ("5") or float, both of which fail the typed decode. An unparseable
-// value or one above the v5 maximum (MaxQueryLimit) is dropped, leaving the query
-// unlimited (the field is optional).
+// normalizeQueryLimit drops a limit above the v5 maximum (MaxQueryLimit); v1 allowed
+// larger/unbounded limits, and an over-max value fails validation. Removing it leaves
+// the query unlimited (the field is optional).
 func normalizeQueryLimit(query map[string]any) {
-	if query["limit"] == nil {
-		return
-	}
 	limit, ok := coerceFloat(query["limit"])
-	if !ok || limit > qb.MaxQueryLimit {
-		delete(query, "limit")
-		return
-	}
-	query["limit"] = int(limit)
-}
-
-// normalizeQueryOffset coerces offset to the int the v5 decode expects; v1 could
-// store it as a string. An unparseable value is dropped (offset defaults to 0).
-func normalizeQueryOffset(query map[string]any) {
-	if query["offset"] == nil {
-		return
-	}
-	offset, ok := coerceFloat(query["offset"])
 	if !ok {
-		delete(query, "offset")
 		return
 	}
-	query["offset"] = int(offset)
+	if limit > qb.MaxQueryLimit {
+		delete(query, "limit")
+	}
 }

pkg/types/querybuildertypes/querybuildertypesv5/builder_elements.go

@@ -118,6 +118,10 @@ const (
 	FilterOperatorHasToken
 	FilterOperatorHasAny
 	FilterOperatorHasAll
+
+	// FilterOperatorSearch backs search('term'): keyless, fanned out by the condition
+	// builder across every searchable column.
+	FilterOperatorSearch
 )
 
 var operatorInverseMapping = map[FilterOperator]FilterOperator{
@@ -186,7 +190,8 @@ func (f FilterOperator) IsNegativeOperator() bool {
 		FilterOperatorIn,
 		FilterOperatorExists,
 		FilterOperatorRegexp,
-		FilterOperatorContains:
+		FilterOperatorContains,
+		FilterOperatorSearch:
 		return false
 	}
 	return true
@@ -243,10 +248,11 @@ func (f FilterOperator) IsArrayFunctionOperator() bool {
 }
 
 // IsFunctionOperator reports whether the operator is a query function
-// (has/hasAny/hasAll/hasToken); these apply to the logs body column only.
+// (has/hasAny/hasAll/hasToken/search) — logs-only, and skipped by the
+// resource-fingerprint builder.
 func (f FilterOperator) IsFunctionOperator() bool {
 	switch f {
-	case FilterOperatorHas, FilterOperatorHasAny, FilterOperatorHasAll, FilterOperatorHasToken:
+	case FilterOperatorHas, FilterOperatorHasAny, FilterOperatorHasAll, FilterOperatorHasToken, FilterOperatorSearch:
 		return true
 	default:
 		return false
@@ -265,6 +271,8 @@ func (f FilterOperator) FunctionName() string {
 		return "hasAll"
 	case FilterOperatorHasToken:
 		return "hasToken"
+	case FilterOperatorSearch:
+		return "search"
 	default:
 		return ""
 	}

pkg/types/querybuildertypes/querybuildertypesv5/qb.go

@@ -57,6 +57,12 @@ type Statement struct {
 	Args           []any
 	Warnings       []string
 	WarningsDocURL string
+	CostGuard      *CostGuard
+}
+
+type CostGuard struct {
+	Warning     string
+	MaxScanRows int64
 }
 
 // StatementBuilder builds the query.

pkg/types/telemetrytypes/field_context.go

@@ -79,6 +79,14 @@ var (
 	}
 )
 
+// FieldContextFromText resolves a context word with the same aliases as key parsing
+// ("tag" -> attribute). ok is false for an unknown word, so callers can reject it
+// rather than get unspecified.
+func FieldContextFromText(text string) (FieldContext, bool) {
+	fc, ok := fieldContexts[strings.ToLower(strings.TrimSpace(text))]
+	return fc, ok
+}
+
 // UnmarshalJSON implements the json.Unmarshaler interface.
 func (f *FieldContext) UnmarshalJSON(data []byte) error {
 	var str string

pkg/variables/variable_replace_visitor.go

@@ -412,10 +412,15 @@ func (v *variableReplacementVisitor) VisitFunctionCall(ctx *grammar.FunctionCall
 }
 
 func (v *variableReplacementVisitor) VisitSearchCall(ctx *grammar.SearchCallContext) any {
-	if ctx.FunctionParamList() == nil {
+	if ctx.ValueList() == nil {
 		return "search()"
 	}
-	return "search(" + v.Visit(ctx.FunctionParamList()).(string) + ")"
+	// VisitValueList already parenthesizes the args and propagates the __all__ marker.
+	result := v.Visit(ctx.ValueList()).(string)
+	if result == specialSkipMarker {
+		return specialSkipMarker
+	}
+	return "search" + result
 }
 
 func (v *variableReplacementVisitor) VisitFunctionParamList(ctx *grammar.FunctionParamListContext) any {

tests/conftest.py

@@ -24,6 +24,7 @@ pytest_plugins = [
     "fixtures.keycloak",
     "fixtures.idp",
     "fixtures.notification_channel",
+    "fixtures.maildev",
     "fixtures.alerts",
     "fixtures.cloudintegrations",
     "fixtures.jsontypes",

tests/fixtures/alerts.py

@@ -1,11 +1,13 @@
 import base64
 import json
+import re
 import time
 import urllib.parse
 import uuid
 from collections.abc import Callable
 from datetime import UTC, datetime, timedelta
 from http import HTTPStatus
+from urllib.parse import urlparse
 
 import pytest
 import requests
@@ -15,6 +17,7 @@ from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
 from fixtures.fs import get_testdata_file_path
 from fixtures.logger import setup_logger
 from fixtures.logs import Logs
+from fixtures.maildev import get_all_mails, verify_email_received
 from fixtures.metrics import Metrics
 from fixtures.traces import Traces
 
@@ -311,3 +314,145 @@ def update_rule_channel_name(rule_data: dict, channel_name: str):
         # loop over all the sepcs and update the channels
         for spec in thresholds["spec"]:
             spec["channels"] = [channel_name]
+
+
+def _is_json_subset(subset, superset) -> bool:
+    """Check if subset is contained within superset recursively.
+    - For dicts: all keys in subset must exist in superset with matching values
+    - For lists: all items in subset must be present in superset
+    - For scalars: exact equality
+    """
+    if isinstance(subset, dict):
+        if not isinstance(superset, dict):
+            return False
+        return all(key in superset and _is_json_subset(value, superset[key]) for key, value in subset.items())
+    if isinstance(subset, list):
+        if not isinstance(superset, list):
+            return False
+        return all(any(_is_json_subset(sub_item, sup_item) for sup_item in superset) for sub_item in subset)
+    if isinstance(subset, re.Pattern):
+        return isinstance(superset, str) and subset.search(superset) is not None
+    return subset == superset
+
+
+def verify_webhook_notification_expectation(
+    notification_channel: types.TestContainerDocker,
+    validation_data: dict,
+) -> bool:
+    """Check if wiremock received a request at the given path
+    whose JSON body is a superset of the expected json_body."""
+    path = validation_data["path"]
+    json_body = validation_data["json_body"]
+
+    url = notification_channel.host_configs["8080"].get("__admin/requests/find")
+    try:
+        res = requests.post(url, json={"method": "POST", "url": path}, timeout=10)
+    except requests.exceptions.RequestException:
+        return False
+    if res.status_code != HTTPStatus.OK:
+        return False
+
+    for req in res.json()["requests"]:
+        body = json.loads(base64.b64decode(req["bodyAsBase64"]).decode("utf-8"))
+        if _is_json_subset(json_body, body):
+            return True
+    return False
+
+
+def _check_notification_validation(
+    validation: types.NotificationValidation,
+    notification_channel: types.TestContainerDocker,
+    maildev: types.TestContainerDocker,
+) -> bool:
+    """Dispatch a single validation check to the appropriate verifier."""
+    if validation.destination_type == "webhook":
+        return verify_webhook_notification_expectation(notification_channel, validation.validation_data)
+    if validation.destination_type == "email":
+        return verify_email_received(maildev, validation.validation_data)
+    raise ValueError(f"Invalid destination type: {validation.destination_type}")
+
+
+def verify_notification_expectation(
+    notification_channel: types.TestContainerDocker,
+    maildev: types.TestContainerDocker,
+    expected_notification: types.AMNotificationExpectation,
+) -> bool:
+    """Poll for expected notifications across webhook and email channels."""
+    time_to_wait = datetime.now() + timedelta(seconds=expected_notification.wait_time_seconds)
+
+    while datetime.now() < time_to_wait:
+        all_found = all(_check_notification_validation(v, notification_channel, maildev) for v in expected_notification.notification_validations)
+
+        if expected_notification.should_notify and all_found:
+            logger.info("All expected notifications found")
+            return True
+
+        time.sleep(1)
+
+    # Timeout reached
+    if not expected_notification.should_notify:
+        # Verify no notifications were received
+        for validation in expected_notification.notification_validations:
+            found = _check_notification_validation(validation, notification_channel, maildev)
+            assert not found, f"Expected no notification but found one for {validation.destination_type} with data {validation.validation_data}"
+        logger.info("No notifications found, as expected")
+        return True
+
+    missing = [v for v in expected_notification.notification_validations if not _check_notification_validation(v, notification_channel, maildev)]
+    assert len(missing) == 0, f"Expected all notifications to be found but missing: {missing}, received: {_received_notifications(notification_channel, maildev, missing)}"
+    return True
+
+
+def _received_notifications(
+    notification_channel: types.TestContainerDocker,
+    maildev: types.TestContainerDocker,
+    missing: list[types.NotificationValidation],
+) -> dict:
+    received = {}
+    if any(v.destination_type == "webhook" for v in missing):
+        webhook_bodies = []
+        for validation in missing:
+            if validation.destination_type != "webhook":
+                continue
+            url = notification_channel.host_configs["8080"].get("__admin/requests/find")
+            try:
+                res = requests.post(url, json={"method": "POST", "url": validation.validation_data["path"]}, timeout=10)
+                webhook_bodies.extend(json.loads(base64.b64decode(req["bodyAsBase64"]).decode("utf-8")) for req in res.json()["requests"])
+            except requests.exceptions.RequestException as exc:
+                webhook_bodies.append(f"<failed to fetch wiremock journal: {exc}>")
+        received["webhook"] = webhook_bodies
+    if any(v.destination_type == "email" for v in missing):
+        received["email"] = get_all_mails(maildev)
+    return received
+
+
+def update_raw_channel_config(
+    channel_config: dict,
+    channel_name: str,
+    notification_channel: types.TestContainerDocker,
+) -> dict:
+    """
+    Updates the channel config to point to the given wiremock
+    notification_channel container to receive notifications.
+    """
+    config = channel_config.copy()
+
+    config["name"] = channel_name
+
+    url_field_map = {
+        "slack_configs": "api_url",
+        "msteamsv2_configs": "webhook_url",
+        "webhook_configs": "url",
+        "pagerduty_configs": "url",
+        "opsgenie_configs": "api_url",
+    }
+
+    for config_key, url_field in url_field_map.items():
+        if config_key in config:
+            for entry in config[config_key]:
+                if url_field in entry:
+                    original_url = entry[url_field]
+                    path = urlparse(original_url).path
+                    entry[url_field] = notification_channel.container_configs["8080"].get(path)
+
+    return config

tests/fixtures/auth.py

@@ -77,7 +77,7 @@ def register_admin(
             timeout=5,
         )
 
-        assert response.status_code == HTTPStatus.OK
+        assert response.status_code == HTTPStatus.OK, f"failed to register admin: {response.status_code} {response.text}"
 
         return types.Operation(name="create_user_admin")
 

tests/fixtures/http.py

@@ -125,13 +125,19 @@ def gateway(
 
 @pytest.fixture(name="make_http_mocks", scope="function")
 def make_http_mocks() -> Callable[[types.TestContainerDocker, list[Mapping]], None]:
+    mocked_containers = []
+
     def _make_http_mocks(container: types.TestContainerDocker, mappings: list[Mapping]) -> None:
         Config.base_url = container.host_configs["8080"].get("/__admin")
 
         for mapping in mappings:
             Mappings.create_mapping(mapping=mapping)
 
+        mocked_containers.append(container)
+
     yield _make_http_mocks
 
-    Mappings.delete_all_mappings()
-    Requests.reset_request_journal()
+    for container in mocked_containers:
+        Config.base_url = container.host_configs["8080"].get("/__admin")
+        Mappings.delete_all_mappings()
+        Requests.reset_request_journal()

tests/fixtures/maildev.py

@@ -0,0 +1,143 @@
+import re
+from http import HTTPStatus
+
+import docker
+import docker.errors
+import pytest
+import requests
+from testcontainers.core.container import DockerContainer, Network
+
+from fixtures import reuse, types
+from fixtures.logger import setup_logger
+
+logger = setup_logger(__name__)
+
+MAILDEV_INCOMING_USER = "apikey"
+MAILDEV_INCOMING_PASS = "integration-smtp-secret"
+
+SMTP_TEST_FROM = "alertmanager@integration.test"
+
+OLD_PROVIDER_SMTP_PASS = "old-provider-smtp-secret"
+NEW_PROVIDER_SMTP_PASS = "new-provider-smtp-secret"
+
+
+def signoz_smtp_env(maildev: "types.TestContainerDocker", password: str = MAILDEV_INCOMING_PASS) -> dict:
+    return {
+        "SIGNOZ_ALERTMANAGER_SIGNOZ_GLOBAL_SMTP__SMARTHOST": f"{maildev.container_configs['1025'].address}:{maildev.container_configs['1025'].port}",
+        "SIGNOZ_ALERTMANAGER_SIGNOZ_GLOBAL_SMTP__FROM": SMTP_TEST_FROM,
+        "SIGNOZ_ALERTMANAGER_SIGNOZ_GLOBAL_SMTP__AUTH__USERNAME": MAILDEV_INCOMING_USER,
+        "SIGNOZ_ALERTMANAGER_SIGNOZ_GLOBAL_SMTP__AUTH__PASSWORD": password,
+        "SIGNOZ_ALERTMANAGER_SIGNOZ_GLOBAL_SMTP__REQUIRE__TLS": "false",
+    }
+
+
+def create_maildev(
+    network: Network,
+    request: pytest.FixtureRequest,
+    pytestconfig: pytest.Config,
+    cache_key: str = "maildev",
+    incoming_user: str = MAILDEV_INCOMING_USER,
+    incoming_pass: str = MAILDEV_INCOMING_PASS,
+) -> types.TestContainerDocker:
+    def create() -> types.TestContainerDocker:
+        container = DockerContainer(image="maildev/maildev:2.2.1")
+        container.with_env("MAILDEV_INCOMING_USER", incoming_user)
+        container.with_env("MAILDEV_INCOMING_PASS", incoming_pass)
+        container.with_exposed_ports(1025, 1080)
+        container.with_network(network=network)
+        container.start()
+
+        return types.TestContainerDocker(
+            id=container.get_wrapped_container().id,
+            host_configs={
+                "1025": types.TestContainerUrlConfig(
+                    scheme="smtp",
+                    address=container.get_container_host_ip(),
+                    port=container.get_exposed_port(1025),
+                ),
+                "1080": types.TestContainerUrlConfig(
+                    scheme="http",
+                    address=container.get_container_host_ip(),
+                    port=container.get_exposed_port(1080),
+                ),
+            },
+            container_configs={
+                "1025": types.TestContainerUrlConfig(
+                    scheme="smtp",
+                    address=container.get_wrapped_container().name,
+                    port=1025,
+                ),
+                "1080": types.TestContainerUrlConfig(
+                    scheme="http",
+                    address=container.get_wrapped_container().name,
+                    port=1080,
+                ),
+            },
+        )
+
+    def delete(container: types.TestContainerDocker):
+        client = docker.from_env()
+        try:
+            client.containers.get(container_id=container.id).stop()
+            client.containers.get(container_id=container.id).remove(v=True)
+        except docker.errors.NotFound:
+            logger.info(
+                "Skipping removal of MailDev, MailDev(%s) not found. Maybe it was manually removed?",
+                {"id": container.id},
+            )
+
+    def restore(cache: dict) -> types.TestContainerDocker:
+        return types.TestContainerDocker.from_cache(cache)
+
+    return reuse.wrap(
+        request,
+        pytestconfig,
+        cache_key,
+        lambda: types.TestContainerDocker(id="", host_configs={}, container_configs={}),
+        create,
+        delete,
+        restore,
+    )
+
+
+@pytest.fixture(name="maildev", scope="package")
+def maildev(network: Network, request: pytest.FixtureRequest, pytestconfig: pytest.Config) -> types.TestContainerDocker:
+    return create_maildev(network, request, pytestconfig)
+
+
+def get_all_mails(_maildev: types.TestContainerDocker) -> list[dict]:
+    url = _maildev.host_configs["1080"].get("/email")
+    response = requests.get(url, timeout=5)
+    assert response.status_code == HTTPStatus.OK, f"Failed to fetch emails from MailDev, status code: {response.status_code}, response: {response.text}"
+
+    def addresses(entries: list[dict]) -> str:
+        return ",".join(sorted(entry.get("address", "") for entry in entries))
+
+    return [
+        {
+            "subject": email.get("subject", ""),
+            "html": email.get("html", ""),
+            "text": email.get("text", ""),
+            "from": addresses(email.get("from", [])),
+            "to": addresses(email.get("to", [])),
+        }
+        for email in response.json()
+    ]
+
+
+def verify_email_received(_maildev: types.TestContainerDocker, filters: dict) -> bool:
+    def matches(expected, actual: str) -> bool:
+        if isinstance(expected, re.Pattern):
+            return expected.search(actual) is not None
+        return expected == actual
+
+    for email in get_all_mails(_maildev):
+        if all(key in email and matches(filter_value, email[key]) for key, filter_value in filters.items()):
+            return True
+    return False
+
+
+def delete_all_mails(_maildev: types.TestContainerDocker) -> None:
+    url = _maildev.host_configs["1080"].get("/email/all")
+    response = requests.delete(url, timeout=5)
+    assert response.status_code == HTTPStatus.OK, f"Failed to delete emails from MailDev, status code: {response.status_code}, response: {response.text}"

tests/fixtures/notification_channel.py

@@ -1,3 +1,6 @@
+# pylint: disable=line-too-long
+import json
+import time
 from collections.abc import Callable
 from http import HTTPStatus
 
@@ -11,10 +14,116 @@ from wiremock.testing.testcontainer import WireMockContainer
 from fixtures import reuse, types
 from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
 from fixtures.logger import setup_logger
+from fixtures.maildev import MAILDEV_INCOMING_PASS, SMTP_TEST_FROM
 
 logger = setup_logger(__name__)
 
 
+EMAIL_TRANSPORT_KEYS = [
+    "from",
+    "hello",
+    "smarthost",
+    "auth_username",
+    "auth_password",
+    "auth_password_file",
+    "auth_secret",
+    "auth_secret_file",
+    "auth_identity",
+    "require_tls",
+    "tls_config",
+    "force_implicit_tls",
+]
+
+
+def assert_email_channel_payload_clean(payload: str) -> None:
+    receiver = json.loads(payload)
+    for email_config in receiver["email_configs"]:
+        transport_keys = set(email_config.keys()) & set(EMAIL_TRANSPORT_KEYS)
+        transport_keys -= {"smarthost"} if email_config.get("smarthost", "") == "" else set()
+        assert not transport_keys, f"email channel payload carries transport keys {transport_keys}: {payload}"
+
+    assert MAILDEV_INCOMING_PASS not in payload
+    assert SMTP_TEST_FROM not in payload
+
+
+"""
+Default notification channel configs shared across alertmanager tests.
+"""
+slack_default_config = {
+    # channel name configured on runtime
+    "slack_configs": [
+        {
+            "api_url": "services/TEAM_ID/BOT_ID/TOKEN_ID",  # base_url configured on runtime
+            "title": '[{{ .Status | toUpper }}{{ if eq .Status "firing" }}:{{ .Alerts.Firing | len }}{{ end }}] {{ .CommonLabels.alertname }} for {{ .CommonLabels.job }}\n {{- if gt (len .CommonLabels) (len .GroupLabels) -}}\n {{" "}}(\n {{- with .CommonLabels.Remove .GroupLabels.Names }}\n {{- range $index, $label := .SortedPairs -}}\n {{ if $index }}, {{ end }}\n {{- $label.Name }}="{{ $label.Value -}}"\n {{- end }}\n {{- end -}}\n )\n {{- end }}',
+            "text": '{{ range .Alerts -}}\r\n *Alert:* {{ .Labels.alertname }}{{ if .Labels.severity }} - {{ .Labels.severity }}{{ end }}\r\n\r\n *Summary:* {{ .Annotations.summary }}\r\n *Description:* {{ .Annotations.description }}\r\n *RelatedLogs:* {{ if gt (len .Annotations.related_logs) 0 -}} View in <{{ .Annotations.related_logs }}|logs explorer> {{- end}}\r\n *RelatedTraces:* {{ if gt (len .Annotations.related_traces) 0 -}} View in <{{ .Annotations.related_traces }}|traces explorer> {{- end}}\r\n\r\n *Details:*\r\n {{ range .Labels.SortedPairs -}}\r\n   {{- if ne .Name "ruleId" -}}\r\n \u2022 *{{ .Name }}:* {{ .Value }}\r\n   {{ end -}}\r\n {{ end -}}\r\n{{ end }}',
+        }
+    ],
+}
+
+msteams_default_config = {
+    "msteamsv2_configs": [
+        {
+            "webhook_url": "msteams/webhook_url",  # base_url configured on runtime
+            "title": '[{{ .Status | toUpper }}{{ if eq .Status "firing" }}:{{ .Alerts.Firing | len }}{{ end }}] {{ .CommonLabels.alertname }} for {{ .CommonLabels.job }}\n {{- if gt (len .CommonLabels) (len .GroupLabels) -}}\n {{" "}}(\n {{- with .CommonLabels.Remove .GroupLabels.Names }}\n {{- range $index, $label := .SortedPairs -}}\n {{ if $index }}, {{ end }}\n {{- $label.Name }}="{{ $label.Value -}}"\n {{- end }}\n {{- end -}}\n )\n {{- end }}',
+            "text": '{{ range .Alerts -}}\r\n *Alert:* {{ .Labels.alertname }}{{ if .Labels.severity }} - {{ .Labels.severity }}{{ end }}\r\n\r\n *Summary:* {{ .Annotations.summary }}\r\n *Description:* {{ .Annotations.description }}\r\n *RelatedLogs:* {{ if gt (len .Annotations.related_logs) 0 -}} View in <{{ .Annotations.related_logs }}|logs explorer> {{- end}}\r\n *RelatedTraces:* {{ if gt (len .Annotations.related_traces) 0 -}} View in <{{ .Annotations.related_traces }}|traces explorer> {{- end}}\r\n\r\n *Details:*\r\n {{ range .Labels.SortedPairs -}}\r\n   {{- if ne .Name "ruleId" -}}\r\n \u2022 *{{ .Name }}:* {{ .Value }}\r\n   {{ end -}}\r\n {{ end -}}\r\n{{ end }}',
+        }
+    ],
+}
+
+pagerduty_default_config = {
+    "pagerduty_configs": [
+        {
+            "routing_key": "PagerDutyRoutingKey",
+            "url": "v2/enqueue",  # base_url configured on runtime
+            "client": "SigNoz Alert Manager",
+            "client_url": "https://enter-signoz-host-n-port-here/alerts",
+            "description": '[{{ .Status | toUpper }}{{ if eq .Status "firing" }}:{{ .Alerts.Firing | len }}{{ end }}] {{ .CommonLabels.alertname }} for {{ .CommonLabels.job }}\n\t{{- if gt (len .CommonLabels) (len .GroupLabels) -}}\n\t {{" "}}(\n\t {{- with .CommonLabels.Remove .GroupLabels.Names }}\n\t\t{{- range $index, $label := .SortedPairs -}}\n\t\t {{ if $index }}, {{ end }}\n\t\t {{- $label.Name }}="{{ $label.Value -}}"\n\t\t{{- end }}\n\t {{- end -}}\n\t )\n\t{{- end }}',
+            "details": {
+                "firing": '{{ template "pagerduty.default.instances" .Alerts.Firing }}',
+                "num_firing": "{{ .Alerts.Firing | len }}",
+                "num_resolved": "{{ .Alerts.Resolved | len }}",
+                "resolved": '{{ template "pagerduty.default.instances" .Alerts.Resolved }}',
+            },
+            "source": "SigNoz Alert Manager",
+            "severity": "{{ (index .Alerts 0).Labels.severity }}",
+        }
+    ],
+}
+
+opsgenie_default_config = {
+    "opsgenie_configs": [
+        {
+            "api_key": "OpsGenieAPIKey",
+            "api_url": "/",  # base_url configured on runtime
+            "description": '{{ if gt (len .Alerts.Firing) 0 -}}\r\n\tAlerts Firing:\r\n\t{{ range .Alerts.Firing }}\r\n\t - Message: {{ .Annotations.description }}\r\n\tLabels:\r\n\t{{ range .Labels.SortedPairs -}}\r\n\t\t{{- if ne .Name "ruleId" }}   - {{ .Name }} = {{ .Value }}\r\n\t{{ end -}}\r\n\t{{- end }}   Annotations:\r\n\t{{ range .Annotations.SortedPairs }}   - {{ .Name }} = {{ .Value }}\r\n\t{{ end }}   Source: {{ .GeneratorURL }}\r\n\t{{ end }}\r\n{{- end }}\r\n{{ if gt (len .Alerts.Resolved) 0 -}}\r\n\tAlerts Resolved:\r\n\t{{ range .Alerts.Resolved }}\r\n\t - Message: {{ .Annotations.description }}\r\n\tLabels:\r\n\t{{ range .Labels.SortedPairs -}}\r\n\t\t{{- if ne .Name "ruleId" }}   - {{ .Name }} = {{ .Value }}\r\n\t{{ end -}}\r\n\t{{- end }}   Annotations:\r\n\t{{ range .Annotations.SortedPairs }}   - {{ .Name }} = {{ .Value }}\r\n\t{{ end }}   Source: {{ .GeneratorURL }}\r\n\t{{ end }}\r\n{{- end }}',
+            "priority": '{{ if eq (index .Alerts 0).Labels.severity "critical" }}P1{{ else if eq (index .Alerts 0).Labels.severity "warning" }}P2{{ else if eq (index .Alerts 0).Labels.severity "info" }}P3{{ else }}P4{{ end }}',
+            "message": "{{ .CommonLabels.alertname }}",
+            "details": {},
+        }
+    ],
+}
+
+webhook_default_config = {
+    "webhook_configs": [
+        {
+            "url": "webhook/webhook_url",  # base_url configured on runtime
+        }
+    ],
+}
+
+email_default_config = {
+    "email_configs": [
+        {
+            "to": "test@example.com",
+            "html": '<html><body>{{ range .Alerts -}}\r\n *Alert:* {{ .Labels.alertname }}{{ if .Labels.severity }} - {{ .Labels.severity }}{{ end }}\r\n\r\n *Summary:* {{ .Annotations.summary }}\r\n *Description:* {{ .Annotations.description }}\r\n *RelatedLogs:* {{ if gt (len .Annotations.related_logs) 0 -}} View in <{{ .Annotations.related_logs }}|logs explorer> {{- end}}\r\n *RelatedTraces:* {{ if gt (len .Annotations.related_traces) 0 -}} View in <{{ .Annotations.related_traces }}|traces explorer> {{- end}}\r\n\r\n *Details:*\r\n {{ range .Labels.SortedPairs -}}\r\n   {{- if ne .Name "ruleId" -}}\r\n \u2022 *{{ .Name }}:* {{ .Value }}\r\n   {{ end -}}\r\n {{ end -}}\r\n{{ end }}</body></html>',
+            "headers": {
+                "Subject": '[{{ .Status | toUpper }}{{ if eq .Status "firing" }}:{{ .Alerts.Firing | len }}{{ end }}] {{ .CommonLabels.alertname }} for {{ .CommonLabels.job }}\n {{- if gt (len .CommonLabels) (len .GroupLabels) -}}\n {{" "}}(\n {{- with .CommonLabels.Remove .GroupLabels.Names }}\n {{- range $index, $label := .SortedPairs -}}\n {{ if $index }}, {{ end }}\n {{- $label.Name }}="{{ $label.Value -}}"\n {{- end }}\n {{- end -}}\n )\n {{- end }}'
+            },
+        }
+    ],
+}
+
+
 @pytest.fixture(name="notification_channel", scope="package")
 def notification_channel(
     network: Network,
@@ -67,6 +176,40 @@ def notification_channel(
     )
 
 
+@pytest.fixture(name="create_notification_channel", scope="function")
+def create_notification_channel(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+) -> Callable[[dict], str]:
+    admin_token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+
+    channel_ids = []
+
+    def _create_notification_channel(channel_config: dict) -> str:
+        response = requests.post(
+            signoz.self.host_configs["8080"].get("/api/v1/channels"),
+            json=channel_config,
+            headers={"Authorization": f"Bearer {admin_token}"},
+            timeout=5,
+        )
+        assert response.status_code == HTTPStatus.CREATED, f"Failed to create channel, Response: {response.text} Response status: {response.status_code}"
+        channel_id = response.json()["data"]["id"]
+        channel_ids.append(channel_id)
+        return channel_id
+
+    yield _create_notification_channel
+
+    for channel_id in channel_ids:
+        response = requests.delete(
+            signoz.self.host_configs["8080"].get(f"/api/v1/channels/{channel_id}"),
+            headers={"Authorization": f"Bearer {admin_token}"},
+            timeout=5,
+        )
+        if response.status_code != HTTPStatus.NO_CONTENT:
+            logger.error("Failed to delete channel: %s", {"channel_id": channel_id, "status": response.status_code, "response": response.text})
+
+
 @pytest.fixture(name="create_webhook_notification_channel", scope="function")
 def create_webhook_notification_channel(
     signoz: types.SigNoz,
@@ -103,3 +246,19 @@ def create_webhook_notification_channel(
         return channel_id
 
     return _create_webhook_notification_channel
+
+
+def send_test_notification(signoz: types.SigNoz, token: str, receiver: dict, wait_seconds: int = 90) -> None:
+    deadline = time.time() + wait_seconds
+    last = None
+    while time.time() < deadline:
+        last = requests.post(
+            signoz.self.host_configs["8080"].get("/api/v1/channels/test"),
+            json=receiver,
+            headers={"Authorization": f"Bearer {token}"},
+            timeout=30,
+        )
+        if last.status_code == HTTPStatus.NO_CONTENT:
+            return
+        time.sleep(2)
+    raise AssertionError(f"test notification did not succeed within {wait_seconds}s, last response: {last.status_code} {last.text}")

tests/fixtures/types.py

@@ -197,3 +197,40 @@ class AlertTestCase:
     alert_data: list[AlertData]
     # list of alert expectations for the test case
     alert_expectation: AlertExpectation
+
+
+@dataclass(frozen=True)
+class NotificationValidation:
+    # destination type of the notification, either webhook or email
+    # slack, msteams, pagerduty, opsgenie, webhook channels send notifications through webhook
+    # email channels send notifications through email
+    destination_type: Literal["webhook", "email"]
+    # validation data for validating the received notification payload
+    validation_data: dict[str, any]
+
+
+@dataclass(frozen=True)
+class AMNotificationExpectation:
+    # whether we expect any notifications to be fired or not, false when testing downtime scenarios
+    # or don't expect any notifications to be fired in given time period
+    should_notify: bool
+    # seconds to wait for the notifications to be fired, if no
+    # notifications are fired in the expected time, the test will fail
+    wait_time_seconds: int
+    # list of notifications to expect, as a single rule can trigger multiple notifications
+    # spanning across different notifiers
+    notification_validations: list[NotificationValidation]
+
+
+@dataclass(frozen=True)
+class AlertManagerNotificationTestCase:
+    # name of the test case
+    name: str
+    # path to the rule file in testdata directory
+    rule_path: str
+    # list of alert data that will be inserted into the database for the rule to be triggered
+    alert_data: list[AlertData]
+    # configuration for the notification channel
+    channel_config: dict[str, any]
+    # notification expectations for the test case
+    notification_expectation: AMNotificationExpectation

tests/integration/bootstrap/setup.py

@@ -39,5 +39,7 @@ def test_teardown(
     idp: types.TestContainerIDP,  # pylint: disable=unused-argument
     create_user_admin: types.Operation,  # pylint: disable=unused-argument
     migrator: types.Operation,  # pylint: disable=unused-argument
+    maildev: types.TestContainerDocker,  # pylint: disable=unused-argument
+    notification_channel: types.TestContainerDocker,  # pylint: disable=unused-argument
 ) -> None:
     pass

tests/integration/testdata/alertmanager/content_templating/logs_data.jsonl

@@ -0,0 +1,20 @@
+{ "timestamp": "2026-01-29T10:00:00.000000Z", "resources": { "service.name": "payment-service" }, "attributes": { "code.file": "payment_handler.py" }, "body": "User login successful", "severity_text": "INFO" }
+{ "timestamp": "2026-01-29T10:00:30.000000Z", "resources": { "service.name": "payment-service" }, "attributes": { "code.file": "payment_handler.py" }, "body": "payment failure: gateway timeout", "severity_text": "ERROR" }
+{ "timestamp": "2026-01-29T10:01:00.000000Z", "resources": { "service.name": "payment-service" }, "attributes": { "code.file": "payment_handler.py" }, "body": "payment failure: card declined", "severity_text": "ERROR" }
+{ "timestamp": "2026-01-29T10:01:30.000000Z", "resources": { "service.name": "payment-service" }, "attributes": { "code.file": "payment_handler.py" }, "body": "Database connection established", "severity_text": "INFO" }
+{ "timestamp": "2026-01-29T10:02:00.000000Z", "resources": { "service.name": "payment-service" }, "attributes": { "code.file": "payment_handler.py" }, "body": "payment failure: insufficient funds", "severity_text": "ERROR" }
+{ "timestamp": "2026-01-29T10:02:30.000000Z", "resources": { "service.name": "payment-service" }, "attributes": { "code.file": "payment_handler.py" }, "body": "payment failure: invalid token", "severity_text": "ERROR" }
+{ "timestamp": "2026-01-29T10:03:00.000000Z", "resources": { "service.name": "payment-service" }, "attributes": { "code.file": "payment_handler.py" }, "body": "API request received", "severity_text": "INFO" }
+{ "timestamp": "2026-01-29T10:03:30.000000Z", "resources": { "service.name": "payment-service" }, "attributes": { "code.file": "payment_handler.py" }, "body": "payment failure: gateway timeout", "severity_text": "ERROR" }
+{ "timestamp": "2026-01-29T10:04:00.000000Z", "resources": { "service.name": "payment-service" }, "attributes": { "code.file": "payment_handler.py" }, "body": "payment failure: card declined", "severity_text": "ERROR" }
+{ "timestamp": "2026-01-29T10:04:30.000000Z", "resources": { "service.name": "payment-service" }, "attributes": { "code.file": "payment_handler.py" }, "body": "payment failure: invalid token", "severity_text": "ERROR" }
+{ "timestamp": "2026-01-29T10:05:00.000000Z", "resources": { "service.name": "payment-service" }, "attributes": { "code.file": "payment_handler.py" }, "body": "payment failure: gateway timeout", "severity_text": "ERROR" }
+{ "timestamp": "2026-01-29T10:05:30.000000Z", "resources": { "service.name": "payment-service" }, "attributes": { "code.file": "payment_handler.py" }, "body": "payment failure: card declined", "severity_text": "ERROR" }
+{ "timestamp": "2026-01-29T10:06:00.000000Z", "resources": { "service.name": "payment-service" }, "attributes": { "code.file": "payment_handler.py" }, "body": "payment failure: gateway timeout", "severity_text": "ERROR" }
+{ "timestamp": "2026-01-29T10:06:30.000000Z", "resources": { "service.name": "payment-service" }, "attributes": { "code.file": "payment_handler.py" }, "body": "payment failure: insufficient funds", "severity_text": "ERROR" }
+{ "timestamp": "2026-01-29T10:07:00.000000Z", "resources": { "service.name": "payment-service" }, "attributes": { "code.file": "payment_handler.py" }, "body": "payment failure: card declined", "severity_text": "ERROR" }
+{ "timestamp": "2026-01-29T10:07:30.000000Z", "resources": { "service.name": "payment-service" }, "attributes": { "code.file": "payment_handler.py" }, "body": "payment failure: gateway timeout", "severity_text": "ERROR" }
+{ "timestamp": "2026-01-29T10:08:00.000000Z", "resources": { "service.name": "payment-service" }, "attributes": { "code.file": "payment_handler.py" }, "body": "Response sent to client", "severity_text": "INFO" }
+{ "timestamp": "2026-01-29T10:08:30.000000Z", "resources": { "service.name": "payment-service" }, "attributes": { "code.file": "payment_handler.py" }, "body": "payment failure: invalid token", "severity_text": "ERROR" }
+{ "timestamp": "2026-01-29T10:09:00.000000Z", "resources": { "service.name": "payment-service" }, "attributes": { "code.file": "payment_handler.py" }, "body": "payment failure: card declined", "severity_text": "ERROR" }
+{ "timestamp": "2026-01-29T10:10:00.000000Z", "resources": { "service.name": "payment-service" }, "attributes": { "code.file": "payment_handler.py" }, "body": "payment failure: gateway timeout", "severity_text": "ERROR" }

tests/integration/testdata/alertmanager/content_templating/logs_rule.json

@@ -0,0 +1,69 @@
+{
+  "alert": "content_templating_logs",
+  "ruleType": "threshold_rule",
+  "alertType": "LOGS_BASED_ALERT",
+  "condition": {
+    "thresholds": {
+      "kind": "basic",
+      "spec": [
+        {
+          "name": "critical",
+          "target": 0,
+          "matchType": "1",
+          "op": "1",
+          "channels": [
+            "test channel"
+          ]
+        }
+      ]
+    },
+    "compositeQuery": {
+      "queryType": "builder",
+      "panelType": "graph",
+      "queries": [
+        {
+          "type": "builder_query",
+          "spec": {
+            "name": "A",
+            "signal": "logs",
+            "filter": {
+              "expression": "body CONTAINS 'payment failure'"
+            },
+            "aggregations": [
+              {
+                "expression": "count()"
+              }
+            ],
+            "groupBy": [
+              {"name": "service.name", "fieldContext": "resource"}
+            ]
+          }
+        }
+      ]
+    },
+    "selectedQueryName": "A"
+  },
+  "evaluation": {
+    "kind": "rolling",
+    "spec": {
+      "evalWindow": "5m0s",
+      "frequency": "15s"
+    }
+  },
+  "labels": {},
+  "annotations": {
+    "description": "Payment failure spike detected on $service_name",
+    "summary": "Payment failures elevated on $service_name"
+  },
+  "notificationSettings": {
+    "groupBy": [],
+    "usePolicy": false,
+    "renotify": {
+      "enabled": false,
+      "interval": "30m",
+      "alertStates": []
+    }
+  },
+  "version": "v5",
+  "schemaVersion": "v2alpha1"
+}

tests/integration/testdata/alertmanager/content_templating/metrics_data.jsonl

@@ -0,0 +1,12 @@
+{"metric_name":"container_memory_bytes_content_templating","labels":{"namespace":"production","pod":"checkout-7d9c8b5f4-x2k9p","container":"checkout","node":"ip-10-0-1-23","severity":"critical","service":"checkout"},"timestamp":"2026-01-29T10:01:00+00:00","value":80,"temporality":"Unspecified","type_":"Gauge","is_monotonic":false,"flags":0,"description":"","unit":"bytes","env":"default","resource_attrs":{},"scope_attrs":{}}
+{"metric_name":"container_memory_bytes_content_templating","labels":{"namespace":"production","pod":"checkout-7d9c8b5f4-x2k9p","container":"checkout","node":"ip-10-0-1-23","severity":"critical","service":"checkout"},"timestamp":"2026-01-29T10:02:00+00:00","value":95,"temporality":"Unspecified","type_":"Gauge","is_monotonic":false,"flags":0,"description":"","unit":"bytes","env":"default","resource_attrs":{},"scope_attrs":{}}
+{"metric_name":"container_memory_bytes_content_templating","labels":{"namespace":"production","pod":"checkout-7d9c8b5f4-x2k9p","container":"checkout","node":"ip-10-0-1-23","severity":"critical","service":"checkout"},"timestamp":"2026-01-29T10:03:00+00:00","value":110,"temporality":"Unspecified","type_":"Gauge","is_monotonic":false,"flags":0,"description":"","unit":"bytes","env":"default","resource_attrs":{},"scope_attrs":{}}
+{"metric_name":"container_memory_bytes_content_templating","labels":{"namespace":"production","pod":"checkout-7d9c8b5f4-x2k9p","container":"checkout","node":"ip-10-0-1-23","severity":"critical","service":"checkout"},"timestamp":"2026-01-29T10:04:00+00:00","value":120,"temporality":"Unspecified","type_":"Gauge","is_monotonic":false,"flags":0,"description":"","unit":"bytes","env":"default","resource_attrs":{},"scope_attrs":{}}
+{"metric_name":"container_memory_bytes_content_templating","labels":{"namespace":"production","pod":"checkout-7d9c8b5f4-x2k9p","container":"checkout","node":"ip-10-0-1-23","severity":"critical","service":"checkout"},"timestamp":"2026-01-29T10:05:00+00:00","value":125,"temporality":"Unspecified","type_":"Gauge","is_monotonic":false,"flags":0,"description":"","unit":"bytes","env":"default","resource_attrs":{},"scope_attrs":{}}
+{"metric_name":"container_memory_bytes_content_templating","labels":{"namespace":"production","pod":"checkout-7d9c8b5f4-x2k9p","container":"checkout","node":"ip-10-0-1-23","severity":"critical","service":"checkout"},"timestamp":"2026-01-29T10:06:00+00:00","value":130,"temporality":"Unspecified","type_":"Gauge","is_monotonic":false,"flags":0,"description":"","unit":"bytes","env":"default","resource_attrs":{},"scope_attrs":{}}
+{"metric_name":"container_memory_bytes_content_templating","labels":{"namespace":"production","pod":"checkout-7d9c8b5f4-x2k9p","container":"checkout","node":"ip-10-0-1-23","severity":"critical","service":"checkout"},"timestamp":"2026-01-29T10:07:00+00:00","value":135,"temporality":"Unspecified","type_":"Gauge","is_monotonic":false,"flags":0,"description":"","unit":"bytes","env":"default","resource_attrs":{},"scope_attrs":{}}
+{"metric_name":"container_memory_bytes_content_templating","labels":{"namespace":"production","pod":"checkout-7d9c8b5f4-x2k9p","container":"checkout","node":"ip-10-0-1-23","severity":"critical","service":"checkout"},"timestamp":"2026-01-29T10:08:00+00:00","value":140,"temporality":"Unspecified","type_":"Gauge","is_monotonic":false,"flags":0,"description":"","unit":"bytes","env":"default","resource_attrs":{},"scope_attrs":{}}
+{"metric_name":"container_memory_bytes_content_templating","labels":{"namespace":"production","pod":"checkout-7d9c8b5f4-x2k9p","container":"checkout","node":"ip-10-0-1-23","severity":"critical","service":"checkout"},"timestamp":"2026-01-29T10:09:00+00:00","value":145,"temporality":"Unspecified","type_":"Gauge","is_monotonic":false,"flags":0,"description":"","unit":"bytes","env":"default","resource_attrs":{},"scope_attrs":{}}
+{"metric_name":"container_memory_bytes_content_templating","labels":{"namespace":"production","pod":"checkout-7d9c8b5f4-x2k9p","container":"checkout","node":"ip-10-0-1-23","severity":"critical","service":"checkout"},"timestamp":"2026-01-29T10:10:00+00:00","value":150,"temporality":"Unspecified","type_":"Gauge","is_monotonic":false,"flags":0,"description":"","unit":"bytes","env":"default","resource_attrs":{},"scope_attrs":{}}
+{"metric_name":"container_memory_bytes_content_templating","labels":{"namespace":"production","pod":"checkout-7d9c8b5f4-x2k9p","container":"checkout","node":"ip-10-0-1-23","severity":"critical","service":"checkout"},"timestamp":"2026-01-29T10:11:00+00:00","value":155,"temporality":"Unspecified","type_":"Gauge","is_monotonic":false,"flags":0,"description":"","unit":"bytes","env":"default","resource_attrs":{},"scope_attrs":{}}
+{"metric_name":"container_memory_bytes_content_templating","labels":{"namespace":"production","pod":"checkout-7d9c8b5f4-x2k9p","container":"checkout","node":"ip-10-0-1-23","severity":"critical","service":"checkout"},"timestamp":"2026-01-29T10:12:00+00:00","value":160,"temporality":"Unspecified","type_":"Gauge","is_monotonic":false,"flags":0,"description":"","unit":"bytes","env":"default","resource_attrs":{},"scope_attrs":{}}

tests/integration/testdata/alertmanager/content_templating/metrics_rule.json

@@ -0,0 +1,72 @@
+{
+  "alert": "content_templating_metrics",
+  "ruleType": "threshold_rule",
+  "alertType": "METRIC_BASED_ALERT",
+  "condition": {
+    "thresholds": {
+      "kind": "basic",
+      "spec": [
+        {
+          "name": "critical",
+          "target": 100,
+          "matchType": "1",
+          "op": "1",
+          "channels": [
+            "test channel"
+          ]
+        }
+      ]
+    },
+    "compositeQuery": {
+      "queryType": "builder",
+      "panelType": "graph",
+      "queries": [
+        {
+          "type": "builder_query",
+          "spec": {
+            "name": "A",
+            "signal": "metrics",
+            "aggregations": [
+              {
+                "metricName": "container_memory_bytes_content_templating",
+                "timeAggregation": "avg",
+                "spaceAggregation": "max"
+              }
+            ],
+            "groupBy": [
+              {"name": "namespace", "fieldContext": "attribute", "fieldDataType": "string"},
+              {"name": "pod", "fieldContext": "attribute", "fieldDataType": "string"},
+              {"name": "container", "fieldContext": "attribute", "fieldDataType": "string"},
+              {"name": "node", "fieldContext": "attribute", "fieldDataType": "string"},
+              {"name": "severity", "fieldContext": "attribute", "fieldDataType": "string"}
+            ]
+          }
+        }
+      ]
+    },
+    "selectedQueryName": "A"
+  },
+  "evaluation": {
+    "kind": "rolling",
+    "spec": {
+      "evalWindow": "5m0s",
+      "frequency": "15s"
+    }
+  },
+  "labels": {},
+  "annotations": {
+    "description": "Container $container in pod $pod ($namespace) exceeded memory threshold",
+    "summary": "High container memory in $namespace/$pod"
+  },
+  "notificationSettings": {
+    "groupBy": [],
+    "usePolicy": false,
+    "renotify": {
+      "enabled": false,
+      "interval": "30m",
+      "alertStates": []
+    }
+  },
+  "version": "v5",
+  "schemaVersion": "v2alpha1"
+}

tests/integration/testdata/alertmanager/content_templating/traces_data.jsonl

@@ -0,0 +1,20 @@
+{ "timestamp": "2026-01-29T10:00:00.000000Z", "duration": "PT1.2S", "trace_id": "591f6d3d6b0a1f9e8a71b2c3d4e5f6a1", "span_id": "c1b2c3d4e5f6a7b8", "parent_span_id": "", "name": "POST /checkout", "kind": 2, "status_code": 1, "status_message": "", "resources": { "deployment.environment": "production", "service.name": "checkout-service", "os.type": "linux", "host.name": "ip-10-0-1-23" }, "attributes": { "net.transport": "IP.TCP", "http.scheme": "http", "http.user_agent": "Integration Test", "http.request.method": "POST", "http.response.status_code": "200", "http.request.path": "/checkout" } }
+{ "timestamp": "2026-01-29T10:00:30.000000Z", "duration": "PT1.4S", "trace_id": "591f6d3d6b0a1f9e8a71b2c3d4e5f6a2", "span_id": "c2b3c4d5e6f7a8b9", "parent_span_id": "", "name": "POST /checkout", "kind": 2, "status_code": 1, "status_message": "", "resources": { "deployment.environment": "production", "service.name": "checkout-service", "os.type": "linux", "host.name": "ip-10-0-1-23" }, "attributes": { "net.transport": "IP.TCP", "http.scheme": "http", "http.user_agent": "Integration Test", "http.request.method": "POST", "http.response.status_code": "200", "http.request.path": "/checkout" } }
+{ "timestamp": "2026-01-29T10:01:00.000000Z", "duration": "PT1.6S", "trace_id": "591f6d3d6b0a1f9e8a71b2c3d4e5f6a3", "span_id": "c3b4c5d6e7f8a9b0", "parent_span_id": "", "name": "POST /checkout", "kind": 2, "status_code": 1, "status_message": "", "resources": { "deployment.environment": "production", "service.name": "checkout-service", "os.type": "linux", "host.name": "ip-10-0-1-23" }, "attributes": { "net.transport": "IP.TCP", "http.scheme": "http", "http.user_agent": "Integration Test", "http.request.method": "POST", "http.response.status_code": "200", "http.request.path": "/checkout" } }
+{ "timestamp": "2026-01-29T10:01:30.000000Z", "duration": "PT1.8S", "trace_id": "591f6d3d6b0a1f9e8a71b2c3d4e5f6a4", "span_id": "c4b5c6d7e8f9a0b1", "parent_span_id": "", "name": "POST /checkout", "kind": 2, "status_code": 1, "status_message": "", "resources": { "deployment.environment": "production", "service.name": "checkout-service", "os.type": "linux", "host.name": "ip-10-0-1-23" }, "attributes": { "net.transport": "IP.TCP", "http.scheme": "http", "http.user_agent": "Integration Test", "http.request.method": "POST", "http.response.status_code": "200", "http.request.path": "/checkout" } }
+{ "timestamp": "2026-01-29T10:02:00.000000Z", "duration": "PT2.1S", "trace_id": "591f6d3d6b0a1f9e8a71b2c3d4e5f6a5", "span_id": "c5b6c7d8e9f0a1b2", "parent_span_id": "", "name": "POST /checkout", "kind": 2, "status_code": 1, "status_message": "", "resources": { "deployment.environment": "production", "service.name": "checkout-service", "os.type": "linux", "host.name": "ip-10-0-1-23" }, "attributes": { "net.transport": "IP.TCP", "http.scheme": "http", "http.user_agent": "Integration Test", "http.request.method": "POST", "http.response.status_code": "200", "http.request.path": "/checkout" } }
+{ "timestamp": "2026-01-29T10:02:30.000000Z", "duration": "PT2.3S", "trace_id": "591f6d3d6b0a1f9e8a71b2c3d4e5f6a6", "span_id": "c6b7c8d9e0f1a2b3", "parent_span_id": "", "name": "POST /checkout", "kind": 2, "status_code": 1, "status_message": "", "resources": { "deployment.environment": "production", "service.name": "checkout-service", "os.type": "linux", "host.name": "ip-10-0-1-23" }, "attributes": { "net.transport": "IP.TCP", "http.scheme": "http", "http.user_agent": "Integration Test", "http.request.method": "POST", "http.response.status_code": "200", "http.request.path": "/checkout" } }
+{ "timestamp": "2026-01-29T10:03:00.000000Z", "duration": "PT2.5S", "trace_id": "591f6d3d6b0a1f9e8a71b2c3d4e5f6a7", "span_id": "c7b8c9d0e1f2a3b4", "parent_span_id": "", "name": "POST /checkout", "kind": 2, "status_code": 1, "status_message": "", "resources": { "deployment.environment": "production", "service.name": "checkout-service", "os.type": "linux", "host.name": "ip-10-0-1-23" }, "attributes": { "net.transport": "IP.TCP", "http.scheme": "http", "http.user_agent": "Integration Test", "http.request.method": "POST", "http.response.status_code": "200", "http.request.path": "/checkout" } }
+{ "timestamp": "2026-01-29T10:03:30.000000Z", "duration": "PT2.7S", "trace_id": "591f6d3d6b0a1f9e8a71b2c3d4e5f6a8", "span_id": "c8b9c0d1e2f3a4b5", "parent_span_id": "", "name": "POST /checkout", "kind": 2, "status_code": 1, "status_message": "", "resources": { "deployment.environment": "production", "service.name": "checkout-service", "os.type": "linux", "host.name": "ip-10-0-1-23" }, "attributes": { "net.transport": "IP.TCP", "http.scheme": "http", "http.user_agent": "Integration Test", "http.request.method": "POST", "http.response.status_code": "200", "http.request.path": "/checkout" } }
+{ "timestamp": "2026-01-29T10:04:00.000000Z", "duration": "PT2.9S", "trace_id": "591f6d3d6b0a1f9e8a71b2c3d4e5f6a9", "span_id": "c9b0c1d2e3f4a5b6", "parent_span_id": "", "name": "POST /checkout", "kind": 2, "status_code": 1, "status_message": "", "resources": { "deployment.environment": "production", "service.name": "checkout-service", "os.type": "linux", "host.name": "ip-10-0-1-23" }, "attributes": { "net.transport": "IP.TCP", "http.scheme": "http", "http.user_agent": "Integration Test", "http.request.method": "POST", "http.response.status_code": "200", "http.request.path": "/checkout" } }
+{ "timestamp": "2026-01-29T10:04:30.000000Z", "duration": "PT3.1S", "trace_id": "591f6d3d6b0a1f9e8a71b2c3d4e5f6b1", "span_id": "d1c2d3e4f5a6b7c8", "parent_span_id": "", "name": "POST /checkout", "kind": 2, "status_code": 1, "status_message": "", "resources": { "deployment.environment": "production", "service.name": "checkout-service", "os.type": "linux", "host.name": "ip-10-0-1-23" }, "attributes": { "net.transport": "IP.TCP", "http.scheme": "http", "http.user_agent": "Integration Test", "http.request.method": "POST", "http.response.status_code": "200", "http.request.path": "/checkout" } }
+{ "timestamp": "2026-01-29T10:05:00.000000Z", "duration": "PT3.3S", "trace_id": "591f6d3d6b0a1f9e8a71b2c3d4e5f6b2", "span_id": "d2c3d4e5f6a7b8c9", "parent_span_id": "", "name": "POST /checkout", "kind": 2, "status_code": 1, "status_message": "", "resources": { "deployment.environment": "production", "service.name": "checkout-service", "os.type": "linux", "host.name": "ip-10-0-1-23" }, "attributes": { "net.transport": "IP.TCP", "http.scheme": "http", "http.user_agent": "Integration Test", "http.request.method": "POST", "http.response.status_code": "200", "http.request.path": "/checkout" } }
+{ "timestamp": "2026-01-29T10:05:30.000000Z", "duration": "PT3.5S", "trace_id": "591f6d3d6b0a1f9e8a71b2c3d4e5f6b3", "span_id": "d3c4d5e6f7a8b9c0", "parent_span_id": "", "name": "POST /checkout", "kind": 2, "status_code": 1, "status_message": "", "resources": { "deployment.environment": "production", "service.name": "checkout-service", "os.type": "linux", "host.name": "ip-10-0-1-23" }, "attributes": { "net.transport": "IP.TCP", "http.scheme": "http", "http.user_agent": "Integration Test", "http.request.method": "POST", "http.response.status_code": "200", "http.request.path": "/checkout" } }
+{ "timestamp": "2026-01-29T10:06:00.000000Z", "duration": "PT3.7S", "trace_id": "591f6d3d6b0a1f9e8a71b2c3d4e5f6b4", "span_id": "d4c5d6e7f8a9b0c1", "parent_span_id": "", "name": "POST /checkout", "kind": 2, "status_code": 1, "status_message": "", "resources": { "deployment.environment": "production", "service.name": "checkout-service", "os.type": "linux", "host.name": "ip-10-0-1-23" }, "attributes": { "net.transport": "IP.TCP", "http.scheme": "http", "http.user_agent": "Integration Test", "http.request.method": "POST", "http.response.status_code": "200", "http.request.path": "/checkout" } }
+{ "timestamp": "2026-01-29T10:06:30.000000Z", "duration": "PT3.9S", "trace_id": "591f6d3d6b0a1f9e8a71b2c3d4e5f6b5", "span_id": "d5c6d7e8f9a0b1c2", "parent_span_id": "", "name": "POST /checkout", "kind": 2, "status_code": 1, "status_message": "", "resources": { "deployment.environment": "production", "service.name": "checkout-service", "os.type": "linux", "host.name": "ip-10-0-1-23" }, "attributes": { "net.transport": "IP.TCP", "http.scheme": "http", "http.user_agent": "Integration Test", "http.request.method": "POST", "http.response.status_code": "200", "http.request.path": "/checkout" } }
+{ "timestamp": "2026-01-29T10:07:00.000000Z", "duration": "PT4.1S", "trace_id": "591f6d3d6b0a1f9e8a71b2c3d4e5f6b6", "span_id": "d6c7d8e9f0a1b2c3", "parent_span_id": "", "name": "POST /checkout", "kind": 2, "status_code": 1, "status_message": "", "resources": { "deployment.environment": "production", "service.name": "checkout-service", "os.type": "linux", "host.name": "ip-10-0-1-23" }, "attributes": { "net.transport": "IP.TCP", "http.scheme": "http", "http.user_agent": "Integration Test", "http.request.method": "POST", "http.response.status_code": "200", "http.request.path": "/checkout" } }
+{ "timestamp": "2026-01-29T10:07:30.000000Z", "duration": "PT4.3S", "trace_id": "591f6d3d6b0a1f9e8a71b2c3d4e5f6b7", "span_id": "d7c8d9e0f1a2b3c4", "parent_span_id": "", "name": "POST /checkout", "kind": 2, "status_code": 1, "status_message": "", "resources": { "deployment.environment": "production", "service.name": "checkout-service", "os.type": "linux", "host.name": "ip-10-0-1-23" }, "attributes": { "net.transport": "IP.TCP", "http.scheme": "http", "http.user_agent": "Integration Test", "http.request.method": "POST", "http.response.status_code": "200", "http.request.path": "/checkout" } }
+{ "timestamp": "2026-01-29T10:08:00.000000Z", "duration": "PT4.5S", "trace_id": "591f6d3d6b0a1f9e8a71b2c3d4e5f6b8", "span_id": "d8c9d0e1f2a3b4c5", "parent_span_id": "", "name": "POST /checkout", "kind": 2, "status_code": 1, "status_message": "", "resources": { "deployment.environment": "production", "service.name": "checkout-service", "os.type": "linux", "host.name": "ip-10-0-1-23" }, "attributes": { "net.transport": "IP.TCP", "http.scheme": "http", "http.user_agent": "Integration Test", "http.request.method": "POST", "http.response.status_code": "200", "http.request.path": "/checkout" } }
+{ "timestamp": "2026-01-29T10:08:30.000000Z", "duration": "PT4.7S", "trace_id": "591f6d3d6b0a1f9e8a71b2c3d4e5f6b9", "span_id": "d9c0d1e2f3a4b5c6", "parent_span_id": "", "name": "POST /checkout", "kind": 2, "status_code": 1, "status_message": "", "resources": { "deployment.environment": "production", "service.name": "checkout-service", "os.type": "linux", "host.name": "ip-10-0-1-23" }, "attributes": { "net.transport": "IP.TCP", "http.scheme": "http", "http.user_agent": "Integration Test", "http.request.method": "POST", "http.response.status_code": "200", "http.request.path": "/checkout" } }
+{ "timestamp": "2026-01-29T10:09:00.000000Z", "duration": "PT4.9S", "trace_id": "591f6d3d6b0a1f9e8a71b2c3d4e5f6c1", "span_id": "e1d2e3f4a5b6c7d8", "parent_span_id": "", "name": "POST /checkout", "kind": 2, "status_code": 1, "status_message": "", "resources": { "deployment.environment": "production", "service.name": "checkout-service", "os.type": "linux", "host.name": "ip-10-0-1-23" }, "attributes": { "net.transport": "IP.TCP", "http.scheme": "http", "http.user_agent": "Integration Test", "http.request.method": "POST", "http.response.status_code": "200", "http.request.path": "/checkout" } }
+{ "timestamp": "2026-01-29T10:10:00.000000Z", "duration": "PT5.1S", "trace_id": "591f6d3d6b0a1f9e8a71b2c3d4e5f6c2", "span_id": "e2d3e4f5a6b7c8d9", "parent_span_id": "", "name": "POST /checkout", "kind": 2, "status_code": 1, "status_message": "", "resources": { "deployment.environment": "production", "service.name": "checkout-service", "os.type": "linux", "host.name": "ip-10-0-1-23" }, "attributes": { "net.transport": "IP.TCP", "http.scheme": "http", "http.user_agent": "Integration Test", "http.request.method": "POST", "http.response.status_code": "200", "http.request.path": "/checkout" } }

tests/integration/testdata/alertmanager/content_templating/traces_rule.json

@@ -0,0 +1,71 @@
+{
+  "alert": "content_templating_traces",
+  "ruleType": "threshold_rule",
+  "alertType": "TRACES_BASED_ALERT",
+  "condition": {
+    "thresholds": {
+      "kind": "basic",
+      "spec": [
+        {
+          "name": "critical",
+          "target": 1,
+          "matchType": "1",
+          "op": "1",
+          "channels": [
+            "test channel"
+          ],
+          "targetUnit": "s"
+        }
+      ]
+    },
+    "compositeQuery": {
+      "queryType": "builder",
+      "unit": "ns",
+      "panelType": "graph",
+      "queries": [
+        {
+          "type": "builder_query",
+          "spec": {
+            "name": "A",
+            "signal": "traces",
+            "filter": {
+              "expression": "http.request.path = '/checkout'"
+            },
+            "aggregations": [
+              {
+                "expression": "p90(duration_nano)"
+              }
+            ],
+            "groupBy": [
+              {"name": "service.name", "fieldContext": "resource"}
+            ]
+          }
+        }
+      ]
+    },
+    "selectedQueryName": "A"
+  },
+  "evaluation": {
+    "kind": "rolling",
+    "spec": {
+      "evalWindow": "5m0s",
+      "frequency": "15s"
+    }
+  },
+  "labels": {},
+  "annotations": {
+    "description": "p90 latency high on $service_name",
+    "summary": "p90 latency exceeded threshold on $service_name"
+  },
+  "notificationSettings": {
+    "groupBy": [],
+    "usePolicy": false,
+    "renotify": {
+      "enabled": false,
+      "interval": "30m",
+      "alertStates": []
+    }
+  },
+  "version": "v5",
+  "schemaVersion": "v2alpha1"
+}

tests/integration/tests/alertmanager/01_channels.py

@@ -0,0 +1,268 @@
+import time
+import uuid
+from collections.abc import Callable
+from http import HTTPStatus
+
+import pytest
+import requests
+from sqlalchemy import text
+
+from fixtures import types
+from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
+from fixtures.maildev import (
+    MAILDEV_INCOMING_PASS,
+    SMTP_TEST_FROM,
+    delete_all_mails,
+    verify_email_received,
+)
+from fixtures.notification_channel import assert_email_channel_payload_clean, send_test_notification
+
+TIMEOUT = 10
+
+
+CHANNEL_TYPE_CASES = [
+    (
+        "webhook",
+        lambda sink: {"webhook_configs": [{"url": sink.container_configs["8080"].get("/webhook/crud-original"), "send_resolved": True}]},
+        lambda sink: {"webhook_configs": [{"url": sink.container_configs["8080"].get("/webhook/crud-updated"), "send_resolved": True}]},
+        "crud-original",
+        "crud-updated",
+    ),
+    (
+        "slack",
+        lambda sink: {"slack_configs": [{"api_url": sink.container_configs["8080"].get("/services/T/B/X"), "channel": "#crud-original"}]},
+        lambda sink: {"slack_configs": [{"api_url": sink.container_configs["8080"].get("/services/T/B/X"), "channel": "#crud-updated"}]},
+        "#crud-original",
+        "#crud-updated",
+    ),
+    (
+        "pagerduty",
+        lambda sink: {"pagerduty_configs": [{"routing_key": "crud-original-routing-key"}]},
+        lambda sink: {"pagerduty_configs": [{"routing_key": "crud-updated-routing-key"}]},
+        "crud-original-routing-key",
+        "crud-updated-routing-key",
+    ),
+    (
+        "opsgenie",
+        lambda sink: {"opsgenie_configs": [{"api_key": "crud-original-api-key", "message": "{{ .CommonLabels.alertname }}"}]},
+        lambda sink: {"opsgenie_configs": [{"api_key": "crud-updated-api-key", "message": "{{ .CommonLabels.alertname }}"}]},
+        "crud-original-api-key",
+        "crud-updated-api-key",
+    ),
+    (
+        "msteamsv2",
+        lambda sink: {"msteamsv2_configs": [{"webhook_url": sink.container_configs["8080"].get("/msteams/crud-original")}]},
+        lambda sink: {"msteamsv2_configs": [{"webhook_url": sink.container_configs["8080"].get("/msteams/crud-updated")}]},
+        "crud-original",
+        "crud-updated",
+    ),
+    (
+        "email",
+        lambda sink: {"email_configs": [{"to": "crud-original@integration.test"}]},
+        lambda sink: {"email_configs": [{"to": "crud-updated@integration.test"}]},
+        "crud-original@integration.test",
+        "crud-updated@integration.test",
+    ),
+]
+
+
+@pytest.mark.parametrize(
+    "channel_type,make_config,make_updated_config,created_marker,updated_marker",
+    CHANNEL_TYPE_CASES,
+    ids=[case[0] for case in CHANNEL_TYPE_CASES],
+)
+def test_channel_crud(  # pylint: disable=too-many-arguments,too-many-positional-arguments
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    notification_channel: types.TestContainerDocker,
+    channel_type: str,
+    make_config: Callable[[types.TestContainerDocker], dict],
+    make_updated_config: Callable[[types.TestContainerDocker], dict],
+    created_marker: str,
+    updated_marker: str,
+) -> None:
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+    name = f"crud-{channel_type}-{uuid.uuid4()}"
+
+    config = {"name": name, **make_config(notification_channel)}
+    response = requests.post(signoz.self.host_configs["8080"].get("/api/v1/channels"), json=config, headers={"Authorization": f"Bearer {token}"}, timeout=TIMEOUT)
+    assert response.status_code == HTTPStatus.CREATED, response.text
+    created = response.json()["data"]
+    channel_id = created["id"]
+    assert created["name"] == name
+    assert created["type"] == channel_type
+
+    response = requests.get(signoz.self.host_configs["8080"].get("/api/v1/channels"), headers={"Authorization": f"Bearer {token}"}, timeout=TIMEOUT)
+    assert response.status_code == HTTPStatus.OK, response.text
+    listed = {channel["name"]: channel for channel in response.json()["data"]}
+    assert name in listed
+    assert listed[name]["type"] == channel_type
+
+    response = requests.get(signoz.self.host_configs["8080"].get(f"/api/v1/channels/{channel_id}"), headers={"Authorization": f"Bearer {token}"}, timeout=TIMEOUT)
+    assert response.status_code == HTTPStatus.OK, response.text
+    assert created_marker in response.json()["data"]["data"]
+
+    updated_config = {"name": name, **make_updated_config(notification_channel)}
+    response = requests.put(signoz.self.host_configs["8080"].get(f"/api/v1/channels/{channel_id}"), json=updated_config, headers={"Authorization": f"Bearer {token}"}, timeout=TIMEOUT)
+    assert response.status_code == HTTPStatus.NO_CONTENT, response.text
+
+    response = requests.get(signoz.self.host_configs["8080"].get(f"/api/v1/channels/{channel_id}"), headers={"Authorization": f"Bearer {token}"}, timeout=TIMEOUT)
+    assert response.status_code == HTTPStatus.OK, response.text
+    data = response.json()["data"]["data"]
+    assert updated_marker in data
+    assert created_marker not in data
+
+    response = requests.delete(signoz.self.host_configs["8080"].get(f"/api/v1/channels/{channel_id}"), headers={"Authorization": f"Bearer {token}"}, timeout=TIMEOUT)
+    assert response.status_code == HTTPStatus.NO_CONTENT, response.text
+
+    response = requests.get(signoz.self.host_configs["8080"].get(f"/api/v1/channels/{channel_id}"), headers={"Authorization": f"Bearer {token}"}, timeout=TIMEOUT)
+    assert response.status_code == HTTPStatus.NOT_FOUND, response.text
+
+
+def test_create_rejects_duplicate_name(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    create_notification_channel: Callable[[dict], str],
+) -> None:
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+    name = f"duplicate-{uuid.uuid4()}"
+
+    create_notification_channel({"name": name, "email_configs": [{"to": "first@integration.test"}]})
+
+    response = requests.post(
+        signoz.self.host_configs["8080"].get("/api/v1/channels"),
+        json={"name": name, "email_configs": [{"to": "second@integration.test"}]},
+        headers={"Authorization": f"Bearer {token}"},
+        timeout=TIMEOUT,
+    )
+    assert response.status_code == HTTPStatus.BAD_REQUEST, response.text
+    assert "unique" in response.text
+
+
+def test_create_rejects_channel_without_configs(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+) -> None:
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+
+    response = requests.post(
+        signoz.self.host_configs["8080"].get("/api/v1/channels"),
+        json={"name": f"empty-{uuid.uuid4()}"},
+        headers={"Authorization": f"Bearer {token}"},
+        timeout=TIMEOUT,
+    )
+    assert response.status_code == HTTPStatus.BAD_REQUEST, response.text
+    assert "notification configuration" in response.text
+
+
+def test_update_rejects_name_change(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    create_notification_channel: Callable[[dict], str],
+) -> None:
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+    name = f"rename-{uuid.uuid4()}"
+    channel_id = create_notification_channel({"name": name, "email_configs": [{"to": "rename@integration.test"}]})
+
+    response = requests.put(
+        signoz.self.host_configs["8080"].get(f"/api/v1/channels/{channel_id}"),
+        json={"name": f"{name}-renamed", "email_configs": [{"to": "rename@integration.test"}]},
+        headers={"Authorization": f"Bearer {token}"},
+        timeout=TIMEOUT,
+    )
+    assert response.status_code == HTTPStatus.BAD_REQUEST, response.text
+    assert "cannot update channel name" in response.text
+
+
+def test_channels_require_authentication(signoz: types.SigNoz) -> None:
+    response = requests.get(signoz.self.host_configs["8080"].get("/api/v1/channels"), timeout=TIMEOUT)
+    assert response.status_code == HTTPStatus.UNAUTHORIZED, response.text
+
+
+def test_email_channel_never_stores_or_serves_smtp_settings(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+) -> None:
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+
+    hostile_name = f"hostile-email-{uuid.uuid4()}"
+    hostile_config = {
+        "name": hostile_name,
+        "email_configs": [
+            {
+                "to": "hostile@integration.test",
+                "from": "spoofed@integration.test",
+                "hello": "attacker.test",
+                "smarthost": "smtp.attacker.test:2525",
+                "auth_username": "attacker",
+                "auth_password": "tenant-posted-secret",
+                "require_tls": False,
+                "headers": {"Subject": "hostile subject"},
+            }
+        ],
+    }
+
+    response = requests.post(signoz.self.host_configs["8080"].get("/api/v1/channels"), json=hostile_config, headers={"Authorization": f"Bearer {token}"}, timeout=TIMEOUT)
+    assert response.status_code == HTTPStatus.CREATED, response.text
+    created = response.json()["data"]
+    assert_email_channel_payload_clean(created["data"])
+
+    response = requests.get(signoz.self.host_configs["8080"].get(f"/api/v1/channels/{created['id']}"), headers={"Authorization": f"Bearer {token}"}, timeout=TIMEOUT)
+    assert response.status_code == HTTPStatus.OK, response.text
+    served = response.json()["data"]["data"]
+    assert_email_channel_payload_clean(served)
+    assert "hostile@integration.test" in served
+    assert "hostile subject" in served
+    assert "smtp.attacker.test" not in served
+    assert "tenant-posted-secret" not in served
+
+    response = requests.get(signoz.self.host_configs["8080"].get("/api/v1/channels"), headers={"Authorization": f"Bearer {token}"}, timeout=TIMEOUT)
+    assert response.status_code == HTTPStatus.OK, response.text
+    assert "tenant-posted-secret" not in response.text
+    assert MAILDEV_INCOMING_PASS not in response.text
+
+    with signoz.sqlstore.conn.connect() as conn:
+        stored = conn.execute(
+            text("SELECT data FROM notification_channel WHERE name = :name"),
+            {"name": hostile_name},
+        ).fetchone()
+        assert stored is not None
+        assert_email_channel_payload_clean(stored[0])
+        assert "tenant-posted-secret" not in stored[0]
+
+        configs = conn.execute(text("SELECT config FROM alertmanager_config")).fetchall()
+        assert len(configs) > 0
+        for (config_raw,) in configs:
+            assert MAILDEV_INCOMING_PASS not in config_raw
+            assert "tenant-posted-secret" not in config_raw
+            assert '"smtp_auth_password"' not in config_raw
+            assert '"auth_password"' not in config_raw
+
+
+def test_email_test_channel_delivers_via_env_transport(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    maildev: types.TestContainerDocker,
+) -> None:
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+    delete_all_mails(maildev)
+
+    recipient = f"delivery-{uuid.uuid4()}@integration.test"
+    send_test_notification(
+        signoz,
+        token,
+        {"name": f"delivery-{uuid.uuid4()}", "email_configs": [{"to": recipient}]},
+    )
+
+    deadline = time.time() + 30
+    while time.time() < deadline:
+        if verify_email_received(maildev, {"to": recipient, "from": SMTP_TEST_FROM}):
+            return
+        time.sleep(1)
+    raise AssertionError(f"no email delivered to {recipient} from {SMTP_TEST_FROM}")

tests/integration/tests/alertmanager/02_notifiers.py

@@ -0,0 +1,360 @@
+import json
+import re
+import time
+import uuid
+from collections.abc import Callable
+from datetime import UTC, datetime, timedelta
+
+import pytest
+from wiremock.client import HttpMethods, Mapping, MappingRequest, MappingResponse
+
+from fixtures import types
+from fixtures.alerts import (
+    get_testdata_file_path,
+    update_raw_channel_config,
+    update_rule_channel_name,
+    verify_notification_expectation,
+)
+from fixtures.logger import setup_logger
+from fixtures.maildev import delete_all_mails
+from fixtures.notification_channel import (
+    email_default_config,
+    msteams_default_config,
+    opsgenie_default_config,
+    pagerduty_default_config,
+    slack_default_config,
+    webhook_default_config,
+)
+
+logger = setup_logger(__name__)
+
+NOTIFIERS_TEST = [
+    types.AlertManagerNotificationTestCase(
+        name="slack_notifier_default_templating",
+        rule_path="alerts/test_scenarios/threshold_above_at_least_once/rule.json",
+        alert_data=[
+            types.AlertData(
+                type="metrics",
+                data_path="alerts/test_scenarios/threshold_above_at_least_once/alert_data.jsonl",
+            ),
+        ],
+        channel_config=slack_default_config,
+        notification_expectation=types.AMNotificationExpectation(
+            should_notify=True,
+            # extra wait for alertmanager server setup
+            wait_time_seconds=120,
+            notification_validations=[
+                types.NotificationValidation(
+                    destination_type="webhook",
+                    validation_data={
+                        "path": "/services/TEAM_ID/BOT_ID/TOKEN_ID",
+                        "json_body": {
+                            "username": "Alertmanager",
+                            "attachments": [
+                                {
+                                    "color": "danger",
+                                    "mrkdwn_in": ["fallback", "pretext", "text"],
+                                }
+                            ],
+                        },
+                    },
+                ),
+            ],
+        ),
+    ),
+    types.AlertManagerNotificationTestCase(
+        name="msteams_notifier_default_templating",
+        rule_path="alerts/test_scenarios/threshold_above_at_least_once/rule.json",
+        alert_data=[
+            types.AlertData(
+                type="metrics",
+                data_path="alerts/test_scenarios/threshold_above_at_least_once/alert_data.jsonl",
+            ),
+        ],
+        channel_config=msteams_default_config,
+        notification_expectation=types.AMNotificationExpectation(
+            should_notify=True,
+            wait_time_seconds=120,
+            notification_validations=[
+                types.NotificationValidation(
+                    destination_type="webhook",
+                    validation_data={
+                        "path": "/msteams/webhook_url",
+                        "json_body": {
+                            "type": "message",
+                            "attachments": [
+                                {
+                                    "contentType": "application/vnd.microsoft.card.adaptive",
+                                    "content": {
+                                        "$schema": "http://adaptivecards.io/schemas/adaptive-card.json",
+                                        "type": "AdaptiveCard",
+                                        "version": "1.2",
+                                        "body": [
+                                            {
+                                                "type": "TextBlock",
+                                                "text": "Alerts",
+                                                "weight": "Bolder",
+                                                "size": "Medium",
+                                                "wrap": True,
+                                                "color": "Attention",
+                                            },
+                                            {
+                                                "type": "TextBlock",
+                                                "text": "Labels",
+                                                "weight": "Bolder",
+                                                "size": "Medium",
+                                            },
+                                            {
+                                                "type": "FactSet",
+                                                "text": "",
+                                                "facts": [
+                                                    {
+                                                        "title": "threshold.name",
+                                                        "value": "critical",
+                                                    }
+                                                ],
+                                            },
+                                            {
+                                                "type": "TextBlock",
+                                                "text": "Annotations",
+                                                "weight": "Bolder",
+                                                "size": "Medium",
+                                            },
+                                            {
+                                                "type": "FactSet",
+                                                "text": "",
+                                                "facts": [
+                                                    {
+                                                        "title": "description",
+                                                        "value": "This alert is fired when the defined metric (current value: 15) crosses the threshold (10)",
+                                                    },
+                                                ],
+                                            },
+                                        ],
+                                        "msteams": {"width": "full"},
+                                        "actions": [
+                                            {
+                                                "type": "Action.OpenUrl",
+                                                "title": "View Alert",
+                                            }
+                                        ],
+                                    },
+                                }
+                            ],
+                        },
+                    },
+                ),
+            ],
+        ),
+    ),
+    types.AlertManagerNotificationTestCase(
+        name="pagerduty_notifier_default_templating",
+        rule_path="alerts/test_scenarios/threshold_above_at_least_once/rule.json",
+        alert_data=[
+            types.AlertData(
+                type="metrics",
+                data_path="alerts/test_scenarios/threshold_above_at_least_once/alert_data.jsonl",
+            ),
+        ],
+        channel_config=pagerduty_default_config,
+        notification_expectation=types.AMNotificationExpectation(
+            should_notify=True,
+            wait_time_seconds=120,
+            notification_validations=[
+                types.NotificationValidation(
+                    destination_type="webhook",
+                    validation_data={
+                        "path": "/v2/enqueue",
+                        "json_body": {
+                            "routing_key": "PagerDutyRoutingKey",
+                            "event_action": "trigger",
+                            "payload": {
+                                "source": "SigNoz Alert Manager",
+                                "severity": "critical",
+                                "custom_details": {
+                                    "firing": {
+                                        "Annotations": [
+                                            {"description = This alert is fired when the defined metric (current value": "15) crosses the threshold (10)"},
+                                        ],
+                                        "Labels": [
+                                            "alertname = threshold_above_at_least_once",
+                                            "severity = critical",
+                                            "threshold.name = critical",
+                                        ],
+                                    }
+                                },
+                            },
+                            "client": "SigNoz Alert Manager",
+                            "client_url": "https://enter-signoz-host-n-port-here/alerts",
+                        },
+                    },
+                ),
+            ],
+        ),
+    ),
+    types.AlertManagerNotificationTestCase(
+        name="opsgenie_notifier_default_templating",
+        rule_path="alerts/test_scenarios/threshold_above_at_least_once/rule.json",
+        alert_data=[
+            types.AlertData(
+                type="metrics",
+                data_path="alerts/test_scenarios/threshold_above_at_least_once/alert_data.jsonl",
+            ),
+        ],
+        channel_config=opsgenie_default_config,
+        notification_expectation=types.AMNotificationExpectation(
+            should_notify=True,
+            wait_time_seconds=120,
+            notification_validations=[
+                types.NotificationValidation(
+                    destination_type="webhook",
+                    validation_data={
+                        "path": "/v2/alerts",
+                        "json_body": {
+                            "message": "threshold_above_at_least_once",
+                            "details": {
+                                "alertname": "threshold_above_at_least_once",
+                                "severity": "critical",
+                                "threshold.name": "critical",
+                            },
+                            "priority": "P1",
+                        },
+                    },
+                ),
+            ],
+        ),
+    ),
+    types.AlertManagerNotificationTestCase(
+        name="webhook_notifier_default_templating",
+        rule_path="alerts/test_scenarios/threshold_above_at_least_once/rule.json",
+        alert_data=[
+            types.AlertData(
+                type="metrics",
+                data_path="alerts/test_scenarios/threshold_above_at_least_once/alert_data.jsonl",
+            ),
+        ],
+        channel_config=webhook_default_config,
+        notification_expectation=types.AMNotificationExpectation(
+            should_notify=True,
+            wait_time_seconds=120,
+            notification_validations=[
+                types.NotificationValidation(
+                    destination_type="webhook",
+                    validation_data={
+                        "path": "/webhook/webhook_url",
+                        "json_body": {
+                            "status": "firing",
+                            "alerts": [
+                                {
+                                    "status": "firing",
+                                    "labels": {
+                                        "alertname": "threshold_above_at_least_once",
+                                        "severity": "critical",
+                                        "threshold.name": "critical",
+                                    },
+                                    "annotations": {
+                                        "description": "This alert is fired when the defined metric (current value: 15) crosses the threshold (10)",
+                                        "summary": "This alert is fired when the defined metric (current value: 15) crosses the threshold (10)",
+                                    },
+                                }
+                            ],
+                            "commonLabels": {
+                                "alertname": "threshold_above_at_least_once",
+                                "severity": "critical",
+                                "threshold.name": "critical",
+                            },
+                            "commonAnnotations": {
+                                "description": "This alert is fired when the defined metric (current value: 15) crosses the threshold (10)",
+                                "summary": "This alert is fired when the defined metric (current value: 15) crosses the threshold (10)",
+                            },
+                        },
+                    },
+                ),
+            ],
+        ),
+    ),
+    types.AlertManagerNotificationTestCase(
+        name="email_notifier_default_templating",
+        rule_path="alerts/test_scenarios/threshold_above_at_least_once/rule.json",
+        alert_data=[
+            types.AlertData(
+                type="metrics",
+                data_path="alerts/test_scenarios/threshold_above_at_least_once/alert_data.jsonl",
+            ),
+        ],
+        channel_config=email_default_config,
+        notification_expectation=types.AMNotificationExpectation(
+            should_notify=True,
+            wait_time_seconds=120,
+            notification_validations=[
+                types.NotificationValidation(
+                    destination_type="email",
+                    validation_data={
+                        "subject": re.compile(r'\[FIRING:1\] threshold_above_at_least_once for  \(alertname="threshold_above_at_least_once", ruleSource="http://localhost:8080/alerts/overview\?ruleId=[0-9a-f-]+", severity="critical", threshold\.name="critical"\)'),
+                    },
+                ),
+            ],
+        ),
+    ),
+]
+
+
+@pytest.mark.parametrize(
+    "notifier_test_case",
+    NOTIFIERS_TEST,
+    ids=lambda notifier_test_case: notifier_test_case.name,
+)
+def test_notifier_templating(  # pylint: disable=too-many-arguments,too-many-positional-arguments
+    notification_channel: types.TestContainerDocker,
+    make_http_mocks: Callable[[types.TestContainerDocker, list[Mapping]], None],
+    create_notification_channel: Callable[[dict], str],
+    create_alert_rule: Callable[[dict], str],
+    insert_alert_data: Callable[[list[types.AlertData], datetime], None],
+    maildev: types.TestContainerDocker,
+    notifier_test_case: types.AlertManagerNotificationTestCase,
+):
+    channel_name = str(uuid.uuid4())
+
+    channel_config = update_raw_channel_config(notifier_test_case.channel_config, channel_name, notification_channel)
+    logger.info("Channel config: %s", {"channel_config": channel_config})
+
+    webhook_validations = [v for v in notifier_test_case.notification_expectation.notification_validations if v.destination_type == "webhook"]
+    if len(webhook_validations) > 0:
+        mock_mappings = [
+            Mapping(
+                request=MappingRequest(method=HttpMethods.POST, url=v.validation_data["path"]),
+                response=MappingResponse(status=200, json_body={}),
+                persistent=False,
+            )
+            for v in webhook_validations
+        ]
+
+        make_http_mocks(notification_channel, mock_mappings)
+        logger.info("Mock mappings created")
+
+    if any(v.destination_type == "email" for v in notifier_test_case.notification_expectation.notification_validations):
+        delete_all_mails(maildev)
+        logger.info("Mails deleted")
+
+    create_notification_channel(channel_config)
+    logger.info("Channel created with name: %s", {"channel_name": channel_name})
+
+    time.sleep(12)
+
+    insert_alert_data(
+        notifier_test_case.alert_data,
+        base_time=datetime.now(tz=UTC) - timedelta(minutes=5),
+    )
+
+    rule_path = get_testdata_file_path(notifier_test_case.rule_path)
+    with open(rule_path, encoding="utf-8") as f:
+        rule_data = json.loads(f.read())
+    update_rule_channel_name(rule_data, channel_name)
+    rule_id = create_alert_rule(rule_data)
+    logger.info("rule created: %s", {"rule_id": rule_id, "rule_name": rule_data["alert"]})
+
+    verify_notification_expectation(
+        notification_channel,
+        maildev,
+        notifier_test_case.notification_expectation,
+    )

tests/integration/tests/alertmanager/03_content_templating.py

@@ -0,0 +1,332 @@
+import json
+import re
+import time
+import uuid
+from collections.abc import Callable
+from datetime import UTC, datetime, timedelta
+
+import pytest
+from wiremock.client import HttpMethods, Mapping, MappingRequest, MappingResponse
+
+from fixtures import types
+from fixtures.alerts import (
+    get_testdata_file_path,
+    update_raw_channel_config,
+    update_rule_channel_name,
+    verify_notification_expectation,
+)
+from fixtures.logger import setup_logger
+from fixtures.maildev import delete_all_mails
+from fixtures.notification_channel import (
+    msteams_default_config,
+    opsgenie_default_config,
+    pagerduty_default_config,
+    slack_default_config,
+    webhook_default_config,
+)
+
+logger = setup_logger(__name__)
+
+
+CONTENT_TEMPLATING_TEST = [
+    types.AlertManagerNotificationTestCase(
+        name="msteams_metrics_default_templating",
+        rule_path="alertmanager/content_templating/metrics_rule.json",
+        alert_data=[
+            types.AlertData(
+                type="metrics",
+                data_path="alertmanager/content_templating/metrics_data.jsonl",
+            ),
+        ],
+        channel_config=msteams_default_config,
+        notification_expectation=types.AMNotificationExpectation(
+            should_notify=True,
+            wait_time_seconds=120,
+            notification_validations=[
+                types.NotificationValidation(
+                    destination_type="webhook",
+                    validation_data={
+                        "path": "/msteams/webhook_url",
+                        "json_body": {
+                            "type": "message",
+                            "attachments": [
+                                {
+                                    "contentType": "application/vnd.microsoft.card.adaptive",
+                                    "content": {
+                                        "type": "AdaptiveCard",
+                                        "body": [
+                                            {
+                                                "type": "TextBlock",
+                                                "text": re.compile(
+                                                    r'\[FIRING:1\] content_templating_metrics for  \(alertname="content_templating_metrics", container="checkout", namespace="production", node="ip-10-0-1-23", pod="checkout-7d9c8b5f4-x2k9p", ruleSource="http://localhost:8080/alerts/overview\?ruleId=[0-9a-f-]+", severity="critical", threshold\.name="critical"\)'
+                                                ),
+                                            },
+                                        ],
+                                    },
+                                }
+                            ],
+                        },
+                    },
+                ),
+            ],
+        ),
+    ),
+    types.AlertManagerNotificationTestCase(
+        name="opsgenie_metrics_default_templating",
+        rule_path="alertmanager/content_templating/metrics_rule.json",
+        alert_data=[
+            types.AlertData(
+                type="metrics",
+                data_path="alertmanager/content_templating/metrics_data.jsonl",
+            ),
+        ],
+        channel_config=opsgenie_default_config,
+        notification_expectation=types.AMNotificationExpectation(
+            should_notify=True,
+            wait_time_seconds=120,
+            notification_validations=[
+                types.NotificationValidation(
+                    destination_type="webhook",
+                    validation_data={
+                        "path": "/v2/alerts",
+                        "json_body": {
+                            "message": "content_templating_metrics",
+                            "details": {
+                                "alertname": "content_templating_metrics",
+                                "container": "checkout",
+                                "namespace": "production",
+                                "node": "ip-10-0-1-23",
+                                "pod": "checkout-7d9c8b5f4-x2k9p",
+                                "severity": "critical",
+                                "threshold.name": "critical",
+                            },
+                            "priority": "P1",
+                        },
+                    },
+                ),
+            ],
+        ),
+    ),
+    types.AlertManagerNotificationTestCase(
+        name="pagerduty_metrics_default_templating",
+        rule_path="alertmanager/content_templating/metrics_rule.json",
+        alert_data=[
+            types.AlertData(
+                type="metrics",
+                data_path="alertmanager/content_templating/metrics_data.jsonl",
+            ),
+        ],
+        channel_config=pagerduty_default_config,
+        notification_expectation=types.AMNotificationExpectation(
+            should_notify=True,
+            wait_time_seconds=120,
+            notification_validations=[
+                types.NotificationValidation(
+                    destination_type="webhook",
+                    validation_data={
+                        "path": "/v2/enqueue",
+                        "json_body": {
+                            "routing_key": "PagerDutyRoutingKey",
+                            "payload": {
+                                "severity": "critical",
+                                "custom_details": {
+                                    "firing": {
+                                        "Labels": [
+                                            "alertname = content_templating_metrics",
+                                            "container = checkout",
+                                            "namespace = production",
+                                            "node = ip-10-0-1-23",
+                                            "pod = checkout-7d9c8b5f4-x2k9p",
+                                            "severity = critical",
+                                            "threshold.name = critical",
+                                        ],
+                                    }
+                                },
+                            },
+                            "client": "SigNoz Alert Manager",
+                            "client_url": "https://enter-signoz-host-n-port-here/alerts",
+                        },
+                    },
+                ),
+            ],
+        ),
+    ),
+    types.AlertManagerNotificationTestCase(
+        name="slack_logs_default_templating",
+        rule_path="alertmanager/content_templating/logs_rule.json",
+        alert_data=[
+            types.AlertData(
+                type="logs",
+                data_path="alertmanager/content_templating/logs_data.jsonl",
+            ),
+        ],
+        channel_config=slack_default_config,
+        notification_expectation=types.AMNotificationExpectation(
+            should_notify=True,
+            wait_time_seconds=120,
+            notification_validations=[
+                types.NotificationValidation(
+                    destination_type="webhook",
+                    validation_data={
+                        "path": "/services/TEAM_ID/BOT_ID/TOKEN_ID",
+                        "json_body": {
+                            "username": "Alertmanager",
+                            "attachments": [
+                                {
+                                    "color": "danger",
+                                    "mrkdwn_in": ["fallback", "pretext", "text"],
+                                }
+                            ],
+                        },
+                    },
+                ),
+            ],
+        ),
+    ),
+    types.AlertManagerNotificationTestCase(
+        name="slack_metrics_default_templating",
+        rule_path="alertmanager/content_templating/metrics_rule.json",
+        alert_data=[
+            types.AlertData(
+                type="metrics",
+                data_path="alertmanager/content_templating/metrics_data.jsonl",
+            ),
+        ],
+        channel_config=slack_default_config,
+        notification_expectation=types.AMNotificationExpectation(
+            should_notify=True,
+            wait_time_seconds=120,
+            notification_validations=[
+                types.NotificationValidation(
+                    destination_type="webhook",
+                    validation_data={
+                        "path": "/services/TEAM_ID/BOT_ID/TOKEN_ID",
+                        "json_body": {
+                            "username": "Alertmanager",
+                            "attachments": [
+                                {
+                                    "color": "danger",
+                                    "mrkdwn_in": ["fallback", "pretext", "text"],
+                                }
+                            ],
+                        },
+                    },
+                ),
+            ],
+        ),
+    ),
+    types.AlertManagerNotificationTestCase(
+        name="webhook_metrics_default_templating",
+        rule_path="alertmanager/content_templating/metrics_rule.json",
+        alert_data=[
+            types.AlertData(
+                type="metrics",
+                data_path="alertmanager/content_templating/metrics_data.jsonl",
+            ),
+        ],
+        channel_config=webhook_default_config,
+        notification_expectation=types.AMNotificationExpectation(
+            should_notify=True,
+            wait_time_seconds=120,
+            notification_validations=[
+                types.NotificationValidation(
+                    destination_type="webhook",
+                    validation_data={
+                        "path": "/webhook/webhook_url",
+                        "json_body": {
+                            "status": "firing",
+                            "alerts": [
+                                {
+                                    "status": "firing",
+                                    "labels": {
+                                        "alertname": "content_templating_metrics",
+                                        "container": "checkout",
+                                        "namespace": "production",
+                                        "node": "ip-10-0-1-23",
+                                        "pod": "checkout-7d9c8b5f4-x2k9p",
+                                        "severity": "critical",
+                                        "threshold.name": "critical",
+                                    },
+                                    "annotations": {
+                                        "description": "Container checkout in pod checkout-7d9c8b5f4-x2k9p (production) exceeded memory threshold",
+                                        "summary": "High container memory in production/checkout-7d9c8b5f4-x2k9p",
+                                    },
+                                }
+                            ],
+                            "commonLabels": {
+                                "alertname": "content_templating_metrics",
+                                "container": "checkout",
+                                "namespace": "production",
+                                "node": "ip-10-0-1-23",
+                                "pod": "checkout-7d9c8b5f4-x2k9p",
+                                "severity": "critical",
+                                "threshold.name": "critical",
+                            },
+                        },
+                    },
+                ),
+            ],
+        ),
+    ),
+]
+
+
+@pytest.mark.parametrize(
+    "content_templating_test_case",
+    CONTENT_TEMPLATING_TEST,
+    ids=lambda content_templating_test_case: content_templating_test_case.name,
+)
+def test_content_templating(  # pylint: disable=too-many-arguments,too-many-positional-arguments
+    notification_channel: types.TestContainerDocker,
+    make_http_mocks: Callable[[types.TestContainerDocker, list[Mapping]], None],
+    create_notification_channel: Callable[[dict], str],
+    create_alert_rule: Callable[[dict], str],
+    insert_alert_data: Callable[[list[types.AlertData], datetime], None],
+    maildev: types.TestContainerDocker,
+    content_templating_test_case: types.AlertManagerNotificationTestCase,
+):
+    channel_name = str(uuid.uuid4())
+
+    channel_config = update_raw_channel_config(content_templating_test_case.channel_config, channel_name, notification_channel)
+    logger.info("Channel config: %s", {"channel_config": channel_config})
+
+    webhook_validations = [v for v in content_templating_test_case.notification_expectation.notification_validations if v.destination_type == "webhook"]
+    if len(webhook_validations) > 0:
+        mock_mappings = [
+            Mapping(
+                request=MappingRequest(method=HttpMethods.POST, url=v.validation_data["path"]),
+                response=MappingResponse(status=200, json_body={}),
+                persistent=False,
+            )
+            for v in webhook_validations
+        ]
+
+        make_http_mocks(notification_channel, mock_mappings)
+        logger.info("Mock mappings created")
+
+    if any(v.destination_type == "email" for v in content_templating_test_case.notification_expectation.notification_validations):
+        delete_all_mails(maildev)
+        logger.info("Mails deleted")
+
+    create_notification_channel(channel_config)
+    logger.info("Channel created with name: %s", {"channel_name": channel_name})
+
+    time.sleep(12)
+
+    insert_alert_data(
+        content_templating_test_case.alert_data,
+        base_time=datetime.now(tz=UTC) - timedelta(minutes=10),
+    )
+
+    rule_path = get_testdata_file_path(content_templating_test_case.rule_path)
+    with open(rule_path, encoding="utf-8") as f:
+        rule_data = json.loads(f.read())
+    update_rule_channel_name(rule_data, channel_name)
+    rule_id = create_alert_rule(rule_data)
+    logger.info("rule created: %s", {"rule_id": rule_id, "rule_name": rule_data["alert"]})
+
+    verify_notification_expectation(
+        notification_channel,
+        maildev,
+        content_templating_test_case.notification_expectation,
+    )

tests/integration/tests/alertmanager/conftest.py

@@ -0,0 +1,35 @@
+import pytest
+from testcontainers.core.container import Network
+
+from fixtures import types
+from fixtures.maildev import signoz_smtp_env
+from fixtures.signoz import create_signoz
+
+
+@pytest.fixture(name="signoz", scope="package")
+def signoz(  # pylint: disable=too-many-arguments,too-many-positional-arguments
+    network: Network,
+    zeus: types.TestContainerDocker,
+    gateway: types.TestContainerDocker,
+    sqlstore: types.TestContainerSQL,
+    clickhouse: types.TestContainerClickhouse,
+    request: pytest.FixtureRequest,
+    pytestconfig: pytest.Config,
+    maildev: types.TestContainerDocker,
+    notification_channel: types.TestContainerDocker,
+) -> types.SigNoz:
+    return create_signoz(
+        network=network,
+        zeus=zeus,
+        gateway=gateway,
+        sqlstore=sqlstore,
+        clickhouse=clickhouse,
+        request=request,
+        pytestconfig=pytestconfig,
+        cache_key="signoz_alertmanager",
+        env_overrides={
+            **signoz_smtp_env(maildev),
+            "SIGNOZ_ALERTMANAGER_SIGNOZ_GLOBAL_PAGERDUTY__URL": notification_channel.container_configs["8080"].get("/v2/enqueue"),
+            "SIGNOZ_ALERTMANAGER_SIGNOZ_GLOBAL_OPSGENIE__API__URL": notification_channel.container_configs["8080"].get("/"),
+        },
+    )

tests/integration/tests/alertmanagerrotation/01_smtp_rotation.py

@@ -0,0 +1,99 @@
+import time
+import uuid
+from collections.abc import Callable
+from http import HTTPStatus
+
+import docker
+import pytest
+import requests
+from testcontainers.core.container import Network
+
+from fixtures import types
+from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD, token_getter
+from fixtures.logger import setup_logger
+from fixtures.maildev import (
+    NEW_PROVIDER_SMTP_PASS,
+    SMTP_TEST_FROM,
+    delete_all_mails,
+    get_all_mails,
+    signoz_smtp_env,
+    verify_email_received,
+)
+from fixtures.notification_channel import send_test_notification
+from fixtures.signoz import create_signoz
+
+logger = setup_logger(__name__)
+
+
+def wait_for_email(maildev: types.TestContainerDocker, filters: dict, wait_seconds: int = 30) -> None:
+    deadline = time.time() + wait_seconds
+    while time.time() < deadline:
+        if verify_email_received(maildev, filters):
+            return
+        time.sleep(1)
+    raise AssertionError(f"no email matching {filters} within {wait_seconds}s, inbox: {get_all_mails(maildev)}")
+
+
+def test_smtp_rotation_applies_to_existing_channels(  # pylint: disable=too-many-arguments,too-many-positional-arguments,too-many-locals
+    network: Network,
+    zeus: types.TestContainerDocker,
+    gateway: types.TestContainerDocker,
+    sqlstore: types.TestContainerSQL,
+    clickhouse: types.TestContainerClickhouse,
+    request: pytest.FixtureRequest,
+    pytestconfig: pytest.Config,
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    maildev_old: types.TestContainerDocker,
+    maildev_new: types.TestContainerDocker,
+) -> None:
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+    channel_name = f"rotation-{uuid.uuid4()}"
+    recipient = f"rotation-{uuid.uuid4()}@integration.test"
+
+    response = requests.post(
+        signoz.self.host_configs["8080"].get("/api/v1/channels"),
+        json={"name": channel_name, "email_configs": [{"to": recipient}]},
+        headers={"Authorization": f"Bearer {token}"},
+        timeout=10,
+    )
+    assert response.status_code == HTTPStatus.CREATED, response.text
+
+    delete_all_mails(maildev_old)
+    recipient_old_probe = f"probe-old-{uuid.uuid4()}@integration.test"
+    send_test_notification(signoz, token, {"name": f"probe-{uuid.uuid4()}", "email_configs": [{"to": recipient_old_probe}]})
+    wait_for_email(maildev_old, {"to": recipient_old_probe, "from": SMTP_TEST_FROM})
+    logger.info("Delivery through the old provider verified")
+
+    docker.from_env().containers.get(signoz.self.id).stop()
+    logger.info("Stopped signoz running against the old provider")
+
+    signoz_new = create_signoz(
+        network=network,
+        zeus=zeus,
+        gateway=gateway,
+        sqlstore=sqlstore,
+        clickhouse=clickhouse,
+        request=request,
+        pytestconfig=pytestconfig,
+        cache_key="signoz_smtp_rotation_new",
+        env_overrides=signoz_smtp_env(maildev_new, password=NEW_PROVIDER_SMTP_PASS),
+    )
+    token_new = token_getter(signoz_new)(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+
+    response = requests.get(
+        signoz_new.self.host_configs["8080"].get("/api/v1/channels"),
+        headers={"Authorization": f"Bearer {token_new}"},
+        timeout=10,
+    )
+    assert response.status_code == HTTPStatus.OK, response.text
+    listed = {channel["name"]: channel for channel in response.json()["data"]}
+    assert channel_name in listed
+
+    delete_all_mails(maildev_new)
+    mails_at_old_provider = len(get_all_mails(maildev_old))
+    recipient_new_probe = f"probe-new-{uuid.uuid4()}@integration.test"
+    send_test_notification(signoz_new, token_new, {"name": f"probe-{uuid.uuid4()}", "email_configs": [{"to": recipient_new_probe}]})
+    wait_for_email(maildev_new, {"to": recipient_new_probe, "from": SMTP_TEST_FROM})
+    assert len(get_all_mails(maildev_old)) == mails_at_old_provider, "old provider must receive nothing after rotation"

tests/integration/tests/alertmanagerrotation/conftest.py

@@ -0,0 +1,40 @@
+import pytest
+from testcontainers.core.container import Network
+
+from fixtures import types
+from fixtures.maildev import NEW_PROVIDER_SMTP_PASS, OLD_PROVIDER_SMTP_PASS, create_maildev, signoz_smtp_env
+from fixtures.signoz import create_signoz
+
+
+@pytest.fixture(name="maildev_old", scope="package")
+def maildev_old(network: Network, request: pytest.FixtureRequest, pytestconfig: pytest.Config) -> types.TestContainerDocker:
+    return create_maildev(network, request, pytestconfig, cache_key="maildev_smtp_old", incoming_pass=OLD_PROVIDER_SMTP_PASS)
+
+
+@pytest.fixture(name="maildev_new", scope="package")
+def maildev_new(network: Network, request: pytest.FixtureRequest, pytestconfig: pytest.Config) -> types.TestContainerDocker:
+    return create_maildev(network, request, pytestconfig, cache_key="maildev_smtp_new", incoming_pass=NEW_PROVIDER_SMTP_PASS)
+
+
+@pytest.fixture(name="signoz", scope="package")
+def signoz(  # pylint: disable=too-many-arguments,too-many-positional-arguments
+    network: Network,
+    zeus: types.TestContainerDocker,
+    gateway: types.TestContainerDocker,
+    sqlstore: types.TestContainerSQL,
+    clickhouse: types.TestContainerClickhouse,
+    request: pytest.FixtureRequest,
+    pytestconfig: pytest.Config,
+    maildev_old: types.TestContainerDocker,
+) -> types.SigNoz:
+    return create_signoz(
+        network=network,
+        zeus=zeus,
+        gateway=gateway,
+        sqlstore=sqlstore,
+        clickhouse=clickhouse,
+        request=request,
+        pytestconfig=pytestconfig,
+        cache_key="signoz_smtp_rotation",
+        env_overrides=signoz_smtp_env(maildev_old, password=OLD_PROVIDER_SMTP_PASS),
+    )

tests/integration/tests/querier_json_body/05_search.py

@@ -0,0 +1,253 @@
+import json
+from collections import namedtuple
+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.logs import Logs
+from fixtures.querier import build_order_by, build_raw_query, get_rows, make_query_request
+
+# querierlogs/15_search.py with use_json_body on (see conftest.py): body matches run against
+# body_v2, the map/log fan-out is unchanged, and the response `body` comes back parsed — a
+# plain-string body is {"message": <body>}.
+
+Bodies = namedtuple("Bodies", ["a", "b", "c", "d"])
+
+
+@pytest.mark.parametrize(
+    "expression,expected",
+    [
+        # ── keyless: fans across every field ────────────────────────────────
+        pytest.param("search('login')", lambda b: {b.a}, id="keyless_body"),
+        pytest.param("search('checkout')", lambda b: {b.a, b.d}, id="keyless_body_and_resource"),
+        pytest.param("search('useast')", lambda b: {b.a, b.c}, id="keyless_resource_value"),
+        pytest.param("search('acme')", lambda b: {b.a, b.c}, id="keyless_attribute_value"),
+        pytest.param("search('tenant')", lambda b: {b.a, b.b, b.c, b.d}, id="keyless_attribute_key"),
+        pytest.param("search('error')", lambda b: {b.b, b.d}, id="keyless_severity_case_insensitive"),
+        pytest.param("search('CHECKOUT')", lambda b: {b.a, b.d}, id="keyless_term_case_insensitive"),
+        # ── scoped: narrows to one context ──────────────────────────────────
+        pytest.param("search('login', body)", lambda b: {b.a}, id="scope_body"),
+        pytest.param("search('login', 'body')", lambda b: {b.a}, id="scope_body_quoted"),
+        pytest.param("search('checkout', body)", lambda b: {b.a}, id="scope_body_excludes_resource"),
+        pytest.param("search('checkout', resource)", lambda b: {b.a, b.d}, id="scope_resource"),
+        pytest.param("search('acme', attribute)", lambda b: {b.a, b.c}, id="scope_attribute"),
+        pytest.param("search('error', log)", lambda b: {b.b, b.d}, id="scope_log_severity"),
+        pytest.param("search('acme', body)", lambda b: set(), id="scope_body_no_match"),
+        pytest.param("search('checkout', attribute)", lambda b: set(), id="scope_attribute_no_match"),
+        # ── multiple scopes: union of the named contexts ────────────────────
+        pytest.param("search('login', body, resource)", lambda b: {b.a}, id="scopes_body_resource_body_only"),
+        pytest.param("search('checkout', body, resource)", lambda b: {b.a, b.d}, id="scopes_body_resource_union"),
+        # ── composition with boolean / field filters ────────────────────────
+        pytest.param("NOT search('login')", lambda b: {b.b, b.c, b.d}, id="negated"),
+        pytest.param("search('useast') AND severity_text = 'INFO'", lambda b: {b.a}, id="and_field_filter"),
+    ],
+)
+def test_search(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    insert_logs: Callable[[list[Logs]], None],
+    expression: str,
+    expected: Callable[[Bodies], set[str]],
+) -> None:
+    """Four self-naming logs, each with a token planted in a distinct place (body,
+    resource, attribute, severity), assert search() reaches exactly the right ones."""
+    body = Bodies(
+        a="alpha checkout login ok",  # service checkout / region useast / tenant acme    / INFO
+        b="bravo declined",  # service payment  / region euwest / tenant globex  / ERROR
+        c="charlie miss",  # service cart     / region useast / tenant acme    / WARN
+        d="delta slow",  # service checkout / region apac   / tenant initech / ERROR
+    )
+    # (body, resources, attributes, severity_text)
+    specs = [
+        (body.a, {"service.name": "checkout", "region": "useast"}, {"tenant": "acme"}, "INFO"),
+        (body.b, {"service.name": "payment", "region": "euwest"}, {"tenant": "globex"}, "ERROR"),
+        (body.c, {"service.name": "cart", "region": "useast"}, {"tenant": "acme"}, "WARN"),
+        (body.d, {"service.name": "checkout", "region": "apac"}, {"tenant": "initech"}, "ERROR"),
+    ]
+
+    now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
+    logs = [Logs(timestamp=now - timedelta(seconds=i + 1), resources=res, attributes=attrs, body=b, severity_text=sev) for i, (b, res, attrs, sev) in enumerate(specs)]
+    insert_logs(logs)
+
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+    response = make_query_request(
+        signoz,
+        token,
+        start_ms=int((now - timedelta(minutes=5)).timestamp() * 1000),
+        end_ms=int(now.timestamp() * 1000),
+        request_type="raw",
+        queries=[
+            build_raw_query(
+                "A",
+                "logs",
+                filter_expression=expression,
+                order=[build_order_by("timestamp", "desc"), build_order_by("id", "desc")],
+                limit=100,
+            )
+        ],
+    )
+
+    assert response.status_code == HTTPStatus.OK, response.text
+    assert response.json()["status"] == "success"
+    # body_v2 comes back parsed; a plain-string body is {"message": <body>}.
+    assert {row["data"]["body"]["message"] for row in get_rows(response)} == expected(body)
+
+
+@pytest.mark.parametrize(
+    "term",
+    [
+        pytest.param("eve@acme.io", id="nested_string_value"),
+        pytest.param("503", id="nested_numeric_value"),
+        pytest.param("status", id="nested_key"),
+    ],
+)
+def test_search_body_reaches_nested_json(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    insert_logs: Callable[[list[Logs]], None],
+    term: str,
+) -> None:
+    """A body-scoped search matches values and keys nested inside the body_v2 JSON."""
+    # searchable content lives only in nested fields, not a top-level message
+    nested_body = json.dumps({"user": {"email": "eve@acme.io"}, "http": {"status": 503}}, separators=(",", ":"))
+
+    now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
+    insert_logs([Logs(timestamp=now - timedelta(seconds=1), resources={"service.name": "api"}, body=nested_body, severity_text="INFO")])
+
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+    response = make_query_request(
+        signoz,
+        token,
+        start_ms=int((now - timedelta(minutes=5)).timestamp() * 1000),
+        end_ms=int(now.timestamp() * 1000),
+        request_type="raw",
+        queries=[build_raw_query("A", "logs", filter_expression=f"search('{term}', body)", order=[build_order_by("timestamp", "desc")], limit=100)],
+    )
+
+    assert response.status_code == HTTPStatus.OK, response.text
+    rows = get_rows(response)
+    assert len(rows) == 1
+    assert rows[0]["data"]["body"]["user"]["email"] == "eve@acme.io"
+
+
+@pytest.mark.parametrize(
+    "expression",
+    [
+        pytest.param("search('login', bogus)", id="unknown_scope_word"),
+        pytest.param("search('login', body.message)", id="qualified_field_not_a_scope"),
+    ],
+)
+def test_search_invalid_scope_rejected(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    expression: str,
+) -> None:
+    """A scope that is not a field context (an unknown word, or a qualified
+    `context.field`) is rejected at build time with a 400 — even when, as with
+    `body.message`, it names a real body path."""
+    now = datetime.now(tz=UTC)
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+    response = make_query_request(
+        signoz,
+        token,
+        start_ms=int((now - timedelta(minutes=5)).timestamp() * 1000),
+        end_ms=int(now.timestamp() * 1000),
+        request_type="raw",
+        queries=[build_raw_query("A", "logs", filter_expression=expression, limit=100)],
+    )
+    assert response.status_code == HTTPStatus.BAD_REQUEST, response.text
+    assert "invalid search scope" in response.text
+
+
+# The querier gates search() on EXPLAIN ESTIMATE against search_max_scan_rows_json_body on
+# this path (50000 here, see conftest.py). Both recoveries the advisory suggests — a
+# selective filter, a narrower range — are exercised below.
+
+
+def test_search_cost_guard_trips_then_passes_with_filter(
+    signoz_search_scan_budget: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    insert_logs: Callable[[list[Logs]], None],
+) -> None:
+    """A broad search() over ~61000 logs exceeds the 50000-row budget and is rejected;
+    the same search narrowed to the 1000-log 'checkout' service scans under budget and
+    succeeds — the advisory's "add a more selective filter" made real."""
+    now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
+    # 60000 'catalog' logs now + 1000 'checkout' logs ~45m back, in an earlier ts_bucket, so
+    # the checkout fingerprint owns its own marks and a resource filter on it prunes the scan.
+    logs = [Logs(timestamp=now - timedelta(seconds=1 + i % 30), resources={"service.name": "catalog"}, body="log line") for i in range(60000)]
+    logs += [Logs(timestamp=now - timedelta(minutes=45, seconds=i % 30), resources={"service.name": "checkout"}, body="log line") for i in range(1000)]
+    insert_logs(logs)
+
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+    start_ms = int((now - timedelta(minutes=60)).timestamp() * 1000)
+    end_ms = int((now + timedelta(minutes=1)).timestamp() * 1000)
+
+    def run(expression: str):
+        return make_query_request(
+            signoz_search_scan_budget,
+            token,
+            start_ms=start_ms,
+            end_ms=end_ms,
+            request_type="raw",
+            queries=[build_raw_query("A", "logs", filter_expression=expression, order=[build_order_by("timestamp", "desc")], limit=100)],
+        )
+
+    # Broad search over the whole range is over budget -> rejected before executing.
+    over_budget = run("search('log')")
+    assert over_budget.status_code == HTTPStatus.BAD_REQUEST, over_budget.text
+    assert "over the per-shard limit" in over_budget.text
+    # The advisory leads the suggestions, then how to get under budget.
+    assert "runs across all fields" in over_budget.text
+    assert "Narrow the time range or add a more selective filter." in over_budget.text
+
+    # Adding a selective resource filter prunes the scan under budget -> runs.
+    within_budget = run("search('log') AND resource.service.name = 'checkout'")
+    assert within_budget.status_code == HTTPStatus.OK, within_budget.text
+    assert len(get_rows(within_budget)) > 0
+
+
+def test_search_cost_guard_passes_with_narrower_time_range(
+    signoz_search_scan_budget: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    insert_logs: Callable[[list[Logs]], None],
+) -> None:
+    """A steady stream of ~80000 logs (~4/sec over the last ~5.5h). A search() over the
+    whole window is over the 50000-row budget and rejected; the same search over the last
+    15 minutes scans only a few thousand rows and runs — the advisory's "narrow the time
+    range" made real."""
+    now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
+    # ~4 logs/sec, oldest ~5.5h back, newest ~now.
+    logs = [Logs(timestamp=now - timedelta(seconds=i // 4), resources={"service.name": "app"}, body="log line") for i in range(80000)]
+    insert_logs(logs)
+
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+
+    def run(lookback_minutes: int):
+        return make_query_request(
+            signoz_search_scan_budget,
+            token,
+            start_ms=int((now - timedelta(minutes=lookback_minutes)).timestamp() * 1000),
+            end_ms=int((now + timedelta(minutes=1)).timestamp() * 1000),
+            request_type="raw",
+            queries=[build_raw_query("A", "logs", filter_expression="search('log')", order=[build_order_by("timestamp", "desc")], limit=100)],
+        )
+
+    # The last 6 hours cover the whole stream (~80000) -> over budget -> rejected.
+    wide = run(360)
+    assert wide.status_code == HTTPStatus.BAD_REQUEST, wide.text
+    assert "over the per-shard limit" in wide.text
+
+    # The last 15 minutes hold only ~3600 logs -> under budget -> runs, returning rows.
+    narrow = run(15)
+    assert narrow.status_code == HTTPStatus.OK, narrow.text
+    assert len(get_rows(narrow)) > 0

tests/integration/tests/querier_json_body/conftest.py

@@ -55,3 +55,34 @@ def signoz_json_body(
             "SIGNOZ_FLAGGER_CONFIG_BOOLEAN_USE__JSON__BODY": True,
         },
     )
+
+
+@pytest.fixture(name="signoz_search_scan_budget", scope="package")
+def signoz_search_scan_budget(
+    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:
+    """The querierlogs budget instance over body_v2: same 50000 rows, but on
+    search_max_scan_rows_json_body. search_max_scan_rows keeps its 60M default, so only the
+    body_v2 budget can trip here. Shares the default instance's sqlstore + clickhouse, so
+    the same admin token and seeded logs work against it."""
+    return create_signoz(
+        network=network,
+        zeus=zeus,
+        gateway=gateway,
+        sqlstore=sqlstore,
+        clickhouse=clickhouse,
+        request=request,
+        pytestconfig=pytestconfig,
+        cache_key="signoz-json-body-search-scan-budget-50k",
+        env_overrides={
+            "SIGNOZ_FLAGGER_CONFIG_BOOLEAN_USE__JSON__BODY": True,
+            "SIGNOZ_QUERIER_SEARCH__MAX__SCAN__ROWS__JSON__BODY": 50000,
+        },
+    )

tests/integration/tests/querierlogs/15_search.py

@@ -0,0 +1,212 @@
+from collections import namedtuple
+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.logs import Logs
+from fixtures.querier import build_order_by, build_raw_query, get_column_data_from_response, get_rows, make_query_request
+
+# search(): keyless fans across every field; scoped search('term', <ctx>...) narrows to
+# the named contexts (body/attribute/resource/log). Flag off here, so body matches the
+# `body` String column (querier_json_body mirrors this over body_v2).
+
+Bodies = namedtuple("Bodies", ["a", "b", "c", "d"])
+
+
+@pytest.mark.parametrize(
+    "expression,expected",
+    [
+        # ── keyless: fans across every field ────────────────────────────────
+        pytest.param("search('login')", lambda b: {b.a}, id="keyless_body"),
+        pytest.param("search('checkout')", lambda b: {b.a, b.d}, id="keyless_body_and_resource"),
+        pytest.param("search('useast')", lambda b: {b.a, b.c}, id="keyless_resource_value"),
+        pytest.param("search('acme')", lambda b: {b.a, b.c}, id="keyless_attribute_value"),
+        pytest.param("search('tenant')", lambda b: {b.a, b.b, b.c, b.d}, id="keyless_attribute_key"),
+        pytest.param("search('error')", lambda b: {b.b, b.d}, id="keyless_severity_case_insensitive"),
+        pytest.param("search('CHECKOUT')", lambda b: {b.a, b.d}, id="keyless_term_case_insensitive"),
+        # ── scoped: narrows to one context ──────────────────────────────────
+        pytest.param("search('login', body)", lambda b: {b.a}, id="scope_body"),
+        pytest.param("search('login', 'body')", lambda b: {b.a}, id="scope_body_quoted"),
+        pytest.param("search('checkout', body)", lambda b: {b.a}, id="scope_body_excludes_resource"),
+        pytest.param("search('checkout', resource)", lambda b: {b.a, b.d}, id="scope_resource"),
+        pytest.param("search('acme', attribute)", lambda b: {b.a, b.c}, id="scope_attribute"),
+        pytest.param("search('error', log)", lambda b: {b.b, b.d}, id="scope_log_severity"),
+        pytest.param("search('acme', body)", lambda b: set(), id="scope_body_no_match"),
+        pytest.param("search('checkout', attribute)", lambda b: set(), id="scope_attribute_no_match"),
+        # ── multiple scopes: union of the named contexts ────────────────────
+        pytest.param("search('login', body, resource)", lambda b: {b.a}, id="scopes_body_resource_body_only"),
+        pytest.param("search('checkout', body, resource)", lambda b: {b.a, b.d}, id="scopes_body_resource_union"),
+        # ── composition with boolean / field filters ────────────────────────
+        pytest.param("NOT search('login')", lambda b: {b.b, b.c, b.d}, id="negated"),
+        pytest.param("search('useast') AND severity_text = 'INFO'", lambda b: {b.a}, id="and_field_filter"),
+    ],
+)
+def test_search(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    insert_logs: Callable[[list[Logs]], None],
+    expression: str,
+    expected: Callable[[Bodies], set[str]],
+) -> None:
+    """Four self-naming logs, each with a token planted in a distinct place (body,
+    resource, attribute, severity), assert search() reaches exactly the right ones."""
+    body = Bodies(
+        a="alpha checkout login ok",  # service checkout / region useast / tenant acme    / INFO
+        b="bravo declined",  # service payment  / region euwest / tenant globex  / ERROR
+        c="charlie miss",  # service cart     / region useast / tenant acme    / WARN
+        d="delta slow",  # service checkout / region apac   / tenant initech / ERROR
+    )
+    # (body, resources, attributes, severity_text)
+    specs = [
+        (body.a, {"service.name": "checkout", "region": "useast"}, {"tenant": "acme"}, "INFO"),
+        (body.b, {"service.name": "payment", "region": "euwest"}, {"tenant": "globex"}, "ERROR"),
+        (body.c, {"service.name": "cart", "region": "useast"}, {"tenant": "acme"}, "WARN"),
+        (body.d, {"service.name": "checkout", "region": "apac"}, {"tenant": "initech"}, "ERROR"),
+    ]
+
+    now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
+    logs = [Logs(timestamp=now - timedelta(seconds=i + 1), resources=res, attributes=attrs, body=b, severity_text=sev) for i, (b, res, attrs, sev) in enumerate(specs)]
+    insert_logs(logs)
+
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+    response = make_query_request(
+        signoz,
+        token,
+        start_ms=int((now - timedelta(minutes=5)).timestamp() * 1000),
+        end_ms=int(now.timestamp() * 1000),
+        request_type="raw",
+        queries=[
+            build_raw_query(
+                "A",
+                "logs",
+                filter_expression=expression,
+                order=[build_order_by("timestamp", "desc"), build_order_by("id", "desc")],
+                limit=100,
+            )
+        ],
+    )
+
+    assert response.status_code == HTTPStatus.OK, response.text
+    assert response.json()["status"] == "success"
+    assert set(get_column_data_from_response(response.json(), "body")) == expected(body)
+
+
+@pytest.mark.parametrize(
+    "expression",
+    [
+        pytest.param("search('login', bogus)", id="unknown_scope_word"),
+        pytest.param("search('login', body.message)", id="qualified_field_not_a_scope"),
+    ],
+)
+def test_search_invalid_scope_rejected(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    expression: str,
+) -> None:
+    """A scope that is not a field context (an unknown word, or a qualified
+    `context.field`) is rejected at build time with a 400."""
+    now = datetime.now(tz=UTC)
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+    response = make_query_request(
+        signoz,
+        token,
+        start_ms=int((now - timedelta(minutes=5)).timestamp() * 1000),
+        end_ms=int(now.timestamp() * 1000),
+        request_type="raw",
+        queries=[build_raw_query("A", "logs", filter_expression=expression, limit=100)],
+    )
+    assert response.status_code == HTTPStatus.BAD_REQUEST, response.text
+    assert "invalid search scope" in response.text
+
+
+# The querier gates search() on EXPLAIN ESTIMATE against search_max_scan_rows (50000 here,
+# see conftest.py). Both recoveries the advisory suggests — a selective filter, a narrower
+# range — are exercised below.
+
+
+def test_search_cost_guard_trips_then_passes_with_filter(
+    signoz_search_scan_budget: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    insert_logs: Callable[[list[Logs]], None],
+) -> None:
+    """A broad search() over ~61000 logs exceeds the 50000-row budget and is rejected;
+    the same search narrowed to the 1000-log 'checkout' service scans under budget and
+    succeeds — the advisory's "add a more selective filter" made real."""
+    now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
+    # 60000 'catalog' logs now + 1000 'checkout' logs ~45m back, in an earlier ts_bucket, so
+    # the checkout fingerprint owns its own marks and a resource filter on it prunes the scan.
+    logs = [Logs(timestamp=now - timedelta(seconds=1 + i % 30), resources={"service.name": "catalog"}, body="log line") for i in range(60000)]
+    logs += [Logs(timestamp=now - timedelta(minutes=45, seconds=i % 30), resources={"service.name": "checkout"}, body="log line") for i in range(1000)]
+    insert_logs(logs)
+
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+    start_ms = int((now - timedelta(minutes=60)).timestamp() * 1000)
+    end_ms = int((now + timedelta(minutes=1)).timestamp() * 1000)
+
+    def run(expression: str):
+        return make_query_request(
+            signoz_search_scan_budget,
+            token,
+            start_ms=start_ms,
+            end_ms=end_ms,
+            request_type="raw",
+            queries=[build_raw_query("A", "logs", filter_expression=expression, order=[build_order_by("timestamp", "desc")], limit=100)],
+        )
+
+    # Broad search over the whole range is over budget -> rejected before executing.
+    over_budget = run("search('log')")
+    assert over_budget.status_code == HTTPStatus.BAD_REQUEST, over_budget.text
+    assert "over the per-shard limit" in over_budget.text
+    # The advisory leads the suggestions, then how to get under budget.
+    assert "runs across all fields" in over_budget.text
+    assert "Narrow the time range or add a more selective filter." in over_budget.text
+
+    # Adding a selective resource filter prunes the scan under budget -> runs.
+    within_budget = run("search('log') AND resource.service.name = 'checkout'")
+    assert within_budget.status_code == HTTPStatus.OK, within_budget.text
+    assert len(get_rows(within_budget)) > 0
+
+
+def test_search_cost_guard_passes_with_narrower_time_range(
+    signoz_search_scan_budget: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    insert_logs: Callable[[list[Logs]], None],
+) -> None:
+    """A steady stream of ~80000 logs (~4/sec over the last ~5.5h). A search() over the
+    whole window is over the 50000-row budget and rejected; the same search over the last
+    15 minutes scans only a few thousand rows and runs — the advisory's "narrow the time
+    range" made real."""
+    now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
+    # ~4 logs/sec, oldest ~5.5h back, newest ~now.
+    logs = [Logs(timestamp=now - timedelta(seconds=i // 4), resources={"service.name": "app"}, body="log line") for i in range(80000)]
+    insert_logs(logs)
+
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+
+    def run(lookback_minutes: int):
+        return make_query_request(
+            signoz_search_scan_budget,
+            token,
+            start_ms=int((now - timedelta(minutes=lookback_minutes)).timestamp() * 1000),
+            end_ms=int((now + timedelta(minutes=1)).timestamp() * 1000),
+            request_type="raw",
+            queries=[build_raw_query("A", "logs", filter_expression="search('log')", order=[build_order_by("timestamp", "desc")], limit=100)],
+        )
+
+    # The last 6 hours cover the whole stream (~80000) -> over budget -> rejected.
+    wide = run(360)
+    assert wide.status_code == HTTPStatus.BAD_REQUEST, wide.text
+    assert "over the per-shard limit" in wide.text
+
+    # The last 15 minutes hold only ~3600 logs -> under budget -> runs, returning rows.
+    narrow = run(15)
+    assert narrow.status_code == HTTPStatus.OK, narrow.text
+    assert len(get_rows(narrow)) > 0

tests/integration/tests/querierlogs/conftest.py

@@ -0,0 +1,34 @@
+import pytest
+from testcontainers.core.container import Network
+
+from fixtures import types
+from fixtures.signoz import create_signoz
+
+
+@pytest.fixture(name="signoz_search_scan_budget", scope="package")
+def signoz_search_scan_budget(
+    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:
+    """SigNoz with a low search_max_scan_rows (50000) so a broad search() trips the cost
+    guard while a selective one stays under it. Shares the default instance's sqlstore +
+    clickhouse, so the same admin token and seeded logs work against it."""
+    return create_signoz(
+        network=network,
+        zeus=zeus,
+        gateway=gateway,
+        sqlstore=sqlstore,
+        clickhouse=clickhouse,
+        request=request,
+        pytestconfig=pytestconfig,
+        cache_key="signoz-search-scan-budget-50k",
+        env_overrides={
+            "SIGNOZ_QUERIER_SEARCH__MAX__SCAN__ROWS": 50000,
+        },
+    )

tests/pyproject.toml

@@ -75,3 +75,7 @@ ignore = [
 
 [tool.ruff.format]
 # Defaults align with black (double quotes, 4-space indent).
+
+[tool.ruff.lint.per-file-ignores]
+"fixtures/notification_channel.py" = ["E501"]
+"integration/tests/alertmanager/*" = ["E501"]