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>

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/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/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/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
 }