Merge branch 'main' into issue_8965

* chore: migrate antd ProgressBar to signoz ui component

* fix: homepage progress bar leaking section, resolve comments

* fix: remove stripe animation from progress bars in api monitoring section

* revert: accidental unrelated files

.github/CODEOWNERS

@@ -118,6 +118,9 @@ go.mod @therealpandey
 
 /tests/integration/ @therealpandey
 
+# e2e tests
+/tests/e2e/ @AshwinBhatkal
+
 # Flagger Owners
 
 /pkg/flagger/ @therealpandey
@@ -162,3 +165,7 @@ go.mod @therealpandey
 /frontend/src/lib/dashboard/ @SigNoz/pulse-frontend
 /frontend/src/lib/dashboardVariables/ @SigNoz/pulse-frontend
 /frontend/src/components/NewSelect/ @SigNoz/pulse-frontend
+
+## Dashboard V2
+/frontend/src/pages/DashboardPageV2/ @SigNoz/pulse-frontend
+/frontend/src/pages/DashboardsListPageV2/ @SigNoz/pulse-frontend

frontend/plugins/rules/no-antd-components.mjs

@@ -18,6 +18,7 @@ const BANNED_COMPONENTS = {
 		'Use @signozhq/ui/typography Typography instead of antd Typography.',
 	Switch: 'Use @signozhq/ui/switch Switch instead of antd Switch.',
 	Badge: 'Use @signozhq/ui/badge instead of antd Badge.',
+	Progress: 'Use @signozhq/ui/progress instead of antd Progress.',
 };
 
 export default {

frontend/src/components/CeleryOverview/CeleryOverviewTable/CeleryOverviewTable.styles.scss

@@ -51,13 +51,6 @@
 				background: var(--l1-background);
 			}
 
-			.progress-container {
-				.ant-progress-bg {
-					height: 8px !important;
-					border-radius: 4px;
-				}
-			}
-
 			.ant-table-tbody > tr:hover > td {
 				background: color-mix(in srgb, var(--l1-foreground) 4%, transparent);
 			}

frontend/src/components/CeleryOverview/CeleryOverviewTable/CeleryOverviewTable.tsx

@@ -9,13 +9,13 @@ import {
 	Flex,
 	Input,
 	InputRef,
-	Progress,
 	Space,
 	Spin,
 	TableColumnsType,
 	TableColumnType,
 	Tooltip,
 } from 'antd';
+import { Progress } from '@signozhq/ui/progress';
 import { Typography } from '@signozhq/ui/typography';
 import type { FilterDropdownProps } from 'antd/lib/table/interface';
 import logEvent from 'api/common/logEvent';
