Merge branch 'main' into has-value-fix

deploy/docker-swarm/docker-compose.ha.yaml

@@ -190,7 +190,7 @@ services:
       # - ../common/clickhouse/storage.xml:/etc/clickhouse-server/config.d/storage.xml
   signoz:
     !!merge <<: *db-depend
-    image: signoz/signoz:v0.118.0
+    image: signoz/signoz:v0.117.1
     ports:
       - "8080:8080" # signoz port
     #   - "6060:6060"     # pprof port

deploy/docker-swarm/docker-compose.yaml

@@ -117,7 +117,7 @@ services:
       # - ../common/clickhouse/storage.xml:/etc/clickhouse-server/config.d/storage.xml
   signoz:
     !!merge <<: *db-depend
-    image: signoz/signoz:v0.118.0
+    image: signoz/signoz:v0.117.1
     ports:
       - "8080:8080" # signoz port
     volumes:

deploy/docker/docker-compose.ha.yaml

@@ -181,7 +181,7 @@ services:
       # - ../common/clickhouse/storage.xml:/etc/clickhouse-server/config.d/storage.xml
   signoz:
     !!merge <<: *db-depend
-    image: signoz/signoz:${VERSION:-v0.118.0}
+    image: signoz/signoz:${VERSION:-v0.117.1}
     container_name: signoz
     ports:
       - "8080:8080" # signoz port

deploy/docker/docker-compose.yaml

@@ -109,7 +109,7 @@ services:
       # - ../common/clickhouse/storage.xml:/etc/clickhouse-server/config.d/storage.xml
   signoz:
     !!merge <<: *db-depend
-    image: signoz/signoz:${VERSION:-v0.118.0}
+    image: signoz/signoz:${VERSION:-v0.117.1}
     container_name: signoz
     ports:
       - "8080:8080" # signoz port

pkg/querier/querier.go

