Merge branch 'main' into replace-prom-table

* fix: add missing filtering for ip address for scalar data

In domain listing api for external api monitoring,
we have option to filter out the IP address but
it only handles timeseries and raw type data while
domain list handler returns scalar data.

* fix: switch to new derived attributes for ip filtering

---------

Co-authored-by: Nityananda Gohain <nityanandagohain@gmail.com>

.github/workflows/integrationci.yaml

@@ -54,7 +54,7 @@ jobs:
           - sqlite
         clickhouse-version:
           - 25.5.6
-          - 25.10.5
+          - 25.12.5
         schema-migrator-version:
           - v0.142.0
         postgres-version:

frontend/src/AppRoutes/pageComponents.ts

@@ -308,3 +308,15 @@ export const PublicDashboardPage = Loadable(
 			/* webpackChunkName: "Public Dashboard Page" */ 'pages/PublicDashboard'
 		),
 );
+
+export const AlertTypeSelectionPage = Loadable(
+	() =>
+		import(
+			/* webpackChunkName: "Alert Type Selection Page" */ 'pages/AlertTypeSelection'
+		),
+);
+
+export const MeterExplorerPage = Loadable(
+	() =>
+		import(/* webpackChunkName: "Meter Explorer Page" */ 'pages/MeterExplorer'),
+);

frontend/src/AppRoutes/routes.ts

@@ -1,12 +1,10 @@
 import { RouteProps } from 'react-router-dom';
 import ROUTES from 'constants/routes';