@@ -59,7 +59,7 @@ function ProgressRender(item: string | number): JSX.Element {
 			<Progress
 				percent={percent}
 				strokeLinecap="butt"
-				size="small"
+				showInfo
 				strokeColor={((): string => {
 					const cpuPercent = percent;
 					if (cpuPercent >= 90) {

frontend/src/container/AlertHistory/Statistics/TopContributorsCard/TopContributorsCard.styles.scss

@@ -45,6 +45,10 @@
 	.contributors-row {
 		height: 80px;
 	}
+	.top-contributors-progress {
+		--progress-background: transparent;
+	}
+
 	&__content {
 		.ant-table {
 			&-cell {

frontend/src/container/AlertHistory/Statistics/TopContributorsCard/TopContributorsRows.tsx

@@ -1,6 +1,7 @@
 import { HTMLAttributes } from 'react';
 import { Color } from '@signozhq/design-tokens';
-import { Progress, Table, TableColumnsType as ColumnsType } from 'antd';
+import { Table, TableColumnsType as ColumnsType } from 'antd';
+import { Progress } from '@signozhq/ui/progress';
 import logEvent from 'api/common/logEvent';
 import { ConditionalAlertPopover } from 'container/AlertHistory/AlertPopover/AlertPopover';
 import AlertLabels from 'pages/AlertDetails/AlertHeader/AlertLabels/AlertLabels';
@@ -51,8 +52,8 @@ function TopContributorsRows({
 					<Progress
 						percent={(count / totalCurrentTriggers) * 100}
 						showInfo={false}
-						trailColor="rgba(255, 255, 255, 0)"
 						strokeColor={Color.BG_ROBIN_500}
+						className="top-contributors-progress"
 					/>
 				</ConditionalAlertPopover>
 			),

frontend/src/container/ApiMonitoring/Explorer/Domains/DomainDetails/DomainDetails.styles.scss

@@ -141,12 +141,9 @@
 
 			.progress-container {
 				width: 158px;
-				.ant-progress {
-					margin: 0;
 
-					.ant-progress-text {
-						font-weight: 600;
-					}
+				span {
+					font-weight: 600;
 				}
 			}
 

frontend/src/container/ApiMonitoring/Explorer/Domains/DomainDetails/components/DomainMetrics.tsx

@@ -1,7 +1,8 @@
 import { useMemo } from 'react';
 import { useQueries } from 'react-query';
 import { Color } from '@signozhq/design-tokens';
-import { Progress, Skeleton, Tooltip } from 'antd';
+import { Skeleton, Tooltip } from 'antd';
+import { Progress } from '@signozhq/ui/progress';
 import { Typography } from '@signozhq/ui/typography';
 import { ENTITY_VERSION_V5 } from 'constants/app';
 import { REACT_QUERY_KEY } from 'constants/reactQueryKeys';
@@ -136,12 +137,11 @@ function DomainMetrics({
 							<Tooltip title={formattedDomainMetricsData.errorRate}>
 								{formattedDomainMetricsData.errorRate !== '-' ? (
 									<Progress
-										status="active"
 										percent={Number(
 											Number(formattedDomainMetricsData.errorRate).toFixed(2),
 										)}
 										strokeLinecap="butt"
-										size="small"
+										showInfo
 										strokeColor={((): string => {
 											const errorRatePercent = Number(
 												Number(formattedDomainMetricsData.errorRate).toFixed(2),

frontend/src/container/ApiMonitoring/Explorer/Domains/DomainDetails/components/EndPointMetrics.tsx

@@ -1,7 +1,8 @@
 import { useMemo } from 'react';
 import { UseQueryResult } from 'react-query';
 import { Color } from '@signozhq/design-tokens';
-import { Progress, Skeleton, Tooltip } from 'antd';
+import { Skeleton, Tooltip } from 'antd';
+import { Progress } from '@signozhq/ui/progress';
 import { Typography } from '@signozhq/ui/typography';
 import {
 	getDisplayValue,
@@ -80,10 +81,9 @@ function EndPointMetrics({
 							<Tooltip title={metricsData?.errorRate}>
 								{metricsData?.errorRate !== '-' ? (
 									<Progress
-										status="active"
 										percent={Number(Number(metricsData?.errorRate ?? 0).toFixed(2))}
 										strokeLinecap="butt"
-										size="small"
+										showInfo
 										strokeColor={((): string => {
 											const errorRatePercent = Number(
 												Number(metricsData?.errorRate ?? 0).toFixed(2),

frontend/src/container/ApiMonitoring/utils.tsx

@@ -1,6 +1,7 @@
 import { ReactNode } from 'react';
 import { Color } from '@signozhq/design-tokens';
-import { Progress, TableColumnType as ColumnType, Tag, Tooltip } from 'antd';
+import { TableColumnType as ColumnType, Tag, Tooltip } from 'antd';
+import { Progress } from '@signozhq/ui/progress';
 import { convertFiltersToExpressionWithExistingQuery } from 'components/QueryBuilderV2/utils';
 import {
 	FiltersType,
@@ -257,10 +258,9 @@ export const columnsConfig: ColumnType<APIDomainsRowData>[] = [
 				errorRate === 'n/a' || errorRate === '-' ? 0 : errorRate;
 			return (
 				<Progress
-					status="active"
 					percent={Number((errorRateValue as number).toFixed(2))}
 					strokeLinecap="butt"
-					size="small"
+					showInfo
 					strokeColor={((): string => {
 						const errorRatePercent = Number((errorRateValue as number).toFixed(2));
 						if (errorRatePercent >= 90) {
@@ -1022,14 +1022,13 @@ export const getEndPointsColumnsConfig = (
 		className: `column`,
 		render: (errorRate: number | string): React.ReactNode => (
 			<Progress
-				status="active"
 				percent={Number(
 					(
 						(errorRate === 'n/a' || errorRate === '-' ? 0 : errorRate) as number
 					).toFixed(1),
 				)}
 				strokeLinecap="butt"
-				size="small"
+				showInfo
 				strokeColor={((): string => {
 					const errorRatePercent = Number((errorRate as number).toFixed(1));
 					if (errorRatePercent >= 90) {
@@ -2514,10 +2513,9 @@ export const dependentServicesColumns: ColumnType<DependentServicesData>[] = [
 		render: (errorPercentage: number | string): React.ReactNode =>
 			errorPercentage !== '-' ? (
 				<Progress
-					status="active"
 					percent={Number((errorPercentage as number).toFixed(2))}
 					strokeLinecap="butt"
-					size="small"
+					showInfo
 					strokeColor={((): string => {
 						const errorPercentagePercent = Number(
 							(errorPercentage as number).toFixed(2),
@@ -3022,14 +3020,13 @@ export const getAllEndpointsWidgetData = (
 		),
 		F1: (errorRate: any): ReactNode => (
 			<Progress
-				status="active"
 				percent={Number(
 					(
 						(errorRate === 'n/a' || errorRate === '-' ? 0 : errorRate) as number
 					).toFixed(2),
 				)}
 				strokeLinecap="butt"
-				size="small"
+				showInfo
 				strokeColor={((): string => {
 					const errorRatePercent = Number(
 						(

frontend/src/container/Home/StepsProgress/StepsProgress.styles.scss

@@ -39,7 +39,5 @@
 
 	width: 100% !important;
 
-	.ant-progress-steps-outer {
-		width: 100% !important;
-	}
+	--progress-width: 100%;
 }

frontend/src/container/Home/StepsProgress/StepsProgress.tsx

@@ -1,4 +1,4 @@
-import { Progress } from 'antd';
+import { Progress } from '@signozhq/ui/progress';
 
 import { ChecklistItem } from '../HomeChecklist/HomeChecklist';
 
@@ -15,9 +15,7 @@ function StepsProgress({
 
 	const totalChecklistItems = checklistItems.length;
 
-	const progress = Math.round(
-		(completedChecklistItems.length / totalChecklistItems) * 100,
-	);
+	const progress = (completedChecklistItems.length / totalChecklistItems) * 100;
 
 	return (
 		<div className="steps-progress-container">

frontend/src/container/InfraMonitoringHosts/constants.tsx

@@ -1,6 +1,7 @@
 import React from 'react';
 import { Color } from '@signozhq/design-tokens';
-import { Progress, Tag } from 'antd';
+import { Tag } from 'antd';
+import { Progress } from '@signozhq/ui/progress';
 import { Typography } from '@signozhq/ui/typography';
 import {
 	getHostLists,
@@ -79,8 +80,8 @@ export const hostDetailsMetadataConfig: K8sDetailsMetadataConfig<HostData>[] = [
 		render: (value): React.ReactNode => (
 			<Progress
 				percent={Number(Number(value).toFixed(1))}
-				size="small"
 				strokeColor={getProgressColor(Number(value))}
+				showInfo
 			/>
 		),
 	},
@@ -90,8 +91,8 @@ export const hostDetailsMetadataConfig: K8sDetailsMetadataConfig<HostData>[] = [
 		render: (value): React.ReactNode => (
 			<Progress
 				percent={Number(Number(value).toFixed(1))}
-				size="small"
 				strokeColor={getMemoryProgressColor(Number(value))}
+				showInfo
 			/>
 		),
 	},

frontend/src/container/InfraMonitoringHosts/table.module.scss

@@ -60,11 +60,6 @@
 	& > div {
 		width: 100%;
 	}
-
-	:global(.ant-progress-bg) {
-		height: 8px !important;
-		border-radius: 4px;
-	}
 }
 
 .progressBar {

frontend/src/container/InfraMonitoringK8s/EntityDetailsUtils/entityDetails.styles.scss

@@ -103,12 +103,8 @@
 			.progress-container {
 				width: 158px;
 
-				.ant-progress {
-					margin: 0;
-
-					.ant-progress-text {
-						font-weight: 600;
-					}
+				span {
+					font-weight: 600;
 				}
 			}
 
@@ -292,10 +288,6 @@
 			}
 
 			.progress-container {
-				.ant-progress-bg {
-					height: 8px !important;
-					border-radius: 4px;
-				}
 			}
 
 			.ant-table-tbody > tr:hover > td {

frontend/src/container/InfraMonitoringK8s/components/EntityProgressBar.tsx

@@ -1,4 +1,4 @@
-import { Progress } from 'antd';
+import { Progress } from '@signozhq/ui/progress';
 import TanStackTable from 'components/TanStackTableView';
 import {
 	getMemoryProgressColor,
@@ -53,7 +53,6 @@ export function EntityProgressBar({
 			<Progress
 				percent={percentage}
 				strokeLinecap="butt"
-				size="small"
 				status="normal"
 				strokeColor={getStrokeColor(type, value)}
 				className={styles.progressBar}

frontend/src/container/MetricsExplorer/Summary/Summary.styles.scss

@@ -142,13 +142,6 @@
 				}
 			}
 
-			.progress-container {
-				.ant-progress-bg {
-					height: 8px !important;
-					border-radius: 4px;
-				}
-			}
-
 			.ant-table-tbody > tr:hover > td {
 				background: color-mix(in srgb, var(--l1-foreground) 4%, transparent);
 			}

frontend/src/container/PaginatedTraceFlamegraph/PaginatedTraceFlamegraph.styles.scss

@@ -87,12 +87,7 @@
 
 					.service-progress-indicator {
 						width: fit-content;
-						margin-inline-end: 0px !important;
-						margin-bottom: 0px !important;
-
-						.ant-progress-inner {
-							width: 30px;
-						}
+						--progress-width: 30px;
 					}
 
 					.percent-value {

frontend/src/container/PaginatedTraceFlamegraph/PaginatedTraceFlamegraph.tsx

@@ -1,6 +1,7 @@
 import { useEffect, useMemo, useState } from 'react';
 import { useParams } from 'react-router-dom';
-import { Progress, Skeleton, Tooltip } from 'antd';
+import { Skeleton, Tooltip } from 'antd';
+import { Progress } from '@signozhq/ui/progress';
 import { Typography } from '@signozhq/ui/typography';
 import { AxiosError } from 'axios';
 import Spinner from 'components/Spinner';

frontend/src/pages/DashboardPageV2/index.tsx

@@ -0,0 +1,8 @@
+function DashboardPageV2(): JSX.Element {
+	return (
+		<div>
+			<h1>Dashboard Page V2</h1>
+		</div>
+	);
+}
+export default DashboardPageV2;

frontend/src/pages/DashboardsListPageV2/index.tsx

@@ -0,0 +1,9 @@
+function DashboardsListPageV2(): JSX.Element {
+	return (
+		<div>
+			<h1>Dashboards List Page V2</h1>
+		</div>
+	);
+}
+
+export default DashboardsListPageV2;

pkg/modules/logspipeline/impllogspipeline/module.go

@@ -0,0 +1,50 @@
+package impllogspipeline
+
+import (
+	"context"
+
+	"github.com/SigNoz/signoz/pkg/modules/logspipeline"
+	"github.com/SigNoz/signoz/pkg/sqlstore"
+	"github.com/SigNoz/signoz/pkg/valuer"
+)
+
+const elementType = "log_pipelines"
+
+type module struct {
+	sqlStore sqlstore.SQLStore
+}
+
+func NewModule(sqlStore sqlstore.SQLStore) logspipeline.Module {
+	return &module{sqlStore: sqlStore}
+}
+
+func (m *module) Collect(ctx context.Context, orgID valuer.UUID) (map[string]any, error) {
+	subq := m.sqlStore.BunDB().NewSelect().
+		TableExpr("agent_config_version").
+		ColumnExpr("MAX(version)").
+		Where("org_id = ?", orgID).
+		Where("element_type = ?", elementType)
+
+	var result struct {
+		Total   int `bun:"total"`
+		Enabled int `bun:"enabled_count"`
+	}
+	err := m.sqlStore.BunDB().NewSelect().
+		TableExpr("agent_config_element AS e").
+		Join("JOIN agent_config_version AS v ON v.id = e.version_id").
+		Join("JOIN pipelines AS p ON p.id = e.element_id").
+		Where("v.org_id = ?", orgID).
+		Where("v.element_type = ?", elementType).
+		Where("v.version = (?)", subq).
+		ColumnExpr("COUNT(*) AS total").
+		ColumnExpr("SUM(CASE WHEN p.enabled THEN 1 ELSE 0 END) AS enabled_count").
+		Scan(ctx, &result)
+	if err != nil {
+		return nil, err
+	}
+
+	return map[string]any{
+		"logs_pipeline.total.count":   result.Total,
+		"logs_pipeline.enabled.count": result.Enabled,
+	}, nil
+}

pkg/modules/logspipeline/logspipeline.go

@@ -0,0 +1,7 @@
+package logspipeline
+
+import "github.com/SigNoz/signoz/pkg/statsreporter"
+
+type Module interface {
+	statsreporter.StatsCollector
+}

pkg/querier/builder_query.go

@@ -199,7 +199,16 @@ func (q *builderQuery[T]) Execute(ctx context.Context) (*qbtypes.Result, error)
 		return q.executeWindowList(ctx)
 	}
 
-	stmt, err := q.stmtBuilder.Build(ctx, q.fromMS, q.toMS, q.kind, q.spec, q.variables)
+	fromMS, toMS := q.fromMS, q.toMS
+	if q.spec.Signal == telemetrytypes.SignalTraces || q.spec.Signal == telemetrytypes.SignalLogs {
+		var overlap bool
+		fromMS, toMS, overlap = q.narrowWindowByTraceID(ctx, fromMS, toMS)
+		if !overlap {
+			return emptyResultFor(q.kind, q.spec.Name), nil
+		}
+	}
+
+	stmt, err := q.stmtBuilder.Build(ctx, fromMS, toMS, q.kind, q.spec, q.variables)
 	if err != nil {
 		return nil, err
 	}
@@ -215,6 +224,81 @@ func (q *builderQuery[T]) Execute(ctx context.Context) (*qbtypes.Result, error)
 	return result, nil
 }
 
+// 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 and overlap=false when the trace lies
+// completely outside the query window — callers should short-circuit in that case.
+//
+// When the trace_id is not present in trace_summary the behaviour differs by
+// signal:
+//   - traces: trace_summary is derived from the spans table, so a missing row
+//     means no spans exist for that trace_id; we short-circuit to empty.
+//   - logs: logs can carry a trace_id even when traces are not ingested at all
+//     (e.g. traces disabled). We must not short-circuit; instead leave the
+//     window untouched and let the query run.
+func (q *builderQuery[T]) narrowWindowByTraceID(ctx context.Context, fromMS, toMS uint64) (uint64, uint64, bool) {
+	if q.spec.Filter == nil || q.spec.Filter.Expression == "" {
+		return fromMS, toMS, true
+	}
+
+	traceIDs, found := telemetrytraces.ExtractTraceIDsFromFilter(q.spec.Filter.Expression)
+	if !found || len(traceIDs) == 0 {
+		return fromMS, toMS, true
+	}
+
+	finder := telemetrytraces.NewTraceTimeRangeFinder(q.telemetryStore)
+	traceStart, traceEnd, ok := finder.GetTraceTimeRangeMulti(ctx, traceIDs)
+	if !ok {
+		if q.spec.Signal == telemetrytypes.SignalTraces {
+			q.logger.DebugContext(ctx, "trace_id not found in trace_summary; short-circuiting traces query to empty",
+				slog.Any("trace_ids", traceIDs))
+			return fromMS, toMS, false
+		}
+		q.logger.DebugContext(ctx, "trace_id not found in trace_summary; leaving time range untouched for logs",
+			slog.Any("trace_ids", traceIDs))
+		return fromMS, toMS, true
+	}
+
+	traceStartMS := uint64(traceStart) / 1_000_000
+	traceEndMS := uint64(traceEnd) / 1_000_000
+	if traceStartMS == 0 || traceEndMS == 0 {
+		return fromMS, toMS, true
+	}
+
+	if traceStartMS > toMS || traceEndMS < fromMS {
+		return fromMS, toMS, false
+	}
+	if traceStartMS > fromMS {
+		fromMS = traceStartMS
+	}
+	if traceEndMS < toMS {
+		toMS = traceEndMS
+	}
+	q.logger.DebugContext(ctx, "optimized time range using trace_id lookup",
+		slog.String("signal", q.spec.Signal.StringValue()),
+		slog.Any("trace_ids", traceIDs),
+		slog.Uint64("start", fromMS),
+		slog.Uint64("end", toMS))
+	return fromMS, toMS, true
+}
+
+// emptyResultFor returns an empty result payload appropriate for the given kind.
+func emptyResultFor(kind qbtypes.RequestType, queryName string) *qbtypes.Result {
+	var value any
+	switch kind {
+	case qbtypes.RequestTypeTimeSeries:
+		value = &qbtypes.TimeSeriesData{QueryName: queryName}
+	case qbtypes.RequestTypeScalar:
+		value = &qbtypes.ScalarData{QueryName: queryName}
+	default:
+		value = &qbtypes.RawData{QueryName: queryName}
+	}
+	return &qbtypes.Result{
+		Type:  kind,
+		Value: value,
+	}
+}
+
 // executeWithContext executes the query with query window and step context for partial value detection.
 func (q *builderQuery[T]) executeWithContext(ctx context.Context, query string, args []any) (*qbtypes.Result, error) {
 	ctx = ctxtypes.NewContextWithCommentVals(ctx, map[string]string{
@@ -310,42 +394,22 @@ func (q *builderQuery[T]) executeWindowList(ctx context.Context) (*qbtypes.Resul
 	totalBytes := uint64(0)
 	start := time.Now()
 
-	// Check if filter contains trace_id(s) and optimize time range if needed
-	if q.spec.Signal == telemetrytypes.SignalTraces &&
-		q.spec.Filter != nil && q.spec.Filter.Expression != "" {
-
-		traceIDs, found := telemetrytraces.ExtractTraceIDsFromFilter(q.spec.Filter.Expression)
-		if found && len(traceIDs) > 0 {
-			finder := telemetrytraces.NewTraceTimeRangeFinder(q.telemetryStore)
-
-			traceStart, traceEnd, ok := finder.GetTraceTimeRangeMulti(ctx, traceIDs)
-			traceStartMS := uint64(traceStart) / 1_000_000
-			traceEndMS := uint64(traceEnd) / 1_000_000
-			if !ok {
-				q.logger.DebugContext(ctx, "failed to get trace time range", slog.Any("trace_ids", traceIDs))
-			} else if traceStartMS > 0 && traceEndMS > 0 {
-				// no overlap — nothing to return
-				if uint64(traceStartMS) > toMS || uint64(traceEndMS) < fromMS {
-					return &qbtypes.Result{
-						Type: qbtypes.RequestTypeRaw,
-						Value: &qbtypes.RawData{
-							QueryName: q.spec.Name,
-						},
-						Stats: qbtypes.ExecStats{
-							DurationMS: uint64(time.Since(start).Milliseconds()),
-						},
-					}, nil
-				}
-
-				// clamp window to trace time range before bucketing
-				if uint64(traceStartMS) > fromMS {
-					fromMS = uint64(traceStartMS)
-				}
-				if uint64(traceEndMS) < toMS {
-					toMS = uint64(traceEndMS)
-				}
-				q.logger.DebugContext(ctx, "optimized time range for traces", slog.Any("trace_ids", traceIDs), slog.Uint64("start", fromMS), slog.Uint64("end", toMS))
-			}
+	// Check if filter contains trace_id(s) and optimize time range if needed.
+	// Applies to both traces (the listing this branch was built for) and logs
+	// (which carry trace_id and benefit from the same clamp before bucketing).
+	if q.spec.Signal == telemetrytypes.SignalTraces || q.spec.Signal == telemetrytypes.SignalLogs {
+		var overlap bool
+		fromMS, toMS, overlap = q.narrowWindowByTraceID(ctx, fromMS, toMS)
+		if !overlap {
+			return &qbtypes.Result{
+				Type: qbtypes.RequestTypeRaw,
+				Value: &qbtypes.RawData{
+					QueryName: q.spec.Name,
+				},
+				Stats: qbtypes.ExecStats{
+					DurationMS: uint64(time.Since(start).Milliseconds()),
+				},
+			}, nil
 		}
 	}
 

pkg/signoz/module.go

@@ -19,6 +19,8 @@ import (
 	"github.com/SigNoz/signoz/pkg/modules/inframonitoring/implinframonitoring"
 	"github.com/SigNoz/signoz/pkg/modules/llmpricingrule"
 	"github.com/SigNoz/signoz/pkg/modules/llmpricingrule/impllmpricingrule"
+	"github.com/SigNoz/signoz/pkg/modules/logspipeline"
+	"github.com/SigNoz/signoz/pkg/modules/logspipeline/impllogspipeline"
 	"github.com/SigNoz/signoz/pkg/modules/metricsexplorer"
 	"github.com/SigNoz/signoz/pkg/modules/metricsexplorer/implmetricsexplorer"
 	"github.com/SigNoz/signoz/pkg/modules/organization"
@@ -85,6 +87,7 @@ type Modules struct {
 	Promote          promote.Module
 	ServiceAccount   serviceaccount.Module
 	CloudIntegration cloudintegration.Module
+	LogsPipeline     logspipeline.Module
 	RuleStateHistory rulestatehistory.Module
 	TraceDetail      tracedetail.Module
 	SpanMapper       spanmapper.Module
@@ -143,6 +146,7 @@ func NewModules(
 		InfraMonitoring:  implinframonitoring.NewModule(telemetryStore, telemetryMetadataStore, querier, providerSettings, config.InfraMonitoring),
 		Promote:          implpromote.NewModule(telemetryMetadataStore, telemetryStore),
 		ServiceAccount:   serviceAccount,
+		LogsPipeline:     impllogspipeline.NewModule(sqlstore),
 		RuleStateHistory: implrulestatehistory.NewModule(implrulestatehistory.NewStore(telemetryStore, telemetryMetadataStore, providerSettings.Logger)),
 		CloudIntegration: cloudIntegrationModule,
 		TraceDetail:      impltracedetail.NewModule(impltracedetail.NewTraceStore(telemetryStore), providerSettings, config.TraceDetail),

pkg/signoz/signoz.go

@@ -497,6 +497,7 @@ func New(
 		modules.AuthDomain,
 		serviceAccount,
 		cloudIntegrationModule,
+		modules.LogsPipeline,
 	}
 
 	// Initialize stats reporter from the available stats reporter provider factories

tests/integration/tests/querier/01_logs.py

@@ -20,6 +20,7 @@ from fixtures.querier import (
     index_series_by_label,
     make_query_request,
 )
+from fixtures.traces import TraceIdGenerator, Traces, TracesKind, TracesStatusCode
 
 
 def test_logs_list(
@@ -2293,3 +2294,331 @@ def test_logs_formula_orderby_and_limit(
     assert len(f3_services) == 3, f"F3: expected 3 rows after limit, got {len(f3_services)}"
     assert f3_values == f4_values[:3], f"F3 values {f3_values} do not match F4[:3] values {f4_values[:3]}"
     assert set(f3_services) == set(f4_services[:3]), f"F3 services {f3_services} do not match F4[:3] services {f4_services[:3]}"
+
+
+def test_logs_list_filter_by_trace_id(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    insert_logs: Callable[[list[Logs]], None],
+    insert_traces: Callable[[list[Traces]], None],
+) -> None:
+    """
+    Tests that filtering logs by trace_id uses the trace_summary lookup to
+    narrow the query window before scanning the logs table:
+    1. Returns the matching log (narrow window, single bucket).
+    2. Does not return duplicate logs when the query window should span multiple
+       exponential buckets (>1 h). But is clamped to the timerange of trace.
+    3. Returns no results when the query window does not contain the trace.
+    4. Logs carrying a trace_id whose trace is NOT in trace_summary (e.g.
+       traces disabled) are still returned — the lookup miss must not
+       short-circuit logs queries.
+    """
+    target_trace_id = TraceIdGenerator.trace_id()
+    other_trace_id = TraceIdGenerator.trace_id()
+    orphan_trace_id = TraceIdGenerator.trace_id()
+    target_root_span_id = TraceIdGenerator.span_id()
+    target_child_span_id = TraceIdGenerator.span_id()
+    other_span_id = TraceIdGenerator.span_id()
+    orphan_span_id = TraceIdGenerator.span_id()
+
+    now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
+
+    common_resources = {
+        "deployment.environment": "production",
+        "service.name": "logs-trace-filter-service",
+        "cloud.provider": "integration",
+    }
+
+    # Populate signoz_traces.distributed_trace_summary by inserting spans for
+    # the target trace_id. trace_summary records min/max of span timestamps
+    # (it ignores span duration), so two spans are inserted to give the trace
+    # a non-trivial recorded window of [now-10s, now-5s].
+    insert_traces(
+        [
+            Traces(
+                timestamp=now - timedelta(seconds=10),
+                duration=timedelta(seconds=1),
+                trace_id=target_trace_id,
+                span_id=target_root_span_id,
+                parent_span_id="",
+                name="root-span",
+                kind=TracesKind.SPAN_KIND_SERVER,
+                status_code=TracesStatusCode.STATUS_CODE_OK,
+                status_message="",
+                resources=common_resources,
+                attributes={},
+            ),
+            Traces(
+                timestamp=now - timedelta(seconds=5),
+                duration=timedelta(seconds=1),
+                trace_id=target_trace_id,
+                span_id=target_child_span_id,
+                parent_span_id=target_root_span_id,
+                name="child-span",
+                kind=TracesKind.SPAN_KIND_CLIENT,
+                status_code=TracesStatusCode.STATUS_CODE_OK,
+                status_message="",
+                resources=common_resources,
+                attributes={},
+            ),
+        ]
+    )
+
+    # Insert logs:
+    # - one with the target trace_id, at a timestamp within the trace's
+    #   recorded window (now-10s..now-5s, padded ±1s).
+    # - one with a different trace_id; must never appear in target_trace_id
+    #   results.
+    # - one with an orphan trace_id whose trace was never ingested — used to
+    #   verify the lookup miss does NOT short-circuit logs queries.
+    insert_logs(
+        [
+            Logs(
+                timestamp=now - timedelta(seconds=7),
+                resources=common_resources,
+                attributes={"http.method": "GET"},
+                body="log inside the target trace window",
+                severity_text="INFO",
+                trace_id=target_trace_id,
+                span_id=target_root_span_id,
+            ),
+            Logs(
+                timestamp=now - timedelta(seconds=3),
+                resources=common_resources,
+                attributes={"http.method": "POST"},
+                body="log with a different trace_id",
+                severity_text="INFO",
+                trace_id=other_trace_id,
+                span_id=other_span_id,
+            ),
+            Logs(
+                timestamp=now - timedelta(seconds=2),
+                resources=common_resources,
+                attributes={"http.method": "PUT"},
+                body="log with a trace_id absent from trace_summary",
+                severity_text="INFO",
+                trace_id=orphan_trace_id,
+                span_id=orphan_span_id,
+            ),
+        ]
+    )
+
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+
+    def _query(start_ms: int, end_ms: int, trace_id: str) -> list:
+        response = make_query_request(
+            signoz,
+            token,
+            start_ms=start_ms,
+            end_ms=end_ms,
+            request_type="raw",
+            queries=[
+                {
+                    "type": "builder_query",
+                    "spec": {
+                        "name": "A",
+                        "signal": "logs",
+                        "disabled": False,
+                        "limit": 100,
+                        "offset": 0,
+                        "filter": {"expression": f"trace_id = '{trace_id}'"},
+                        "order": [
+                            {"key": {"name": "timestamp"}, "direction": "desc"},
+                            {"key": {"name": "id"}, "direction": "desc"},
+                        ],
+                    },
+                }
+            ],
+        )
+        assert response.status_code == HTTPStatus.OK
+        assert response.json()["status"] == "success"
+        return response.json()["data"]["data"]["results"][0]["rows"] or []
+
+    now_ms = int(now.timestamp() * 1000)
+
+    # --- Test 1: narrow window (single bucket, <1 h) ---
+    narrow_start_ms = int((now - timedelta(minutes=5)).timestamp() * 1000)
+    narrow_rows = _query(narrow_start_ms, now_ms, target_trace_id)
+
+    assert len(narrow_rows) == 1, f"Expected 1 log for trace_id filter (narrow window), got {len(narrow_rows)}"
+    assert narrow_rows[0]["data"]["trace_id"] == target_trace_id
+    assert narrow_rows[0]["data"]["span_id"] == target_root_span_id
+
+    # --- Test 2: wide window (>1 h, camp to the timerange from trace_summary) ---
+    # Should still return exactly one log — no duplicates from multi-bucket scan.
+    wide_start_ms = int((now - timedelta(hours=12)).timestamp() * 1000)
+    wide_rows = _query(wide_start_ms, now_ms, target_trace_id)
+
+    assert len(wide_rows) == 1, f"Expected 1 log for trace_id filter (wide window, multi-bucket), got {len(wide_rows)} — possible duplicate-log regression"
+    assert wide_rows[0]["data"]["trace_id"] == target_trace_id
+    assert wide_rows[0]["data"]["span_id"] == target_root_span_id
+
+    # --- Test 3: window that does not contain the trace returns no results ---
+    past_start_ms = int((now - timedelta(hours=6)).timestamp() * 1000)
+    past_end_ms = int((now - timedelta(hours=2)).timestamp() * 1000)
+    past_rows = _query(past_start_ms, past_end_ms, target_trace_id)
+
+    assert len(past_rows) == 0, f"Expected 0 logs for trace_id filter outside time window, got {len(past_rows)}"
+
+    # --- Test 4: trace_id not present in trace_summary still returns logs ---
+    orphan_rows = _query(narrow_start_ms, now_ms, orphan_trace_id)
+
+    assert len(orphan_rows) == 1, f"Expected 1 log for orphan trace_id (no trace_summary entry), got {len(orphan_rows)} — logs query may have been incorrectly short-circuited"
+    assert orphan_rows[0]["data"]["trace_id"] == orphan_trace_id
+
+
+def test_logs_aggregation_filter_by_trace_id(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    insert_logs: Callable[[list[Logs]], None],
+    insert_traces: Callable[[list[Traces]], None],
+) -> None:
+    """
+    Tests that the trace_id time-range optimization also applies to
+    non-window-list (time_series / aggregation) logs queries:
+    1. Wide query window containing the trace returns the correct count.
+    2. Query window outside the trace's time range short-circuits to an
+       empty result.
+    3. A trace_id with no row in trace_summary (e.g. traces disabled) still
+       returns the matching logs — the lookup miss must not short-circuit
+       logs aggregation queries.
+    """
+    target_trace_id = TraceIdGenerator.trace_id()
+    orphan_trace_id = TraceIdGenerator.trace_id()
+    target_root_span_id = TraceIdGenerator.span_id()
+    target_child_span_id = TraceIdGenerator.span_id()
+    orphan_span_id = TraceIdGenerator.span_id()
+
+    now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
+
+    common_resources = {
+        "deployment.environment": "production",
+        "service.name": "logs-trace-agg-service",
+        "cloud.provider": "integration",
+    }
+
+    # trace_summary records min/max of span timestamps (it ignores duration),
+    # so insert two spans to give the trace a recorded window wide enough to
+    # comfortably contain the log timestamps below.
+    insert_traces(
+        [
+            Traces(
+                timestamp=now - timedelta(seconds=10),
+                duration=timedelta(seconds=1),
+                trace_id=target_trace_id,
+                span_id=target_root_span_id,
+                parent_span_id="",
+                name="root-span",
+                kind=TracesKind.SPAN_KIND_SERVER,
+                status_code=TracesStatusCode.STATUS_CODE_OK,
+                status_message="",
+                resources=common_resources,
+                attributes={},
+            ),
+            Traces(
+                timestamp=now - timedelta(seconds=5),
+                duration=timedelta(seconds=1),
+                trace_id=target_trace_id,
+                span_id=target_child_span_id,
+                parent_span_id=target_root_span_id,
+                name="child-span",
+                kind=TracesKind.SPAN_KIND_CLIENT,
+                status_code=TracesStatusCode.STATUS_CODE_OK,
+                status_message="",
+                resources=common_resources,
+                attributes={},
+            ),
+        ]
+    )
+
+    # Two logs for the target trace_id, both inside the recorded trace window.
+    # One additional log carries an orphan trace_id with no row in
+    # trace_summary — used to verify that the lookup miss does not
+    # short-circuit logs aggregations.
+    insert_logs(
+        [
+            Logs(
+                timestamp=now - timedelta(seconds=9),
+                resources=common_resources,
+                attributes={},
+                body="log A inside trace window",
+                severity_text="INFO",
+                trace_id=target_trace_id,
+                span_id=target_root_span_id,
+            ),
+            Logs(
+                timestamp=now - timedelta(seconds=6),
+                resources=common_resources,
+                attributes={},
+                body="log B inside trace window",
+                severity_text="INFO",
+                trace_id=target_trace_id,
+                span_id=target_root_span_id,
+            ),
+            Logs(
+                timestamp=now - timedelta(seconds=2),
+                resources=common_resources,
+                attributes={},
+                body="log with a trace_id absent from trace_summary",
+                severity_text="INFO",
+                trace_id=orphan_trace_id,
+                span_id=orphan_span_id,
+            ),
+        ]
+    )
+
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+
+    def _count(start_ms: int, end_ms: int, trace_id: str) -> float:
+        response = make_query_request(
+            signoz,
+            token,
+            start_ms=start_ms,
+            end_ms=end_ms,
+            request_type="time_series",
+            queries=[
+                {
+                    "type": "builder_query",
+                    "spec": {
+                        "name": "A",
+                        "signal": "logs",
+                        "stepInterval": 60,
+                        "disabled": False,
+                        "filter": {"expression": f"trace_id = '{trace_id}'"},
+                        "having": {"expression": ""},
+                        "aggregations": [{"expression": "count()"}],
+                    },
+                }
+            ],
+        )
+        assert response.status_code == HTTPStatus.OK
+        assert response.json()["status"] == "success"
+        results = response.json()["data"]["data"]["results"]
+        assert len(results) == 1
+        aggregations = results[0].get("aggregations") or []
+        if not aggregations:
+            return 0
+        series = aggregations[0].get("series") or []
+        if not series:
+            return 0
+        return sum(v["value"] for v in series[0]["values"])
+
+    now_ms = int(now.timestamp() * 1000)
+    narrow_start_ms = int((now - timedelta(minutes=5)).timestamp() * 1000)
+
+    # --- Test 1: wide window (>1 h) containing the trace returns 2 logs ---
+    wide_start_ms = int((now - timedelta(hours=12)).timestamp() * 1000)
+    wide_count = _count(wide_start_ms, now_ms, target_trace_id)
+    assert wide_count == 2, f"Expected count=2 for trace_id aggregation (wide window), got {wide_count}"
+
+    # --- Test 2: window outside the trace short-circuits to empty ---
+    past_start_ms = int((now - timedelta(hours=6)).timestamp() * 1000)
+    past_end_ms = int((now - timedelta(hours=2)).timestamp() * 1000)
+    past_count = _count(past_start_ms, past_end_ms, target_trace_id)
+    assert past_count == 0, f"Expected count=0 for trace_id aggregation outside time window, got {past_count}"
+
+    # --- Test 3: trace_id not present in trace_summary still returns logs ---
+    orphan_count = _count(narrow_start_ms, now_ms, orphan_trace_id)
+    assert orphan_count == 1, f"Expected count=1 for orphan trace_id aggregation, got {orphan_count} — query may have been incorrectly short-circuited"

tests/integration/tests/querier/04_traces.py

@@ -2123,3 +2123,115 @@ def test_traces_list_filter_by_trace_id(
     past_rows = _query(past_start_ms, past_end_ms)
 
     assert len(past_rows) == 0, f"Expected 0 spans for trace_id filter outside time window, got {len(past_rows)}"
+
+
+def test_traces_aggregation_filter_by_trace_id(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    insert_traces: Callable[[list[Traces]], None],
+) -> None:
+    """
+    Tests that the trace_id time-range optimization also applies to
+    non-window-list (time_series / aggregation) traces queries:
+    1. Wide query window containing the trace returns the correct count.
+    2. Query window outside the trace's time range short-circuits to empty.
+    3. Filter referencing a trace_id with no row in trace_summary
+       short-circuits to empty (trace_summary is authoritative for traces).
+    """
+    target_trace_id = TraceIdGenerator.trace_id()
+    target_root_span_id = TraceIdGenerator.span_id()
+    target_child_span_id = TraceIdGenerator.span_id()
+    missing_trace_id = TraceIdGenerator.trace_id()
+
+    now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
+
+    common_resources = {
+        "deployment.environment": "production",
+        "service.name": "traces-agg-filter-service",
+        "cloud.provider": "integration",
+    }
+
+    insert_traces(
+        [
+            Traces(
+                timestamp=now - timedelta(seconds=10),
+                duration=timedelta(seconds=5),
+                trace_id=target_trace_id,
+                span_id=target_root_span_id,
+                parent_span_id="",
+                name="root-span",
+                kind=TracesKind.SPAN_KIND_SERVER,
+                status_code=TracesStatusCode.STATUS_CODE_OK,
+                status_message="",
+                resources=common_resources,
+                attributes={"http.request.method": "GET"},
+            ),
+            Traces(
+                timestamp=now - timedelta(seconds=9),
+                duration=timedelta(seconds=1),
+                trace_id=target_trace_id,
+                span_id=target_child_span_id,
+                parent_span_id=target_root_span_id,
+                name="child-span",
+                kind=TracesKind.SPAN_KIND_CLIENT,
+                status_code=TracesStatusCode.STATUS_CODE_OK,
+                status_message="",
+                resources=common_resources,
+                attributes={},
+            ),
+        ]
+    )
+
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+
+    def _count(start_ms: int, end_ms: int, trace_id: str) -> float:
+        response = make_query_request(
+            signoz,
+            token,
+            start_ms=start_ms,
+            end_ms=end_ms,
+            request_type="time_series",
+            queries=[
+                {
+                    "type": "builder_query",
+                    "spec": {
+                        "name": "A",
+                        "signal": "traces",
+                        "stepInterval": 60,
+                        "disabled": False,
+                        "filter": {"expression": f"trace_id = '{trace_id}'"},
+                        "aggregations": [{"expression": "count()"}],
+                    },
+                }
+            ],
+        )
+        assert response.status_code == HTTPStatus.OK
+        assert response.json()["status"] == "success"
+        results = response.json()["data"]["data"]["results"]
+        assert len(results) == 1
+        aggregations = results[0].get("aggregations") or []
+        if not aggregations:
+            return 0
+        series = aggregations[0].get("series") or []
+        if not series:
+            return 0
+        return sum(v["value"] for v in series[0]["values"])
+
+    now_ms = int(now.timestamp() * 1000)
+
+    # --- Test 1: wide window (>1 h) containing the trace returns both spans ---
+    wide_start_ms = int((now - timedelta(hours=12)).timestamp() * 1000)
+    wide_count = _count(wide_start_ms, now_ms, target_trace_id)
+    assert wide_count == 2, f"Expected count=2 for trace_id aggregation (wide window), got {wide_count}"
+
+    # --- Test 2: window outside the trace short-circuits to empty ---
+    past_start_ms = int((now - timedelta(hours=6)).timestamp() * 1000)
+    past_end_ms = int((now - timedelta(hours=2)).timestamp() * 1000)
+    past_count = _count(past_start_ms, past_end_ms, target_trace_id)
+    assert past_count == 0, f"Expected count=0 for trace_id aggregation outside time window, got {past_count}"
+
+    # --- Test 3: trace_id with no entry in trace_summary short-circuits ---
+    missing_start_ms = int((now - timedelta(minutes=5)).timestamp() * 1000)
+    missing_count = _count(missing_start_ms, now_ms, missing_trace_id)
+    assert missing_count == 0, f"Expected count=0 for trace_id absent from trace_summary, got {missing_count}"