@@ -40,7 +40,6 @@ type querier struct {
 	promEngine               prometheus.Prometheus
 	traceStmtBuilder         qbtypes.StatementBuilder[qbtypes.TraceAggregation]
 	logStmtBuilder           qbtypes.StatementBuilder[qbtypes.LogAggregation]
-	auditStmtBuilder         qbtypes.StatementBuilder[qbtypes.LogAggregation]
 	metricStmtBuilder        qbtypes.StatementBuilder[qbtypes.MetricAggregation]
 	meterStmtBuilder         qbtypes.StatementBuilder[qbtypes.MetricAggregation]
 	traceOperatorStmtBuilder qbtypes.TraceOperatorStatementBuilder
@@ -57,7 +56,6 @@ func New(
 	promEngine prometheus.Prometheus,
 	traceStmtBuilder qbtypes.StatementBuilder[qbtypes.TraceAggregation],
 	logStmtBuilder qbtypes.StatementBuilder[qbtypes.LogAggregation],
-	auditStmtBuilder qbtypes.StatementBuilder[qbtypes.LogAggregation],
 	metricStmtBuilder qbtypes.StatementBuilder[qbtypes.MetricAggregation],
 	meterStmtBuilder qbtypes.StatementBuilder[qbtypes.MetricAggregation],
 	traceOperatorStmtBuilder qbtypes.TraceOperatorStatementBuilder,
@@ -71,7 +69,6 @@ func New(
 		promEngine:               promEngine,
 		traceStmtBuilder:         traceStmtBuilder,
 		logStmtBuilder:           logStmtBuilder,
-		auditStmtBuilder:         auditStmtBuilder,
 		metricStmtBuilder:        metricStmtBuilder,
 		meterStmtBuilder:         meterStmtBuilder,
 		traceOperatorStmtBuilder: traceOperatorStmtBuilder,
@@ -364,11 +361,7 @@ func (q *querier) QueryRange(ctx context.Context, orgID valuer.UUID, req *qbtype
 			case qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]:
 				spec.ShiftBy = extractShiftFromBuilderQuery(spec)
 				timeRange := adjustTimeRangeForShift(spec, qbtypes.TimeRange{From: req.Start, To: req.End}, req.RequestType)
-				stmtBuilder := q.logStmtBuilder
-				if spec.Source == telemetrytypes.SourceAudit {
-					stmtBuilder = q.auditStmtBuilder
-				}
-				bq := newBuilderQuery(q.logger, q.telemetryStore, stmtBuilder, spec, timeRange, req.RequestType, tmplVars)
+				bq := newBuilderQuery(q.logger, q.telemetryStore, q.logStmtBuilder, spec, timeRange, req.RequestType, tmplVars)
 				queries[spec.Name] = bq
 				steps[spec.Name] = spec.StepInterval
 			case qbtypes.QueryBuilderQuery[qbtypes.MetricAggregation]:
@@ -557,11 +550,7 @@ func (q *querier) QueryRawStream(ctx context.Context, orgID valuer.UUID, req *qb
 		case <-tick:
 			// timestamp end is not specified here
 			timeRange := adjustTimeRangeForShift(spec, qbtypes.TimeRange{From: tsStart}, req.RequestType)
-			liveTailStmtBuilder := q.logStmtBuilder
-			if spec.Source == telemetrytypes.SourceAudit {
-				liveTailStmtBuilder = q.auditStmtBuilder
-			}
-			bq := newBuilderQuery(q.logger, q.telemetryStore, liveTailStmtBuilder, spec, timeRange, req.RequestType, map[string]qbtypes.VariableItem{
+			bq := newBuilderQuery(q.logger, q.telemetryStore, q.logStmtBuilder, spec, timeRange, req.RequestType, map[string]qbtypes.VariableItem{
 				"id": {
 					Value: updatedLogID,
 				},
@@ -861,11 +850,7 @@ func (q *querier) createRangedQuery(originalQuery qbtypes.Query, timeRange qbtyp
 		specCopy := qt.spec.Copy()
 		specCopy.ShiftBy = extractShiftFromBuilderQuery(specCopy)
 		adjustedTimeRange := adjustTimeRangeForShift(specCopy, timeRange, qt.kind)
-		shiftStmtBuilder := q.logStmtBuilder
-		if qt.spec.Source == telemetrytypes.SourceAudit {
-			shiftStmtBuilder = q.auditStmtBuilder
-		}
-		return newBuilderQuery(q.logger, q.telemetryStore, shiftStmtBuilder, specCopy, adjustedTimeRange, qt.kind, qt.variables)
+		return newBuilderQuery(q.logger, q.telemetryStore, q.logStmtBuilder, specCopy, adjustedTimeRange, qt.kind, qt.variables)
 
 	case *builderQuery[qbtypes.MetricAggregation]:
 		specCopy := qt.spec.Copy()

pkg/querier/signozquerier/provider.go

@@ -9,7 +9,6 @@ import (
 	"github.com/SigNoz/signoz/pkg/prometheus"
 	"github.com/SigNoz/signoz/pkg/querier"
 	"github.com/SigNoz/signoz/pkg/querybuilder"
-	"github.com/SigNoz/signoz/pkg/telemetryaudit"
 	"github.com/SigNoz/signoz/pkg/telemetrylogs"
 	"github.com/SigNoz/signoz/pkg/telemetrymetadata"
 	"github.com/SigNoz/signoz/pkg/telemetrymeter"
@@ -64,11 +63,6 @@ func newProvider(
 		telemetrylogs.TagAttributesV2TableName,
 		telemetrylogs.LogAttributeKeysTblName,
 		telemetrylogs.LogResourceKeysTblName,
-		telemetryaudit.DBName,
-		telemetryaudit.AuditLogsTableName,
-		telemetryaudit.TagAttributesTableName,
-		telemetryaudit.LogAttributeKeysTblName,
-		telemetryaudit.LogResourceKeysTblName,
 		telemetrymetadata.DBName,
 		telemetrymetadata.AttributesMetadataLocalTableName,
 		telemetrymetadata.ColumnEvolutionMetadataTableName,
@@ -88,13 +82,13 @@ func newProvider(
 		telemetryStore,
 	)
 
-	// Create trace operator statement builder
+	// ADD: Create trace operator statement builder
 	traceOperatorStmtBuilder := telemetrytraces.NewTraceOperatorStatementBuilder(
 		settings,
 		telemetryMetadataStore,
 		traceFieldMapper,
 		traceConditionBuilder,
-		traceStmtBuilder,
+		traceStmtBuilder, // Pass the regular trace statement builder
 		traceAggExprRewriter,
 	)
 
@@ -118,26 +112,6 @@ func newProvider(
 		telemetrylogs.GetBodyJSONKey,
 	)
 
-	// Create audit statement builder
-	auditFieldMapper := telemetryaudit.NewFieldMapper()
-	auditConditionBuilder := telemetryaudit.NewConditionBuilder(auditFieldMapper)
-	auditAggExprRewriter := querybuilder.NewAggExprRewriter(
-		settings,
-		telemetryaudit.DefaultFullTextColumn,
-		auditFieldMapper,
-		auditConditionBuilder,
-		nil,
-	)
-	auditStmtBuilder := telemetryaudit.NewAuditQueryStatementBuilder(
-		settings,
-		telemetryMetadataStore,
-		auditFieldMapper,
-		auditConditionBuilder,
-		auditAggExprRewriter,
-		telemetryaudit.DefaultFullTextColumn,
-		nil,
-	)
-
 	// Create metric statement builder
 	metricFieldMapper := telemetrymetrics.NewFieldMapper()
 	metricConditionBuilder := telemetrymetrics.NewConditionBuilder(metricFieldMapper)
@@ -174,7 +148,6 @@ func newProvider(
 		prometheus,
 		traceStmtBuilder,
 		logStmtBuilder,
-		auditStmtBuilder,
 		metricStmtBuilder,
 		meterStmtBuilder,
 		traceOperatorStmtBuilder,

pkg/query-service/rules/setups_test.go

@@ -46,7 +46,6 @@ func prepareQuerierForMetrics(t *testing.T, telemetryStore telemetrystore.Teleme
 		nil, // prometheus
 		nil, // traceStmtBuilder
 		nil, // logStmtBuilder
-		nil, // auditStmtBuilder
 		metricStmtBuilder,
 		nil, // meterStmtBuilder
 		nil, // traceOperatorStmtBuilder
@@ -92,7 +91,6 @@ func prepareQuerierForLogs(telemetryStore telemetrystore.TelemetryStore, keysMap
 		nil,            // prometheus
 		nil,            // traceStmtBuilder
 		logStmtBuilder, // logStmtBuilder
-		nil,            // auditStmtBuilder
 		nil,            // metricStmtBuilder
 		nil,            // meterStmtBuilder
 		nil,            // traceOperatorStmtBuilder
@@ -133,7 +131,6 @@ func prepareQuerierForTraces(telemetryStore telemetrystore.TelemetryStore, keysM
 		nil,              // prometheus
 		traceStmtBuilder, // traceStmtBuilder
 		nil,              // logStmtBuilder
-		nil,              // auditStmtBuilder
 		nil,              // metricStmtBuilder
 		nil,              // meterStmtBuilder
 		nil,              // traceOperatorStmtBuilder

pkg/querybuilder/where_clause_visitor.go

@@ -818,9 +818,9 @@ func (v *filterExpressionVisitor) VisitFunctionCall(ctx *grammar.FunctionCallCon
 			case "has":
 				cond = fmt.Sprintf("has(%s, %s)", fieldName, v.builder.Var(value[0]))
 			case "hasAny":
-				cond = fmt.Sprintf("hasAny(%s, %s)", fieldName, v.builder.Var(value))
+				cond = fmt.Sprintf("hasAny(%s, %s)", fieldName, v.builder.Var(value[0]))
 			case "hasAll":
-				cond = fmt.Sprintf("hasAll(%s, %s)", fieldName, v.builder.Var(value))
+				cond = fmt.Sprintf("hasAll(%s, %s)", fieldName, v.builder.Var(value[0]))
 			}
 			conds = append(conds, cond)
 		}

pkg/signoz/signoz.go

@@ -33,7 +33,6 @@ import (
 	"github.com/SigNoz/signoz/pkg/sqlschema"
 	"github.com/SigNoz/signoz/pkg/sqlstore"
 	"github.com/SigNoz/signoz/pkg/statsreporter"
-	"github.com/SigNoz/signoz/pkg/telemetryaudit"
 	"github.com/SigNoz/signoz/pkg/telemetrylogs"
 	"github.com/SigNoz/signoz/pkg/telemetrymetadata"
 	"github.com/SigNoz/signoz/pkg/telemetrymeter"
@@ -396,11 +395,6 @@ func New(
 		telemetrylogs.TagAttributesV2TableName,
 		telemetrylogs.LogAttributeKeysTblName,
 		telemetrylogs.LogResourceKeysTblName,
-		telemetryaudit.DBName,
-		telemetryaudit.AuditLogsTableName,
-		telemetryaudit.TagAttributesTableName,
-		telemetryaudit.LogAttributeKeysTblName,
-		telemetryaudit.LogResourceKeysTblName,
 		telemetrymetadata.DBName,
 		telemetrymetadata.AttributesMetadataLocalTableName,
 		telemetrymetadata.ColumnEvolutionMetadataTableName,

pkg/telemetryaudit/condition_builder.go

@@ -1,200 +0,0 @@
-package telemetryaudit
-
-import (
-	"context"
-	"fmt"
-
-	schema "github.com/SigNoz/signoz-otel-collector/cmd/signozschemamigrator/schema_migrator"
-	"github.com/SigNoz/signoz/pkg/errors"
-	"github.com/SigNoz/signoz/pkg/querybuilder"
-	qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
-	"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
-	"github.com/huandu/go-sqlbuilder"
-)
-
-type conditionBuilder struct {
-	fm qbtypes.FieldMapper
-}
-
-func NewConditionBuilder(fm qbtypes.FieldMapper) *conditionBuilder {
-	return &conditionBuilder{fm: fm}
-}
-
-func (c *conditionBuilder) conditionFor(
-	ctx context.Context,
-	startNs, endNs uint64,
-	key *telemetrytypes.TelemetryFieldKey,
-	operator qbtypes.FilterOperator,
-	value any,
-	sb *sqlbuilder.SelectBuilder,
-) (string, error) {
-	columns, err := c.fm.ColumnFor(ctx, startNs, endNs, key)
-	if err != nil {
-		return "", err
-	}
-
-	if operator.IsStringSearchOperator() {
-		value = querybuilder.FormatValueForContains(value)
-	}
-
-	fieldExpression, err := c.fm.FieldFor(ctx, startNs, endNs, key)
-	if err != nil {
-		return "", err
-	}
-
-	fieldExpression, value = querybuilder.DataTypeCollisionHandledFieldName(key, value, fieldExpression, operator)
-
-	switch operator {
-	case qbtypes.FilterOperatorEqual:
-		return sb.E(fieldExpression, value), nil
-	case qbtypes.FilterOperatorNotEqual:
-		return sb.NE(fieldExpression, value), nil
-	case qbtypes.FilterOperatorGreaterThan:
-		return sb.G(fieldExpression, value), nil
-	case qbtypes.FilterOperatorGreaterThanOrEq:
-		return sb.GE(fieldExpression, value), nil
-	case qbtypes.FilterOperatorLessThan:
-		return sb.LT(fieldExpression, value), nil
-	case qbtypes.FilterOperatorLessThanOrEq:
-		return sb.LE(fieldExpression, value), nil
-	case qbtypes.FilterOperatorLike:
-		return sb.Like(fieldExpression, value), nil
-	case qbtypes.FilterOperatorNotLike:
-		return sb.NotLike(fieldExpression, value), nil
-	case qbtypes.FilterOperatorILike:
-		return sb.ILike(fieldExpression, value), nil
-	case qbtypes.FilterOperatorNotILike:
-		return sb.NotILike(fieldExpression, value), nil
-	case qbtypes.FilterOperatorContains:
-		return sb.ILike(fieldExpression, fmt.Sprintf("%%%s%%", value)), nil
-	case qbtypes.FilterOperatorNotContains:
-		return sb.NotILike(fieldExpression, fmt.Sprintf("%%%s%%", value)), nil
-	case qbtypes.FilterOperatorRegexp:
-		return fmt.Sprintf(`match(%s, %s)`, sqlbuilder.Escape(fieldExpression), sb.Var(value)), nil
-	case qbtypes.FilterOperatorNotRegexp:
-		return fmt.Sprintf(`NOT match(%s, %s)`, sqlbuilder.Escape(fieldExpression), sb.Var(value)), nil
-	case qbtypes.FilterOperatorBetween:
-		values, ok := value.([]any)
-		if !ok {
-			return "", qbtypes.ErrBetweenValues
-		}
-		if len(values) != 2 {
-			return "", qbtypes.ErrBetweenValues
-		}
-		return sb.Between(fieldExpression, values[0], values[1]), nil
-	case qbtypes.FilterOperatorNotBetween:
-		values, ok := value.([]any)
-		if !ok {
-			return "", qbtypes.ErrBetweenValues
-		}
-		if len(values) != 2 {
-			return "", qbtypes.ErrBetweenValues
-		}
-		return sb.NotBetween(fieldExpression, values[0], values[1]), nil
-	case qbtypes.FilterOperatorIn:
-		values, ok := value.([]any)
-		if !ok {
-			return "", qbtypes.ErrInValues
-		}
-		conditions := []string{}
-		for _, value := range values {
-			conditions = append(conditions, sb.E(fieldExpression, value))
-		}
-		return sb.Or(conditions...), nil
-	case qbtypes.FilterOperatorNotIn:
-		values, ok := value.([]any)
-		if !ok {
-			return "", qbtypes.ErrInValues
-		}
-		conditions := []string{}
-		for _, value := range values {
-			conditions = append(conditions, sb.NE(fieldExpression, value))
-		}
-		return sb.And(conditions...), nil
-	case qbtypes.FilterOperatorExists, qbtypes.FilterOperatorNotExists:
-		var value any
-		column := columns[0]
-
-		switch column.Type.GetType() {
-		case schema.ColumnTypeEnumJSON:
-			if operator == qbtypes.FilterOperatorExists {
-				return sb.IsNotNull(fieldExpression), nil
-			}
-			return sb.IsNull(fieldExpression), nil
-		case schema.ColumnTypeEnumLowCardinality:
-			switch elementType := column.Type.(schema.LowCardinalityColumnType).ElementType; elementType.GetType() {
-			case schema.ColumnTypeEnumString:
-				value = ""
-				if operator == qbtypes.FilterOperatorExists {
-					return sb.NE(fieldExpression, value), nil
-				}
-				return sb.E(fieldExpression, value), nil
-			default:
-				return "", errors.NewInvalidInputf(errors.CodeInvalidInput, "exists operator is not supported for low cardinality column type %s", elementType)
-			}
-		case schema.ColumnTypeEnumString:
-			value = ""
-			if operator == qbtypes.FilterOperatorExists {
-				return sb.NE(fieldExpression, value), nil
-			}
-			return sb.E(fieldExpression, value), nil
-		case schema.ColumnTypeEnumUInt64, schema.ColumnTypeEnumUInt32, schema.ColumnTypeEnumUInt8:
-			value = 0
-			if operator == qbtypes.FilterOperatorExists {
-				return sb.NE(fieldExpression, value), nil
-			}
-			return sb.E(fieldExpression, value), nil
-		case schema.ColumnTypeEnumMap:
-			keyType := column.Type.(schema.MapColumnType).KeyType
-			if _, ok := keyType.(schema.LowCardinalityColumnType); !ok {
-				return "", errors.NewInvalidInputf(errors.CodeInvalidInput, "key type %s is not supported for map column type %s", keyType, column.Type)
-			}
-
-			switch valueType := column.Type.(schema.MapColumnType).ValueType; valueType.GetType() {
-			case schema.ColumnTypeEnumString, schema.ColumnTypeEnumBool, schema.ColumnTypeEnumFloat64:
-				leftOperand := fmt.Sprintf("mapContains(%s, '%s')", column.Name, key.Name)
-				if key.Materialized {
-					leftOperand = telemetrytypes.FieldKeyToMaterializedColumnNameForExists(key)
-				}
-				if operator == qbtypes.FilterOperatorExists {
-					return sb.E(leftOperand, true), nil
-				}
-				return sb.NE(leftOperand, true), nil
-			default:
-				return "", errors.NewInvalidInputf(errors.CodeInvalidInput, "exists operator is not supported for map column type %s", valueType)
-			}
-		default:
-			return "", errors.NewInvalidInputf(errors.CodeInvalidInput, "exists operator is not supported for column type %s", column.Type)
-		}
-	}
-	return "", errors.NewInvalidInputf(errors.CodeInvalidInput, "unsupported operator: %v", operator)
-}
-
-func (c *conditionBuilder) ConditionFor(
-	ctx context.Context,
-	startNs uint64,
-	endNs uint64,
-	key *telemetrytypes.TelemetryFieldKey,
-	operator qbtypes.FilterOperator,
-	value any,
-	sb *sqlbuilder.SelectBuilder,
-) (string, error) {
-	condition, err := c.conditionFor(ctx, startNs, endNs, key, operator, value, sb)
-	if err != nil {
-		return "", err
-	}
-
-	if key.FieldContext == telemetrytypes.FieldContextLog || key.FieldContext == telemetrytypes.FieldContextScope {
-		return condition, nil
-	}
-
-	if operator.AddDefaultExistsFilter() {
-		existsCondition, err := c.conditionFor(ctx, startNs, endNs, key, qbtypes.FilterOperatorExists, nil, sb)
-		if err != nil {
-			return "", err
-		}
-		return sb.And(condition, existsCondition), nil
-	}
-
-	return condition, nil
-}

pkg/telemetryaudit/const.go

@@ -1,129 +0,0 @@
-package telemetryaudit
-
-import (
-	schema "github.com/SigNoz/signoz-otel-collector/cmd/signozschemamigrator/schema_migrator"
-	qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
-	"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
-)
-
-const (
-	// Internal Columns.
-	IDColumn                   = "id"
-	TimestampBucketStartColumn = "ts_bucket_start"
-	ResourceFingerPrintColumn  = "resource_fingerprint"
-
-	// Intrinsic Columns.
-	TimestampColumn         = "timestamp"
-	ObservedTimestampColumn = "observed_timestamp"
-	BodyColumn              = "body"
-	EventNameColumn         = "event_name"
-	TraceIDColumn           = "trace_id"
-	SpanIDColumn            = "span_id"
-	TraceFlagsColumn        = "trace_flags"
-	SeverityTextColumn      = "severity_text"
-	SeverityNumberColumn    = "severity_number"
-	ScopeNameColumn         = "scope_name"
-	ScopeVersionColumn      = "scope_version"
-
-	// Contextual Columns.
-	AttributesStringColumn = "attributes_string"
-	AttributesNumberColumn = "attributes_number"
-	AttributesBoolColumn   = "attributes_bool"
-	ResourceColumn         = "resource"
-	ScopeStringColumn      = "scope_string"
-)
-
-var (
-	DefaultFullTextColumn = &telemetrytypes.TelemetryFieldKey{
-		Name:          "body",
-		Signal:        telemetrytypes.SignalLogs,
-		FieldContext:  telemetrytypes.FieldContextLog,
-		FieldDataType: telemetrytypes.FieldDataTypeString,
-	}
-
-	IntrinsicFields = map[string]telemetrytypes.TelemetryFieldKey{
-		"body": {
-			Name:          "body",
-			Signal:        telemetrytypes.SignalLogs,
-			FieldContext:  telemetrytypes.FieldContextLog,
-			FieldDataType: telemetrytypes.FieldDataTypeString,
-		},
-		"trace_id": {
-			Name:          "trace_id",
-			Signal:        telemetrytypes.SignalLogs,
-			FieldContext:  telemetrytypes.FieldContextLog,
-			FieldDataType: telemetrytypes.FieldDataTypeString,
-		},
-		"span_id": {
-			Name:          "span_id",
-			Signal:        telemetrytypes.SignalLogs,
-			FieldContext:  telemetrytypes.FieldContextLog,
-			FieldDataType: telemetrytypes.FieldDataTypeString,
-		},
-		"trace_flags": {
-			Name:          "trace_flags",
-			Signal:        telemetrytypes.SignalLogs,
-			FieldContext:  telemetrytypes.FieldContextLog,
-			FieldDataType: telemetrytypes.FieldDataTypeNumber,
-		},
-		"severity_text": {
-			Name:          "severity_text",
-			Signal:        telemetrytypes.SignalLogs,
-			FieldContext:  telemetrytypes.FieldContextLog,
-			FieldDataType: telemetrytypes.FieldDataTypeString,
-		},
-		"severity_number": {
-			Name:          "severity_number",
-			Signal:        telemetrytypes.SignalLogs,
-			FieldContext:  telemetrytypes.FieldContextLog,
-			FieldDataType: telemetrytypes.FieldDataTypeNumber,
-		},
-		"event_name": {
-			Name:          "event_name",
-			Signal:        telemetrytypes.SignalLogs,
-			FieldContext:  telemetrytypes.FieldContextLog,
-			FieldDataType: telemetrytypes.FieldDataTypeString,
-		},
-	}
-
-	DefaultSortingOrder = []qbtypes.OrderBy{
-		{
-			Key: qbtypes.OrderByKey{
-				TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{
-					Name: TimestampColumn,
-				},
-			},
-			Direction: qbtypes.OrderDirectionDesc,
-		},
-		{
-			Key: qbtypes.OrderByKey{
-				TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{
-					Name: IDColumn,
-				},
-			},
-			Direction: qbtypes.OrderDirectionDesc,
-		},
-	}
-)
-
-var auditLogColumns = map[string]*schema.Column{
-	"ts_bucket_start":      {Name: "ts_bucket_start", Type: schema.ColumnTypeUInt64},
-	"resource_fingerprint": {Name: "resource_fingerprint", Type: schema.ColumnTypeString},
-	"timestamp":            {Name: "timestamp", Type: schema.ColumnTypeUInt64},
-	"observed_timestamp":   {Name: "observed_timestamp", Type: schema.ColumnTypeUInt64},
-	"id":                   {Name: "id", Type: schema.ColumnTypeString},
-	"trace_id":             {Name: "trace_id", Type: schema.ColumnTypeString},
-	"span_id":              {Name: "span_id", Type: schema.ColumnTypeString},
-	"trace_flags":          {Name: "trace_flags", Type: schema.ColumnTypeUInt32},
-	"severity_text":        {Name: "severity_text", Type: schema.LowCardinalityColumnType{ElementType: schema.ColumnTypeString}},
-	"severity_number":      {Name: "severity_number", Type: schema.ColumnTypeUInt8},
-	"body":                 {Name: "body", Type: schema.ColumnTypeString},
-	"attributes_string":    {Name: "attributes_string", Type: schema.MapColumnType{KeyType: schema.LowCardinalityColumnType{ElementType: schema.ColumnTypeString}, ValueType: schema.ColumnTypeString}},
-	"attributes_number":    {Name: "attributes_number", Type: schema.MapColumnType{KeyType: schema.LowCardinalityColumnType{ElementType: schema.ColumnTypeString}, ValueType: schema.ColumnTypeFloat64}},
-	"attributes_bool":      {Name: "attributes_bool", Type: schema.MapColumnType{KeyType: schema.LowCardinalityColumnType{ElementType: schema.ColumnTypeString}, ValueType: schema.ColumnTypeBool}},
-	"resource":             {Name: "resource", Type: schema.JSONColumnType{}},
-	"event_name":           {Name: "event_name", Type: schema.ColumnTypeString},
-	"scope_name":           {Name: "scope_name", Type: schema.ColumnTypeString},
-	"scope_version":        {Name: "scope_version", Type: schema.ColumnTypeString},
-	"scope_string":         {Name: "scope_string", Type: schema.MapColumnType{KeyType: schema.LowCardinalityColumnType{ElementType: schema.ColumnTypeString}, ValueType: schema.ColumnTypeString}},
-}

pkg/telemetryaudit/field_mapper.go

@@ -1,124 +0,0 @@
-package telemetryaudit
-
-import (
-	"context"
-	"fmt"
-
-	schema "github.com/SigNoz/signoz-otel-collector/cmd/signozschemamigrator/schema_migrator"
-	"github.com/SigNoz/signoz/pkg/errors"
-	qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
-	"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
-	"github.com/huandu/go-sqlbuilder"
-
-	"golang.org/x/exp/maps"
-)
-
-type fieldMapper struct{}
-
-func NewFieldMapper() qbtypes.FieldMapper {
-	return &fieldMapper{}
-}
-
-func (m *fieldMapper) getColumn(_ context.Context, key *telemetrytypes.TelemetryFieldKey) ([]*schema.Column, error) {
-	switch key.FieldContext {
-	case telemetrytypes.FieldContextResource:
-		return []*schema.Column{auditLogColumns["resource"]}, nil
-	case telemetrytypes.FieldContextScope:
-		switch key.Name {
-		case "name", "scope.name", "scope_name":
-			return []*schema.Column{auditLogColumns["scope_name"]}, nil
-		case "version", "scope.version", "scope_version":
-			return []*schema.Column{auditLogColumns["scope_version"]}, nil
-		}
-		return []*schema.Column{auditLogColumns["scope_string"]}, nil
-	case telemetrytypes.FieldContextAttribute:
-		switch key.FieldDataType {
-		case telemetrytypes.FieldDataTypeString:
-			return []*schema.Column{auditLogColumns["attributes_string"]}, nil
-		case telemetrytypes.FieldDataTypeInt64, telemetrytypes.FieldDataTypeFloat64, telemetrytypes.FieldDataTypeNumber:
-			return []*schema.Column{auditLogColumns["attributes_number"]}, nil
-		case telemetrytypes.FieldDataTypeBool:
-			return []*schema.Column{auditLogColumns["attributes_bool"]}, nil
-		}
-	case telemetrytypes.FieldContextLog, telemetrytypes.FieldContextUnspecified:
-		col, ok := auditLogColumns[key.Name]
-		if !ok {
-			return nil, qbtypes.ErrColumnNotFound
-		}
-		return []*schema.Column{col}, nil
-	}
-
-	return nil, qbtypes.ErrColumnNotFound
-}
-
-func (m *fieldMapper) FieldFor(ctx context.Context, _, _ uint64, key *telemetrytypes.TelemetryFieldKey) (string, error) {
-	columns, err := m.getColumn(ctx, key)
-	if err != nil {
-		return "", err
-	}
-	if len(columns) != 1 {
-		return "", errors.Newf(errors.TypeInternal, errors.CodeInternal, "expected exactly 1 column, got %d", len(columns))
-	}
-	column := columns[0]
-
-	switch column.Type.GetType() {
-	case schema.ColumnTypeEnumJSON:
-		if key.FieldContext != telemetrytypes.FieldContextResource {
-			return "", errors.Newf(errors.TypeInvalidInput, errors.CodeInvalidInput, "only resource context fields are supported for json columns in audit, got %s", key.FieldContext.String)
-		}
-		return fmt.Sprintf("%s.`%s`::String", column.Name, key.Name), nil
-	case schema.ColumnTypeEnumLowCardinality:
-		return column.Name, nil
-	case schema.ColumnTypeEnumString, schema.ColumnTypeEnumUInt64, schema.ColumnTypeEnumUInt32, schema.ColumnTypeEnumUInt8:
-		return column.Name, nil
-	case schema.ColumnTypeEnumMap:
-		keyType := column.Type.(schema.MapColumnType).KeyType
-		if _, ok := keyType.(schema.LowCardinalityColumnType); !ok {
-			return "", errors.NewInvalidInputf(errors.CodeInvalidInput, "key type %s is not supported for map column type %s", keyType, column.Type)
-		}
-
-		switch valueType := column.Type.(schema.MapColumnType).ValueType; valueType.GetType() {
-		case schema.ColumnTypeEnumString, schema.ColumnTypeEnumBool, schema.ColumnTypeEnumFloat64:
-			if key.Materialized {
-				return telemetrytypes.FieldKeyToMaterializedColumnName(key), nil
-			}
-			return fmt.Sprintf("%s['%s']", column.Name, key.Name), nil
-		default:
-			return "", errors.NewInvalidInputf(errors.CodeInvalidInput, "unsupported map value type %s", valueType)
-		}
-	}
-
-	return column.Name, nil
-}
-
-func (m *fieldMapper) ColumnFor(ctx context.Context, _, _ uint64, key *telemetrytypes.TelemetryFieldKey) ([]*schema.Column, error) {
-	return m.getColumn(ctx, key)
-}
-
-func (m *fieldMapper) ColumnExpressionFor(
-	ctx context.Context,
-	tsStart, tsEnd uint64,
-	field *telemetrytypes.TelemetryFieldKey,
-	keys map[string][]*telemetrytypes.TelemetryFieldKey,
-) (string, error) {
-	fieldExpression, err := m.FieldFor(ctx, tsStart, tsEnd, field)
-	if errors.Is(err, qbtypes.ErrColumnNotFound) {
-		keysForField := keys[field.Name]
-		if len(keysForField) == 0 {
-			if _, ok := auditLogColumns[field.Name]; ok {
-				field.FieldContext = telemetrytypes.FieldContextLog
-				fieldExpression, _ = m.FieldFor(ctx, tsStart, tsEnd, field)
-			} else {
-				correction, found := telemetrytypes.SuggestCorrection(field.Name, maps.Keys(keys))
-				if found {
-					return "", errors.Wrap(err, errors.TypeInvalidInput, errors.CodeInvalidInput, correction)
-				}
-				return "", errors.Wrapf(err, errors.TypeInvalidInput, errors.CodeInvalidInput, "field `%s` not found", field.Name)
-			}
-		} else {
-			fieldExpression, _ = m.FieldFor(ctx, tsStart, tsEnd, keysForField[0])
-		}
-	}
-
-	return fmt.Sprintf("%s AS `%s`", sqlbuilder.Escape(fieldExpression), field.Name), nil
-}

pkg/telemetryaudit/statement_builder.go

@@ -1,612 +0,0 @@
-package telemetryaudit
-
-import (
-	"context"
-	"fmt"
-	"log/slog"
-	"strings"
-
-	"github.com/SigNoz/signoz/pkg/errors"
-	"github.com/SigNoz/signoz/pkg/factory"
-	"github.com/SigNoz/signoz/pkg/querybuilder"
-	"github.com/SigNoz/signoz/pkg/telemetryresourcefilter"
-	qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
-	"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
-	"github.com/huandu/go-sqlbuilder"
-)
-
-type auditQueryStatementBuilder struct {
-	logger                    *slog.Logger
-	metadataStore             telemetrytypes.MetadataStore
-	fm                        qbtypes.FieldMapper
-	cb                        qbtypes.ConditionBuilder
-	resourceFilterStmtBuilder qbtypes.StatementBuilder[qbtypes.LogAggregation]
-	aggExprRewriter           qbtypes.AggExprRewriter
-	fullTextColumn            *telemetrytypes.TelemetryFieldKey
-	jsonKeyToKey              qbtypes.JsonKeyToFieldFunc
-}
-
-var _ qbtypes.StatementBuilder[qbtypes.LogAggregation] = (*auditQueryStatementBuilder)(nil)
-
-func NewAuditQueryStatementBuilder(
-	settings factory.ProviderSettings,
-	metadataStore telemetrytypes.MetadataStore,
-	fieldMapper qbtypes.FieldMapper,
-	conditionBuilder qbtypes.ConditionBuilder,
-	aggExprRewriter qbtypes.AggExprRewriter,
-	fullTextColumn *telemetrytypes.TelemetryFieldKey,
-	jsonKeyToKey qbtypes.JsonKeyToFieldFunc,
-) *auditQueryStatementBuilder {
-	auditSettings := factory.NewScopedProviderSettings(settings, "github.com/SigNoz/signoz/pkg/telemetryaudit")
-
-	resourceFilterStmtBuilder := telemetryresourcefilter.New[qbtypes.LogAggregation](
-		settings,
-		DBName,
-		LogsResourceTableName,
-		telemetrytypes.SignalLogs,
-		telemetrytypes.SourceAudit,
-		metadataStore,
-		fullTextColumn,
-		jsonKeyToKey,
-	)
-
-	return &auditQueryStatementBuilder{
-		logger:                    auditSettings.Logger(),
-		metadataStore:             metadataStore,
-		fm:                        fieldMapper,
-		cb:                        conditionBuilder,
-		resourceFilterStmtBuilder: resourceFilterStmtBuilder,
-		aggExprRewriter:           aggExprRewriter,
-		fullTextColumn:            fullTextColumn,
-		jsonKeyToKey:              jsonKeyToKey,
-	}
-}
-
-func (b *auditQueryStatementBuilder) Build(
-	ctx context.Context,
-	start uint64,
-	end uint64,
-	requestType qbtypes.RequestType,
-	query qbtypes.QueryBuilderQuery[qbtypes.LogAggregation],
-	variables map[string]qbtypes.VariableItem,
-) (*qbtypes.Statement, error) {
-	start = querybuilder.ToNanoSecs(start)
-	end = querybuilder.ToNanoSecs(end)
-
-	keySelectors := getKeySelectors(query)
-	keys, _, err := b.metadataStore.GetKeysMulti(ctx, keySelectors)
-	if err != nil {
-		return nil, err
-	}
-
-	query = b.adjustKeys(ctx, keys, query, requestType)
-
-	q := sqlbuilder.NewSelectBuilder()
-
-	var stmt *qbtypes.Statement
-	switch requestType {
-	case qbtypes.RequestTypeRaw, qbtypes.RequestTypeRawStream:
-		stmt, err = b.buildListQuery(ctx, q, query, start, end, keys, variables)
-	case qbtypes.RequestTypeTimeSeries:
-		stmt, err = b.buildTimeSeriesQuery(ctx, q, query, start, end, keys, variables)
-	case qbtypes.RequestTypeScalar:
-		stmt, err = b.buildScalarQuery(ctx, q, query, start, end, keys, false, variables)
-	default:
-		return nil, errors.NewInvalidInputf(errors.CodeInvalidInput, "unsupported request type: %s", requestType)
-	}
-
-	if err != nil {
-		return nil, err
-	}
-
-	return stmt, nil
-}
-
-func getKeySelectors(query qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]) []*telemetrytypes.FieldKeySelector {
-	var keySelectors []*telemetrytypes.FieldKeySelector
-
-	for idx := range query.Aggregations {
-		aggExpr := query.Aggregations[idx]
-		selectors := querybuilder.QueryStringToKeysSelectors(aggExpr.Expression)
-		keySelectors = append(keySelectors, selectors...)
-	}
-
-	if query.Filter != nil && query.Filter.Expression != "" {
-		whereClauseSelectors := querybuilder.QueryStringToKeysSelectors(query.Filter.Expression)
-		keySelectors = append(keySelectors, whereClauseSelectors...)
-	}
-
-	for idx := range query.GroupBy {
-		groupBy := query.GroupBy[idx]
-		keySelectors = append(keySelectors, &telemetrytypes.FieldKeySelector{
-			Name:          groupBy.Name,
-			Signal:        telemetrytypes.SignalLogs,
-			FieldContext:  groupBy.FieldContext,
-			FieldDataType: groupBy.FieldDataType,
-		})
-	}
-
-	for idx := range query.SelectFields {
-		selectField := query.SelectFields[idx]
-		keySelectors = append(keySelectors, &telemetrytypes.FieldKeySelector{
-			Name:          selectField.Name,
-			Signal:        telemetrytypes.SignalLogs,
-			FieldContext:  selectField.FieldContext,
-			FieldDataType: selectField.FieldDataType,
-		})
-	}
-
-	for idx := range query.Order {
-		keySelectors = append(keySelectors, &telemetrytypes.FieldKeySelector{
-			Name:          query.Order[idx].Key.Name,
-			Signal:        telemetrytypes.SignalLogs,
-			FieldContext:  query.Order[idx].Key.FieldContext,
-			FieldDataType: query.Order[idx].Key.FieldDataType,
-		})
-	}
-
-	for idx := range keySelectors {
-		keySelectors[idx].Signal = telemetrytypes.SignalLogs
-		keySelectors[idx].Source = telemetrytypes.SourceAudit
-		keySelectors[idx].SelectorMatchType = telemetrytypes.FieldSelectorMatchTypeExact
-	}
-
-	return keySelectors
-}
-
-func (b *auditQueryStatementBuilder) adjustKeys(ctx context.Context, keys map[string][]*telemetrytypes.TelemetryFieldKey, query qbtypes.QueryBuilderQuery[qbtypes.LogAggregation], requestType qbtypes.RequestType) qbtypes.QueryBuilderQuery[qbtypes.LogAggregation] {
-	keys["id"] = append([]*telemetrytypes.TelemetryFieldKey{{
-		Name:          "id",
-		Signal:        telemetrytypes.SignalLogs,
-		FieldContext:  telemetrytypes.FieldContextLog,
-		FieldDataType: telemetrytypes.FieldDataTypeString,
-	}}, keys["id"]...)
-
-	keys["timestamp"] = append([]*telemetrytypes.TelemetryFieldKey{{
-		Name:          "timestamp",
-		Signal:        telemetrytypes.SignalLogs,
-		FieldContext:  telemetrytypes.FieldContextLog,
-		FieldDataType: telemetrytypes.FieldDataTypeNumber,
-	}}, keys["timestamp"]...)
-
-	actions := querybuilder.AdjustKeysForAliasExpressions(&query, requestType)
-	actions = append(actions, querybuilder.AdjustDuplicateKeys(&query)...)
-
-	for idx := range query.SelectFields {
-		actions = append(actions, b.adjustKey(&query.SelectFields[idx], keys)...)
-	}
-	for idx := range query.GroupBy {
-		actions = append(actions, b.adjustKey(&query.GroupBy[idx].TelemetryFieldKey, keys)...)
-	}
-	for idx := range query.Order {
-		actions = append(actions, b.adjustKey(&query.Order[idx].Key.TelemetryFieldKey, keys)...)
-	}
-
-	for _, action := range actions {
-		b.logger.InfoContext(ctx, "key adjustment action", slog.String("action", action))
-	}
-
-	return query
-}
-
-func (b *auditQueryStatementBuilder) adjustKey(key *telemetrytypes.TelemetryFieldKey, keys map[string][]*telemetrytypes.TelemetryFieldKey) []string {
-	if _, ok := IntrinsicFields[key.Name]; ok {
-		intrinsicField := IntrinsicFields[key.Name]
-		return querybuilder.AdjustKey(key, keys, &intrinsicField)
-	}
-	return querybuilder.AdjustKey(key, keys, nil)
-}
-
-func (b *auditQueryStatementBuilder) buildListQuery(
-	ctx context.Context,
-	sb *sqlbuilder.SelectBuilder,
-	query qbtypes.QueryBuilderQuery[qbtypes.LogAggregation],
-	start, end uint64,
-	keys map[string][]*telemetrytypes.TelemetryFieldKey,
-	variables map[string]qbtypes.VariableItem,
-) (*qbtypes.Statement, error) {
-	var (
-		cteFragments []string
-		cteArgs      [][]any
-	)
-
-	if frag, args, err := b.maybeAttachResourceFilter(ctx, sb, query, start, end, variables); err != nil {
-		return nil, err
-	} else if frag != "" {
-		cteFragments = append(cteFragments, frag)
-		cteArgs = append(cteArgs, args)
-	}
-
-	sb.Select(TimestampColumn)
-	sb.SelectMore(IDColumn)
-	if len(query.SelectFields) == 0 {
-		sb.SelectMore(TraceIDColumn)
-		sb.SelectMore(SpanIDColumn)
-		sb.SelectMore(TraceFlagsColumn)
-		sb.SelectMore(SeverityTextColumn)
-		sb.SelectMore(SeverityNumberColumn)
-		sb.SelectMore(ScopeNameColumn)
-		sb.SelectMore(ScopeVersionColumn)
-		sb.SelectMore(BodyColumn)
-		sb.SelectMore(EventNameColumn)
-		sb.SelectMore(AttributesStringColumn)
-		sb.SelectMore(AttributesNumberColumn)
-		sb.SelectMore(AttributesBoolColumn)
-		sb.SelectMore(ResourceColumn)
-		sb.SelectMore(ScopeStringColumn)
-	} else {
-		for index := range query.SelectFields {
-			if query.SelectFields[index].Name == TimestampColumn || query.SelectFields[index].Name == IDColumn {
-				continue
-			}
-
-			colExpr, err := b.fm.ColumnExpressionFor(ctx, start, end, &query.SelectFields[index], keys)
-			if err != nil {
-				return nil, err
-			}
-			sb.SelectMore(colExpr)
-		}
-	}
-
-	sb.From(fmt.Sprintf("%s.%s", DBName, AuditLogsTableName))
-
-	preparedWhereClause, err := b.addFilterCondition(ctx, sb, start, end, query, keys, variables)
-	if err != nil {
-		return nil, err
-	}
-
-	for _, orderBy := range query.Order {
-		colExpr, err := b.fm.ColumnExpressionFor(ctx, start, end, &orderBy.Key.TelemetryFieldKey, keys)
-		if err != nil {
-			return nil, err
-		}
-		sb.OrderBy(fmt.Sprintf("%s %s", colExpr, orderBy.Direction.StringValue()))
-	}
-
-	if query.Limit > 0 {
-		sb.Limit(query.Limit)
-	} else {
-		sb.Limit(100)
-	}
-
-	if query.Offset > 0 {
-		sb.Offset(query.Offset)
-	}
-
-	mainSQL, mainArgs := sb.BuildWithFlavor(sqlbuilder.ClickHouse)
-
-	finalSQL := querybuilder.CombineCTEs(cteFragments) + mainSQL
-	finalArgs := querybuilder.PrependArgs(cteArgs, mainArgs)
-
-	stmt := &qbtypes.Statement{
-		Query: finalSQL,
-		Args:  finalArgs,
-	}
-	if preparedWhereClause != nil {
-		stmt.Warnings = preparedWhereClause.Warnings
-		stmt.WarningsDocURL = preparedWhereClause.WarningsDocURL
-	}
-
-	return stmt, nil
-}
-
-func (b *auditQueryStatementBuilder) buildTimeSeriesQuery(
-	ctx context.Context,
-	sb *sqlbuilder.SelectBuilder,
-	query qbtypes.QueryBuilderQuery[qbtypes.LogAggregation],
-	start, end uint64,
-	keys map[string][]*telemetrytypes.TelemetryFieldKey,
-	variables map[string]qbtypes.VariableItem,
-) (*qbtypes.Statement, error) {
-	var (
-		cteFragments []string
-		cteArgs      [][]any
-	)
-
-	if frag, args, err := b.maybeAttachResourceFilter(ctx, sb, query, start, end, variables); err != nil {
-		return nil, err
-	} else if frag != "" {
-		cteFragments = append(cteFragments, frag)
-		cteArgs = append(cteArgs, args)
-	}
-
-	sb.SelectMore(fmt.Sprintf(
-		"toStartOfInterval(fromUnixTimestamp64Nano(timestamp), INTERVAL %d SECOND) AS ts",
-		int64(query.StepInterval.Seconds()),
-	))
-
-	var allGroupByArgs []any
-
-	fieldNames := make([]string, 0, len(query.GroupBy))
-	for _, gb := range query.GroupBy {
-		expr, args, err := querybuilder.CollisionHandledFinalExpr(ctx, start, end, &gb.TelemetryFieldKey, b.fm, b.cb, keys, telemetrytypes.FieldDataTypeString, b.jsonKeyToKey)
-		if err != nil {
-			return nil, err
-		}
-
-		colExpr := fmt.Sprintf("toString(%s) AS `%s`", expr, gb.Name)
-		allGroupByArgs = append(allGroupByArgs, args...)
-		sb.SelectMore(colExpr)
-		fieldNames = append(fieldNames, fmt.Sprintf("`%s`", gb.Name))
-	}
-
-	allAggChArgs := make([]any, 0)
-	for i, agg := range query.Aggregations {
-		rewritten, chArgs, err := b.aggExprRewriter.Rewrite(ctx, start, end, agg.Expression, uint64(query.StepInterval.Seconds()), keys)
-		if err != nil {
-			return nil, err
-		}
-		allAggChArgs = append(allAggChArgs, chArgs...)
-		sb.SelectMore(fmt.Sprintf("%s AS __result_%d", rewritten, i))
-	}
-
-	sb.From(fmt.Sprintf("%s.%s", DBName, AuditLogsTableName))
-
-	preparedWhereClause, err := b.addFilterCondition(ctx, sb, start, end, query, keys, variables)
-	if err != nil {
-		return nil, err
-	}
-
-	var finalSQL string
-	var finalArgs []any
-
-	if query.Limit > 0 && len(query.GroupBy) > 0 {
-		cteSB := sqlbuilder.NewSelectBuilder()
-		cteStmt, err := b.buildScalarQuery(ctx, cteSB, query, start, end, keys, true, variables)
-		if err != nil {
-			return nil, err
-		}
-
-		cteFragments = append(cteFragments, fmt.Sprintf("__limit_cte AS (%s)", cteStmt.Query))
-		cteArgs = append(cteArgs, cteStmt.Args)
-
-		tuple := fmt.Sprintf("(%s)", strings.Join(fieldNames, ", "))
-		sb.Where(fmt.Sprintf("%s GLOBAL IN (SELECT %s FROM __limit_cte)", tuple, strings.Join(fieldNames, ", ")))
-
-		sb.GroupBy("ts")
-		sb.GroupBy(querybuilder.GroupByKeys(query.GroupBy)...)
-		if query.Having != nil && query.Having.Expression != "" {
-			rewriter := querybuilder.NewHavingExpressionRewriter()
-			rewrittenExpr, err := rewriter.RewriteForLogs(query.Having.Expression, query.Aggregations)
-			if err != nil {
-				return nil, err
-			}
-			sb.Having(rewrittenExpr)
-		}
-
-		if len(query.Order) != 0 {
-			for _, orderBy := range query.Order {
-				_, ok := aggOrderBy(orderBy, query)
-				if !ok {
-					sb.OrderBy(fmt.Sprintf("`%s` %s", orderBy.Key.Name, orderBy.Direction.StringValue()))
-				}
-			}
-			sb.OrderBy("ts desc")
-		}
-
-		combinedArgs := append(allGroupByArgs, allAggChArgs...)
-		mainSQL, mainArgs := sb.BuildWithFlavor(sqlbuilder.ClickHouse, combinedArgs...)
-
-		finalSQL = querybuilder.CombineCTEs(cteFragments) + mainSQL
-		finalArgs = querybuilder.PrependArgs(cteArgs, mainArgs)
-	} else {
-		sb.GroupBy("ts")
-		sb.GroupBy(querybuilder.GroupByKeys(query.GroupBy)...)
-		if query.Having != nil && query.Having.Expression != "" {
-			rewriter := querybuilder.NewHavingExpressionRewriter()
-			rewrittenExpr, err := rewriter.RewriteForLogs(query.Having.Expression, query.Aggregations)
-			if err != nil {
-				return nil, err
-			}
-			sb.Having(rewrittenExpr)
-		}
-
-		if len(query.Order) != 0 {
-			for _, orderBy := range query.Order {
-				_, ok := aggOrderBy(orderBy, query)
-				if !ok {
-					sb.OrderBy(fmt.Sprintf("`%s` %s", orderBy.Key.Name, orderBy.Direction.StringValue()))
-				}
-			}
-			sb.OrderBy("ts desc")
-		}
-
-		combinedArgs := append(allGroupByArgs, allAggChArgs...)
-		mainSQL, mainArgs := sb.BuildWithFlavor(sqlbuilder.ClickHouse, combinedArgs...)
-
-		finalSQL = querybuilder.CombineCTEs(cteFragments) + mainSQL
-		finalArgs = querybuilder.PrependArgs(cteArgs, mainArgs)
-	}
-
-	stmt := &qbtypes.Statement{
-		Query: finalSQL,
-		Args:  finalArgs,
-	}
-	if preparedWhereClause != nil {
-		stmt.Warnings = preparedWhereClause.Warnings
-		stmt.WarningsDocURL = preparedWhereClause.WarningsDocURL
-	}
-
-	return stmt, nil
-}
-
-func (b *auditQueryStatementBuilder) buildScalarQuery(
-	ctx context.Context,
-	sb *sqlbuilder.SelectBuilder,
-	query qbtypes.QueryBuilderQuery[qbtypes.LogAggregation],
-	start, end uint64,
-	keys map[string][]*telemetrytypes.TelemetryFieldKey,
-	skipResourceCTE bool,
-	variables map[string]qbtypes.VariableItem,
-) (*qbtypes.Statement, error) {
-	var (
-		cteFragments []string
-		cteArgs      [][]any
-	)
-
-	if frag, args, err := b.maybeAttachResourceFilter(ctx, sb, query, start, end, variables); err != nil {
-		return nil, err
-	} else if frag != "" && !skipResourceCTE {
-		cteFragments = append(cteFragments, frag)
-		cteArgs = append(cteArgs, args)
-	}
-
-	allAggChArgs := []any{}
-
-	var allGroupByArgs []any
-
-	for _, gb := range query.GroupBy {
-		expr, args, err := querybuilder.CollisionHandledFinalExpr(ctx, start, end, &gb.TelemetryFieldKey, b.fm, b.cb, keys, telemetrytypes.FieldDataTypeString, b.jsonKeyToKey)
-		if err != nil {
-			return nil, err
-		}
-
-		colExpr := fmt.Sprintf("toString(%s) AS `%s`", expr, gb.Name)
-		allGroupByArgs = append(allGroupByArgs, args...)
-		sb.SelectMore(colExpr)
-	}
-
-	rateInterval := (end - start) / querybuilder.NsToSeconds
-
-	if len(query.Aggregations) > 0 {
-		for idx := range query.Aggregations {
-			aggExpr := query.Aggregations[idx]
-			rewritten, chArgs, err := b.aggExprRewriter.Rewrite(ctx, start, end, aggExpr.Expression, rateInterval, keys)
-			if err != nil {
-				return nil, err
-			}
-			allAggChArgs = append(allAggChArgs, chArgs...)
-			sb.SelectMore(fmt.Sprintf("%s AS __result_%d", rewritten, idx))
-		}
-	}
-
-	sb.From(fmt.Sprintf("%s.%s", DBName, AuditLogsTableName))
-
-	preparedWhereClause, err := b.addFilterCondition(ctx, sb, start, end, query, keys, variables)
-	if err != nil {
-		return nil, err
-	}
-
-	sb.GroupBy(querybuilder.GroupByKeys(query.GroupBy)...)
-
-	if query.Having != nil && query.Having.Expression != "" {
-		rewriter := querybuilder.NewHavingExpressionRewriter()
-		rewrittenExpr, err := rewriter.RewriteForLogs(query.Having.Expression, query.Aggregations)
-		if err != nil {
-			return nil, err
-		}
-		sb.Having(rewrittenExpr)
-	}
-
-	for _, orderBy := range query.Order {
-		idx, ok := aggOrderBy(orderBy, query)
-		if ok {
-			sb.OrderBy(fmt.Sprintf("__result_%d %s", idx, orderBy.Direction.StringValue()))
-		} else {
-			sb.OrderBy(fmt.Sprintf("`%s` %s", orderBy.Key.Name, orderBy.Direction.StringValue()))
-		}
-	}
-
-	if len(query.Order) == 0 {
-		sb.OrderBy("__result_0 DESC")
-	}
-
-	if query.Limit > 0 {
-		sb.Limit(query.Limit)
-	}
-
-	combinedArgs := append(allGroupByArgs, allAggChArgs...)
-
-	mainSQL, mainArgs := sb.BuildWithFlavor(sqlbuilder.ClickHouse, combinedArgs...)
-
-	finalSQL := querybuilder.CombineCTEs(cteFragments) + mainSQL
-	finalArgs := querybuilder.PrependArgs(cteArgs, mainArgs)
-
-	stmt := &qbtypes.Statement{
-		Query: finalSQL,
-		Args:  finalArgs,
-	}
-	if preparedWhereClause != nil {
-		stmt.Warnings = preparedWhereClause.Warnings
-		stmt.WarningsDocURL = preparedWhereClause.WarningsDocURL
-	}
-
-	return stmt, nil
-}
-
-func (b *auditQueryStatementBuilder) addFilterCondition(
-	ctx context.Context,
-	sb *sqlbuilder.SelectBuilder,
-	start, end uint64,
-	query qbtypes.QueryBuilderQuery[qbtypes.LogAggregation],
-	keys map[string][]*telemetrytypes.TelemetryFieldKey,
-	variables map[string]qbtypes.VariableItem,
-) (*querybuilder.PreparedWhereClause, error) {
-	var preparedWhereClause *querybuilder.PreparedWhereClause
-	var err error
-
-	if query.Filter != nil && query.Filter.Expression != "" {
-		preparedWhereClause, err = querybuilder.PrepareWhereClause(query.Filter.Expression, querybuilder.FilterExprVisitorOpts{
-			Context:            ctx,
-			Logger:             b.logger,
-			FieldMapper:        b.fm,
-			ConditionBuilder:   b.cb,
-			FieldKeys:          keys,
-			SkipResourceFilter: true,
-			FullTextColumn:     b.fullTextColumn,
-			JsonKeyToKey:       b.jsonKeyToKey,
-			Variables:          variables,
-			StartNs:            start,
-			EndNs:              end,
-		})
-
-		if err != nil {
-			return nil, err
-		}
-	}
-
-	if preparedWhereClause != nil {
-		sb.AddWhereClause(preparedWhereClause.WhereClause)
-	}
-
-	startBucket := start/querybuilder.NsToSeconds - querybuilder.BucketAdjustment
-	var endBucket uint64
-	if end != 0 {
-		endBucket = end / querybuilder.NsToSeconds
-	}
-
-	if start != 0 {
-		sb.Where(sb.GE("timestamp", fmt.Sprintf("%d", start)), sb.GE("ts_bucket_start", startBucket))
-	}
-	if end != 0 {
-		sb.Where(sb.L("timestamp", fmt.Sprintf("%d", end)), sb.LE("ts_bucket_start", endBucket))
-	}
-
-	return preparedWhereClause, nil
-}
-
-func aggOrderBy(k qbtypes.OrderBy, q qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]) (int, bool) {
-	for i, agg := range q.Aggregations {
-		if k.Key.Name == agg.Alias || k.Key.Name == agg.Expression || k.Key.Name == fmt.Sprintf("%d", i) {
-			return i, true
-		}
-	}
-	return 0, false
-}
-
-func (b *auditQueryStatementBuilder) maybeAttachResourceFilter(
-	ctx context.Context,
-	sb *sqlbuilder.SelectBuilder,
-	query qbtypes.QueryBuilderQuery[qbtypes.LogAggregation],
-	start, end uint64,
-	variables map[string]qbtypes.VariableItem,
-) (cteSQL string, cteArgs []any, err error) {
-	stmt, err := b.resourceFilterStmtBuilder.Build(ctx, start, end, qbtypes.RequestTypeRaw, query, variables)
-	if err != nil {
-		return "", nil, err
-	}
-
-	sb.Where("resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter)")
-
-	return fmt.Sprintf("__resource_filter AS (%s)", stmt.Query), stmt.Args, nil
-}

pkg/telemetryaudit/statement_builder_test.go

@@ -1,223 +0,0 @@
-package telemetryaudit
-
-import (
-	"context"
-	"testing"
-	"time"
-
-	"github.com/SigNoz/signoz/pkg/instrumentation/instrumentationtest"
-	"github.com/SigNoz/signoz/pkg/querybuilder"
-	qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
-	"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
-	"github.com/SigNoz/signoz/pkg/types/telemetrytypes/telemetrytypestest"
-	"github.com/stretchr/testify/require"
-)
-
-func auditFieldKeyMap() map[string][]*telemetrytypes.TelemetryFieldKey {
-	key := func(name string, ctx telemetrytypes.FieldContext, dt telemetrytypes.FieldDataType, materialized bool) *telemetrytypes.TelemetryFieldKey {
-		return &telemetrytypes.TelemetryFieldKey{
-			Name:          name,
-			Signal:        telemetrytypes.SignalLogs,
-			FieldContext:  ctx,
-			FieldDataType: dt,
-			Materialized:  materialized,
-		}
-	}
-
-	attr := telemetrytypes.FieldContextAttribute
-	res := telemetrytypes.FieldContextResource
-	str := telemetrytypes.FieldDataTypeString
-	i64 := telemetrytypes.FieldDataTypeInt64
-
-	return map[string][]*telemetrytypes.TelemetryFieldKey{
-		"service.name":                 {key("service.name", res, str, false)},
-		"signoz.audit.action":          {key("signoz.audit.action", attr, str, true)},
-		"signoz.audit.outcome":         {key("signoz.audit.outcome", attr, str, true)},
-		"signoz.audit.principal.email": {key("signoz.audit.principal.email", attr, str, true)},
-		"signoz.audit.principal.id":    {key("signoz.audit.principal.id", attr, str, true)},
-		"signoz.audit.principal.type":  {key("signoz.audit.principal.type", attr, str, true)},
-		"signoz.audit.resource.kind":   {key("signoz.audit.resource.kind", res, str, false)},
-		"signoz.audit.resource.id":     {key("signoz.audit.resource.id", res, str, false)},
-		"signoz.audit.action_category": {key("signoz.audit.action_category", attr, str, false)},
-		"signoz.audit.error.type":      {key("signoz.audit.error.type", attr, str, false)},
-		"signoz.audit.error.code":      {key("signoz.audit.error.code", attr, str, false)},
-		"http.request.method":          {key("http.request.method", attr, str, false)},
-		"http.response.status_code":    {key("http.response.status_code", attr, i64, false)},
-	}
-}
-
-func newTestAuditStatementBuilder() *auditQueryStatementBuilder {
-	mockMetadataStore := telemetrytypestest.NewMockMetadataStore()
-	mockMetadataStore.KeysMap = auditFieldKeyMap()
-
-	fm := NewFieldMapper()
-	cb := NewConditionBuilder(fm)
-	aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, fm, cb, nil)
-
-	return NewAuditQueryStatementBuilder(
-		instrumentationtest.New().ToProviderSettings(),
-		mockMetadataStore,
-		fm,
-		cb,
-		aggExprRewriter,
-		DefaultFullTextColumn,
-		nil,
-	)
-}
-
-func TestStatementBuilder(t *testing.T) {
-	statementBuilder := newTestAuditStatementBuilder()
-	ctx := context.Background()
-
-	testCases := []struct {
-		name        string
-		requestType qbtypes.RequestType
-		query       qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]
-		expected    qbtypes.Statement
-		expectedErr error
-	}{
-		// List: all actions by a specific user (materialized principal.id filter)
-		{
-			name:        "ListByPrincipalID",
-			requestType: qbtypes.RequestTypeRaw,
-			query: qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]{
-				Signal: telemetrytypes.SignalLogs,
-				Source: telemetrytypes.SourceAudit,
-				Filter: &qbtypes.Filter{
-					Expression: "signoz.audit.principal.id = '019a-1234-abcd-5678'",
-				},
-				Limit: 100,
-			},
-			expected: qbtypes.Statement{
-				Query: "WITH __resource_filter AS (SELECT fingerprint FROM signoz_audit.distributed_logs_resource WHERE true AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ?) SELECT timestamp, id, trace_id, span_id, trace_flags, severity_text, severity_number, scope_name, scope_version, body, event_name, attributes_string, attributes_number, attributes_bool, resource, scope_string FROM signoz_audit.distributed_logs WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter) AND (`attribute_string_signoz$$audit$$principal$$id` = ? AND `attribute_string_signoz$$audit$$principal$$id_exists` = ?) AND timestamp >= ? AND ts_bucket_start >= ? AND timestamp < ? AND ts_bucket_start <= ? LIMIT ?",
-				Args:  []any{uint64(1747945619), uint64(1747983448), "019a-1234-abcd-5678", true, "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 100},
-			},
-		},
-		// List: all failed actions (materialized outcome filter)
-		{
-			name:        "ListByOutcomeFailure",
-			requestType: qbtypes.RequestTypeRaw,
-			query: qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]{
-				Signal: telemetrytypes.SignalLogs,
-				Source: telemetrytypes.SourceAudit,
-				Filter: &qbtypes.Filter{
-					Expression: "signoz.audit.outcome = 'failure'",
-				},
-				Limit: 100,
-			},
-			expected: qbtypes.Statement{
-				Query: "WITH __resource_filter AS (SELECT fingerprint FROM signoz_audit.distributed_logs_resource WHERE true AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ?) SELECT timestamp, id, trace_id, span_id, trace_flags, severity_text, severity_number, scope_name, scope_version, body, event_name, attributes_string, attributes_number, attributes_bool, resource, scope_string FROM signoz_audit.distributed_logs WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter) AND (`attribute_string_signoz$$audit$$outcome` = ? AND `attribute_string_signoz$$audit$$outcome_exists` = ?) AND timestamp >= ? AND ts_bucket_start >= ? AND timestamp < ? AND ts_bucket_start <= ? LIMIT ?",
-				Args:  []any{uint64(1747945619), uint64(1747983448), "failure", true, "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 100},
-			},
-		},
-		// List: change history of a specific dashboard (two materialized column AND)
-		{
-			name:        "ListByResourceKindAndID",
-			requestType: qbtypes.RequestTypeRaw,
-			query: qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]{
-				Signal: telemetrytypes.SignalLogs,
-				Source: telemetrytypes.SourceAudit,
-				Filter: &qbtypes.Filter{
-					Expression: "signoz.audit.resource.kind = 'dashboard' AND signoz.audit.resource.id = '019b-5678-efgh-9012'",
-				},
-				Limit: 100,
-			},
-			expected: qbtypes.Statement{
-				Query: "WITH __resource_filter AS (SELECT fingerprint FROM signoz_audit.distributed_logs_resource WHERE ((simpleJSONExtractString(labels, 'signoz.audit.resource.kind') = ? AND labels LIKE ? AND labels LIKE ?) AND (simpleJSONExtractString(labels, 'signoz.audit.resource.id') = ? AND labels LIKE ? AND labels LIKE ?)) AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ?) SELECT timestamp, id, trace_id, span_id, trace_flags, severity_text, severity_number, scope_name, scope_version, body, event_name, attributes_string, attributes_number, attributes_bool, resource, scope_string FROM signoz_audit.distributed_logs WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter) AND true AND timestamp >= ? AND ts_bucket_start >= ? AND timestamp < ? AND ts_bucket_start <= ? LIMIT ?",
-				Args:  []any{"dashboard", "%signoz.audit.resource.kind%", "%signoz.audit.resource.kind\":\"dashboard%", "019b-5678-efgh-9012", "%signoz.audit.resource.id%", "%signoz.audit.resource.id\":\"019b-5678-efgh-9012%", uint64(1747945619), uint64(1747983448), "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 100},
-			},
-		},
-		// List: all dashboard deletions (compliance — resource.kind + action AND)
-		{
-			name:        "ListByResourceKindAndAction",
-			requestType: qbtypes.RequestTypeRaw,
-			query: qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]{
-				Signal: telemetrytypes.SignalLogs,
-				Source: telemetrytypes.SourceAudit,
-				Filter: &qbtypes.Filter{
-					Expression: "signoz.audit.resource.kind = 'dashboard' AND signoz.audit.action = 'delete'",
-				},
-				Limit: 100,
-			},
-			expected: qbtypes.Statement{
-				Query: "WITH __resource_filter AS (SELECT fingerprint FROM signoz_audit.distributed_logs_resource WHERE (simpleJSONExtractString(labels, 'signoz.audit.resource.kind') = ? AND labels LIKE ? AND labels LIKE ?) AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ?) SELECT timestamp, id, trace_id, span_id, trace_flags, severity_text, severity_number, scope_name, scope_version, body, event_name, attributes_string, attributes_number, attributes_bool, resource, scope_string FROM signoz_audit.distributed_logs WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter) AND (`attribute_string_signoz$$audit$$action` = ? AND `attribute_string_signoz$$audit$$action_exists` = ?) AND timestamp >= ? AND ts_bucket_start >= ? AND timestamp < ? AND ts_bucket_start <= ? LIMIT ?",
-				Args:  []any{"dashboard", "%signoz.audit.resource.kind%", "%signoz.audit.resource.kind\":\"dashboard%", uint64(1747945619), uint64(1747983448), "delete", true, "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 100},
-			},
-		},
-		// List: all actions by service accounts (materialized principal.type)
-		{
-			name:        "ListByPrincipalType",
-			requestType: qbtypes.RequestTypeRaw,
-			query: qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]{
-				Signal: telemetrytypes.SignalLogs,
-				Source: telemetrytypes.SourceAudit,
-				Filter: &qbtypes.Filter{
-					Expression: "signoz.audit.principal.type = 'service_account'",
-				},
-				Limit: 100,
-			},
-			expected: qbtypes.Statement{
-				Query: "WITH __resource_filter AS (SELECT fingerprint FROM signoz_audit.distributed_logs_resource WHERE true AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ?) SELECT timestamp, id, trace_id, span_id, trace_flags, severity_text, severity_number, scope_name, scope_version, body, event_name, attributes_string, attributes_number, attributes_bool, resource, scope_string FROM signoz_audit.distributed_logs WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter) AND (`attribute_string_signoz$$audit$$principal$$type` = ? AND `attribute_string_signoz$$audit$$principal$$type_exists` = ?) AND timestamp >= ? AND ts_bucket_start >= ? AND timestamp < ? AND ts_bucket_start <= ? LIMIT ?",
-				Args:  []any{uint64(1747945619), uint64(1747983448), "service_account", true, "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 100},
-			},
-		},
-		// Scalar: alert — count forbidden errors (outcome + action AND)
-		{
-			name:        "ScalarCountByOutcomeAndAction",
-			requestType: qbtypes.RequestTypeScalar,
-			query: qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]{
-				Signal:       telemetrytypes.SignalLogs,
-				Source:       telemetrytypes.SourceAudit,
-				StepInterval: qbtypes.Step{Duration: 60 * time.Second},
-				Filter: &qbtypes.Filter{
-					Expression: "signoz.audit.outcome = 'failure' AND signoz.audit.action = 'update'",
-				},
-				Aggregations: []qbtypes.LogAggregation{
-					{Expression: "count()"},
-				},
-			},
-			expected: qbtypes.Statement{
-				Query: "WITH __resource_filter AS (SELECT fingerprint FROM signoz_audit.distributed_logs_resource WHERE true AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ?) SELECT count() AS __result_0 FROM signoz_audit.distributed_logs WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter) AND ((`attribute_string_signoz$$audit$$outcome` = ? AND `attribute_string_signoz$$audit$$outcome_exists` = ?) AND (`attribute_string_signoz$$audit$$action` = ? AND `attribute_string_signoz$$audit$$action_exists` = ?)) AND timestamp >= ? AND ts_bucket_start >= ? AND timestamp < ? AND ts_bucket_start <= ? ORDER BY __result_0 DESC",
-				Args:  []any{uint64(1747945619), uint64(1747983448), "failure", true, "update", true, "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448)},
-			},
-		},
-		// TimeSeries: failures grouped by principal email with top-N limit
-		{
-			name:        "TimeSeriesFailuresGroupedByPrincipal",
-			requestType: qbtypes.RequestTypeTimeSeries,
-			query: qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]{
-				Signal:       telemetrytypes.SignalLogs,
-				Source:       telemetrytypes.SourceAudit,
-				StepInterval: qbtypes.Step{Duration: 60 * time.Second},
-				Aggregations: []qbtypes.LogAggregation{
-					{Expression: "count()"},
-				},
-				Filter: &qbtypes.Filter{
-					Expression: "signoz.audit.outcome = 'failure'",
-				},
-				GroupBy: []qbtypes.GroupByKey{
-					{TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{Name: "signoz.audit.principal.email"}},
-				},
-				Limit: 5,
-			},
-			expected: qbtypes.Statement{
-				Query: "WITH __resource_filter AS (SELECT fingerprint FROM signoz_audit.distributed_logs_resource WHERE true AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ?), __limit_cte AS (SELECT toString(multiIf(`attribute_string_signoz$$audit$$principal$$email_exists` = ?, `attribute_string_signoz$$audit$$principal$$email`, NULL)) AS `signoz.audit.principal.email`, count() AS __result_0 FROM signoz_audit.distributed_logs WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter) AND (`attribute_string_signoz$$audit$$outcome` = ? AND `attribute_string_signoz$$audit$$outcome_exists` = ?) AND timestamp >= ? AND ts_bucket_start >= ? AND timestamp < ? AND ts_bucket_start <= ? GROUP BY `signoz.audit.principal.email` ORDER BY __result_0 DESC LIMIT ?) SELECT toStartOfInterval(fromUnixTimestamp64Nano(timestamp), INTERVAL 60 SECOND) AS ts, toString(multiIf(`attribute_string_signoz$$audit$$principal$$email_exists` = ?, `attribute_string_signoz$$audit$$principal$$email`, NULL)) AS `signoz.audit.principal.email`, count() AS __result_0 FROM signoz_audit.distributed_logs WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter) AND (`attribute_string_signoz$$audit$$outcome` = ? AND `attribute_string_signoz$$audit$$outcome_exists` = ?) AND timestamp >= ? AND ts_bucket_start >= ? AND timestamp < ? AND ts_bucket_start <= ? AND (`signoz.audit.principal.email`) GLOBAL IN (SELECT `signoz.audit.principal.email` FROM __limit_cte) GROUP BY ts, `signoz.audit.principal.email`",
-				Args:  []any{uint64(1747945619), uint64(1747983448), true, "failure", true, "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 5, true, "failure", true, "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448)},
-			},
-		},
-	}
-
-	for _, testCase := range testCases {
-		t.Run(testCase.name, func(t *testing.T) {
-			q, err := statementBuilder.Build(ctx, 1747947419000, 1747983448000, testCase.requestType, testCase.query, nil)
-			if testCase.expectedErr != nil {
-				require.Error(t, err)
-				require.Contains(t, err.Error(), testCase.expectedErr.Error())
-			} else {
-				require.NoError(t, err)
-				require.Equal(t, testCase.expected.Query, q.Query)
-				require.Equal(t, testCase.expected.Args, q.Args)
-			}
-		})
-	}
-}

pkg/telemetryaudit/tables.go

@@ -1,12 +0,0 @@
-package telemetryaudit
-
-const (
-	DBName                      = "signoz_audit"
-	AuditLogsTableName      = "distributed_logs"
-	AuditLogsLocalTableName = "logs"
-	TagAttributesTableName      = "distributed_tag_attributes"
-	TagAttributesLocalTableName = "tag_attributes"
-	LogAttributeKeysTblName     = "distributed_logs_attribute_keys"
-	LogResourceKeysTblName      = "distributed_logs_resource_keys"
-	LogsResourceTableName       = "distributed_logs_resource"
-)

pkg/telemetrylogs/json_stmt_builder_test.go

@@ -631,7 +631,7 @@ func TestJSONStmtBuilder_ArrayPaths(t *testing.T) {
 			filter: "hasAll(body.user.permissions, ['read', 'write'])",
 			expected: TestExpected{
 				WhereClause: "hasAll(dynamicElement(body_v2.`user.permissions`, 'Array(Nullable(String))'), ?)",
-				Args:        []any{uint64(1747945619), uint64(1747983448), []any{[]any{"read", "write"}}, "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 10},
+				Args:        []any{uint64(1747945619), uint64(1747983448), []any{"read", "write"}, "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 10},
 			},
 		},
 
@@ -757,7 +757,7 @@ func TestJSONStmtBuilder_ArrayPaths(t *testing.T) {
 			filter: "hasAny(education[].awards[].participated[].members, ['Piyush', 'Tushar'])",
 			expected: TestExpected{
 				WhereClause: "hasAny(arrayFlatten(arrayConcat(arrayMap(`body_v2.education`->arrayConcat(arrayMap(`body_v2.education[].awards`->arrayConcat(arrayMap(`body_v2.education[].awards[].participated`->dynamicElement(`body_v2.education[].awards[].participated`.`members`, 'Array(Nullable(String))'), dynamicElement(`body_v2.education[].awards`.`participated`, 'Array(JSON(max_dynamic_types=4, max_dynamic_paths=0))')), arrayMap(`body_v2.education[].awards[].participated`->dynamicElement(`body_v2.education[].awards[].participated`.`members`, 'Array(Nullable(String))'), arrayMap(x->assumeNotNull(dynamicElement(x, 'JSON')), arrayFilter(x->(dynamicType(x) = 'JSON'), dynamicElement(`body_v2.education[].awards`.`participated`, 'Array(Dynamic)'))))), dynamicElement(`body_v2.education`.`awards`, 'Array(JSON(max_dynamic_types=8, max_dynamic_paths=0))')), arrayMap(`body_v2.education[].awards`->arrayConcat(arrayMap(`body_v2.education[].awards[].participated`->dynamicElement(`body_v2.education[].awards[].participated`.`members`, 'Array(Nullable(String))'), dynamicElement(`body_v2.education[].awards`.`participated`, 'Array(JSON(max_dynamic_types=16, max_dynamic_paths=256))')), arrayMap(`body_v2.education[].awards[].participated`->dynamicElement(`body_v2.education[].awards[].participated`.`members`, 'Array(Nullable(String))'), arrayMap(x->assumeNotNull(dynamicElement(x, 'JSON')), arrayFilter(x->(dynamicType(x) = 'JSON'), dynamicElement(`body_v2.education[].awards`.`participated`, 'Array(Dynamic)'))))), arrayMap(x->assumeNotNull(dynamicElement(x, 'JSON')), arrayFilter(x->(dynamicType(x) = 'JSON'), dynamicElement(`body_v2.education`.`awards`, 'Array(Dynamic)'))))), dynamicElement(body_v2.`education`, 'Array(JSON(max_dynamic_types=16, max_dynamic_paths=0))')))), ?)",
-				Args:        []any{uint64(1747945619), uint64(1747983448), []any{[]any{"Piyush", "Tushar"}}, "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 10},
+				Args:        []any{uint64(1747945619), uint64(1747983448), []any{"Piyush", "Tushar"}, "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 10},
 			},
 		},
 		{

pkg/telemetrylogs/statement_builder.go

@@ -45,7 +45,6 @@ func NewLogQueryStatementBuilder(
 		DBName,
 		LogsResourceV2TableName,
 		telemetrytypes.SignalLogs,
-		telemetrytypes.SourceUnspecified,
 		metadataStore,
 		fullTextColumn,
 		jsonKeyToKey,

pkg/telemetrymetadata/metadata.go

@@ -13,7 +13,6 @@ import (
 	"github.com/SigNoz/signoz/pkg/errors"
 	"github.com/SigNoz/signoz/pkg/factory"
 	"github.com/SigNoz/signoz/pkg/querybuilder"
-	"github.com/SigNoz/signoz/pkg/telemetryaudit"
 	"github.com/SigNoz/signoz/pkg/telemetrylogs"
 	"github.com/SigNoz/signoz/pkg/telemetrymetrics"
 	"github.com/SigNoz/signoz/pkg/telemetrystore"
@@ -28,7 +27,6 @@ import (
 var (
 	ErrFailedToGetTracesKeys    = errors.Newf(errors.TypeInternal, errors.CodeInternal, "failed to get traces keys")
 	ErrFailedToGetLogsKeys      = errors.Newf(errors.TypeInternal, errors.CodeInternal, "failed to get logs keys")
-	ErrFailedToGetAuditKeys     = errors.Newf(errors.TypeInternal, errors.CodeInternal, "failed to get audit keys")
 	ErrFailedToGetTblStatement  = errors.Newf(errors.TypeInternal, errors.CodeInternal, "failed to get tbl statement")
 	ErrFailedToGetMetricsKeys   = errors.Newf(errors.TypeInternal, errors.CodeInternal, "failed to get metrics keys")
 	ErrFailedToGetMeterKeys     = errors.Newf(errors.TypeInternal, errors.CodeInternal, "failed to get meter keys")
@@ -52,11 +50,6 @@ type telemetryMetaStore struct {
 	logAttributeKeysTblName        string
 	logResourceKeysTblName         string
 	logsV2TblName                  string
-	auditDBName                    string
-	auditLogsTblName                 string
-	auditFieldsTblName             string
-	auditAttributeKeysTblName      string
-	auditResourceKeysTblName       string
 	relatedMetadataDBName          string
 	relatedMetadataTblName         string
 	columnEvolutionMetadataTblName string
@@ -86,11 +79,6 @@ func NewTelemetryMetaStore(
 	logsFieldsTblName string,
 	logAttributeKeysTblName string,
 	logResourceKeysTblName string,
-	auditDBName string,
-	auditLogsTblName string,
-	auditFieldsTblName string,
-	auditAttributeKeysTblName string,
-	auditResourceKeysTblName string,
 	relatedMetadataDBName string,
 	relatedMetadataTblName string,
 	columnEvolutionMetadataTblName string,
@@ -113,11 +101,6 @@ func NewTelemetryMetaStore(
 		logsFieldsTblName:              logsFieldsTblName,
 		logAttributeKeysTblName:        logAttributeKeysTblName,
 		logResourceKeysTblName:         logResourceKeysTblName,
-		auditDBName:                    auditDBName,
-		auditLogsTblName:                 auditLogsTblName,
-		auditFieldsTblName:             auditFieldsTblName,
-		auditAttributeKeysTblName:      auditAttributeKeysTblName,
-		auditResourceKeysTblName:       auditResourceKeysTblName,
 		relatedMetadataDBName:          relatedMetadataDBName,
 		relatedMetadataTblName:         relatedMetadataTblName,
 		columnEvolutionMetadataTblName: columnEvolutionMetadataTblName,
@@ -609,227 +592,6 @@ func (t *telemetryMetaStore) getLogsKeys(ctx context.Context, fieldKeySelectors
 	return keys, complete, nil
 }
 
-func (t *telemetryMetaStore) auditTblStatementToFieldKeys(ctx context.Context) ([]*telemetrytypes.TelemetryFieldKey, error) {
-	ctx = ctxtypes.NewContextWithCommentVals(ctx, map[string]string{
-		instrumentationtypes.TelemetrySignal:  telemetrytypes.SignalLogs.StringValue(),
-		instrumentationtypes.CodeNamespace:    "metadata",
-		instrumentationtypes.CodeFunctionName: "auditTblStatementToFieldKeys",
-	})
-
-	query := fmt.Sprintf("SHOW CREATE TABLE %s.%s", t.auditDBName, t.auditLogsTblName)
-	statements := []telemetrytypes.ShowCreateTableStatement{}
-	err := t.telemetrystore.ClickhouseDB().Select(ctx, &statements, query)
-	if err != nil {
-		return nil, errors.Wrap(err, errors.TypeInternal, errors.CodeInternal, ErrFailedToGetTblStatement.Error())
-	}
-
-	materialisedKeys, err := ExtractFieldKeysFromTblStatement(statements[0].Statement)
-	if err != nil {
-		return nil, errors.Wrap(err, errors.TypeInternal, errors.CodeInternal, ErrFailedToGetAuditKeys.Error())
-	}
-
-	for idx := range materialisedKeys {
-		materialisedKeys[idx].Signal = telemetrytypes.SignalLogs
-	}
-
-	return materialisedKeys, nil
-}
-
-func (t *telemetryMetaStore) getAuditKeys(ctx context.Context, fieldKeySelectors []*telemetrytypes.FieldKeySelector) ([]*telemetrytypes.TelemetryFieldKey, bool, error) {
-	ctx = ctxtypes.NewContextWithCommentVals(ctx, map[string]string{
-		instrumentationtypes.TelemetrySignal:  telemetrytypes.SignalLogs.StringValue(),
-		instrumentationtypes.CodeNamespace:    "metadata",
-		instrumentationtypes.CodeFunctionName: "getAuditKeys",
-	})
-
-	if len(fieldKeySelectors) == 0 {
-		return nil, true, nil
-	}
-
-	matKeys, err := t.auditTblStatementToFieldKeys(ctx)
-	if err != nil {
-		return nil, false, err
-	}
-	mapOfKeys := make(map[string]*telemetrytypes.TelemetryFieldKey)
-	for _, key := range matKeys {
-		mapOfKeys[key.Name+";"+key.FieldContext.StringValue()+";"+key.FieldDataType.StringValue()] = key
-	}
-
-	var queries []string
-	var allArgs []any
-
-	queryAttributeTable := false
-	queryResourceTable := false
-
-	for _, selector := range fieldKeySelectors {
-		if selector.FieldContext == telemetrytypes.FieldContextUnspecified {
-			queryAttributeTable = true
-			queryResourceTable = true
-			break
-		} else if selector.FieldContext == telemetrytypes.FieldContextAttribute {
-			queryAttributeTable = true
-		} else if selector.FieldContext == telemetrytypes.FieldContextResource {
-			queryResourceTable = true
-		}
-	}
-
-	tablesToQuery := []struct {
-		fieldContext telemetrytypes.FieldContext
-		shouldQuery  bool
-		tblName     string
-	}{
-		{telemetrytypes.FieldContextAttribute, queryAttributeTable, t.auditDBName + "." + t.auditAttributeKeysTblName},
-		{telemetrytypes.FieldContextResource, queryResourceTable, t.auditDBName + "." + t.auditResourceKeysTblName},
-	}
-
-	for _, table := range tablesToQuery {
-		if !table.shouldQuery {
-			continue
-		}
-
-		fieldContext := table.fieldContext
-		tblName := table.tblName
-
-		sb := sqlbuilder.Select(
-			"name AS tag_key",
-			fmt.Sprintf("'%s' AS tag_type", fieldContext.TagType()),
-			"lower(datatype) AS tag_data_type",
-			fmt.Sprintf("%d AS priority", getPriorityForContext(fieldContext)),
-		).From(tblName)
-
-		var limit int
-		conds := []string{}
-
-		for _, fieldKeySelector := range fieldKeySelectors {
-			if fieldKeySelector.FieldContext != telemetrytypes.FieldContextUnspecified && fieldKeySelector.FieldContext != fieldContext {
-				continue
-			}
-
-			fieldKeyConds := []string{}
-			if fieldKeySelector.SelectorMatchType == telemetrytypes.FieldSelectorMatchTypeExact {
-				fieldKeyConds = append(fieldKeyConds, sb.E("name", fieldKeySelector.Name))
-			} else {
-				fieldKeyConds = append(fieldKeyConds, sb.ILike("name", "%"+escapeForLike(fieldKeySelector.Name)+"%"))
-			}
-
-			if fieldKeySelector.FieldDataType != telemetrytypes.FieldDataTypeUnspecified {
-				fieldKeyConds = append(fieldKeyConds, sb.E("datatype", fieldKeySelector.FieldDataType.TagDataType()))
-			}
-
-			if len(fieldKeyConds) > 0 {
-				conds = append(conds, sb.And(fieldKeyConds...))
-			}
-			limit += fieldKeySelector.Limit
-		}
-
-		if len(conds) > 0 {
-			sb.Where(sb.Or(conds...))
-		}
-
-		sb.GroupBy("name", "datatype")
-		if limit == 0 {
-			limit = 1000
-		}
-
-		query, args := sb.BuildWithFlavor(sqlbuilder.ClickHouse)
-		queries = append(queries, query)
-		allArgs = append(allArgs, args...)
-	}
-
-	if len(queries) == 0 {
-		return []*telemetrytypes.TelemetryFieldKey{}, true, nil
-	}
-
-	var limit int
-	for _, fieldKeySelector := range fieldKeySelectors {
-		limit += fieldKeySelector.Limit
-	}
-	if limit == 0 {
-		limit = 1000
-	}
-
-	mainQuery := fmt.Sprintf(`
-		SELECT tag_key, tag_type, tag_data_type, max(priority) as priority
-		FROM (
-			%s
-		) AS combined_results
-		GROUP BY tag_key, tag_type, tag_data_type
-		ORDER BY priority
-		LIMIT %d
-	`, strings.Join(queries, " UNION ALL "), limit+1)
-
-	rows, err := t.telemetrystore.ClickhouseDB().Query(ctx, mainQuery, allArgs...)
-	if err != nil {
-		return nil, false, errors.Wrap(err, errors.TypeInternal, errors.CodeInternal, ErrFailedToGetAuditKeys.Error())
-	}
-	defer rows.Close()
-
-	keys := []*telemetrytypes.TelemetryFieldKey{}
-	rowCount := 0
-	searchTexts := []string{}
-
-	for _, fieldKeySelector := range fieldKeySelectors {
-		searchTexts = append(searchTexts, fieldKeySelector.Name)
-	}
-
-	for rows.Next() {
-		rowCount++
-		if rowCount > limit {
-			break
-		}
-
-		var name string
-		var fieldContext telemetrytypes.FieldContext
-		var fieldDataType telemetrytypes.FieldDataType
-		var priority uint8
-		err = rows.Scan(&name, &fieldContext, &fieldDataType, &priority)
-		if err != nil {
-			return nil, false, errors.Wrap(err, errors.TypeInternal, errors.CodeInternal, ErrFailedToGetAuditKeys.Error())
-		}
-		key, ok := mapOfKeys[name+";"+fieldContext.StringValue()+";"+fieldDataType.StringValue()]
-
-		if !ok {
-			key = &telemetrytypes.TelemetryFieldKey{
-				Name:          name,
-				Signal:        telemetrytypes.SignalLogs,
-				FieldContext:  fieldContext,
-				FieldDataType: fieldDataType,
-			}
-		}
-
-		keys = append(keys, key)
-		mapOfKeys[name+";"+fieldContext.StringValue()+";"+fieldDataType.StringValue()] = key
-	}
-
-	if rows.Err() != nil {
-		return nil, false, errors.Wrap(rows.Err(), errors.TypeInternal, errors.CodeInternal, ErrFailedToGetAuditKeys.Error())
-	}
-
-	complete := rowCount <= limit
-
-	// Add intrinsic audit fields (same as logs intrinsics: body, severity_text, etc.)
-	staticKeys := maps.Keys(telemetryaudit.IntrinsicFields)
-	for _, key := range staticKeys {
-		found := false
-		for _, v := range searchTexts {
-			if v == "" || strings.Contains(key, v) {
-				found = true
-				break
-			}
-		}
-
-		if found {
-			if field, exists := telemetryaudit.IntrinsicFields[key]; exists {
-				if _, added := mapOfKeys[field.Name+";"+field.FieldContext.StringValue()+";"+field.FieldDataType.StringValue()]; !added {
-					keys = append(keys, &field)
-				}
-			}
-		}
-	}
-
-	return keys, complete, nil
-}
-
 func getPriorityForContext(ctx telemetrytypes.FieldContext) int {
 	switch ctx {
 	case telemetrytypes.FieldContextLog:
@@ -1127,11 +889,7 @@ func (t *telemetryMetaStore) GetKeys(ctx context.Context, fieldKeySelector *tele
 	case telemetrytypes.SignalTraces:
 		keys, complete, err = t.getTracesKeys(ctx, selectors)
 	case telemetrytypes.SignalLogs:
-		if fieldKeySelector.Source == telemetrytypes.SourceAudit {
-			keys, complete, err = t.getAuditKeys(ctx, selectors)
-		} else {
-			keys, complete, err = t.getLogsKeys(ctx, selectors)
-		}
+		keys, complete, err = t.getLogsKeys(ctx, selectors)
 	case telemetrytypes.SignalMetrics:
 		if fieldKeySelector.Source == telemetrytypes.SourceMeter {
 			keys, complete, err = t.getMeterSourceMetricKeys(ctx, selectors)
@@ -1180,7 +938,6 @@ func (t *telemetryMetaStore) GetKeys(ctx context.Context, fieldKeySelector *tele
 func (t *telemetryMetaStore) GetKeysMulti(ctx context.Context, fieldKeySelectors []*telemetrytypes.FieldKeySelector) (map[string][]*telemetrytypes.TelemetryFieldKey, bool, error) {
 
 	logsSelectors := []*telemetrytypes.FieldKeySelector{}
-	auditSelectors := []*telemetrytypes.FieldKeySelector{}
 	tracesSelectors := []*telemetrytypes.FieldKeySelector{}
 	metricsSelectors := []*telemetrytypes.FieldKeySelector{}
 	meterSourceMetricsSelectors := []*telemetrytypes.FieldKeySelector{}
@@ -1188,11 +945,7 @@ func (t *telemetryMetaStore) GetKeysMulti(ctx context.Context, fieldKeySelectors
 	for _, fieldKeySelector := range fieldKeySelectors {
 		switch fieldKeySelector.Signal {
 		case telemetrytypes.SignalLogs:
-			if fieldKeySelector.Source == telemetrytypes.SourceAudit {
-				auditSelectors = append(auditSelectors, fieldKeySelector)
-			} else {
-				logsSelectors = append(logsSelectors, fieldKeySelector)
-			}
+			logsSelectors = append(logsSelectors, fieldKeySelector)
 		case telemetrytypes.SignalTraces:
 			tracesSelectors = append(tracesSelectors, fieldKeySelector)
 		case telemetrytypes.SignalMetrics:
@@ -1212,10 +965,6 @@ func (t *telemetryMetaStore) GetKeysMulti(ctx context.Context, fieldKeySelectors
 	if err != nil {
 		return nil, false, err
 	}
-	auditKeys, auditComplete, err := t.getAuditKeys(ctx, auditSelectors)
-	if err != nil {
-		return nil, false, err
-	}
 	tracesKeys, tracesComplete, err := t.getTracesKeys(ctx, tracesSelectors)
 	if err != nil {
 		return nil, false, err
@@ -1230,15 +979,12 @@ func (t *telemetryMetaStore) GetKeysMulti(ctx context.Context, fieldKeySelectors
 		return nil, false, err
 	}
 	// Complete only if all queries are complete
-	complete := logsComplete && auditComplete && tracesComplete && metricsComplete
+	complete := logsComplete && tracesComplete && metricsComplete
 
 	mapOfKeys := make(map[string][]*telemetrytypes.TelemetryFieldKey)
 	for _, key := range logsKeys {
 		mapOfKeys[key.Name] = append(mapOfKeys[key.Name], key)
 	}
-	for _, key := range auditKeys {
-		mapOfKeys[key.Name] = append(mapOfKeys[key.Name], key)
-	}
 	for _, key := range tracesKeys {
 		mapOfKeys[key.Name] = append(mapOfKeys[key.Name], key)
 	}
@@ -1592,97 +1338,6 @@ func (t *telemetryMetaStore) getLogFieldValues(ctx context.Context, fieldValueSe
 	return values, complete, nil
 }
 
-func (t *telemetryMetaStore) getAuditFieldValues(ctx context.Context, fieldValueSelector *telemetrytypes.FieldValueSelector) (*telemetrytypes.TelemetryFieldValues, bool, error) {
-	ctx = ctxtypes.NewContextWithCommentVals(ctx, map[string]string{
-		instrumentationtypes.TelemetrySignal:  telemetrytypes.SignalLogs.StringValue(),
-		instrumentationtypes.CodeNamespace:    "metadata",
-		instrumentationtypes.CodeFunctionName: "getAuditFieldValues",
-	})
-
-	limit := fieldValueSelector.Limit
-	if limit == 0 {
-		limit = 50
-	}
-
-	sb := sqlbuilder.Select("DISTINCT string_value, number_value").From(t.auditDBName + "." + t.auditFieldsTblName)
-
-	if fieldValueSelector.Name != "" {
-		sb.Where(sb.E("tag_key", fieldValueSelector.Name))
-	}
-
-	if fieldValueSelector.FieldContext != telemetrytypes.FieldContextUnspecified {
-		sb.Where(sb.E("tag_type", fieldValueSelector.FieldContext.TagType()))
-	}
-
-	if fieldValueSelector.FieldDataType != telemetrytypes.FieldDataTypeUnspecified {
-		sb.Where(sb.E("tag_data_type", fieldValueSelector.FieldDataType.TagDataType()))
-	}
-
-	if fieldValueSelector.Value != "" {
-		switch fieldValueSelector.FieldDataType {
-		case telemetrytypes.FieldDataTypeString:
-			sb.Where(sb.ILike("string_value", "%"+escapeForLike(fieldValueSelector.Value)+"%"))
-		case telemetrytypes.FieldDataTypeNumber:
-			sb.Where(sb.IsNotNull("number_value"))
-			sb.Where(sb.ILike("toString(number_value)", "%"+escapeForLike(fieldValueSelector.Value)+"%"))
-		case telemetrytypes.FieldDataTypeUnspecified:
-			sb.Where(sb.Or(
-				sb.ILike("string_value", "%"+escapeForLike(fieldValueSelector.Value)+"%"),
-				sb.ILike("toString(number_value)", "%"+escapeForLike(fieldValueSelector.Value)+"%"),
-			))
-		}
-	}
-
-	// fetch one extra row to detect whether the result set is complete
-	sb.Limit(limit + 1)
-
-	query, args := sb.BuildWithFlavor(sqlbuilder.ClickHouse)
-
-	rows, err := t.telemetrystore.ClickhouseDB().Query(ctx, query, args...)
-	if err != nil {
-		return nil, false, errors.Wrap(err, errors.TypeInternal, errors.CodeInternal, ErrFailedToGetAuditKeys.Error())
-	}
-	defer rows.Close()
-
-	values := &telemetrytypes.TelemetryFieldValues{}
-	seen := make(map[string]bool)
-	rowCount := 0
-	totalCount := 0
-
-	for rows.Next() {
-		rowCount++
-
-		var stringValue string
-		var numberValue float64
-		err = rows.Scan(&stringValue, &numberValue)
-		if err != nil {
-			return nil, false, errors.Wrap(err, errors.TypeInternal, errors.CodeInternal, ErrFailedToGetAuditKeys.Error())
-		}
-		if stringValue != "" && !seen[stringValue] {
-			if totalCount >= limit {
-				break
-			}
-			values.StringValues = append(values.StringValues, stringValue)
-			seen[stringValue] = true
-			totalCount++
-		}
-		if numberValue != 0 {
-			if totalCount >= limit {
-				break
-			}
-			if !seen[fmt.Sprintf("%f", numberValue)] {
-				values.NumberValues = append(values.NumberValues, numberValue)
-				seen[fmt.Sprintf("%f", numberValue)] = true
-				totalCount++
-			}
-		}
-	}
-
-	complete := rowCount <= limit
-
-	return values, complete, nil
-}
-
 // getMetricFieldValues returns field values and whether the result is complete.
 func (t *telemetryMetaStore) getMetricFieldValues(ctx context.Context, fieldValueSelector *telemetrytypes.FieldValueSelector) (*telemetrytypes.TelemetryFieldValues, bool, error) {
 	ctx = ctxtypes.NewContextWithCommentVals(ctx, map[string]string{
@@ -1973,11 +1628,7 @@ func (t *telemetryMetaStore) GetAllValues(ctx context.Context, fieldValueSelecto
 	case telemetrytypes.SignalTraces:
 		values, complete, err = t.getSpanFieldValues(ctx, fieldValueSelector)
 	case telemetrytypes.SignalLogs:
-		if fieldValueSelector.Source == telemetrytypes.SourceAudit {
-			values, complete, err = t.getAuditFieldValues(ctx, fieldValueSelector)
-		} else {
-			values, complete, err = t.getLogFieldValues(ctx, fieldValueSelector)
-		}
+		values, complete, err = t.getLogFieldValues(ctx, fieldValueSelector)
 	case telemetrytypes.SignalMetrics:
 		if fieldValueSelector.Source == telemetrytypes.SourceMeter {
 			values, complete, err = t.getMeterSourceMetricFieldValues(ctx, fieldValueSelector)

pkg/telemetrymetadata/metadata_query_test.go

@@ -5,7 +5,6 @@ import (
 	"testing"
 
 	"github.com/SigNoz/signoz/pkg/instrumentation/instrumentationtest"
-	"github.com/SigNoz/signoz/pkg/telemetryaudit"
 	"github.com/SigNoz/signoz/pkg/telemetrylogs"
 	"github.com/SigNoz/signoz/pkg/telemetrymeter"
 	"github.com/SigNoz/signoz/pkg/telemetrymetrics"
@@ -38,11 +37,6 @@ func TestGetFirstSeenFromMetricMetadata(t *testing.T) {
 		telemetrylogs.TagAttributesV2TableName,
 		telemetrylogs.LogAttributeKeysTblName,
 		telemetrylogs.LogResourceKeysTblName,
-		telemetryaudit.DBName,
-		telemetryaudit.AuditLogsTableName,
-		telemetryaudit.TagAttributesTableName,
-		telemetryaudit.LogAttributeKeysTblName,
-		telemetryaudit.LogResourceKeysTblName,
 		DBName,
 		AttributesMetadataLocalTableName,
 		ColumnEvolutionMetadataTableName,

pkg/telemetrymetadata/metadata_test.go

@@ -7,7 +7,6 @@ import (
 
 	"github.com/SigNoz/signoz/pkg/errors"
 	"github.com/SigNoz/signoz/pkg/instrumentation/instrumentationtest"
-	"github.com/SigNoz/signoz/pkg/telemetryaudit"
 	"github.com/SigNoz/signoz/pkg/telemetrylogs"
 	"github.com/SigNoz/signoz/pkg/telemetrymeter"
 	"github.com/SigNoz/signoz/pkg/telemetrymetrics"
@@ -37,11 +36,6 @@ func newTestTelemetryMetaStoreTestHelper(store telemetrystore.TelemetryStore) te
 		telemetrylogs.TagAttributesV2TableName,
 		telemetrylogs.LogAttributeKeysTblName,
 		telemetrylogs.LogResourceKeysTblName,
-		telemetryaudit.DBName,
-		telemetryaudit.AuditLogsTableName,
-		telemetryaudit.TagAttributesTableName,
-		telemetryaudit.LogAttributeKeysTblName,
-		telemetryaudit.LogResourceKeysTblName,
 		DBName,
 		AttributesMetadataLocalTableName,
 		ColumnEvolutionMetadataTableName,

pkg/telemetryresourcefilter/statement_builder.go

@@ -21,7 +21,6 @@ type resourceFilterStatementBuilder[T any] struct {
 	conditionBuilder qbtypes.ConditionBuilder
 	metadataStore    telemetrytypes.MetadataStore
 	signal           telemetrytypes.Signal
-	source           telemetrytypes.Source
 
 	fullTextColumn *telemetrytypes.TelemetryFieldKey
 	jsonKeyToKey   qbtypes.JsonKeyToFieldFunc
@@ -38,7 +37,6 @@ func New[T any](
 	dbName string,
 	tableName string,
 	signal telemetrytypes.Signal,
-	source telemetrytypes.Source,
 	metadataStore telemetrytypes.MetadataStore,
 	fullTextColumn *telemetrytypes.TelemetryFieldKey,
 	jsonKeyToKey qbtypes.JsonKeyToFieldFunc,
@@ -54,7 +52,6 @@ func New[T any](
 		conditionBuilder: cb,
 		metadataStore:    metadataStore,
 		signal:           signal,
-		source:           source,
 		fullTextColumn:   fullTextColumn,
 		jsonKeyToKey:     jsonKeyToKey,
 	}
@@ -75,7 +72,6 @@ func (b *resourceFilterStatementBuilder[T]) getKeySelectors(query qbtypes.QueryB
 			continue
 		}
 		keySelectors[idx].Signal = b.signal
-		keySelectors[idx].Source = b.source
 		keySelectors[idx].SelectorMatchType = telemetrytypes.FieldSelectorMatchTypeExact
 		filteredKeySelectors = append(filteredKeySelectors, keySelectors[idx])
 	}

pkg/telemetryresourcefilter/statement_builder_test.go

@@ -375,7 +375,6 @@ func TestResourceFilterStatementBuilder_Traces(t *testing.T) {
 		"signoz_traces",
 		"distributed_traces_v3_resource",
 		telemetrytypes.SignalTraces,
-		telemetrytypes.SourceUnspecified,
 		mockMetadataStore,
 		nil,
 		nil,
@@ -593,7 +592,6 @@ func TestResourceFilterStatementBuilder_Logs(t *testing.T) {
 		"signoz_logs",
 		"distributed_logs_v2_resource",
 		telemetrytypes.SignalLogs,
-		telemetrytypes.SourceUnspecified,
 		mockMetadataStore,
 		nil,
 		nil,
@@ -655,7 +653,6 @@ func TestResourceFilterStatementBuilder_Variables(t *testing.T) {
 		"signoz_traces",
 		"distributed_traces_v3_resource",
 		telemetrytypes.SignalTraces,
-		telemetrytypes.SourceUnspecified,
 		mockMetadataStore,
 		nil,
 		nil,

pkg/telemetrytraces/statement_builder.go

@@ -49,7 +49,6 @@ func NewTraceQueryStatementBuilder(
 		DBName,
 		TracesResourceV3TableName,
 		telemetrytypes.SignalTraces,
-		telemetrytypes.SourceUnspecified,
 		metadataStore,
 		nil,
 		nil,

pkg/telemetrytraces/trace_operator_statement_builder.go

@@ -39,7 +39,6 @@ func NewTraceOperatorStatementBuilder(
 		DBName,
 		TracesResourceV3TableName,
 		telemetrytypes.SignalTraces,
-		telemetrytypes.SourceUnspecified,
 		metadataStore,
 		nil,
 		nil,

pkg/types/telemetrytypes/source.go

@@ -7,13 +7,11 @@ type Source struct {
 }
 
 var (
-	SourceAudit       = Source{valuer.NewString("audit")}
 	SourceMeter       = Source{valuer.NewString("meter")}
 	SourceUnspecified = Source{valuer.NewString("")}
 )
 
 // Enum returns the acceptable values for Source.
-// TODO: Add SourceAudit once the frontend is ready for consumption.
 func (Source) Enum() []any {
 	return []any{
 		SourceMeter,

tests/integration/conftest.py

@@ -12,7 +12,6 @@ pytest_plugins = [
     "fixtures.sqlite",
     "fixtures.zookeeper",
     "fixtures.signoz",
-    "fixtures.audit",
     "fixtures.logs",
     "fixtures.traces",
     "fixtures.metrics",

tests/integration/fixtures/audit.py

@@ -1,404 +0,0 @@
-import datetime
-import json
-from abc import ABC
-from typing import Any, Callable, Generator, List, Optional
-
-import numpy as np
-import pytest
-from ksuid import KsuidMs
-
-from fixtures import types
-from fixtures.fingerprint import LogsOrTracesFingerprint
-
-
-class AuditResource(ABC):
-    labels: str
-    fingerprint: str
-    seen_at_ts_bucket_start: np.int64
-
-    def __init__(
-        self,
-        labels: dict[str, str],
-        fingerprint: str,
-        seen_at_ts_bucket_start: np.int64,
-    ) -> None:
-        self.labels = json.dumps(labels, separators=(",", ":"))
-        self.fingerprint = fingerprint
-        self.seen_at_ts_bucket_start = seen_at_ts_bucket_start
-
-    def np_arr(self) -> np.array:
-        return np.array(
-            [
-                self.labels,
-                self.fingerprint,
-                self.seen_at_ts_bucket_start,
-            ]
-        )
-
-
-class AuditResourceOrAttributeKeys(ABC):
-    name: str
-    datatype: str
-
-    def __init__(self, name: str, datatype: str) -> None:
-        self.name = name
-        self.datatype = datatype
-
-    def np_arr(self) -> np.array:
-        return np.array([self.name, self.datatype])
-
-
-class AuditTagAttributes(ABC):
-    unix_milli: np.int64
-    tag_key: str
-    tag_type: str
-    tag_data_type: str
-    string_value: str
-    int64_value: Optional[np.int64]
-    float64_value: Optional[np.float64]
-
-    def __init__(
-        self,
-        timestamp: datetime.datetime,
-        tag_key: str,
-        tag_type: str,
-        tag_data_type: str,
-        string_value: Optional[str],
-        int64_value: Optional[np.int64],
-        float64_value: Optional[np.float64],
-    ) -> None:
-        self.unix_milli = np.int64(int(timestamp.timestamp() * 1e3))
-        self.tag_key = tag_key
-        self.tag_type = tag_type
-        self.tag_data_type = tag_data_type
-        self.string_value = string_value or ""
-        self.int64_value = int64_value
-        self.float64_value = float64_value
-
-    def np_arr(self) -> np.array:
-        return np.array(
-            [
-                self.unix_milli,
-                self.tag_key,
-                self.tag_type,
-                self.tag_data_type,
-                self.string_value,
-                self.int64_value,
-                self.float64_value,
-            ]
-        )
-
-
-class AuditLog(ABC):
-    """Represents a single audit log event in signoz_audit.
-
-    Matches the ClickHouse DDL from the schema migration (ticket #1936):
-    - Database: signoz_audit
-    - Local table: logs
-    - Distributed table: distributed_logs
-    - No resources_string column (resource JSON only)
-    - Has event_name column
-    - 7 materialized columns auto-populated from attributes_string at INSERT time
-    """
-
-    ts_bucket_start: np.uint64
-    resource_fingerprint: str
-    timestamp: np.uint64
-    observed_timestamp: np.uint64
-    id: str
-    trace_id: str
-    span_id: str
-    trace_flags: np.uint32
-    severity_text: str
-    severity_number: np.uint8
-    body: str
-    scope_name: str
-    scope_version: str
-    scope_string: dict[str, str]
-    attributes_string: dict[str, str]
-    attributes_number: dict[str, np.float64]
-    attributes_bool: dict[str, bool]
-    resource_json: dict[str, str]
-    event_name: str
-
-    resource: List[AuditResource]
-    tag_attributes: List[AuditTagAttributes]
-    resource_keys: List[AuditResourceOrAttributeKeys]
-    attribute_keys: List[AuditResourceOrAttributeKeys]
-
-    def __init__(
-        self,
-        timestamp: Optional[datetime.datetime] = None,
-        resources: dict[str, Any] = {},
-        attributes: dict[str, Any] = {},
-        body: str = "",
-        event_name: str = "",
-        severity_text: str = "INFO",
-        trace_id: str = "",
-        span_id: str = "",
-        trace_flags: np.uint32 = 0,
-        scope_name: str = "signoz.audit",
-        scope_version: str = "",
-    ) -> None:
-        if timestamp is None:
-            timestamp = datetime.datetime.now()
-        self.tag_attributes = []
-        self.attribute_keys = []
-        self.resource_keys = []
-
-        self.timestamp = np.uint64(int(timestamp.timestamp() * 1e9))
-        self.observed_timestamp = self.timestamp
-
-        minute = timestamp.minute
-        bucket_minute = 0 if minute < 30 else 30
-        bucket_start = timestamp.replace(minute=bucket_minute, second=0, microsecond=0)
-        self.ts_bucket_start = np.uint64(int(bucket_start.timestamp()))
-
-        self.id = str(KsuidMs(datetime=timestamp))
-
-        self.trace_id = trace_id
-        self.span_id = span_id
-        self.trace_flags = trace_flags
-
-        self.severity_text = severity_text
-        self.severity_number = np.uint8(9 if severity_text == "INFO" else 17)
-
-        self.body = body
-        self.event_name = event_name
-
-        # Resources — JSON column only (no resources_string in audit DDL)
-        self.resource_json = {k: str(v) for k, v in resources.items()}
-        for k, v in self.resource_json.items():
-            self.tag_attributes.append(
-                AuditTagAttributes(
-                    timestamp=timestamp,
-                    tag_key=k,
-                    tag_type="resource",
-                    tag_data_type="string",
-                    string_value=str(v),
-                    int64_value=None,
-                    float64_value=None,
-                )
-            )
-            self.resource_keys.append(
-                AuditResourceOrAttributeKeys(name=k, datatype="string")
-            )
-
-        self.resource_fingerprint = LogsOrTracesFingerprint(
-            self.resource_json
-        ).calculate()
-
-        # Process attributes by type
-        self.attributes_string = {}
-        self.attributes_number = {}
-        self.attributes_bool = {}
-
-        for k, v in attributes.items():
-            if isinstance(v, bool):
-                self.attributes_bool[k] = v
-                self.tag_attributes.append(
-                    AuditTagAttributes(
-                        timestamp=timestamp,
-                        tag_key=k,
-                        tag_type="tag",
-                        tag_data_type="bool",
-                        string_value=None,
-                        int64_value=None,
-                        float64_value=None,
-                    )
-                )
-                self.attribute_keys.append(
-                    AuditResourceOrAttributeKeys(name=k, datatype="bool")
-                )
-            elif isinstance(v, int):
-                self.attributes_number[k] = np.float64(v)
-                self.tag_attributes.append(
-                    AuditTagAttributes(
-                        timestamp=timestamp,
-                        tag_key=k,
-                        tag_type="tag",
-                        tag_data_type="int64",
-                        string_value=None,
-                        int64_value=np.int64(v),
-                        float64_value=None,
-                    )
-                )
-                self.attribute_keys.append(
-                    AuditResourceOrAttributeKeys(name=k, datatype="int64")
-                )
-            elif isinstance(v, float):
-                self.attributes_number[k] = np.float64(v)
-                self.tag_attributes.append(
-                    AuditTagAttributes(
-                        timestamp=timestamp,
-                        tag_key=k,
-                        tag_type="tag",
-                        tag_data_type="float64",
-                        string_value=None,
-                        int64_value=None,
-                        float64_value=np.float64(v),
-                    )
-                )
-                self.attribute_keys.append(
-                    AuditResourceOrAttributeKeys(name=k, datatype="float64")
-                )
-            else:
-                self.attributes_string[k] = str(v)
-                self.tag_attributes.append(
-                    AuditTagAttributes(
-                        timestamp=timestamp,
-                        tag_key=k,
-                        tag_type="tag",
-                        tag_data_type="string",
-                        string_value=str(v),
-                        int64_value=None,
-                        float64_value=None,
-                    )
-                )
-                self.attribute_keys.append(
-                    AuditResourceOrAttributeKeys(name=k, datatype="string")
-                )
-
-        self.scope_name = scope_name
-        self.scope_version = scope_version
-        self.scope_string = {}
-
-        self.resource = [
-            AuditResource(
-                labels=self.resource_json,
-                fingerprint=self.resource_fingerprint,
-                seen_at_ts_bucket_start=self.ts_bucket_start,
-            )
-        ]
-
-    def np_arr(self) -> np.array:
-        return np.array(
-            [
-                self.ts_bucket_start,
-                self.resource_fingerprint,
-                self.timestamp,
-                self.observed_timestamp,
-                self.id,
-                self.trace_id,
-                self.span_id,
-                self.trace_flags,
-                self.severity_text,
-                self.severity_number,
-                self.body,
-                self.scope_name,
-                self.scope_version,
-                self.scope_string,
-                self.attributes_string,
-                self.attributes_number,
-                self.attributes_bool,
-                self.resource_json,
-                self.event_name,
-            ]
-        )
-
-
-@pytest.fixture(name="insert_audit_logs", scope="function")
-def insert_audit_logs(
-    clickhouse: types.TestContainerClickhouse,
-) -> Generator[Callable[[List[AuditLog]], None], Any, None]:
-    def _insert_audit_logs(logs: List[AuditLog]) -> None:
-        resources: List[AuditResource] = []
-        for log in logs:
-            resources.extend(log.resource)
-
-        if len(resources) > 0:
-            clickhouse.conn.insert(
-                database="signoz_audit",
-                table="distributed_logs_resource",
-                data=[resource.np_arr() for resource in resources],
-                column_names=[
-                    "labels",
-                    "fingerprint",
-                    "seen_at_ts_bucket_start",
-                ],
-            )
-
-        tag_attributes: List[AuditTagAttributes] = []
-        for log in logs:
-            tag_attributes.extend(log.tag_attributes)
-
-        if len(tag_attributes) > 0:
-            clickhouse.conn.insert(
-                database="signoz_audit",
-                table="distributed_tag_attributes",
-                data=[ta.np_arr() for ta in tag_attributes],
-                column_names=[
-                    "unix_milli",
-                    "tag_key",
-                    "tag_type",
-                    "tag_data_type",
-                    "string_value",
-                    "int64_value",
-                    "float64_value",
-                ],
-            )
-
-        attribute_keys: List[AuditResourceOrAttributeKeys] = []
-        for log in logs:
-            attribute_keys.extend(log.attribute_keys)
-
-        if len(attribute_keys) > 0:
-            clickhouse.conn.insert(
-                database="signoz_audit",
-                table="distributed_logs_attribute_keys",
-                data=[ak.np_arr() for ak in attribute_keys],
-                column_names=["name", "datatype"],
-            )
-
-        resource_keys: List[AuditResourceOrAttributeKeys] = []
-        for log in logs:
-            resource_keys.extend(log.resource_keys)
-
-        if len(resource_keys) > 0:
-            clickhouse.conn.insert(
-                database="signoz_audit",
-                table="distributed_logs_resource_keys",
-                data=[rk.np_arr() for rk in resource_keys],
-                column_names=["name", "datatype"],
-            )
-
-        clickhouse.conn.insert(
-            database="signoz_audit",
-            table="distributed_logs",
-            data=[log.np_arr() for log in logs],
-            column_names=[
-                "ts_bucket_start",
-                "resource_fingerprint",
-                "timestamp",
-                "observed_timestamp",
-                "id",
-                "trace_id",
-                "span_id",
-                "trace_flags",
-                "severity_text",
-                "severity_number",
-                "body",
-                "scope_name",
-                "scope_version",
-                "scope_string",
-                "attributes_string",
-                "attributes_number",
-                "attributes_bool",
-                "resource",
-                "event_name",
-            ],
-        )
-
-    yield _insert_audit_logs
-
-    cluster = clickhouse.env["SIGNOZ_TELEMETRYSTORE_CLICKHOUSE_CLUSTER"]
-    for table in [
-        "logs",
-        "logs_resource",
-        "tag_attributes",
-        "logs_attribute_keys",
-        "logs_resource_keys",
-    ]:
-        clickhouse.conn.query(
-            f"TRUNCATE TABLE signoz_audit.{table} ON CLUSTER '{cluster}' SYNC"
-        )

tests/integration/fixtures/querier.py

@@ -38,7 +38,6 @@ class OrderBy:
 class BuilderQuery:
     signal: str
     name: str = "A"
-    source: Optional[str] = None
     limit: Optional[int] = None
     filter_expression: Optional[str] = None
     select_fields: Optional[List[TelemetryFieldKey]] = None
@@ -49,8 +48,6 @@ class BuilderQuery:
             "signal": self.signal,
             "name": self.name,
         }
-        if self.source:
-            spec["source"] = self.source
         if self.limit is not None:
             spec["limit"] = self.limit
         if self.filter_expression:
@@ -58,9 +55,7 @@ class BuilderQuery:
         if self.select_fields:
             spec["selectFields"] = [f.to_dict() for f in self.select_fields]
         if self.order:
-            spec["order"] = [
-                o.to_dict() if hasattr(o, "to_dict") else o for o in self.order
-            ]
+            spec["order"] = [o.to_dict() for o in self.order]
         return {"type": "builder_query", "spec": spec}
 
 
@@ -81,9 +76,7 @@ class TraceOperatorQuery:
         if self.limit is not None:
             spec["limit"] = self.limit
         if self.order:
-            spec["order"] = [
-                o.to_dict() if hasattr(o, "to_dict") else o for o in self.order
-            ]
+            spec["order"] = [o.to_dict() for o in self.order]
         return {"type": "builder_trace_operator", "spec": spec}
 
 
@@ -449,7 +442,6 @@ def build_scalar_query(
     signal: str,
     aggregations: List[Dict],
     *,
-    source: Optional[str] = None,
     group_by: Optional[List[Dict]] = None,
     order: Optional[List[Dict]] = None,
     limit: Optional[int] = None,
@@ -466,9 +458,6 @@ def build_scalar_query(
         "aggregations": aggregations,
     }
 
-    if source:
-        spec["source"] = source
-
     if group_by:
         spec["groupBy"] = group_by
 

tests/integration/src/auditquerier/01_audit_logs.py

@@ -1,441 +0,0 @@
-from datetime import datetime, timedelta, timezone
-from http import HTTPStatus
-from typing import Callable, List
-
-import pytest
-
-from fixtures import types
-from fixtures.audit import AuditLog
-from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
-from fixtures.querier import (
-    BuilderQuery,
-    build_logs_aggregation,
-    build_order_by,
-    build_scalar_query,
-    make_query_request,
-)
-
-
-def test_audit_list_all(
-    signoz: types.SigNoz,
-    create_user_admin: None,  # pylint: disable=unused-argument
-    get_token: Callable[[str, str], str],
-    insert_audit_logs: Callable[[List[AuditLog]], None],
-) -> None:
-    """List audit events across multiple resource types — verify count, ordering, and fields."""
-    now = datetime.now(tz=timezone.utc)
-    insert_audit_logs(
-        [
-            AuditLog(
-                timestamp=now - timedelta(seconds=3),
-                resources={
-                    "service.name": "signoz",
-                    "signoz.audit.resource.kind": "alert-rule",
-                    "signoz.audit.resource.id": "alert-001",
-                },
-                attributes={
-                    "signoz.audit.principal.id": "user-010",
-                    "signoz.audit.principal.email": "ops@acme.com",
-                    "signoz.audit.principal.type": "user",
-                    "signoz.audit.action": "create",
-                    "signoz.audit.outcome": "success",
-                },
-                body="ops@acme.com (user-010) created alert-rule (alert-001)",
-                event_name="alert-rule.created",
-                severity_text="INFO",
-            ),
-            AuditLog(
-                timestamp=now - timedelta(seconds=2),
-                resources={
-                    "service.name": "signoz",
-                    "signoz.audit.resource.kind": "saved-view",
-                    "signoz.audit.resource.id": "view-001",
-                },
-                attributes={
-                    "signoz.audit.principal.id": "user-010",
-                    "signoz.audit.principal.email": "ops@acme.com",
-                    "signoz.audit.principal.type": "user",
-                    "signoz.audit.action": "update",
-                    "signoz.audit.outcome": "success",
-                },
-                body="ops@acme.com (user-010) updated saved-view (view-001)",
-                event_name="saved-view.updated",
-                severity_text="INFO",
-            ),
-            AuditLog(
-                timestamp=now - timedelta(seconds=1),
-                resources={
-                    "service.name": "signoz",
-                    "signoz.audit.resource.kind": "user",
-                    "signoz.audit.resource.id": "user-020",
-                },
-                attributes={
-                    "signoz.audit.principal.id": "user-010",
-                    "signoz.audit.principal.email": "ops@acme.com",
-                    "signoz.audit.principal.type": "user",
-                    "signoz.audit.action": "update",
-                    "signoz.audit.action_category": "access_control",
-                    "signoz.audit.outcome": "success",
-                },
-                body="ops@acme.com (user-010) updated user (user-020)",
-                event_name="user.role.changed",
-                severity_text="INFO",
-            ),
-        ]
-    )
-    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
-
-    now = datetime.now(tz=timezone.utc)
-    response = make_query_request(
-        signoz,
-        token,
-        start_ms=int((now - timedelta(seconds=30)).timestamp() * 1000),
-        end_ms=int(now.timestamp() * 1000),
-        queries=[
-            BuilderQuery(
-                signal="logs",
-                source="audit",
-                limit=100,
-                order=[build_order_by("timestamp"), build_order_by("id")],
-            ).to_dict()
-        ],
-        request_type="raw",
-    )
-
-    assert response.status_code == HTTPStatus.OK
-    assert response.json()["status"] == "success"
-
-    rows = response.json()["data"]["data"]["results"][0]["rows"]
-    assert len(rows) == 3
-
-    # Most recent first
-    assert rows[0]["data"]["event_name"] == "user.role.changed"
-    assert rows[1]["data"]["event_name"] == "saved-view.updated"
-    assert rows[2]["data"]["event_name"] == "alert-rule.created"
-
-    # Verify event_name and body are present
-    assert rows[0]["data"]["body"] == "ops@acme.com (user-010) updated user (user-020)"
-    assert rows[0]["data"]["severity_text"] == "INFO"
-
-
-@pytest.mark.parametrize(
-    "filter_expression,expected_count,expected_event_names",
-    [
-        pytest.param(
-            "signoz.audit.principal.id = 'user-001'",
-            3,
-            {"session.login", "dashboard.updated", "dashboard.created"},
-            id="filter_by_principal_id",
-        ),
-        pytest.param(
-            "signoz.audit.outcome = 'failure'",
-            1,
-            {"dashboard.deleted"},
-            id="filter_by_outcome_failure",
-        ),
-        pytest.param(
-            "signoz.audit.resource.kind = 'dashboard'"
-            " AND signoz.audit.resource.id = 'dash-001'",
-            3,
-            {"dashboard.deleted", "dashboard.updated", "dashboard.created"},
-            id="filter_by_resource_kind_and_id",
-        ),
-        pytest.param(
-            "signoz.audit.principal.type = 'service_account'",
-            1,
-            {"serviceaccount.apikey.created"},
-            id="filter_by_principal_type",
-        ),
-        pytest.param(
-            "signoz.audit.resource.kind = 'dashboard'"
-            " AND signoz.audit.action = 'delete'",
-            1,
-            {"dashboard.deleted"},
-            id="filter_by_resource_kind_and_action",
-        ),
-    ],
-)
-def test_audit_filter(
-    signoz: types.SigNoz,
-    create_user_admin: None,  # pylint: disable=unused-argument
-    get_token: Callable[[str, str], str],
-    insert_audit_logs: Callable[[List[AuditLog]], None],
-    filter_expression: str,
-    expected_count: int,
-    expected_event_names: set,
-) -> None:
-    """Parametrized audit filter tests covering the documented query patterns."""
-    now = datetime.now(tz=timezone.utc)
-    insert_audit_logs(
-        [
-            AuditLog(
-                timestamp=now - timedelta(seconds=5),
-                resources={
-                    "service.name": "signoz",
-                    "signoz.audit.resource.kind": "dashboard",
-                    "signoz.audit.resource.id": "dash-001",
-                },
-                attributes={
-                    "signoz.audit.principal.id": "user-001",
-                    "signoz.audit.principal.email": "alice@acme.com",
-                    "signoz.audit.principal.type": "user",
-                    "signoz.audit.action": "create",
-                    "signoz.audit.action_category": "configuration_change",
-                    "signoz.audit.outcome": "success",
-                },
-                body="alice@acme.com created dashboard",
-                event_name="dashboard.created",
-            ),
-            AuditLog(
-                timestamp=now - timedelta(seconds=4),
-                resources={
-                    "service.name": "signoz",
-                    "signoz.audit.resource.kind": "dashboard",
-                    "signoz.audit.resource.id": "dash-001",
-                },
-                attributes={
-                    "signoz.audit.principal.id": "user-001",
-                    "signoz.audit.principal.email": "alice@acme.com",
-                    "signoz.audit.principal.type": "user",
-                    "signoz.audit.action": "update",
-                    "signoz.audit.action_category": "configuration_change",
-                    "signoz.audit.outcome": "success",
-                },
-                body="alice@acme.com updated dashboard",
-                event_name="dashboard.updated",
-            ),
-            AuditLog(
-                timestamp=now - timedelta(seconds=3),
-                resources={
-                    "service.name": "signoz",
-                    "signoz.audit.resource.kind": "dashboard",
-                    "signoz.audit.resource.id": "dash-001",
-                },
-                attributes={
-                    "signoz.audit.principal.id": "user-002",
-                    "signoz.audit.principal.email": "viewer@acme.com",
-                    "signoz.audit.principal.type": "user",
-                    "signoz.audit.action": "delete",
-                    "signoz.audit.action_category": "configuration_change",
-                    "signoz.audit.outcome": "failure",
-                    "signoz.audit.error.type": "forbidden",
-                    "signoz.audit.error.code": "authz_forbidden",
-                },
-                body="viewer@acme.com failed to delete dashboard",
-                event_name="dashboard.deleted",
-                severity_text="ERROR",
-            ),
-            AuditLog(
-                timestamp=now - timedelta(seconds=2),
-                resources={
-                    "service.name": "signoz",
-                    "signoz.audit.resource.kind": "serviceaccount",
-                    "signoz.audit.resource.id": "sa-001",
-                },
-                attributes={
-                    "signoz.audit.principal.id": "sa-001",
-                    "signoz.audit.principal.email": "",
-                    "signoz.audit.principal.type": "service_account",
-                    "signoz.audit.action": "create",
-                    "signoz.audit.action_category": "access_control",
-                    "signoz.audit.outcome": "success",
-                },
-                body="sa-001 created serviceaccount",
-                event_name="serviceaccount.apikey.created",
-            ),
-            AuditLog(
-                timestamp=now - timedelta(seconds=1),
-                resources={
-                    "service.name": "signoz",
-                    "signoz.audit.resource.kind": "session",
-                    "signoz.audit.resource.id": "*",
-                },
-                attributes={
-                    "signoz.audit.principal.id": "user-001",
-                    "signoz.audit.principal.email": "alice@acme.com",
-                    "signoz.audit.principal.type": "user",
-                    "signoz.audit.action": "login",
-                    "signoz.audit.action_category": "access_control",
-                    "signoz.audit.outcome": "success",
-                },
-                body="alice@acme.com login session",
-                event_name="session.login",
-            ),
-        ]
-    )
-    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
-
-    now = datetime.now(tz=timezone.utc)
-    response = make_query_request(
-        signoz,
-        token,
-        start_ms=int((now - timedelta(seconds=30)).timestamp() * 1000),
-        end_ms=int(now.timestamp() * 1000),
-        queries=[
-            BuilderQuery(
-                signal="logs",
-                source="audit",
-                limit=100,
-                filter_expression=filter_expression,
-                order=[build_order_by("timestamp"), build_order_by("id")],
-            ).to_dict()
-        ],
-        request_type="raw",
-    )
-
-    assert response.status_code == HTTPStatus.OK
-
-    rows = response.json()["data"]["data"]["results"][0]["rows"]
-    assert len(rows) == expected_count
-
-    actual_event_names = {row["data"]["event_name"] for row in rows}
-    assert actual_event_names == expected_event_names
-
-
-def test_audit_scalar_count_failures(
-    signoz: types.SigNoz,
-    create_user_admin: None,  # pylint: disable=unused-argument
-    get_token: Callable[[str, str], str],
-    insert_audit_logs: Callable[[List[AuditLog]], None],
-) -> None:
-    """Alert query — count multiple failures from different principals."""
-    now = datetime.now(tz=timezone.utc)
-    insert_audit_logs(
-        [
-            AuditLog(
-                timestamp=now - timedelta(seconds=3),
-                resources={
-                    "service.name": "signoz",
-                    "signoz.audit.resource.kind": "dashboard",
-                    "signoz.audit.resource.id": "dash-100",
-                },
-                attributes={
-                    "signoz.audit.principal.id": "user-050",
-                    "signoz.audit.principal.type": "user",
-                    "signoz.audit.action": "delete",
-                    "signoz.audit.outcome": "failure",
-                },
-                body="user-050 failed to delete dashboard",
-                event_name="dashboard.deleted",
-                severity_text="ERROR",
-            ),
-            AuditLog(
-                timestamp=now - timedelta(seconds=2),
-                resources={
-                    "service.name": "signoz",
-                    "signoz.audit.resource.kind": "alert-rule",
-                    "signoz.audit.resource.id": "alert-200",
-                },
-                attributes={
-                    "signoz.audit.principal.id": "user-060",
-                    "signoz.audit.principal.type": "user",
-                    "signoz.audit.action": "update",
-                    "signoz.audit.outcome": "failure",
-                },
-                body="user-060 failed to update alert-rule",
-                event_name="alert-rule.updated",
-                severity_text="ERROR",
-            ),
-            AuditLog(
-                timestamp=now - timedelta(seconds=1),
-                resources={
-                    "service.name": "signoz",
-                    "signoz.audit.resource.kind": "dashboard",
-                    "signoz.audit.resource.id": "dash-100",
-                },
-                attributes={
-                    "signoz.audit.principal.id": "user-050",
-                    "signoz.audit.principal.type": "user",
-                    "signoz.audit.action": "update",
-                    "signoz.audit.outcome": "success",
-                },
-                body="user-050 updated dashboard",
-                event_name="dashboard.updated",
-            ),
-        ]
-    )
-    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
-
-    now = datetime.now(tz=timezone.utc)
-    response = make_query_request(
-        signoz,
-        token,
-        start_ms=int((now - timedelta(seconds=30)).timestamp() * 1000),
-        end_ms=int(now.timestamp() * 1000),
-        queries=[
-            build_scalar_query(
-                name="A",
-                signal="logs",
-                source="audit",
-                aggregations=[build_logs_aggregation("count()")],
-                filter_expression="signoz.audit.outcome = 'failure'",
-            )
-        ],
-        request_type="scalar",
-    )
-
-    assert response.status_code == HTTPStatus.OK
-    assert response.json()["status"] == "success"
-
-    scalar_data = response.json()["data"]["data"]["results"][0].get("data", [])
-    assert len(scalar_data) == 1
-    assert scalar_data[0][0] == 2
-
-
-def test_audit_does_not_leak_into_logs(
-    signoz: types.SigNoz,
-    create_user_admin: None,  # pylint: disable=unused-argument
-    get_token: Callable[[str, str], str],
-    insert_audit_logs: Callable[[List[AuditLog]], None],
-) -> None:
-    """A single audit event in signoz_audit must not appear in regular log queries."""
-    now = datetime.now(tz=timezone.utc)
-    insert_audit_logs(
-        [
-            AuditLog(
-                timestamp=now - timedelta(seconds=1),
-                resources={
-                    "service.name": "signoz",
-                    "signoz.audit.resource.kind": "organization",
-                    "signoz.audit.resource.id": "org-999",
-                },
-                attributes={
-                    "signoz.audit.principal.id": "user-admin",
-                    "signoz.audit.principal.type": "user",
-                    "signoz.audit.action": "update",
-                    "signoz.audit.outcome": "success",
-                },
-                body="user-admin updated organization (org-999)",
-                event_name="organization.updated",
-            ),
-        ]
-    )
-    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
-
-    now = datetime.now(tz=timezone.utc)
-    response = make_query_request(
-        signoz,
-        token,
-        start_ms=int((now - timedelta(seconds=30)).timestamp() * 1000),
-        end_ms=int(now.timestamp() * 1000),
-        queries=[
-            BuilderQuery(
-                signal="logs",
-                limit=100,
-                order=[build_order_by("timestamp"), build_order_by("id")],
-            ).to_dict()
-        ],
-        request_type="raw",
-    )
-
-    assert response.status_code == HTTPStatus.OK
-
-    rows = response.json()["data"]["data"]["results"][0].get("rows") or []
-
-    audit_bodies = [
-        row["data"]["body"]
-        for row in rows
-        if "signoz.audit"
-        in row["data"].get("attributes_string", {}).get("signoz.audit.action", "")
-    ]
-    assert len(audit_bodies) == 0