-import AlertTypeSelectionPage from 'pages/AlertTypeSelection';
-import MessagingQueues from 'pages/MessagingQueues';
-import MeterExplorer from 'pages/MeterExplorer';
 
 import {
 	AlertHistory,
 	AlertOverview,
+	AlertTypeSelectionPage,
 	AllAlertChannels,
 	AllErrors,
 	ApiMonitoring,
@@ -29,6 +27,8 @@ import {
 	LogsExplorer,
 	LogsIndexToFields,
 	LogsSaveViews,
+	MessagingQueuesMainPage,
+	MeterExplorerPage,
 	MetricsExplorer,
 	OldLogsExplorer,
 	Onboarding,
@@ -399,28 +399,28 @@ const routes: AppRoutes[] = [
 	{
 		path: ROUTES.MESSAGING_QUEUES_KAFKA,
 		exact: true,
-		component: MessagingQueues,
+		component: MessagingQueuesMainPage,
 		key: 'MESSAGING_QUEUES_KAFKA',
 		isPrivate: true,
 	},
 	{
 		path: ROUTES.MESSAGING_QUEUES_CELERY_TASK,
 		exact: true,
-		component: MessagingQueues,
+		component: MessagingQueuesMainPage,
 		key: 'MESSAGING_QUEUES_CELERY_TASK',
 		isPrivate: true,
 	},
 	{
 		path: ROUTES.MESSAGING_QUEUES_OVERVIEW,
 		exact: true,
-		component: MessagingQueues,
+		component: MessagingQueuesMainPage,
 		key: 'MESSAGING_QUEUES_OVERVIEW',
 		isPrivate: true,
 	},
 	{
 		path: ROUTES.MESSAGING_QUEUES_KAFKA_DETAIL,
 		exact: true,
-		component: MessagingQueues,
+		component: MessagingQueuesMainPage,
 		key: 'MESSAGING_QUEUES_KAFKA_DETAIL',
 		isPrivate: true,
 	},
@@ -463,21 +463,21 @@ const routes: AppRoutes[] = [
 	{
 		path: ROUTES.METER,
 		exact: true,
-		component: MeterExplorer,
+		component: MeterExplorerPage,
 		key: 'METER',
 		isPrivate: true,
 	},
 	{
 		path: ROUTES.METER_EXPLORER,
 		exact: true,
-		component: MeterExplorer,
+		component: MeterExplorerPage,
 		key: 'METER_EXPLORER',
 		isPrivate: true,
 	},
 	{
 		path: ROUTES.METER_EXPLORER_VIEWS,
 		exact: true,
-		component: MeterExplorer,
+		component: MeterExplorerPage,
 		key: 'METER_EXPLORER_VIEWS',
 		isPrivate: true,
 	},

pkg/apiserver/signozapiserver/logspipeline.go

@@ -1,12 +0,0 @@
-package signozapiserver
-
-import (
-	"net/http"
-
-	"github.com/gorilla/mux"
-)
-
-func (provider *provider) addLogspipelineRoutes(router *mux.Router) error {
-	router.HandleFunc("/api/v2/pipelines", provider.logspipelineHandler.ListPipelines).Methods(http.MethodGet)
-	return nil
-}

pkg/apiserver/signozapiserver/provider.go

@@ -14,7 +14,6 @@ import (
 	"github.com/SigNoz/signoz/pkg/modules/authdomain"
 	"github.com/SigNoz/signoz/pkg/modules/dashboard"
 	"github.com/SigNoz/signoz/pkg/modules/fields"
-	"github.com/SigNoz/signoz/pkg/modules/logspipeline"
 	"github.com/SigNoz/signoz/pkg/modules/metricsexplorer"
 	"github.com/SigNoz/signoz/pkg/modules/organization"
 	"github.com/SigNoz/signoz/pkg/modules/preference"
@@ -49,7 +48,6 @@ type provider struct {
 	authzHandler           authz.Handler
 	zeusHandler            zeus.Handler
 	querierHandler         querier.Handler
-	logspipelineHandler    logspipeline.Handler
 }
 
 func NewFactory(
@@ -71,7 +69,6 @@ func NewFactory(
 	authzHandler authz.Handler,
 	zeusHandler zeus.Handler,
 	querierHandler querier.Handler,
-	logspipelineHandler logspipeline.Handler,
 ) factory.ProviderFactory[apiserver.APIServer, apiserver.Config] {
 	return factory.NewProviderFactory(factory.MustNewName("signoz"), func(ctx context.Context, providerSettings factory.ProviderSettings, config apiserver.Config) (apiserver.APIServer, error) {
 		return newProvider(
@@ -96,7 +93,6 @@ func NewFactory(
 			authzHandler,
 			zeusHandler,
 			querierHandler,
-			logspipelineHandler,
 		)
 	})
 }
@@ -123,7 +119,6 @@ func newProvider(
 	authzHandler authz.Handler,
 	zeusHandler zeus.Handler,
 	querierHandler querier.Handler,
-	logspipelineHandler logspipeline.Handler,
 ) (apiserver.APIServer, error) {
 	settings := factory.NewScopedProviderSettings(providerSettings, "github.com/SigNoz/signoz/pkg/apiserver/signozapiserver")
 	router := mux.NewRouter().UseEncodedPath()
@@ -148,7 +143,6 @@ func newProvider(
 		authzHandler:           authzHandler,
 		zeusHandler:            zeusHandler,
 		querierHandler:         querierHandler,
-		logspipelineHandler:    logspipelineHandler,
 	}
 
 	provider.authZ = middleware.NewAuthZ(settings.Logger(), orgGetter, authz)
@@ -229,14 +223,9 @@ func (provider *provider) AddToRouter(router *mux.Router) error {
 		return err
 	}
 
-	if err := provider.addLogspipelineRoutes(router); err != nil {
-		return err
-	}
-
 	return nil
 }
 
-
 func newSecuritySchemes(role types.Role) []handler.OpenAPISecurityScheme {
 	return []handler.OpenAPISecurityScheme{
 		{Name: ctxtypes.AuthTypeAPIKey.StringValue(), Scopes: []string{role.String()}},

pkg/modules/logspipeline/impllogspipeline/handler.go

@@ -1,91 +0,0 @@
-package impllogspipeline
-
-import (
-	"net/http"
-	"strconv"
-
-	"github.com/SigNoz/signoz/pkg/errors"
-	"github.com/SigNoz/signoz/pkg/http/render"
-	"github.com/SigNoz/signoz/pkg/modules/logspipeline"
-	"github.com/SigNoz/signoz/pkg/types/authtypes"
-	"github.com/SigNoz/signoz/pkg/valuer"
-	"github.com/gorilla/mux"
-)
-
-type handler struct {
-	module logspipeline.Module
-}
-
-func NewHandler(module logspipeline.Module) logspipeline.Handler {
-	return &handler{module: module}
-}
-
-func (h *handler) ListPipelines(w http.ResponseWriter, r *http.Request) {
-	claims, err := authtypes.ClaimsFromContext(r.Context())
-	if err != nil {
-		render.Error(w, err)
-		return
-	}
-
-	orgID, errv2 := valuer.NewUUID(claims.OrgID)
-	if errv2 != nil {
-		render.Error(w, errv2)
-		return
-	}
-
-	version, err := ParseAgentConfigVersion(r)
-	if err != nil {
-		render.Error(w, err)
-		return
-	}
-
-	if version != -1 {
-		pipelines, err := h.module.ListPipelinesByVersion(r.Context(), orgID, version)
-		if err != nil {
-			render.Error(w, err)
-			return
-		}
-		render.Success(w, http.StatusOK, pipelines)
-		return
-	}
-
-	pipelines, err := h.module.ListPipelines(r.Context(), orgID)
-	if err != nil {
-		render.Error(w, err)
-		return
-	}
-
-	render.Success(w, http.StatusOK, pipelines)
-}
-
-func (h *handler) GetPipeline(w http.ResponseWriter, r *http.Request) {
-}
-
-func (h *handler) CreatePipeline(w http.ResponseWriter, r *http.Request) {
-}
-
-func (h *handler) UpdatePipeline(w http.ResponseWriter, r *http.Request) {
-}
-
-func (h *handler) DeletePipeline(w http.ResponseWriter, r *http.Request) {
-}
-
-func ParseAgentConfigVersion(r *http.Request) (int, error) {
-	versionString := mux.Vars(r)["version"]
-
-	if versionString == "latest" {
-		return -1, nil
-	}
-
-	version64, err := strconv.ParseInt(versionString, 0, 8)
-
-	if err != nil {
-		return 0, errors.WrapInvalidInputf(err, errors.CodeInvalidInput, "invalid version number")
-	}
-
-	if version64 <= 0 {
-		return 0, errors.NewInvalidInputf(errors.CodeInvalidInput, "invalid version number")
-	}
-
-	return int(version64), nil
-}

pkg/modules/logspipeline/impllogspipeline/module.go

@@ -1,230 +0,0 @@
-package impllogspipeline
-
-import (
-	"context"
-	"time"
-
-	"github.com/SigNoz/signoz/pkg/errors"
-	"github.com/SigNoz/signoz/pkg/modules/logspipeline"
-	"github.com/SigNoz/signoz/pkg/query-service/agentConf"
-	"github.com/SigNoz/signoz/pkg/sqlstore"
-	"github.com/SigNoz/signoz/pkg/types"
-	"github.com/SigNoz/signoz/pkg/types/authtypes"
-	"github.com/SigNoz/signoz/pkg/types/opamptypes"
-	"github.com/SigNoz/signoz/pkg/types/pipelinetypes"
-	"github.com/SigNoz/signoz/pkg/valuer"
-	"github.com/uptrace/bun"
-	"go.uber.org/zap"
-)
-
-type module struct {
-	sqlstore sqlstore.SQLStore
-}
-
-func NewModule(sqlstore sqlstore.SQLStore) logspipeline.Module {
-	return &module{sqlstore: sqlstore}
-}
-
-func (m *module) ListPipelines(ctx context.Context, orgID valuer.UUID) ([]pipelinetypes.GettablePipeline, error) {
-	latestVersion := -1
-	// get latest agent config
-	lastestConfig, err := agentConf.GetLatestVersion(ctx, orgID, opamptypes.ElementTypeLogPipelines)
-	if err != nil && !errors.Ast(err, errors.TypeNotFound) {
-		return nil, err
-	}
-
-	if lastestConfig != nil {
-		latestVersion = lastestConfig.Version
-	}
-	return m.ListPipelinesByVersion(ctx, orgID, latestVersion)
-}
-
-func (m *module) ListPipelinesByVersion(ctx context.Context, orgID valuer.UUID, version int) ([]pipelinetypes.GettablePipeline, error) {
-	var stored []pipelinetypes.StoreablePipeline
-	err := m.sqlstore.BunDB().NewSelect().
-		Model(&stored).
-		Join("JOIN agent_config_element e ON p.id = e.element_id").
-		Join("JOIN agent_config_version v ON v.id = e.version_id").
-		Where("e.element_type = ?", opamptypes.ElementTypeLogPipelines.StringValue()).
-		Where("v.version = ?", version).
-		Where("v.org_id = ?", orgID.StringValue()).
-		Order("p.order_id ASC").
-		Scan(ctx)
-	if err != nil {
-		return nil, err
-	}
-
-	pipelines := make([]pipelinetypes.GettablePipeline, len(stored))
-	if len(stored) == 0 {
-		return pipelines, nil
-	}
-
-	for i := range stored {
-		pipelines[i].StoreablePipeline = stored[i]
-		if err := pipelines[i].ParseRawConfig(); err != nil {
-			return nil, err
-		}
-		if err := pipelines[i].ParseFilter(); err != nil {
-			return nil, err
-		}
-	}
-
-	return pipelines, nil
-}
-
-func (m *module) GetPipeline(ctx context.Context, orgID valuer.UUID, id string) (*pipelinetypes.GettablePipeline, error) {
-	return nil, nil
-}
-
-func (m *module) CreatePipeline(ctx context.Context, orgID valuer.UUID, claims *authtypes.Claims, pipeline *pipelinetypes.PostablePipeline) (*pipelinetypes.GettablePipeline, error) {
-	storeable, err := pipeline.ToStoreablePipeline()
-	if err != nil {
-		return nil, err
-	}
-
-	// regenerate the id and set other fields
-	storeable.Identifiable.ID = valuer.GenerateUUID()
-	storeable.OrgID = orgID.String()
-	storeable.TimeAuditable = types.TimeAuditable{
-		CreatedAt: time.Now(),
-	}
-	storeable.UserAuditable = types.UserAuditable{
-		CreatedBy: claims.Email,
-	}
-
-	_, err = m.sqlstore.BunDB().NewInsert().
-		Model(&storeable).
-		Exec(ctx)
-	if err != nil {
-		zap.L().Error("error in inserting pipeline data", zap.Error(err))
-		return nil, errors.WrapInternalf(err, errors.CodeInternal, "failed to insert pipeline")
-	}
-
-	return &pipelinetypes.GettablePipeline{
-		StoreablePipeline: *storeable,
-		Filter:            pipeline.Filter,
-		Config:            pipeline.Config,
-	}, nil
-}
-
-func (m *module) UpdatePipeline(ctx context.Context, orgID valuer.UUID, claims *authtypes.Claims, pipeline *pipelinetypes.PostablePipeline) (*pipelinetypes.GettablePipeline, error) {
-	if err := pipeline.IsValid(); err != nil {
-		return nil, err
-	}
-
-	storeable, err := pipeline.ToStoreablePipeline()
-	if err != nil {
-		return nil, err
-	}
-
-	storeable.OrgID = orgID.String()
-	storeable.TimeAuditable = types.TimeAuditable{
-		UpdatedAt: time.Now(),
-	}
-	storeable.UserAuditable = types.UserAuditable{
-		UpdatedBy: claims.Email,
-	}
-
-	// get id from storeable pipeline
-	id := storeable.ID.StringValue()
-
-	// depending on the order_id update the rest of the table
-	// example 1: total available pipelines are 6, and order_id 5 is moved to 2, then we need to update the rest of the table
-	// old: 1, 2, 3, 4, 5, 6
-	//         ^         |
-	//         |_________|
-	// So pipelines starting from 2nd position till 4th position shift to right (or increase their order_id) by 1 position
-	// example 2: total available pipelines are 6, and order_id 2 is moved to 4, then we need to update the rest of the table
-	// old: 1, 2, 3, 4, 5, 6
-	//         |     ^
-	//         |_____|
-	// So pipelines starting from 3rd position till 4th position shift to left (or decrease their order_id) by 1 position
-	if err := m.sqlstore.RunInTxCtx(ctx, nil, func(ctx context.Context) error {
-		db := m.sqlstore.BunDBCtx(ctx)
-
-		var existing pipelinetypes.StoreablePipeline
-		if err := db.NewSelect().
-			Model(&existing).
-			Where("id = ?", id).
-			Where("org_id = ?", orgID.StringValue()).
-			Scan(ctx); err != nil {
-			return m.sqlstore.WrapNotFoundErrf(
-				err,
-				errors.CodeNotFound,
-				"pipeline with id %s does not exist in org %s",
-				id,
-				orgID.StringValue(),
-			)
-		}
-
-		oldOrderID := existing.OrderID
-		newOrderID := storeable.OrderID
-
-		// Reorder other pipelines if the order has changed.
-		if newOrderID != oldOrderID {
-			if err := reorderPipelinesInTx(ctx, db, orgID.StringValue(), oldOrderID, newOrderID); err != nil {
-				return err
-			}
-		}
-
-		// Preserve primary key and immutable fields.
-		storeable.ID = existing.ID
-
-		// Persist the updated pipeline (including its new order).
-		if _, err := db.NewUpdate().
-			Model(storeable).
-			Where("id = ?", id).
-			Where("org_id = ?", orgID.StringValue()).
-			Exec(ctx); err != nil {
-			return err
-		}
-
-		return nil
-	}); err != nil {
-		return nil, err
-	}
-
-	return &pipelinetypes.GettablePipeline{
-		StoreablePipeline: *storeable,
-		Filter:            pipeline.Filter,
-		Config:            pipeline.Config,
-	}, nil
-}
-
-// reorderPipelinesInTx updates order_id of other pipelines in a transaction-aware way.
-// It assumes that all pipelines for a given org have consecutive order_id values starting from 1.
-// The logic is:
-//   - When moving a pipeline from a higher position to a lower position (e.g., 5 -> 2),
-//     all pipelines in [newOrderID, oldOrderID) are shifted right by +1.
-//   - When moving from a lower position to a higher position (e.g., 2 -> 4),
-//     all pipelines in (oldOrderID, newOrderID] are shifted left by -1.
-func reorderPipelinesInTx(ctx context.Context, tx bun.IDB, orgID string, oldOrderID, newOrderID int) error {
-	switch {
-	case newOrderID < oldOrderID:
-		// Move up: shift affected pipelines down (order_id + 1).
-		_, err := tx.NewUpdate().
-			Model((*pipelinetypes.StoreablePipeline)(nil)).
-			Set("order_id = order_id + 1").
-			Where("org_id = ?", orgID).
-			Where("order_id >= ?", newOrderID).
-			Where("order_id < ?", oldOrderID).
-			Exec(ctx)
-		return err
-	case newOrderID > oldOrderID:
-		// Move down: shift affected pipelines up (order_id - 1).
-		_, err := tx.NewUpdate().
-			Model((*pipelinetypes.StoreablePipeline)(nil)).
-			Set("order_id = order_id - 1").
-			Where("org_id = ?", orgID).
-			Where("order_id > ?", oldOrderID).
-			Where("order_id <= ?", newOrderID).
-			Exec(ctx)
-		return err
-	default:
-		return nil
-	}
-}
-
-func (m *module) DeletePipeline(ctx context.Context, orgID valuer.UUID, id string) error {
-	return nil
-}

pkg/modules/logspipeline/logspipeline.go

@@ -1,27 +0,0 @@
-package logspipeline
-
-import (
-	"context"
-	"net/http"
-
-	"github.com/SigNoz/signoz/pkg/types/authtypes"
-	"github.com/SigNoz/signoz/pkg/types/pipelinetypes"
-	"github.com/SigNoz/signoz/pkg/valuer"
-)
-
-type Module interface {
-	ListPipelines(ctx context.Context, orgID valuer.UUID) ([]pipelinetypes.GettablePipeline, error)
-	ListPipelinesByVersion(ctx context.Context, orgID valuer.UUID, version int) ([]pipelinetypes.GettablePipeline, error)
-	GetPipeline(ctx context.Context, orgID valuer.UUID, id string) (*pipelinetypes.GettablePipeline, error)
-	CreatePipeline(ctx context.Context, orgID valuer.UUID, claims *authtypes.Claims, pipeline *pipelinetypes.PostablePipeline) (*pipelinetypes.GettablePipeline, error)
-	UpdatePipeline(ctx context.Context, orgID valuer.UUID, claims *authtypes.Claims, pipeline *pipelinetypes.PostablePipeline) (*pipelinetypes.GettablePipeline, error)
-	DeletePipeline(ctx context.Context, orgID valuer.UUID, id string) error
-}
-
-type Handler interface {
-	ListPipelines(w http.ResponseWriter, r *http.Request)
-	GetPipeline(w http.ResponseWriter, r *http.Request)
-	CreatePipeline(w http.ResponseWriter, r *http.Request)
-	UpdatePipeline(w http.ResponseWriter, r *http.Request)
-	DeletePipeline(w http.ResponseWriter, r *http.Request)
-}

pkg/modules/promote/implpromote/module.go

@@ -87,7 +87,7 @@ func (m *module) ListPromotedAndIndexedPaths(ctx context.Context) ([]promotetype
 }
 
 func (m *module) listPromotedPaths(ctx context.Context) ([]string, error) {
-	paths, err := m.metadataStore.ListPromotedPaths(ctx)
+	paths, err := m.metadataStore.GetPromotedPaths(ctx)
 	if err != nil {
 		return nil, err
 	}
@@ -142,7 +142,7 @@ func (m *module) PromoteAndIndexPaths(
 		pathsStr = append(pathsStr, path.Path)
 	}
 
-	existingPromotedPaths, err := m.metadataStore.ListPromotedPaths(ctx, pathsStr...)
+	existingPromotedPaths, err := m.metadataStore.GetPromotedPaths(ctx, pathsStr...)
 	if err != nil {
 		return err
 	}

pkg/modules/thirdpartyapi/translator.go

@@ -120,6 +120,8 @@ func FilterResponse(results []*qbtypes.QueryRangeResponse) []*qbtypes.QueryRange
 					}
 				}
 				resultData.Rows = filteredRows
+			case *qbtypes.ScalarData:
+				resultData.Data = filterScalarDataIPs(resultData.Columns, resultData.Data)
 			}
 
 			filteredData = append(filteredData, result)
@@ -145,6 +147,39 @@ func shouldIncludeSeries(series *qbtypes.TimeSeries) bool {
 	return true
 }
 
+func filterScalarDataIPs(columns []*qbtypes.ColumnDescriptor, data [][]any) [][]any {
+	// Find column indices for server address fields
+	serverColIndices := make([]int, 0)
+	for i, col := range columns {
+		if col.Name == derivedKeyHTTPHost {
+			serverColIndices = append(serverColIndices, i)
+		}
+	}
+
+	if len(serverColIndices) == 0 {
+		return data
+	}
+
+	filtered := make([][]any, 0, len(data))
+	for _, row := range data {
+		includeRow := true
+		for _, colIdx := range serverColIndices {
+			if colIdx < len(row) {
+				if strVal, ok := row[colIdx].(string); ok {
+					if net.ParseIP(strVal) != nil {
+						includeRow = false
+						break
+					}
+				}
+			}
+		}
+		if includeRow {
+			filtered = append(filtered, row)
+		}
+	}
+	return filtered
+}
+
 func shouldIncludeRow(row *qbtypes.RawRow) bool {
 	if row.Data != nil {
 		if domainVal, ok := row.Data[derivedKeyHTTPHost]; ok {

pkg/modules/thirdpartyapi/translator_test.go

@@ -117,6 +117,59 @@ func TestFilterResponse(t *testing.T) {
 				},
 			},
 		},
+		{
+			name: "should filter out IP addresses from scalar data",
+			input: []*qbtypes.QueryRangeResponse{
+				{
+					Data: qbtypes.QueryData{
+						Results: []any{
+							&qbtypes.ScalarData{
+								QueryName: "endpoints",
+								Columns: []*qbtypes.ColumnDescriptor{
+									{
+										TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{Name: derivedKeyHTTPHost},
+										Type:              qbtypes.ColumnTypeGroup,
+									},
+									{
+										TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{Name: "endpoints"},
+										Type:              qbtypes.ColumnTypeAggregation,
+									},
+								},
+								Data: [][]any{
+									{"192.168.1.1", 10},
+									{"example.com", 20},
+									{"10.0.0.1", 5},
+								},
+							},
+						},
+					},
+				},
+			},
+			expected: []*qbtypes.QueryRangeResponse{
+				{
+					Data: qbtypes.QueryData{
+						Results: []any{
+							&qbtypes.ScalarData{
+								QueryName: "endpoints",
+								Columns: []*qbtypes.ColumnDescriptor{
+									{
+										TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{Name: derivedKeyHTTPHost},
+										Type:              qbtypes.ColumnTypeGroup,
+									},
+									{
+										TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{Name: "endpoints"},
+										Type:              qbtypes.ColumnTypeAggregation,
+									},
+								},
+								Data: [][]any{
+									{"example.com", 20},
+								},
+							},
+						},
+					},
+				},
+			},
+		},
 	}
 
 	for _, tt := range tests {

pkg/query-service/app/http_handler.go

@@ -9,7 +9,6 @@ import (
 
 	"github.com/SigNoz/signoz/pkg/errors"
 	"github.com/SigNoz/signoz/pkg/flagger"
-	"github.com/SigNoz/signoz/pkg/modules/logspipeline/impllogspipeline"
 	"github.com/SigNoz/signoz/pkg/modules/thirdpartyapi"
 	"github.com/SigNoz/signoz/pkg/queryparser"
 
@@ -4049,6 +4048,26 @@ func (aH *APIHandler) logAggregate(w http.ResponseWriter, r *http.Request) {
 	aH.WriteJSON(w, r, model.GetLogsAggregatesResponse{})
 }
 
+func parseAgentConfigVersion(r *http.Request) (int, error) {
+	versionString := mux.Vars(r)["version"]
+
+	if versionString == "latest" {
+		return -1, nil
+	}
+
+	version64, err := strconv.ParseInt(versionString, 0, 8)
+
+	if err != nil {
+		return 0, errors.WrapInvalidInputf(err, errors.CodeInvalidInput, "invalid version number")
+	}
+
+	if version64 <= 0 {
+		return 0, errors.NewInvalidInputf(errors.CodeInvalidInput, "invalid version number")
+	}
+
+	return int(version64), nil
+}
+
 func (aH *APIHandler) PreviewLogsPipelinesHandler(w http.ResponseWriter, r *http.Request) {
 	req := logparsingpipeline.PipelinesPreviewRequest{}
 
@@ -4079,7 +4098,7 @@ func (aH *APIHandler) ListLogsPipelinesHandler(w http.ResponseWriter, r *http.Re
 		return
 	}
 
-	version, err := impllogspipeline.ParseAgentConfigVersion(r)
+	version, err := parseAgentConfigVersion(r)
 	if err != nil {
 		render.Error(w, err)
 		return

pkg/signoz/module.go

@@ -13,8 +13,6 @@ import (
 	"github.com/SigNoz/signoz/pkg/modules/authdomain"
 	"github.com/SigNoz/signoz/pkg/modules/authdomain/implauthdomain"
 	"github.com/SigNoz/signoz/pkg/modules/dashboard"
-	"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"
@@ -68,7 +66,6 @@ type Modules struct {
 	SpanPercentile  spanpercentile.Module
 	MetricsExplorer metricsexplorer.Module
 	Promote         promote.Module
-	LogsPipeline    logspipeline.Module
 }
 
 func NewModules(
@@ -113,6 +110,5 @@ func NewModules(
 		Services:        implservices.NewModule(querier, telemetryStore),
 		MetricsExplorer: implmetricsexplorer.NewModule(telemetryStore, telemetryMetadataStore, cache, ruleStore, dashboard, providerSettings, config.MetricsExplorer),
 		Promote:         implpromote.NewModule(telemetryMetadataStore, telemetryStore),
-		LogsPipeline:    impllogspipeline.NewModule(sqlstore),
 	}
 }

pkg/signoz/openapi.go

@@ -18,7 +18,6 @@ import (
 	"github.com/SigNoz/signoz/pkg/modules/authdomain"
 	"github.com/SigNoz/signoz/pkg/modules/dashboard"
 	"github.com/SigNoz/signoz/pkg/modules/fields"
-	"github.com/SigNoz/signoz/pkg/modules/logspipeline"
 	"github.com/SigNoz/signoz/pkg/modules/metricsexplorer"
 	"github.com/SigNoz/signoz/pkg/modules/organization"
 	"github.com/SigNoz/signoz/pkg/modules/preference"
@@ -60,7 +59,6 @@ func NewOpenAPI(ctx context.Context, instrumentation instrumentation.Instrumenta
 		struct{ authz.Handler }{},
 		struct{ zeus.Handler }{},
 		struct{ querier.Handler }{},
-		struct{ logspipeline.Handler }{},
 	).New(ctx, instrumentation.ToProviderSettings(), apiserver.Config{})
 	if err != nil {
 		return nil, err

pkg/signoz/provider.go

@@ -23,7 +23,6 @@ import (
 	"github.com/SigNoz/signoz/pkg/global"
 	"github.com/SigNoz/signoz/pkg/global/signozglobal"
 	"github.com/SigNoz/signoz/pkg/modules/authdomain/implauthdomain"
-	"github.com/SigNoz/signoz/pkg/modules/logspipeline/impllogspipeline"
 	"github.com/SigNoz/signoz/pkg/modules/organization"
 	"github.com/SigNoz/signoz/pkg/modules/organization/implorganization"
 	"github.com/SigNoz/signoz/pkg/modules/preference/implpreference"
@@ -255,7 +254,6 @@ func NewAPIServerProviderFactories(orgGetter organization.Getter, authz authz.Au
 			handlers.AuthzHandler,
 			handlers.ZeusHandler,
 			handlers.QuerierHandler,
-			impllogspipeline.NewHandler(modules.LogsPipeline),
 		),
 	)
 }

pkg/telemetrymetadata/body_json_metadata.go

@@ -10,7 +10,6 @@ import (
 	"github.com/ClickHouse/clickhouse-go/v2/lib/chcol"
 	schemamigrator "github.com/SigNoz/signoz-otel-collector/cmd/signozschemamigrator/schema_migrator"
 	"github.com/SigNoz/signoz-otel-collector/constants"
-	"github.com/SigNoz/signoz-otel-collector/utils"
 	"github.com/SigNoz/signoz/pkg/errors"
 	"github.com/SigNoz/signoz/pkg/querybuilder"
 	"github.com/SigNoz/signoz/pkg/telemetrylogs"
@@ -113,7 +112,7 @@ func (t *telemetryMetaStore) buildBodyJSONPaths(ctx context.Context,
 
 	for _, fieldKey := range fieldKeys {
 		promotedKey := strings.Split(fieldKey.Name, telemetrytypes.ArraySep)[0]
-		fieldKey.Materialized = promoted.Contains(promotedKey)
+		fieldKey.Materialized = promoted[promotedKey]
 		fieldKey.Indexes = indexes[fieldKey.Name]
 	}
 
@@ -295,33 +294,6 @@ func (t *telemetryMetaStore) ListLogsJSONIndexes(ctx context.Context, filters ..
 	return indexes, nil
 }
 
-func (t *telemetryMetaStore) ListPromotedPaths(ctx context.Context, paths ...string) (map[string]struct{}, error) {
-	sb := sqlbuilder.Select("path").From(fmt.Sprintf("%s.%s", DBName, PromotedPathsTableName))
-	pathConditions := []string{}
-	for _, path := range paths {
-		pathConditions = append(pathConditions, sb.Equal("path", path))
-	}
-	sb.Where(sb.Or(pathConditions...))
-	query, args := sb.BuildWithFlavor(sqlbuilder.ClickHouse)
-
-	rows, err := t.telemetrystore.ClickhouseDB().Query(ctx, query, args...)
-	if err != nil {
-		return nil, errors.WrapInternalf(err, CodeFailLoadPromotedPaths, "failed to load promoted paths")
-	}
-	defer rows.Close()
-
-	next := make(map[string]struct{})
-	for rows.Next() {
-		var path string
-		if err := rows.Scan(&path); err != nil {
-			return nil, errors.WrapInternalf(err, CodeFailLoadPromotedPaths, "failed to scan promoted path")
-		}
-		next[path] = struct{}{}
-	}
-
-	return next, nil
-}
-
 // TODO(Piyush): Remove this if not used in future
 func (t *telemetryMetaStore) ListJSONValues(ctx context.Context, path string, limit int) (*telemetrytypes.TelemetryFieldValues, bool, error) {
 	path = CleanPathPrefixes(path)
@@ -484,11 +456,12 @@ func derefValue(v any) any {
 	return val.Interface()
 }
 
-// IsPathPromoted checks if a specific path is promoted
+// IsPathPromoted checks if a specific path is promoted (Column Evolution table: field_name for logs body).
 func (t *telemetryMetaStore) IsPathPromoted(ctx context.Context, path string) (bool, error) {
 	split := strings.Split(path, telemetrytypes.ArraySep)
-	query := fmt.Sprintf("SELECT 1 FROM %s.%s WHERE path = ? LIMIT 1", DBName, PromotedPathsTableName)
-	rows, err := t.telemetrystore.ClickhouseDB().Query(ctx, query, split[0])
+	pathSegment := split[0]
+	query := fmt.Sprintf("SELECT 1 FROM %s.%s WHERE signal = ? AND column_name = ? AND field_context = ? AND field_name = ? LIMIT 1", DBName, PromotedPathsTableName)
+	rows, err := t.telemetrystore.ClickhouseDB().Query(ctx, query, telemetrytypes.SignalLogs, telemetrylogs.LogsV2BodyPromotedColumn, telemetrytypes.FieldContextBody, pathSegment)
 	if err != nil {
 		return false, errors.WrapInternalf(err, CodeFailCheckPathPromoted, "failed to check if path %s is promoted", path)
 	}
@@ -497,15 +470,24 @@ func (t *telemetryMetaStore) IsPathPromoted(ctx context.Context, path string) (b
 	return rows.Next(), nil
 }
 
-// GetPromotedPaths checks if a specific path is promoted
-func (t *telemetryMetaStore) GetPromotedPaths(ctx context.Context, paths ...string) (*utils.ConcurrentSet[string], error) {
-	sb := sqlbuilder.Select("path").From(fmt.Sprintf("%s.%s", DBName, PromotedPathsTableName))
-	pathConditions := []string{}
-	for _, path := range paths {
-		split := strings.Split(path, telemetrytypes.ArraySep)
-		pathConditions = append(pathConditions, sb.Equal("path", split[0]))
+// GetPromotedPaths returns promoted paths from the Column Evolution table (field_name for logs body).
+func (t *telemetryMetaStore) GetPromotedPaths(ctx context.Context, paths ...string) (map[string]bool, error) {
+	sb := sqlbuilder.Select("field_name").From(fmt.Sprintf("%s.%s", DBName, PromotedPathsTableName))
+	conditions := []string{
+		sb.Equal("signal", telemetrytypes.SignalLogs),
+		sb.Equal("column_name", telemetrylogs.LogsV2BodyPromotedColumn),
+		sb.Equal("field_context", telemetrytypes.FieldContextBody),
+		sb.NotEqual("field_name", "__all__"),
 	}
-	sb.Where(sb.Or(pathConditions...))
+	if len(paths) > 0 {
+		pathArgs := make([]interface{}, len(paths))
+		for i, path := range paths {
+			split := strings.Split(path, telemetrytypes.ArraySep)
+			pathArgs[i] = split[0]
+		}
+		conditions = append(conditions, sb.In("field_name", pathArgs))
+	}
+	sb.Where(sb.And(conditions...))
 
 	query, args := sb.BuildWithFlavor(sqlbuilder.ClickHouse)
 	rows, err := t.telemetrystore.ClickhouseDB().Query(ctx, query, args...)
@@ -514,13 +496,13 @@ func (t *telemetryMetaStore) GetPromotedPaths(ctx context.Context, paths ...stri
 	}
 	defer rows.Close()
 
-	promotedPaths := utils.NewConcurrentSet[string]()
+	promotedPaths := make(map[string]bool)
 	for rows.Next() {
-		var path string
-		if err := rows.Scan(&path); err != nil {
+		var fieldName string
+		if err := rows.Scan(&fieldName); err != nil {
 			return nil, errors.WrapInternalf(err, CodeFailCheckPathPromoted, "failed to scan promoted path")
 		}
-		promotedPaths.Insert(path)
+		promotedPaths[fieldName] = true
 	}
 
 	return promotedPaths, nil
@@ -534,21 +516,22 @@ func CleanPathPrefixes(path string) string {
 	return path
 }
 
+// PromotePaths inserts promoted paths into the Column Evolution table (same schema as signoz-otel-collector metadata_migrations).
 func (t *telemetryMetaStore) PromotePaths(ctx context.Context, paths ...string) error {
 	batch, err := t.telemetrystore.ClickhouseDB().PrepareBatch(ctx,
-		fmt.Sprintf("INSERT INTO %s.%s (path, created_at) VALUES", DBName,
+		fmt.Sprintf("INSERT INTO %s.%s (signal, column_name, column_type, field_context, field_name, version, release_time) VALUES", DBName,
 			PromotedPathsTableName))
 	if err != nil {
 		return errors.WrapInternalf(err, CodeFailedToPrepareBatch, "failed to prepare batch")
 	}
 
-	nowMs := uint64(time.Now().UnixMilli())
+	releaseTime := time.Now().UnixNano()
 	for _, p := range paths {
 		trimmed := strings.TrimSpace(p)
 		if trimmed == "" {
 			continue
 		}
-		if err := batch.Append(trimmed, nowMs); err != nil {
+		if err := batch.Append(telemetrytypes.SignalLogs, telemetrylogs.LogsV2BodyPromotedColumn, "JSON()", telemetrytypes.FieldContextBody, trimmed, 0, releaseTime); err != nil {
 			_ = batch.Abort()
 			return errors.WrapInternalf(err, CodeFailedToAppendPath, "failed to append path")
 		}

pkg/telemetrymetadata/tables.go

@@ -7,6 +7,7 @@ const (
 	AttributesMetadataTableName      = "distributed_attributes_metadata"
 	AttributesMetadataLocalTableName = "attributes_metadata"
 	PathTypesTableName               = otelcollectorconst.DistributedPathTypesTable
-	PromotedPathsTableName           = otelcollectorconst.DistributedPromotedPathsTable
+	// Column Evolution table stores promoted paths as (signal, column_name, field_context, field_name); see signoz-otel-collector metadata_migrations.
+	PromotedPathsTableName           = "distributed_column_evolution_metadata"
 	SkipIndexTableName               = "system.data_skipping_indices"
 )

pkg/types/pipelinetypes/pipeline.go

@@ -10,7 +10,6 @@ import (
 	v3 "github.com/SigNoz/signoz/pkg/query-service/model/v3"
 	"github.com/SigNoz/signoz/pkg/query-service/queryBuilderToExpr"
 	"github.com/SigNoz/signoz/pkg/types"
-	"github.com/SigNoz/signoz/pkg/valuer"
 	"github.com/uptrace/bun"
 )
 
@@ -267,37 +266,6 @@ func (p *PostablePipeline) IsValid() error {
 	return nil
 }
 
-func (p *PostablePipeline) ToStoreablePipeline() (*StoreablePipeline, error) {
-	rawConfig, err := json.Marshal(p.Config)
-	if err != nil {
-		return nil, errors.WrapInternalf(err, errors.CodeInternal, "failed to unmarshal postable pipeline config")
-	}
-
-	filter, err := json.Marshal(p.Filter)
-	if err != nil {
-		return nil, errors.WrapInternalf(err, errors.CodeInternal, "failed to marshal postable pipeline filter")
-	}
-	identifier := valuer.GenerateUUID()
-	if p.ID != "" {
-		identifier, err = valuer.NewUUID(p.ID)
-		if err != nil {
-			return nil, errors.WithAdditionalf(err, "failed to parse postable pipeline id")
-		}
-	}
-	return &StoreablePipeline{
-		Identifiable: types.Identifiable{
-			ID: identifier,
-		},
-		OrderID:      p.OrderID,
-		Enabled:      p.Enabled,
-		Name:         p.Name,
-		Alias:        p.Alias,
-		Description:  p.Description,
-		FilterString: string(filter),
-		ConfigJSON:   string(rawConfig),
-	}, nil
-}
-
 func isValidOperator(op PipelineOperator) error {
 	if op.ID == "" {
 		return errors.New(errors.TypeInvalidInput, errors.CodeInvalidInput, "PipelineOperator.ID is required")

pkg/types/telemetrytypes/store.go

@@ -36,7 +36,7 @@ type MetadataStore interface {
 	ListLogsJSONIndexes(ctx context.Context, filters ...string) (map[string][]schemamigrator.Index, error)
 
 	// ListPromotedPaths lists the promoted paths.
-	ListPromotedPaths(ctx context.Context, paths ...string) (map[string]struct{}, error)
+	GetPromotedPaths(ctx context.Context, paths ...string) (map[string]bool, error)
 
 	// PromotePaths promotes the paths.
 	PromotePaths(ctx context.Context, paths ...string) error

pkg/types/telemetrytypes/telemetrytypestest/metadata_store.go

@@ -16,7 +16,7 @@ type MockMetadataStore struct {
 	RelatedValuesMap   map[string][]string
 	AllValuesMap       map[string]*telemetrytypes.TelemetryFieldValues
 	TemporalityMap     map[string]metrictypes.Temporality
-	PromotedPathsMap   map[string]struct{}
+	PromotedPathsMap   map[string]bool
 	LogsJSONIndexesMap map[string][]schemamigrator.Index
 	LookupKeysMap      map[telemetrytypes.MetricMetadataLookupKey]int64
 }
@@ -28,7 +28,7 @@ func NewMockMetadataStore() *MockMetadataStore {
 		RelatedValuesMap:   make(map[string][]string),
 		AllValuesMap:       make(map[string]*telemetrytypes.TelemetryFieldValues),
 		TemporalityMap:     make(map[string]metrictypes.Temporality),
-		PromotedPathsMap:   make(map[string]struct{}),
+		PromotedPathsMap:   make(map[string]bool),
 		LogsJSONIndexesMap: make(map[string][]schemamigrator.Index),
 		LookupKeysMap:      make(map[telemetrytypes.MetricMetadataLookupKey]int64),
 	}
@@ -295,13 +295,13 @@ func (m *MockMetadataStore) SetTemporality(metricName string, temporality metric
 // PromotePaths promotes the paths.
 func (m *MockMetadataStore) PromotePaths(ctx context.Context, paths ...string) error {
 	for _, path := range paths {
-		m.PromotedPathsMap[path] = struct{}{}
+		m.PromotedPathsMap[path] = true
 	}
 	return nil
 }
 
-// ListPromotedPaths lists the promoted paths.
-func (m *MockMetadataStore) ListPromotedPaths(ctx context.Context, paths ...string) (map[string]struct{}, error) {
+// GetPromotedPaths returns the promoted paths.
+func (m *MockMetadataStore) GetPromotedPaths(ctx context.Context, paths ...string) (map[string]bool, error) {
 	return m.PromotedPathsMap, nil
 }