Compare commits

..

1 Commits

Author SHA1 Message Date
Tushar Vats
8a7d32e701 refactor(querier): 3-layer per-signal query architecture
Split the five telemetry<signal> packages, which mixed three concerns, into
three per-signal layers with a cycle-free dependency direction:

- telemetryschema/<signal>telemetryschema — primitives (const + table
  selection, field_mapper, condition_builder, trace helpers); leaf layer.
- statementbuilder/<signal>statementbuilder — SQL generation, each with a
  cohesive New() that internalizes FieldMapper/ConditionBuilder/AggExprRewriter.
  telemetryresourcefilter moves here as statementbuilder/resourcefilter.
- telemetrymetadata — key/value resolution; NewTelemetryMetaStore collapses
  from 24 args to (settings, telemetrystore, flagger), sourcing table names
  from the schema constants.

Centralize query-stack assembly in signoz.go via newQueryStack: build a single
metadata store, the statementbuilder.Builders bundle, and the bucket cache
once, then inject them. This removes the duplicate metadata store that
signozquerier used to build, so signozquerier is now a thin
querier.New(*statementbuilder.Builders) adapter.

Also:
- statementbuilder.Config owns SkipResourceFingerprint (moved off
  querier.Config). YAML key moves querier.skip_resource_fingerprint ->
  statementbuilder.skip_resource_fingerprint.
- Querier interface moves into querier.go (interfaces.go removed); BucketCache
  -> bucket_cache.go, Handler -> api.go.
- Add pkg/querier/queriertest.MockQuerier.
2026-07-28 01:56:30 +05:30
110 changed files with 1168 additions and 1175 deletions

View File

@@ -2,9 +2,9 @@ package main
import (
"github.com/SigNoz/signoz/pkg/metercollector"
"github.com/SigNoz/signoz/pkg/telemetrylogs"
"github.com/SigNoz/signoz/pkg/telemetrymetrics"
"github.com/SigNoz/signoz/pkg/telemetrytraces"
"github.com/SigNoz/signoz/pkg/telemetryschema/logstelemetryschema"
"github.com/SigNoz/signoz/pkg/telemetryschema/metricstelemetryschema"
"github.com/SigNoz/signoz/pkg/telemetryschema/tracestelemetryschema"
"github.com/SigNoz/signoz/pkg/types/retentiontypes"
"github.com/SigNoz/signoz/pkg/types/zeustypes"
)
@@ -25,8 +25,8 @@ var meterConfigs = []metercollector.Config{
Name: zeustypes.MeterLogSize,
Unit: zeustypes.MeterUnitBytes,
Aggregation: zeustypes.MeterAggregationSum,
DBName: telemetrylogs.DBName,
TableName: telemetrylogs.LogsV2LocalTableName,
DBName: logstelemetryschema.DBName,
TableName: logstelemetryschema.LogsV2LocalTableName,
DefaultRetentionDays: retentiontypes.DefaultLogsRetentionDays,
},
},
@@ -36,8 +36,8 @@ var meterConfigs = []metercollector.Config{
Name: zeustypes.MeterSpanSize,
Unit: zeustypes.MeterUnitBytes,
Aggregation: zeustypes.MeterAggregationSum,
DBName: telemetrytraces.DBName,
TableName: telemetrytraces.SpanIndexV3LocalTableName,
DBName: tracestelemetryschema.DBName,
TableName: tracestelemetryschema.SpanIndexV3LocalTableName,
DefaultRetentionDays: retentiontypes.DefaultTracesRetentionDays,
},
},
@@ -47,8 +47,8 @@ var meterConfigs = []metercollector.Config{
Name: zeustypes.MeterDatapointCount,
Unit: zeustypes.MeterUnitCount,
Aggregation: zeustypes.MeterAggregationSum,
DBName: telemetrymetrics.DBName,
TableName: telemetrymetrics.SamplesV4LocalTableName,
DBName: metricstelemetryschema.DBName,
TableName: metricstelemetryschema.SamplesV4LocalTableName,
DefaultRetentionDays: retentiontypes.DefaultMetricsRetentionDays,
},
},

View File

@@ -16,7 +16,7 @@ import (
"github.com/SigNoz/signoz/pkg/factory"
"github.com/SigNoz/signoz/pkg/metercollector"
"github.com/SigNoz/signoz/pkg/modules/retention"
"github.com/SigNoz/signoz/pkg/telemetrymeter"
"github.com/SigNoz/signoz/pkg/telemetryschema/metertelemetryschema"
"github.com/SigNoz/signoz/pkg/telemetrystore"
"github.com/SigNoz/signoz/pkg/types/licensetypes"
"github.com/SigNoz/signoz/pkg/types/retentiontypes"
@@ -153,7 +153,7 @@ func (provider *Provider) Collect(
func buildOriginQuery(meterName string) (string, []any) {
sb := sqlbuilder.NewSelectBuilder()
sb.Select("toInt64(ifNull(min(unix_milli), 0))")
sb.From(telemetrymeter.DBName + "." + telemetrymeter.SamplesTableName)
sb.From(metertelemetryschema.DBName + "." + metertelemetryschema.SamplesTableName)
sb.Where(sb.Equal("metric_name", meterName))
return sb.BuildWithFlavor(sqlbuilder.ClickHouse)
}
@@ -171,7 +171,7 @@ func buildQuery(meterName string, segment *retentiontypes.RetentionPolicySegment
sb := sqlbuilder.NewSelectBuilder()
sb.Select(selects...)
sb.From(telemetrymeter.DBName + "." + telemetrymeter.SamplesTableName)
sb.From(metertelemetryschema.DBName + "." + metertelemetryschema.SamplesTableName)
sb.Where(
sb.Equal("metric_name", meterName),
sb.GTE("unix_milli", segment.StartMs),

View File

@@ -8,16 +8,16 @@ import (
sqlbuilder "github.com/huandu/go-sqlbuilder"
"github.com/SigNoz/signoz/pkg/errors"
"github.com/SigNoz/signoz/pkg/telemetrymetrics"
"github.com/SigNoz/signoz/pkg/telemetryschema/metricstelemetryschema"
"github.com/SigNoz/signoz/pkg/telemetrystore"
"github.com/SigNoz/signoz/pkg/types/ctxtypes"
"github.com/SigNoz/signoz/pkg/types/metricreductionruletypes"
)
var (
reductionRulesTable = telemetrymetrics.DBName + "." + telemetrymetrics.ReductionRulesTableName
metadataTable = telemetrymetrics.DBName + "." + telemetrymetrics.AttributesMetadataTableName
bufferSeriesTable = telemetrymetrics.DBName + "." + telemetrymetrics.TimeseriesV4BufferTableName
reductionRulesTable = metricstelemetryschema.DBName + "." + metricstelemetryschema.ReductionRulesTableName
metadataTable = metricstelemetryschema.DBName + "." + metricstelemetryschema.AttributesMetadataTableName
bufferSeriesTable = metricstelemetryschema.DBName + "." + metricstelemetryschema.TimeseriesV4BufferTableName
)
const timeSeriesBucketMilli = int64(time.Hour / time.Millisecond)
@@ -192,7 +192,7 @@ func (c *clickhouse) ingestedSeriesCount(ctx context.Context, metricNames []stri
sb := sqlbuilder.NewSelectBuilder()
sb.Select("metric_name", "uniq(fingerprint)")
sb.From(telemetrymetrics.DBName + "." + telemetrymetrics.SamplesV4BufferTableName)
sb.From(metricstelemetryschema.DBName + "." + metricstelemetryschema.SamplesV4BufferTableName)
conds := []string{
sb.In("metric_name", names...),
sb.GE("unix_milli", startMs),
@@ -229,8 +229,8 @@ func (c *clickhouse) ingestedSeriesCount(ctx context.Context, metricNames []stri
// reduced sample tables.
func (c *clickhouse) reducedSeriesCount(ctx context.Context, metricNames []string, effectiveFrom map[string]int64, startMs, endMs int64) (map[string]uint64, error) {
out := make(map[string]uint64, len(metricNames))
for _, table := range []string{telemetrymetrics.SamplesV4ReducedLastTableName, telemetrymetrics.SamplesV4ReducedSumTableName} {
counts, err := c.reducedSeriesCountForTable(ctx, telemetrymetrics.DBName+"."+table, metricNames, effectiveFrom, startMs, endMs)
for _, table := range []string{metricstelemetryschema.SamplesV4ReducedLastTableName, metricstelemetryschema.SamplesV4ReducedSumTableName} {
counts, err := c.reducedSeriesCountForTable(ctx, metricstelemetryschema.DBName+"."+table, metricNames, effectiveFrom, startMs, endMs)
if err != nil {
return nil, err
}
@@ -300,9 +300,9 @@ func (c *clickhouse) RankByVolume(ctx context.Context, metricNames []string, eff
direction = "DESC"
}
ingestedTable := telemetrymetrics.DBName + "." + telemetrymetrics.SamplesV4BufferTableName
reducedLast := telemetrymetrics.DBName + "." + telemetrymetrics.SamplesV4ReducedLastTableName
reducedSum := telemetrymetrics.DBName + "." + telemetrymetrics.SamplesV4ReducedSumTableName
ingestedTable := metricstelemetryschema.DBName + "." + metricstelemetryschema.SamplesV4BufferTableName
reducedLast := metricstelemetryschema.DBName + "." + metricstelemetryschema.SamplesV4ReducedLastTableName
reducedSum := metricstelemetryschema.DBName + "." + metricstelemetryschema.SamplesV4ReducedSumTableName
sb := sqlbuilder.NewSelectBuilder()
sb.Select("base.metric_name AS metric_name", "ifNull(i.cnt, 0) AS ingested", "ifNull(d.cnt, 0) AS reduced")
@@ -352,16 +352,16 @@ func (c *clickhouse) SampleVolumeByMetric(ctx context.Context, metricNames []str
}
ctx = c.withThreads(ctx)
ingested, err := c.countSamplesByMetric(ctx, telemetrymetrics.DBName+"."+telemetrymetrics.SamplesV4BufferTableName, "count()", metricNames, effectiveFrom, startMs, endMs)
ingested, err := c.countSamplesByMetric(ctx, metricstelemetryschema.DBName+"."+metricstelemetryschema.SamplesV4BufferTableName, "count()", metricNames, effectiveFrom, startMs, endMs)
if err != nil {
return nil, err
}
last, err := c.countSamplesByMetric(ctx, telemetrymetrics.DBName+"."+telemetrymetrics.SamplesV4ReducedLastTableName, "uniq(reduced_fingerprint, unix_milli)", metricNames, effectiveFrom, startMs, endMs)
last, err := c.countSamplesByMetric(ctx, metricstelemetryschema.DBName+"."+metricstelemetryschema.SamplesV4ReducedLastTableName, "uniq(reduced_fingerprint, unix_milli)", metricNames, effectiveFrom, startMs, endMs)
if err != nil {
return nil, err
}
sum, err := c.countSamplesByMetric(ctx, telemetrymetrics.DBName+"."+telemetrymetrics.SamplesV4ReducedSumTableName, "uniq(reduced_fingerprint, unix_milli)", metricNames, effectiveFrom, startMs, endMs)
sum, err := c.countSamplesByMetric(ctx, metricstelemetryschema.DBName+"."+metricstelemetryschema.SamplesV4ReducedSumTableName, "uniq(reduced_fingerprint, unix_milli)", metricNames, effectiveFrom, startMs, endMs)
if err != nil {
return nil, err
}
@@ -419,7 +419,7 @@ func (c *clickhouse) TotalVolume(ctx context.Context, startMs, endMs int64) (uin
sb := sqlbuilder.NewSelectBuilder()
sb.Select("uniq(fingerprint)", "count()")
sb.From(telemetrymetrics.DBName + "." + telemetrymetrics.SamplesV4BufferTableName)
sb.From(metricstelemetryschema.DBName + "." + metricstelemetryschema.SamplesV4BufferTableName)
sb.Where(sb.GE("unix_milli", startMs), sb.LT("unix_milli", endMs))
query, args := sb.BuildWithFlavor(sqlbuilder.ClickHouse)
@@ -470,7 +470,7 @@ func (c *clickhouse) SampleTimeseries(ctx context.Context, ruledMetrics []string
func (c *clickhouse) totalSamplesByBucket(ctx context.Context, startMs, endMs int64) (map[int64]uint64, error) {
sb := sqlbuilder.NewSelectBuilder()
sb.Select(sampleBucketExpr, "count()")
sb.From(telemetrymetrics.DBName + "." + telemetrymetrics.SamplesV4BufferTableName)
sb.From(metricstelemetryschema.DBName + "." + metricstelemetryschema.SamplesV4BufferTableName)
sb.Where(sb.GE("unix_milli", startMs), sb.LT("unix_milli", endMs))
sb.GroupBy("bucket")
@@ -485,7 +485,7 @@ func (c *clickhouse) ruledIngestedSamplesByBucket(ctx context.Context, metricNam
sb := sqlbuilder.NewSelectBuilder()
sb.Select(sampleBucketExpr, "count()")
sb.From(telemetrymetrics.DBName + "." + telemetrymetrics.SamplesV4BufferTableName)
sb.From(metricstelemetryschema.DBName + "." + metricstelemetryschema.SamplesV4BufferTableName)
conds := []string{sb.In("metric_name", names...), sb.GE("unix_milli", startMs), sb.LT("unix_milli", endMs)}
if len(effectiveFrom) > 0 {
conds = append(conds, strictEffectiveFrom(sb, metricNames, effectiveFrom))
@@ -499,7 +499,7 @@ func (c *clickhouse) ruledIngestedSamplesByBucket(ctx context.Context, metricNam
// reduced 60s rows are versioned by computed_at, so count distinct buckets.
func (c *clickhouse) ruledRetainedSamplesByBucket(ctx context.Context, metricNames []string, effectiveFrom map[string]int64, startMs, endMs int64) (map[int64]uint64, error) {
out := make(map[int64]uint64)
for _, table := range []string{telemetrymetrics.SamplesV4ReducedLastTableName, telemetrymetrics.SamplesV4ReducedSumTableName} {
for _, table := range []string{metricstelemetryschema.SamplesV4ReducedLastTableName, metricstelemetryschema.SamplesV4ReducedSumTableName} {
names := make([]any, len(metricNames))
for i, name := range metricNames {
names[i] = name
@@ -507,7 +507,7 @@ func (c *clickhouse) ruledRetainedSamplesByBucket(ctx context.Context, metricNam
sb := sqlbuilder.NewSelectBuilder()
sb.Select(sampleBucketExpr, "uniq(reduced_fingerprint, unix_milli)")
sb.From(telemetrymetrics.DBName + "." + table)
sb.From(metricstelemetryschema.DBName + "." + table)
conds := []string{sb.In("metric_name", names...), sb.GE("unix_milli", startMs), sb.LT("unix_milli", endMs)}
if len(effectiveFrom) > 0 {
conds = append(conds, strictEffectiveFrom(sb, metricNames, effectiveFrom))

4
go.mod
View File

@@ -4,7 +4,7 @@ go 1.25.7
require (
dario.cat/mergo v1.0.2
github.com/AfterShip/clickhouse-sql-parser v0.5.2
github.com/AfterShip/clickhouse-sql-parser v0.4.16
github.com/ClickHouse/clickhouse-go/v2 v2.44.0
github.com/DATA-DOG/go-sqlmock v1.5.2
github.com/SigNoz/clickhouse-go-mock v0.14.0
@@ -223,7 +223,7 @@ require (
github.com/golang/protobuf v1.5.4 // indirect
github.com/golang/snappy v1.0.0 // indirect
github.com/google/btree v1.1.3 // indirect
github.com/google/cel-go v0.29.0 // indirect
github.com/google/cel-go v0.28.0 // indirect
github.com/google/s2a-go v0.1.9 // indirect
github.com/googleapis/enterprise-certificate-proxy v0.3.14 // indirect
github.com/googleapis/gax-go/v2 v2.18.0 // indirect

8
go.sum
View File

@@ -66,8 +66,8 @@ dario.cat/mergo v1.0.2/go.mod h1:E/hbnu0NxMFBjpMIE34DRGLWqDy0g5FuKDhCb31ngxA=
dmitri.shuralyov.com/gpu/mtl v0.0.0-20190408044501-666a987793e9/go.mod h1:H6x//7gZCb22OMCxBHrMx7a5I7Hp++hsVxbQ4BYO7hU=
filippo.io/edwards25519 v1.2.0 h1:crnVqOiS4jqYleHd9vaKZ+HKtHfllngJIiOpNpoJsjo=
filippo.io/edwards25519 v1.2.0/go.mod h1:xzAOLCNug/yB62zG1bQ8uziwrIqIuxhctzJT18Q77mc=
github.com/AfterShip/clickhouse-sql-parser v0.5.2 h1:KCxdmvuVawVF4yl0ZS33q7olgmLZwvZG5BjWHp6o414=
github.com/AfterShip/clickhouse-sql-parser v0.5.2/go.mod h1:Qi3qvPTfZb/aFwI5V4WFOahgjsLJa4MzVijIAfwOhDw=
github.com/AfterShip/clickhouse-sql-parser v0.4.16 h1:gpl+wXclYUKT0p4+gBq22XeRYWwEoZ9f35vogqMvkLQ=
github.com/AfterShip/clickhouse-sql-parser v0.4.16/go.mod h1:W0Z82wJWkJxz2RVun/RMwxue3g7ut47Xxl+SFqdJGus=
github.com/Azure/azure-sdk-for-go v68.0.0+incompatible h1:fcYLmCpyNYRnvJbPerq7U0hS+6+I79yEDJBqVNcqUzU=
github.com/Azure/azure-sdk-for-go/sdk/azcore v1.21.0 h1:fou+2+WFTib47nS+nz/ozhEBnvU96bKHy6LjRsY4E28=
github.com/Azure/azure-sdk-for-go/sdk/azcore v1.21.0/go.mod h1:t76Ruy8AHvUAC8GfMWJMa0ElSbuIcO03NLpynfbgsPA=
@@ -507,8 +507,8 @@ github.com/google/btree v0.0.0-20180813153112-4030bb1f1f0c/go.mod h1:lNA+9X1NB3Z
github.com/google/btree v1.0.0/go.mod h1:lNA+9X1NB3Zf8V7Ke586lFgjr2dZNuvo3lPJSGZ5JPQ=
github.com/google/btree v1.1.3 h1:CVpQJjYgC4VbzxeGVHfvZrv1ctoYCAI8vbl07Fcxlyg=
github.com/google/btree v1.1.3/go.mod h1:qOPhT0dTNdNzV6Z/lhRX0YXUafgPLFUh+gZMl761Gm4=
github.com/google/cel-go v0.29.0 h1:fEG+Ja3YRwNOqnQxTyJwoByAUAvTuxUGiro/jhrm4F4=
github.com/google/cel-go v0.29.0/go.mod h1:X0bD6iVNR8pkROSOoHVdgTkzmRcosof7WQqCD6wcMc8=
github.com/google/cel-go v0.28.0 h1:KjSWstCpz/MN5t4a8gnGJNIYUsJRpdi/r97xWDphIQc=
github.com/google/cel-go v0.28.0/go.mod h1:X0bD6iVNR8pkROSOoHVdgTkzmRcosof7WQqCD6wcMc8=
github.com/google/gnostic-models v0.7.0 h1:qwTtogB15McXDaNqTZdzPJRHvaVJlAl+HVQnLmJEJxo=
github.com/google/gnostic-models v0.7.0/go.mod h1:whL5G0m6dmc5cPxKc5bdKdEN3UjI7OUGxBlw57miDrQ=
github.com/google/go-cmp v0.2.0/go.mod h1:oXzfMopK8JAjlY9xF4vHSVASa0yLyX7SntLO5aqRK0M=

View File

@@ -5,14 +5,14 @@ import (
"fmt"
"github.com/SigNoz/signoz/pkg/errors"
"github.com/SigNoz/signoz/pkg/telemetrymetrics"
"github.com/SigNoz/signoz/pkg/telemetryschema/metricstelemetryschema"
"github.com/SigNoz/signoz/pkg/valuer"
)
func (m *module) Collect(ctx context.Context, _ valuer.UUID) (map[string]any, error) {
stats := make(map[string]any)
metadataTable := fmt.Sprintf("%s.%s", telemetrymetrics.DBName, telemetrymetrics.AttributesMetadataTableName)
metadataTable := fmt.Sprintf("%s.%s", metricstelemetryschema.DBName, metricstelemetryschema.AttributesMetadataTableName)
var (
systemMetricCount uint64
k8sMetricCount uint64

View File

@@ -7,7 +7,7 @@ import (
"strings"
"github.com/SigNoz/signoz/pkg/querybuilder"
"github.com/SigNoz/signoz/pkg/telemetrymetrics"
"github.com/SigNoz/signoz/pkg/telemetryschema/metricstelemetryschema"
"github.com/SigNoz/signoz/pkg/types/inframonitoringtypes"
qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
"github.com/SigNoz/signoz/pkg/valuer"
@@ -281,7 +281,7 @@ func (m *module) getPerGroupContainerStatusCounts(
mergedFilterExpr := mergeFilterExpressions(userFilterExpr, buildPageGroupsFilterExpr(pageGroups))
samplesStartMs, flooredEndMs, tsAdjustedStart, _, localTimeSeriesTable, distributedSamplesTable, _ := alignedMetricWindow(start, end)
valueCol := telemetrymetrics.ValueColumnForSamplesTable(distributedSamplesTable)
valueCol := metricstelemetryschema.ValueColumnForSamplesTable(distributedSamplesTable)
// Built once; identical across the two fps CTEs (buildFilterClause hits the
// metadata store + parses the expression). AddWhereClause only reads it.
@@ -310,7 +310,7 @@ func (m *module) getPerGroupContainerStatusCounts(
)
}
stateFps.Select(stateFpsCols...)
stateFps.From(fmt.Sprintf("%s.%s", telemetrymetrics.DBName, localTimeSeriesTable))
stateFps.From(fmt.Sprintf("%s.%s", metricstelemetryschema.DBName, localTimeSeriesTable))
stateFps.Where(
stateFps.E("metric_name", containerStatusStateMetricName),
stateFps.GE("unix_milli", tsAdjustedStart),
@@ -342,7 +342,7 @@ func (m *module) getPerGroupContainerStatusCounts(
containerState.Select(containerStateCols...)
containerState.From(fmt.Sprintf(
"%s.%s AS samples INNER JOIN state_fps AS fps ON samples.fingerprint = fps.fingerprint",
telemetrymetrics.DBName, distributedSamplesTable,
metricstelemetryschema.DBName, distributedSamplesTable,
))
containerState.Where(
containerState.E("samples.metric_name", containerStatusStateMetricName),
@@ -361,7 +361,7 @@ func (m *module) getPerGroupContainerStatusCounts(
fmt.Sprintf("JSONExtractString(labels, %s) AS container_name", reasonFps.Var(containerNameAttrKey)),
fmt.Sprintf("JSONExtractString(labels, %s) AS reason", reasonFps.Var(containerStatusReasonAttrKey)),
)
reasonFps.From(fmt.Sprintf("%s.%s", telemetrymetrics.DBName, localTimeSeriesTable))
reasonFps.From(fmt.Sprintf("%s.%s", metricstelemetryschema.DBName, localTimeSeriesTable))
reasonFps.Where(
reasonFps.E("metric_name", containerStatusReasonMetricName),
reasonFps.GE("unix_milli", tsAdjustedStart),
@@ -391,7 +391,7 @@ func (m *module) getPerGroupContainerStatusCounts(
)
reasonInner.From(fmt.Sprintf(
"%s.%s AS samples INNER JOIN reason_fps AS fps ON samples.fingerprint = fps.fingerprint",
telemetrymetrics.DBName, distributedSamplesTable,
metricstelemetryschema.DBName, distributedSamplesTable,
))
reasonInner.Where(
reasonInner.E("samples.metric_name", containerStatusReasonMetricName),
@@ -546,7 +546,7 @@ func (m *module) getPerGroupContainerRestartCounts(
mergedFilterExpr := mergeFilterExpressions(userFilterExpr, buildPageGroupsFilterExpr(pageGroups))
samplesStartMs, flooredEndMs, tsAdjustedStart, _, localTimeSeriesTable, distributedSamplesTable, _ := alignedMetricWindow(start, end)
valueCol := telemetrymetrics.ValueColumnForSamplesTable(distributedSamplesTable)
valueCol := metricstelemetryschema.ValueColumnForSamplesTable(distributedSamplesTable)
var (
filterClause *sqlbuilder.WhereClause
@@ -572,7 +572,7 @@ func (m *module) getPerGroupContainerRestartCounts(
)
}
restartFps.Select(restartFpsCols...)
restartFps.From(fmt.Sprintf("%s.%s", telemetrymetrics.DBName, localTimeSeriesTable))
restartFps.From(fmt.Sprintf("%s.%s", metricstelemetryschema.DBName, localTimeSeriesTable))
restartFps.Where(
restartFps.E("metric_name", containerRestartsMetricName),
restartFps.GE("unix_milli", tsAdjustedStart),
@@ -602,7 +602,7 @@ func (m *module) getPerGroupContainerRestartCounts(
containerRestarts.Select(containerRestartsCols...)
containerRestarts.From(fmt.Sprintf(
"%s.%s AS samples INNER JOIN restart_fps AS fps ON samples.fingerprint = fps.fingerprint",
telemetrymetrics.DBName, distributedSamplesTable,
metricstelemetryschema.DBName, distributedSamplesTable,
))
containerRestarts.Where(
containerRestarts.E("samples.metric_name", containerRestartsMetricName),
@@ -690,7 +690,7 @@ func (m *module) getPerGroupContainerReadyCounts(
mergedFilterExpr := mergeFilterExpressions(userFilterExpr, buildPageGroupsFilterExpr(pageGroups))
samplesStartMs, flooredEndMs, tsAdjustedStart, _, localTimeSeriesTable, distributedSamplesTable, _ := alignedMetricWindow(start, end)
valueCol := telemetrymetrics.ValueColumnForSamplesTable(distributedSamplesTable)
valueCol := metricstelemetryschema.ValueColumnForSamplesTable(distributedSamplesTable)
var (
filterClause *sqlbuilder.WhereClause
@@ -716,7 +716,7 @@ func (m *module) getPerGroupContainerReadyCounts(
)
}
readyFps.Select(readyFpsCols...)
readyFps.From(fmt.Sprintf("%s.%s", telemetrymetrics.DBName, localTimeSeriesTable))
readyFps.From(fmt.Sprintf("%s.%s", metricstelemetryschema.DBName, localTimeSeriesTable))
readyFps.Where(
readyFps.E("metric_name", containerReadyMetricName),
readyFps.GE("unix_milli", tsAdjustedStart),
@@ -746,7 +746,7 @@ func (m *module) getPerGroupContainerReadyCounts(
containerReady.Select(containerReadyCols...)
containerReady.From(fmt.Sprintf(
"%s.%s AS samples INNER JOIN ready_fps AS fps ON samples.fingerprint = fps.fingerprint",
telemetrymetrics.DBName, distributedSamplesTable,
metricstelemetryschema.DBName, distributedSamplesTable,
))
containerReady.Where(
containerReady.E("samples.metric_name", containerReadyMetricName),

View File

@@ -9,7 +9,7 @@ import (
"github.com/SigNoz/signoz/pkg/errors"
"github.com/SigNoz/signoz/pkg/flagger"
"github.com/SigNoz/signoz/pkg/querybuilder"
"github.com/SigNoz/signoz/pkg/telemetrymetrics"
"github.com/SigNoz/signoz/pkg/telemetryschema/metricstelemetryschema"
"github.com/SigNoz/signoz/pkg/types/featuretypes"
"github.com/SigNoz/signoz/pkg/types/inframonitoringtypes"
"github.com/SigNoz/signoz/pkg/types/metrictypes"
@@ -344,11 +344,11 @@ func alignedMetricWindow(startMs, endMs int64) (
flooredEndMs = flooredEndMs - (flooredEndMs % (adjustStep * 1000))
}
tsAdjustedStartMs, _, distributedTSTable, localTSTable := telemetrymetrics.WhichTSTableToUse(
tsAdjustedStartMs, _, distributedTSTable, localTSTable := metricstelemetryschema.WhichTSTableToUse(
samplesAdjustedStartMs, flooredEndMs, false, nil,
)
distributedSamplesTable, localSamplesTable := telemetrymetrics.WhichSamplesTableToUse(
distributedSamplesTable, localSamplesTable := metricstelemetryschema.WhichSamplesTableToUse(
samplesAdjustedStartMs, flooredEndMs,
metrictypes.UnspecifiedType, metrictypes.TimeAggregationUnspecified, false, nil,
)
@@ -362,7 +362,7 @@ func alignedMetricWindow(startMs, endMs int64) (
func (m *module) buildSamplesTblFingerprintSubQuery(metricNames []string, samplesTable string, flooredStart, flooredEnd uint64) *sqlbuilder.SelectBuilder {
fpSB := sqlbuilder.NewSelectBuilder()
fpSB.Select("DISTINCT fingerprint")
fpSB.From(fmt.Sprintf("%s.%s", telemetrymetrics.DBName, samplesTable))
fpSB.From(fmt.Sprintf("%s.%s", metricstelemetryschema.DBName, samplesTable))
fpSB.Where(
fpSB.In("metric_name", sqlbuilder.List(metricNames)),
fpSB.GE("unix_milli", flooredStart),
@@ -376,7 +376,7 @@ func (m *module) buildSamplesTblFingerprintSubQuery(metricNames []string, sample
func (m *module) buildReducedSamplesTblFingerprintSubQuery(metricNames []string, flooredStart, flooredEnd uint64) *sqlbuilder.SelectBuilder {
lastSB := sqlbuilder.NewSelectBuilder()
lastSB.Select("reduced_fingerprint")
lastSB.From(fmt.Sprintf("%s.%s", telemetrymetrics.DBName, telemetrymetrics.SamplesV4ReducedLastTableName))
lastSB.From(fmt.Sprintf("%s.%s", metricstelemetryschema.DBName, metricstelemetryschema.SamplesV4ReducedLastTableName))
lastSB.Where(
lastSB.In("metric_name", sqlbuilder.List(metricNames)),
lastSB.GE("unix_milli", flooredStart),
@@ -385,7 +385,7 @@ func (m *module) buildReducedSamplesTblFingerprintSubQuery(metricNames []string,
sumSB := sqlbuilder.NewSelectBuilder()
sumSB.Select("reduced_fingerprint")
sumSB.From(fmt.Sprintf("%s.%s", telemetrymetrics.DBName, telemetrymetrics.SamplesV4ReducedSumTableName))
sumSB.From(fmt.Sprintf("%s.%s", metricstelemetryschema.DBName, metricstelemetryschema.SamplesV4ReducedSumTableName))
sumSB.Where(
sumSB.In("metric_name", sqlbuilder.List(metricNames)),
sumSB.GE("unix_milli", flooredStart),
@@ -478,7 +478,7 @@ func (m *module) getEarliestMetricTime(ctx context.Context, metricNames []string
sb := sqlbuilder.NewSelectBuilder()
sb.Select("min(first_reported_unix_milli) AS min_first_reported")
sb.From(fmt.Sprintf("%s.%s", telemetrymetrics.DBName, telemetrymetrics.AttributesMetadataTableName))
sb.From(fmt.Sprintf("%s.%s", metricstelemetryschema.DBName, metricstelemetryschema.AttributesMetadataTableName))
sb.Where(sb.In("metric_name", sqlbuilder.List(metricNames)))
query, args := sb.BuildWithFlavor(sqlbuilder.ClickHouse)
@@ -504,7 +504,7 @@ func (m *module) getMetricsExistence(ctx context.Context, metricNames []string)
sb := sqlbuilder.NewSelectBuilder()
sb.Select("metric_name", "count(*) AS cnt")
sb.From(fmt.Sprintf("%s.%s", telemetrymetrics.DBName, telemetrymetrics.AttributesMetadataTableName))
sb.From(fmt.Sprintf("%s.%s", metricstelemetryschema.DBName, metricstelemetryschema.AttributesMetadataTableName))
sb.Where(sb.In("metric_name", sqlbuilder.List(metricNames)))
sb.GroupBy("metric_name")
@@ -549,7 +549,7 @@ func (m *module) getAttributesExistence(ctx context.Context, metricNames, attrNa
}
sb := sqlbuilder.NewSelectBuilder()
sb.Select("attr_name", "count(*) AS cnt")
sb.From(fmt.Sprintf("%s.%s", telemetrymetrics.DBName, telemetrymetrics.AttributesMetadataTableName))
sb.From(fmt.Sprintf("%s.%s", metricstelemetryschema.DBName, metricstelemetryschema.AttributesMetadataTableName))
sb.Where(
sb.In("metric_name", sqlbuilder.List(metricNames)),
sb.In("attr_name", sqlbuilder.List(attrNames)),
@@ -656,7 +656,7 @@ func (m *module) getMetadata(
rawSrc := sqlbuilder.NewSelectBuilder()
rawSrc.Select("labels", "unix_milli")
rawSrc.From(fmt.Sprintf("%s.%s", telemetrymetrics.DBName, distributedTableName))
rawSrc.From(fmt.Sprintf("%s.%s", metricstelemetryschema.DBName, distributedTableName))
rawSrc.Where(
rawSrc.In("metric_name", sqlbuilder.List(metricNames)),
rawSrc.GE("unix_milli", tsAdjustedStartMs),
@@ -669,7 +669,7 @@ func (m *module) getMetadata(
reducedSrc := sqlbuilder.NewSelectBuilder()
reducedSrc.Select("labels", "unix_milli")
reducedSrc.From(fmt.Sprintf("%s.%s", telemetrymetrics.DBName, telemetrymetrics.TimeseriesV4ReducedTableName))
reducedSrc.From(fmt.Sprintf("%s.%s", metricstelemetryschema.DBName, metricstelemetryschema.TimeseriesV4ReducedTableName))
reducedSrc.Where(
reducedSrc.In("metric_name", sqlbuilder.List(metricNames)),
reducedSrc.GE("unix_milli", tsAdjustedStartMs),
@@ -683,7 +683,7 @@ func (m *module) getMetadata(
// Inner query reads over the union of raw + reduced series.
innerSB.From(innerSB.BuilderAs(sqlbuilder.UnionAll(rawSrc, reducedSrc), "series"))
} else {
innerSB.From(fmt.Sprintf("%s.%s", telemetrymetrics.DBName, distributedTableName))
innerSB.From(fmt.Sprintf("%s.%s", metricstelemetryschema.DBName, distributedTableName))
innerSB.Where(
innerSB.In("metric_name", sqlbuilder.List(metricNames)),
innerSB.GE("unix_milli", tsAdjustedStartMs),
@@ -870,7 +870,7 @@ func (m *module) getPerGroupDistinctCounts(
rawSrc := sqlbuilder.NewSelectBuilder()
rawSrc.Select("labels")
rawSrc.From(fmt.Sprintf("%s.%s", telemetrymetrics.DBName, distributedTimeSeriesTbl))
rawSrc.From(fmt.Sprintf("%s.%s", metricstelemetryschema.DBName, distributedTimeSeriesTbl))
rawSrc.Where(
rawSrc.In("metric_name", sqlbuilder.List(metricNames)),
rawSrc.GE("unix_milli", tsAdjustedStartMs),
@@ -883,7 +883,7 @@ func (m *module) getPerGroupDistinctCounts(
reducedSrc := sqlbuilder.NewSelectBuilder()
reducedSrc.Select("labels")
reducedSrc.From(fmt.Sprintf("%s.%s", telemetrymetrics.DBName, telemetrymetrics.TimeseriesV4ReducedTableName))
reducedSrc.From(fmt.Sprintf("%s.%s", metricstelemetryschema.DBName, metricstelemetryschema.TimeseriesV4ReducedTableName))
reducedSrc.Where(
reducedSrc.In("metric_name", sqlbuilder.List(metricNames)),
reducedSrc.GE("unix_milli", tsAdjustedStartMs),
@@ -896,7 +896,7 @@ func (m *module) getPerGroupDistinctCounts(
sb.From(sb.BuilderAs(sqlbuilder.UnionAll(rawSrc, reducedSrc), "series"))
} else {
sb.From(fmt.Sprintf("%s.%s", telemetrymetrics.DBName, distributedTimeSeriesTbl))
sb.From(fmt.Sprintf("%s.%s", metricstelemetryschema.DBName, distributedTimeSeriesTbl))
sb.Where(
sb.In("metric_name", sqlbuilder.List(metricNames)),
sb.GE("unix_milli", tsAdjustedStartMs),

View File

@@ -7,7 +7,7 @@ import (
"strings"
"github.com/SigNoz/signoz/pkg/flagger"
"github.com/SigNoz/signoz/pkg/telemetrymetrics"
"github.com/SigNoz/signoz/pkg/telemetryschema/metricstelemetryschema"
"github.com/SigNoz/signoz/pkg/types/featuretypes"
"github.com/SigNoz/signoz/pkg/types/inframonitoringtypes"
qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
@@ -72,7 +72,7 @@ func (m *module) getPerGroupHostStatusCounts(
rawSrc := sqlbuilder.NewSelectBuilder()
rawSrc.Select("labels")
rawSrc.From(fmt.Sprintf("%s.%s", telemetrymetrics.DBName, distributedTimeSeriesTableName))
rawSrc.From(fmt.Sprintf("%s.%s", metricstelemetryschema.DBName, distributedTimeSeriesTableName))
rawSrc.Where(
rawSrc.In("metric_name", sqlbuilder.List(metricNames)),
rawSrc.GE("unix_milli", tsAdjustedStartMs),
@@ -85,7 +85,7 @@ func (m *module) getPerGroupHostStatusCounts(
reducedSrc := sqlbuilder.NewSelectBuilder()
reducedSrc.Select("labels")
reducedSrc.From(fmt.Sprintf("%s.%s", telemetrymetrics.DBName, telemetrymetrics.TimeseriesV4ReducedTableName))
reducedSrc.From(fmt.Sprintf("%s.%s", metricstelemetryschema.DBName, metricstelemetryschema.TimeseriesV4ReducedTableName))
reducedSrc.Where(
reducedSrc.In("metric_name", sqlbuilder.List(metricNames)),
reducedSrc.GE("unix_milli", tsAdjustedStartMs),
@@ -101,7 +101,7 @@ func (m *module) getPerGroupHostStatusCounts(
fpSB := m.buildSamplesTblFingerprintSubQuery(metricNames, localSamplesTable, samplesStartMs, flooredEndMs)
sb.From(fmt.Sprintf("%s.%s", telemetrymetrics.DBName, distributedTimeSeriesTableName))
sb.From(fmt.Sprintf("%s.%s", metricstelemetryschema.DBName, distributedTimeSeriesTableName))
sb.Where(
sb.In("metric_name", sqlbuilder.List(metricNames)),
sb.GE("unix_milli", tsAdjustedStartMs),
@@ -357,7 +357,7 @@ func (m *module) getActiveHostsQuery(metricNames []string, hostNameAttr string,
sb := sqlbuilder.NewSelectBuilder()
sb.Distinct()
sb.Select("attr_string_value")
sb.From(fmt.Sprintf("%s.%s", telemetrymetrics.DBName, telemetrymetrics.AttributesMetadataTableName))
sb.From(fmt.Sprintf("%s.%s", metricstelemetryschema.DBName, metricstelemetryschema.AttributesMetadataTableName))
sb.Where(
sb.In("metric_name", sqlbuilder.List(metricNames)),
sb.E("attr_name", hostNameAttr),

View File

@@ -9,7 +9,7 @@ import (
"github.com/SigNoz/signoz/pkg/flagger"
"github.com/SigNoz/signoz/pkg/modules/inframonitoring"
"github.com/SigNoz/signoz/pkg/querier"
"github.com/SigNoz/signoz/pkg/telemetrymetrics"
"github.com/SigNoz/signoz/pkg/telemetryschema/metricstelemetryschema"
"github.com/SigNoz/signoz/pkg/telemetrystore"
"github.com/SigNoz/signoz/pkg/types/ctxtypes"
"github.com/SigNoz/signoz/pkg/types/inframonitoringtypes"
@@ -40,8 +40,8 @@ func NewModule(
providerSettings factory.ProviderSettings,
cfg inframonitoring.Config,
) inframonitoring.Module {
fieldMapper := telemetrymetrics.NewFieldMapper()
condBuilder := telemetrymetrics.NewConditionBuilder(fieldMapper)
fieldMapper := metricstelemetryschema.NewFieldMapper()
condBuilder := metricstelemetryschema.NewConditionBuilder(fieldMapper)
return &module{
telemetryStore: telemetryStore,
telemetryMetadataStore: telemetryMetadataStore,

View File

@@ -7,7 +7,7 @@ import (
"strings"
"github.com/SigNoz/signoz/pkg/querybuilder"
"github.com/SigNoz/signoz/pkg/telemetrymetrics"
"github.com/SigNoz/signoz/pkg/telemetryschema/metricstelemetryschema"
"github.com/SigNoz/signoz/pkg/types/inframonitoringtypes"
qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
"github.com/SigNoz/signoz/pkg/valuer"
@@ -186,7 +186,7 @@ func (m *module) getPerGroupNodeConditionCounts(
// Step-floor bounds + resolve tables in one shot to match QB v5 querier.
samplesStartMs, flooredEndMs, tsAdjustedStartMs, _, localTimeSeriesTable, distributedSamplesTable, _ := alignedMetricWindow(start, end)
valueCol := telemetrymetrics.ValueColumnForSamplesTable(distributedSamplesTable)
valueCol := metricstelemetryschema.ValueColumnForSamplesTable(distributedSamplesTable)
// ----- timeSeriesFPs -----
timeSeriesFPs := sqlbuilder.NewSelectBuilder()
@@ -200,7 +200,7 @@ func (m *module) getPerGroupNodeConditionCounts(
)
}
timeSeriesFPs.Select(timeSeriesFPsSelectCols...)
timeSeriesFPs.From(fmt.Sprintf("%s.%s", telemetrymetrics.DBName, localTimeSeriesTable))
timeSeriesFPs.From(fmt.Sprintf("%s.%s", metricstelemetryschema.DBName, localTimeSeriesTable))
timeSeriesFPs.Where(
timeSeriesFPs.E("metric_name", nodeConditionMetricName),
timeSeriesFPs.GE("unix_milli", tsAdjustedStartMs),
@@ -237,7 +237,7 @@ func (m *module) getPerGroupNodeConditionCounts(
latestConditionPerNode.Select(latestConditionPerNodeSelectCols...)
latestConditionPerNode.From(fmt.Sprintf(
"%s.%s AS samples INNER JOIN time_series_fps AS tsfp ON samples.fingerprint = tsfp.fingerprint",
telemetrymetrics.DBName, distributedSamplesTable,
metricstelemetryschema.DBName, distributedSamplesTable,
))
latestConditionPerNode.Where(
latestConditionPerNode.E("samples.metric_name", nodeConditionMetricName),

View File

@@ -8,7 +8,7 @@ import (
"time"
"github.com/SigNoz/signoz/pkg/querybuilder"
"github.com/SigNoz/signoz/pkg/telemetrymetrics"
"github.com/SigNoz/signoz/pkg/telemetryschema/metricstelemetryschema"
"github.com/SigNoz/signoz/pkg/types/inframonitoringtypes"
qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
"github.com/SigNoz/signoz/pkg/valuer"
@@ -284,7 +284,7 @@ func (m *module) getPerGroupPodStatusCounts(
mergedFilterExpr := mergeFilterExpressions(userFilterExpr, buildPageGroupsFilterExpr(pageGroups))
samplesStartMs, flooredEndMs, tsAdjustedStart, _, localTimeSeriesTable, distributedSamplesTable, _ := alignedMetricWindow(start, end)
valueCol := telemetrymetrics.ValueColumnForSamplesTable(distributedSamplesTable)
valueCol := metricstelemetryschema.ValueColumnForSamplesTable(distributedSamplesTable)
// Build the merged filter clause once; it's identical across the three fps
// CTEs, and buildFilterClause hits the metadata store + parses the
@@ -313,7 +313,7 @@ func (m *module) getPerGroupPodStatusCounts(
)
}
phaseFps.Select(phaseFpsCols...)
phaseFps.From(fmt.Sprintf("%s.%s", telemetrymetrics.DBName, localTimeSeriesTable))
phaseFps.From(fmt.Sprintf("%s.%s", metricstelemetryschema.DBName, localTimeSeriesTable))
phaseFps.Where(
phaseFps.E("metric_name", podPhaseMetricName),
phaseFps.GE("unix_milli", tsAdjustedStart),
@@ -340,7 +340,7 @@ func (m *module) getPerGroupPodStatusCounts(
phasePerPod.Select(phasePerPodCols...)
phasePerPod.From(fmt.Sprintf(
"%s.%s AS samples INNER JOIN phase_fps AS fps ON samples.fingerprint = fps.fingerprint",
telemetrymetrics.DBName, distributedSamplesTable,
metricstelemetryschema.DBName, distributedSamplesTable,
))
phasePerPod.Where(
phasePerPod.E("samples.metric_name", podPhaseMetricName),
@@ -357,7 +357,7 @@ func (m *module) getPerGroupPodStatusCounts(
"fingerprint",
fmt.Sprintf("JSONExtractString(labels, %s) AS pod_uid", podReasonFps.Var(podUIDAttrKey)),
)
podReasonFps.From(fmt.Sprintf("%s.%s", telemetrymetrics.DBName, localTimeSeriesTable))
podReasonFps.From(fmt.Sprintf("%s.%s", metricstelemetryschema.DBName, localTimeSeriesTable))
podReasonFps.Where(
podReasonFps.E("metric_name", podStatusReasonMetricName),
podReasonFps.GE("unix_milli", tsAdjustedStart),
@@ -377,7 +377,7 @@ func (m *module) getPerGroupPodStatusCounts(
)
podReasonPerPod.From(fmt.Sprintf(
"%s.%s AS samples INNER JOIN pod_reason_fps AS fps ON samples.fingerprint = fps.fingerprint",
telemetrymetrics.DBName, distributedSamplesTable,
metricstelemetryschema.DBName, distributedSamplesTable,
))
podReasonPerPod.Where(
podReasonPerPod.E("samples.metric_name", podStatusReasonMetricName),
@@ -396,7 +396,7 @@ func (m *module) getPerGroupPodStatusCounts(
fmt.Sprintf("JSONExtractString(labels, %s) AS container_name", containerReasonFps.Var(containerNameAttrKey)),
fmt.Sprintf("JSONExtractString(labels, %s) AS reason", containerReasonFps.Var(containerStatusReasonAttrKey)),
)
containerReasonFps.From(fmt.Sprintf("%s.%s", telemetrymetrics.DBName, localTimeSeriesTable))
containerReasonFps.From(fmt.Sprintf("%s.%s", metricstelemetryschema.DBName, localTimeSeriesTable))
containerReasonFps.Where(
containerReasonFps.E("metric_name", containerStatusReasonMetricName),
containerReasonFps.GE("unix_milli", tsAdjustedStart),
@@ -432,7 +432,7 @@ func (m *module) getPerGroupPodStatusCounts(
)
containerInner.From(fmt.Sprintf(
"%s.%s AS samples INNER JOIN container_reason_fps AS fps ON samples.fingerprint = fps.fingerprint",
telemetrymetrics.DBName, distributedSamplesTable,
metricstelemetryschema.DBName, distributedSamplesTable,
))
containerInner.Where(
containerInner.E("samples.metric_name", containerStatusReasonMetricName),
@@ -611,7 +611,7 @@ func (m *module) getPerGroupPodRestartCounts(
mergedFilterExpr := mergeFilterExpressions(userFilterExpr, buildPageGroupsFilterExpr(pageGroups))
samplesStartMs, flooredEndMs, tsAdjustedStart, _, localTimeSeriesTable, distributedSamplesTable, _ := alignedMetricWindow(start, end)
valueCol := telemetrymetrics.ValueColumnForSamplesTable(distributedSamplesTable)
valueCol := metricstelemetryschema.ValueColumnForSamplesTable(distributedSamplesTable)
var (
filterClause *sqlbuilder.WhereClause
@@ -637,7 +637,7 @@ func (m *module) getPerGroupPodRestartCounts(
)
}
restartFps.Select(restartFpsCols...)
restartFps.From(fmt.Sprintf("%s.%s", telemetrymetrics.DBName, localTimeSeriesTable))
restartFps.From(fmt.Sprintf("%s.%s", metricstelemetryschema.DBName, localTimeSeriesTable))
restartFps.Where(
restartFps.E("metric_name", containerRestartsMetricName),
restartFps.GE("unix_milli", tsAdjustedStart),
@@ -667,7 +667,7 @@ func (m *module) getPerGroupPodRestartCounts(
containerRestarts.Select(containerRestartsCols...)
containerRestarts.From(fmt.Sprintf(
"%s.%s AS samples INNER JOIN restart_fps AS fps ON samples.fingerprint = fps.fingerprint",
telemetrymetrics.DBName, distributedSamplesTable,
metricstelemetryschema.DBName, distributedSamplesTable,
))
containerRestarts.Where(
containerRestarts.E("samples.metric_name", containerRestartsMetricName),

View File

@@ -19,8 +19,8 @@ import (
"github.com/SigNoz/signoz/pkg/modules/dashboard"
"github.com/SigNoz/signoz/pkg/modules/metricsexplorer"
"github.com/SigNoz/signoz/pkg/querybuilder"
"github.com/SigNoz/signoz/pkg/telemetrymeter"
"github.com/SigNoz/signoz/pkg/telemetrymetrics"
"github.com/SigNoz/signoz/pkg/telemetryschema/metertelemetryschema"
"github.com/SigNoz/signoz/pkg/telemetryschema/metricstelemetryschema"
"github.com/SigNoz/signoz/pkg/telemetrystore"
"github.com/SigNoz/signoz/pkg/types/ctxtypes"
"github.com/SigNoz/signoz/pkg/types/dashboardtypes"
@@ -49,8 +49,8 @@ type module struct {
// NewModule constructs the metrics module with the provided dependencies.
func NewModule(ts telemetrystore.TelemetryStore, telemetryMetadataStore telemetrytypes.MetadataStore, cache cache.Cache, ruleStore ruletypes.RuleStore, dashboardModule dashboard.Module, fl flagger.Flagger, providerSettings factory.ProviderSettings, cfg metricsexplorer.Config) metricsexplorer.Module {
fieldMapper := telemetrymetrics.NewFieldMapper()
condBuilder := telemetrymetrics.NewConditionBuilder(fieldMapper)
fieldMapper := metricstelemetryschema.NewFieldMapper()
condBuilder := metricstelemetryschema.NewConditionBuilder(fieldMapper)
return &module{
telemetryStore: ts,
fieldMapper: fieldMapper,
@@ -89,7 +89,7 @@ func (m *module) listMeterMetrics(ctx context.Context, params *metricsexplorerty
"argMax(temporality, unix_milli) AS temporality",
"argMax(is_monotonic, unix_milli) AS is_monotonic",
)
sb.From(fmt.Sprintf("%s.%s", telemetrymeter.DBName, telemetrymeter.SamplesTableName))
sb.From(fmt.Sprintf("%s.%s", metertelemetryschema.DBName, metertelemetryschema.SamplesTableName))
if params.Start != nil && params.End != nil {
sb.Where(sb.Between("unix_milli", *params.Start, *params.End))
@@ -160,11 +160,11 @@ func (m *module) listMetrics(ctx context.Context, orgID valuer.UUID, params *met
hasRange := params.Start != nil && params.End != nil
if hasRange {
var distributedTsTable string
rangeStart, rangeEnd, distributedTsTable, _ = telemetrymetrics.WhichTSTableToUse(uint64(*params.Start), uint64(*params.End), false, nil)
rawSB.From(fmt.Sprintf("%s.%s", telemetrymetrics.DBName, distributedTsTable))
rangeStart, rangeEnd, distributedTsTable, _ = metricstelemetryschema.WhichTSTableToUse(uint64(*params.Start), uint64(*params.End), false, nil)
rawSB.From(fmt.Sprintf("%s.%s", metricstelemetryschema.DBName, distributedTsTable))
rawSB.Where(rawSB.Between("unix_milli", rangeStart, rangeEnd))
} else {
rawSB.From(fmt.Sprintf("%s.%s", telemetrymetrics.DBName, telemetrymetrics.TimeseriesV41weekTableName))
rawSB.From(fmt.Sprintf("%s.%s", metricstelemetryschema.DBName, metricstelemetryschema.TimeseriesV41weekTableName))
}
applyListFilters(rawSB)
@@ -174,7 +174,7 @@ func (m *module) listMetrics(ctx context.Context, orgID valuer.UUID, params *met
if reductionEnabled {
reducedSB := sqlbuilder.NewSelectBuilder()
reducedSB.Select("DISTINCT metric_name")
reducedSB.From(fmt.Sprintf("%s.%s", telemetrymetrics.DBName, telemetrymetrics.TimeseriesV4ReducedTableName))
reducedSB.From(fmt.Sprintf("%s.%s", metricstelemetryschema.DBName, metricstelemetryschema.TimeseriesV4ReducedTableName))
if hasRange {
reducedSB.Where(reducedSB.Between("unix_milli", rangeStart, rangeEnd))
}
@@ -512,7 +512,7 @@ func (m *module) CheckMetricExists(ctx context.Context, orgID valuer.UUID, metri
sb := sqlbuilder.NewSelectBuilder()
sb.Select("count(*) > 0 as metricExists")
sb.From(fmt.Sprintf("%s.%s", telemetrymetrics.DBName, telemetrymetrics.AttributesMetadataTableName))
sb.From(fmt.Sprintf("%s.%s", metricstelemetryschema.DBName, metricstelemetryschema.AttributesMetadataTableName))
sb.Where(sb.E("metric_name", metricName))
query, args := sb.BuildWithFlavor(sqlbuilder.ClickHouse)
@@ -534,7 +534,7 @@ func (m *module) HasNonSigNozMetrics(ctx context.Context) (bool, error) {
sb := sqlbuilder.NewSelectBuilder()
sb.Select("count(*) > 0")
sb.From(fmt.Sprintf("%s.%s", telemetrymetrics.DBName, telemetrymetrics.TimeseriesV41weekTableName))
sb.From(fmt.Sprintf("%s.%s", metricstelemetryschema.DBName, metricstelemetryschema.TimeseriesV41weekTableName))
sb.Where("metric_name NOT LIKE 'signoz_%'")
query, args := sb.BuildWithFlavor(sqlbuilder.ClickHouse)
@@ -570,10 +570,10 @@ func (m *module) InspectMetrics(
return nil, err
}
tsStart, _, tsTable, _ := telemetrymetrics.WhichTSTableToUse(start, end, false, nil)
tsStart, _, tsTable, _ := metricstelemetryschema.WhichTSTableToUse(start, end, false, nil)
tsSb := sqlbuilder.NewSelectBuilder()
tsSb.Select("fingerprint", "labels")
tsSb.From(fmt.Sprintf("%s.%s", telemetrymetrics.DBName, tsTable))
tsSb.From(fmt.Sprintf("%s.%s", metricstelemetryschema.DBName, tsTable))
tsSb.Where(
tsSb.E("metric_name", req.MetricName),
tsSb.GE("unix_milli", tsStart),
@@ -625,7 +625,7 @@ func (m *module) InspectMetrics(
samplesSb := sqlbuilder.NewSelectBuilder()
samplesSb.Select("fingerprint", "unix_milli", "value")
samplesSb.From(fmt.Sprintf("%s.%s", telemetrymetrics.DBName, telemetrymetrics.SamplesV4TableName))
samplesSb.From(fmt.Sprintf("%s.%s", metricstelemetryschema.DBName, metricstelemetryschema.SamplesV4TableName))
samplesSb.Where(
samplesSb.In("fingerprint", fingerprints...),
samplesSb.E("metric_name", req.MetricName),
@@ -717,7 +717,7 @@ func (m *module) fetchUpdatedMetadata(ctx context.Context, orgID valuer.UUID, me
"argMax(temporality, created_at) AS temporality",
"argMax(is_monotonic, created_at) AS is_monotonic",
)
sb.From(fmt.Sprintf("%s.%s", telemetrymetrics.DBName, telemetrymetrics.UpdatedMetadataTableName))
sb.From(fmt.Sprintf("%s.%s", metricstelemetryschema.DBName, metricstelemetryschema.UpdatedMetadataTableName))
sb.Where(sb.In("metric_name", args...))
sb.GroupBy("metric_name")
@@ -777,7 +777,7 @@ func (m *module) fetchTimeseriesMetadata(ctx context.Context, orgID valuer.UUID,
"anyLast(temporality) AS temporality",
"argMax(is_monotonic, unix_milli) AS is_monotonic",
)
sb.From(fmt.Sprintf("%s.%s", telemetrymetrics.DBName, telemetrymetrics.TimeseriesV4TableName))
sb.From(fmt.Sprintf("%s.%s", metricstelemetryschema.DBName, metricstelemetryschema.TimeseriesV4TableName))
sb.Where(sb.In("metric_name", args...))
sb.GroupBy("metric_name")
@@ -891,7 +891,7 @@ func (m *module) checkForLabelInMetric(ctx context.Context, metricName string, l
sb := sqlbuilder.NewSelectBuilder()
sb.Select("count(*) > 0 AS has_label")
sb.From(fmt.Sprintf("%s.%s", telemetrymetrics.DBName, telemetrymetrics.AttributesMetadataTableName))
sb.From(fmt.Sprintf("%s.%s", metricstelemetryschema.DBName, metricstelemetryschema.AttributesMetadataTableName))
sb.Where(sb.E("metric_name", metricName))
sb.Where(sb.E("attr_name", label))
sb.Limit(1)
@@ -914,7 +914,7 @@ func (m *module) insertMetricsMetadata(ctx context.Context, orgID valuer.UUID, r
createdAt := time.Now().UnixMilli()
ib := sqlbuilder.NewInsertBuilder()
ib.InsertInto(fmt.Sprintf("%s.%s", telemetrymetrics.DBName, telemetrymetrics.UpdatedMetadataTableName))
ib.InsertInto(fmt.Sprintf("%s.%s", metricstelemetryschema.DBName, metricstelemetryschema.UpdatedMetadataTableName))
ib.Cols("metric_name", "temporality", "is_monotonic", "type", "description", "unit", "created_at")
ib.Values(
req.MetricName,
@@ -1016,9 +1016,9 @@ func (m *module) fetchMetricsStatsWithSamples(
}
}
start, end, distributedTsTable, localTsTable := telemetrymetrics.WhichTSTableToUse(uint64(req.Start), uint64(req.End), false, nil)
distributedSamplesTable, _ := telemetrymetrics.WhichSamplesTableToUse(uint64(req.Start), uint64(req.End), metrictypes.UnspecifiedType, metrictypes.TimeAggregationUnspecified, false, nil)
countExp := telemetrymetrics.CountExpressionForSamplesTable(distributedSamplesTable)
start, end, distributedTsTable, localTsTable := metricstelemetryschema.WhichTSTableToUse(uint64(req.Start), uint64(req.End), false, nil)
distributedSamplesTable, _ := metricstelemetryschema.WhichSamplesTableToUse(uint64(req.Start), uint64(req.End), metrictypes.UnspecifiedType, metrictypes.TimeAggregationUnspecified, false, nil)
countExp := metricstelemetryschema.CountExpressionForSamplesTable(distributedSamplesTable)
// Timeseries counts per metric
tsSB := sqlbuilder.NewSelectBuilder()
@@ -1026,7 +1026,7 @@ func (m *module) fetchMetricsStatsWithSamples(
"metric_name",
"uniq(fingerprint) AS timeseries",
)
tsSB.From(fmt.Sprintf("%s.%s", telemetrymetrics.DBName, distributedTsTable))
tsSB.From(fmt.Sprintf("%s.%s", metricstelemetryschema.DBName, distributedTsTable))
tsSB.Where(tsSB.Between("unix_milli", start, end))
tsSB.Where("NOT startsWith(metric_name, 'signoz')")
if filterWhereClause != nil {
@@ -1040,7 +1040,7 @@ func (m *module) fetchMetricsStatsWithSamples(
"metric_name",
fmt.Sprintf("%s AS samples", countExp),
)
samplesSB.From(fmt.Sprintf("%s.%s", telemetrymetrics.DBName, distributedSamplesTable))
samplesSB.From(fmt.Sprintf("%s.%s", metricstelemetryschema.DBName, distributedSamplesTable))
samplesSB.Where(samplesSB.Between("unix_milli", req.Start, req.End))
samplesSB.Where("NOT startsWith(metric_name, 'signoz')")
@@ -1053,7 +1053,7 @@ func (m *module) fetchMetricsStatsWithSamples(
if filterWhereClause != nil {
fingerprintSB := sqlbuilder.NewSelectBuilder()
fingerprintSB.Select("fingerprint")
fingerprintSB.From(fmt.Sprintf("%s.%s", telemetrymetrics.DBName, localTsTable))
fingerprintSB.From(fmt.Sprintf("%s.%s", metricstelemetryschema.DBName, localTsTable))
fingerprintSB.Where(fingerprintSB.Between("unix_milli", start, end))
fingerprintSB.Where("NOT startsWith(metric_name, 'signoz')")
fingerprintSB.AddWhereClause(sqlbuilder.CopyWhereClause(filterWhereClause))
@@ -1064,7 +1064,7 @@ func (m *module) fetchMetricsStatsWithSamples(
} else {
metricNamesSB := sqlbuilder.NewSelectBuilder()
metricNamesSB.Select("DISTINCT metric_name")
metricNamesSB.From(fmt.Sprintf("%s.%s", telemetrymetrics.DBName, localTsTable))
metricNamesSB.From(fmt.Sprintf("%s.%s", metricstelemetryschema.DBName, localTsTable))
metricNamesSB.Where(metricNamesSB.Between("unix_milli", start, end))
metricNamesSB.Where("NOT startsWith(metric_name, 'signoz')")
@@ -1090,18 +1090,18 @@ func (m *module) fetchMetricsStatsWithSamples(
// the data lands either of the _reduced_last/_reduced_sum tables
reducedLastSB := sqlbuilder.NewSelectBuilder()
reducedLastSB.Select("metric_name", "uniq(reduced_fingerprint, unix_milli) AS cnt")
reducedLastSB.From(fmt.Sprintf("%s.%s", telemetrymetrics.DBName, telemetrymetrics.SamplesV4ReducedLastTableName))
reducedLastSB.From(fmt.Sprintf("%s.%s", metricstelemetryschema.DBName, metricstelemetryschema.SamplesV4ReducedLastTableName))
reducedLastSB.Where(reducedLastSB.Between("unix_milli", req.Start, req.End))
reducedLastSB.Where("NOT startsWith(metric_name, 'signoz')")
reducedSumSB := sqlbuilder.NewSelectBuilder()
reducedSumSB.Select("metric_name", "uniq(reduced_fingerprint, unix_milli) AS cnt")
reducedSumSB.From(fmt.Sprintf("%s.%s", telemetrymetrics.DBName, telemetrymetrics.SamplesV4ReducedSumTableName))
reducedSumSB.From(fmt.Sprintf("%s.%s", metricstelemetryschema.DBName, metricstelemetryschema.SamplesV4ReducedSumTableName))
reducedSumSB.Where(reducedSumSB.Between("unix_milli", req.Start, req.End))
reducedSumSB.Where("NOT startsWith(metric_name, 'signoz')")
// separate query for reduced series counts
reducedTsSB.From(fmt.Sprintf("%s.%s", telemetrymetrics.DBName, telemetrymetrics.TimeseriesV4ReducedTableName))
reducedTsSB.From(fmt.Sprintf("%s.%s", metricstelemetryschema.DBName, metricstelemetryschema.TimeseriesV4ReducedTableName))
reducedTsSB.Where(reducedTsSB.Between("unix_milli", start, end))
reducedTsSB.Where("NOT startsWith(metric_name, 'signoz')")
reducedTsSB.GroupBy("metric_name")
@@ -1112,7 +1112,7 @@ func (m *module) fetchMetricsStatsWithSamples(
// samples uses a separate cte with local table
reducedFpSB := sqlbuilder.NewSelectBuilder()
reducedFpSB.Select("fingerprint")
reducedFpSB.From(fmt.Sprintf("%s.%s", telemetrymetrics.DBName, telemetrymetrics.TimeseriesV4ReducedLocalTableName))
reducedFpSB.From(fmt.Sprintf("%s.%s", metricstelemetryschema.DBName, metricstelemetryschema.TimeseriesV4ReducedLocalTableName))
reducedFpSB.Where(reducedFpSB.Between("unix_milli", start, end))
reducedFpSB.Where("NOT startsWith(metric_name, 'signoz')")
reducedFpSB.AddWhereClause(sqlbuilder.CopyWhereClause(filterWhereClause))
@@ -1220,11 +1220,11 @@ func (m *module) computeTimeseriesTreemap(ctx context.Context, orgID valuer.UUID
}
}
start, end, distributedTsTable, _ := telemetrymetrics.WhichTSTableToUse(uint64(req.Start), uint64(req.End), false, nil)
start, end, distributedTsTable, _ := metricstelemetryschema.WhichTSTableToUse(uint64(req.Start), uint64(req.End), false, nil)
totalTSBuilder := sqlbuilder.NewSelectBuilder()
totalTSBuilder.Select("uniq(fingerprint) AS total_time_series")
totalTSBuilder.From(fmt.Sprintf("%s.%s", telemetrymetrics.DBName, distributedTsTable))
totalTSBuilder.From(fmt.Sprintf("%s.%s", metricstelemetryschema.DBName, distributedTsTable))
totalTSBuilder.Where(totalTSBuilder.Between("unix_milli", start, end))
metricsSB := sqlbuilder.NewSelectBuilder()
@@ -1232,7 +1232,7 @@ func (m *module) computeTimeseriesTreemap(ctx context.Context, orgID valuer.UUID
"metric_name",
"uniq(fingerprint) AS total_value",
)
metricsSB.From(fmt.Sprintf("%s.%s", telemetrymetrics.DBName, distributedTsTable))
metricsSB.From(fmt.Sprintf("%s.%s", metricstelemetryschema.DBName, distributedTsTable))
metricsSB.Where(metricsSB.Between("unix_milli", start, end))
metricsSB.Where("NOT startsWith(metric_name, 'signoz')")
if filterWhereClause != nil {
@@ -1244,7 +1244,7 @@ func (m *module) computeTimeseriesTreemap(ctx context.Context, orgID valuer.UUID
if reductionEnabled {
reducedTotalTSBuilder := sqlbuilder.NewSelectBuilder()
reducedTotalTSBuilder.Select("uniq(fingerprint) AS total_time_series")
reducedTotalTSBuilder.From(fmt.Sprintf("%s.%s", telemetrymetrics.DBName, telemetrymetrics.TimeseriesV4ReducedTableName))
reducedTotalTSBuilder.From(fmt.Sprintf("%s.%s", metricstelemetryschema.DBName, metricstelemetryschema.TimeseriesV4ReducedTableName))
reducedTotalTSBuilder.Where(reducedTotalTSBuilder.Between("unix_milli", start, end))
reducedMetricsSB := sqlbuilder.NewSelectBuilder()
@@ -1252,7 +1252,7 @@ func (m *module) computeTimeseriesTreemap(ctx context.Context, orgID valuer.UUID
"metric_name",
"uniq(fingerprint) AS total_value",
)
reducedMetricsSB.From(fmt.Sprintf("%s.%s", telemetrymetrics.DBName, telemetrymetrics.TimeseriesV4ReducedTableName))
reducedMetricsSB.From(fmt.Sprintf("%s.%s", metricstelemetryschema.DBName, metricstelemetryschema.TimeseriesV4ReducedTableName))
reducedMetricsSB.Where(reducedMetricsSB.Between("unix_milli", start, end))
reducedMetricsSB.Where("NOT startsWith(metric_name, 'signoz')")
if filterWhereClause != nil {
@@ -1338,15 +1338,15 @@ func (m *module) computeSamplesTreemap(ctx context.Context, orgID valuer.UUID, r
}
}
start, end, distributedTsTable, localTsTable := telemetrymetrics.WhichTSTableToUse(uint64(req.Start), uint64(req.End), false, nil)
distributedSamplesTable, _ := telemetrymetrics.WhichSamplesTableToUse(uint64(req.Start), uint64(req.End), metrictypes.UnspecifiedType, metrictypes.TimeAggregationUnspecified, false, nil)
countExp := telemetrymetrics.CountExpressionForSamplesTable(distributedSamplesTable)
start, end, distributedTsTable, localTsTable := metricstelemetryschema.WhichTSTableToUse(uint64(req.Start), uint64(req.End), false, nil)
distributedSamplesTable, _ := metricstelemetryschema.WhichSamplesTableToUse(uint64(req.Start), uint64(req.End), metrictypes.UnspecifiedType, metrictypes.TimeAggregationUnspecified, false, nil)
countExp := metricstelemetryschema.CountExpressionForSamplesTable(distributedSamplesTable)
candidateLimit := req.Limit + 50
metricCandidatesSB := sqlbuilder.NewSelectBuilder()
metricCandidatesSB.Select("metric_name")
metricCandidatesSB.From(fmt.Sprintf("%s.%s", telemetrymetrics.DBName, distributedTsTable))
metricCandidatesSB.From(fmt.Sprintf("%s.%s", metricstelemetryschema.DBName, distributedTsTable))
metricCandidatesSB.Where("NOT startsWith(metric_name, 'signoz')")
metricCandidatesSB.Where(metricCandidatesSB.Between("unix_milli", start, end))
if filterWhereClause != nil {
@@ -1364,7 +1364,7 @@ func (m *module) computeSamplesTreemap(ctx context.Context, orgID valuer.UUID, r
if reductionEnabled {
reducedCandidatesSB = sqlbuilder.NewSelectBuilder()
reducedCandidatesSB.Select("metric_name")
reducedCandidatesSB.From(fmt.Sprintf("%s.%s", telemetrymetrics.DBName, telemetrymetrics.TimeseriesV4ReducedTableName))
reducedCandidatesSB.From(fmt.Sprintf("%s.%s", metricstelemetryschema.DBName, metricstelemetryschema.TimeseriesV4ReducedTableName))
reducedCandidatesSB.Where("NOT startsWith(metric_name, 'signoz')")
reducedCandidatesSB.Where(reducedCandidatesSB.Between("unix_milli", start, end))
if filterWhereClause != nil {
@@ -1379,7 +1379,7 @@ func (m *module) computeSamplesTreemap(ctx context.Context, orgID valuer.UUID, r
totalSamplesSB := sqlbuilder.NewSelectBuilder()
totalSamplesSB.Select(fmt.Sprintf("%s AS total_samples", countExp))
totalSamplesSB.From(fmt.Sprintf("%s.%s", telemetrymetrics.DBName, distributedSamplesTable))
totalSamplesSB.From(fmt.Sprintf("%s.%s", metricstelemetryschema.DBName, distributedSamplesTable))
totalSamplesSB.Where(totalSamplesSB.Between("unix_milli", req.Start, req.End))
sampleCountsSB := sqlbuilder.NewSelectBuilder()
@@ -1387,7 +1387,7 @@ func (m *module) computeSamplesTreemap(ctx context.Context, orgID valuer.UUID, r
"metric_name",
fmt.Sprintf("%s AS samples", countExp),
)
sampleCountsSB.From(fmt.Sprintf("%s.%s", telemetrymetrics.DBName, distributedSamplesTable))
sampleCountsSB.From(fmt.Sprintf("%s.%s", metricstelemetryschema.DBName, distributedSamplesTable))
sampleCountsSB.Where(sampleCountsSB.Between("unix_milli", req.Start, req.End))
sampleCountsSB.Where("metric_name GLOBAL IN (SELECT metric_name FROM __metric_candidates)")
@@ -1397,13 +1397,13 @@ func (m *module) computeSamplesTreemap(ctx context.Context, orgID valuer.UUID, r
if reductionEnabled {
reducedLastSB = sqlbuilder.NewSelectBuilder()
reducedLastSB.Select("metric_name", "uniq(reduced_fingerprint, unix_milli) AS cnt")
reducedLastSB.From(fmt.Sprintf("%s.%s", telemetrymetrics.DBName, telemetrymetrics.SamplesV4ReducedLastTableName))
reducedLastSB.From(fmt.Sprintf("%s.%s", metricstelemetryschema.DBName, metricstelemetryschema.SamplesV4ReducedLastTableName))
reducedLastSB.Where(reducedLastSB.Between("unix_milli", req.Start, req.End))
reducedLastSB.Where("metric_name GLOBAL IN (SELECT metric_name FROM __reduced_metric_candidates)")
reducedSumSB = sqlbuilder.NewSelectBuilder()
reducedSumSB.Select("metric_name", "uniq(reduced_fingerprint, unix_milli) AS cnt")
reducedSumSB.From(fmt.Sprintf("%s.%s", telemetrymetrics.DBName, telemetrymetrics.SamplesV4ReducedSumTableName))
reducedSumSB.From(fmt.Sprintf("%s.%s", metricstelemetryschema.DBName, metricstelemetryschema.SamplesV4ReducedSumTableName))
reducedSumSB.Where(reducedSumSB.Between("unix_milli", req.Start, req.End))
reducedSumSB.Where("metric_name GLOBAL IN (SELECT metric_name FROM __reduced_metric_candidates)")
}
@@ -1411,7 +1411,7 @@ func (m *module) computeSamplesTreemap(ctx context.Context, orgID valuer.UUID, r
if filterWhereClause != nil {
fingerprintSB := sqlbuilder.NewSelectBuilder()
fingerprintSB.Select("fingerprint")
fingerprintSB.From(fmt.Sprintf("%s.%s", telemetrymetrics.DBName, localTsTable))
fingerprintSB.From(fmt.Sprintf("%s.%s", metricstelemetryschema.DBName, localTsTable))
fingerprintSB.Where(fingerprintSB.Between("unix_milli", start, end))
fingerprintSB.Where("NOT startsWith(metric_name, 'signoz')")
fingerprintSB.AddWhereClause(sqlbuilder.CopyWhereClause(filterWhereClause))
@@ -1423,7 +1423,7 @@ func (m *module) computeSamplesTreemap(ctx context.Context, orgID valuer.UUID, r
if reductionEnabled {
reducedFingerprintSB := sqlbuilder.NewSelectBuilder()
reducedFingerprintSB.Select("fingerprint")
reducedFingerprintSB.From(fmt.Sprintf("%s.%s", telemetrymetrics.DBName, telemetrymetrics.TimeseriesV4ReducedLocalTableName))
reducedFingerprintSB.From(fmt.Sprintf("%s.%s", metricstelemetryschema.DBName, metricstelemetryschema.TimeseriesV4ReducedLocalTableName))
reducedFingerprintSB.Where(reducedFingerprintSB.Between("unix_milli", start, end))
reducedFingerprintSB.Where("NOT startsWith(metric_name, 'signoz')")
reducedFingerprintSB.AddWhereClause(sqlbuilder.CopyWhereClause(filterWhereClause))
@@ -1457,12 +1457,12 @@ func (m *module) computeSamplesTreemap(ctx context.Context, orgID valuer.UUID, r
// Grand total includes reduced sample volume so percentages stay consistent.
reducedTotalLastSB := sqlbuilder.NewSelectBuilder()
reducedTotalLastSB.Select("uniq(reduced_fingerprint, unix_milli) AS cnt")
reducedTotalLastSB.From(fmt.Sprintf("%s.%s", telemetrymetrics.DBName, telemetrymetrics.SamplesV4ReducedLastTableName))
reducedTotalLastSB.From(fmt.Sprintf("%s.%s", metricstelemetryschema.DBName, metricstelemetryschema.SamplesV4ReducedLastTableName))
reducedTotalLastSB.Where(reducedTotalLastSB.Between("unix_milli", req.Start, req.End))
reducedTotalSumSB := sqlbuilder.NewSelectBuilder()
reducedTotalSumSB.Select("uniq(reduced_fingerprint, unix_milli) AS cnt")
reducedTotalSumSB.From(fmt.Sprintf("%s.%s", telemetrymetrics.DBName, telemetrymetrics.SamplesV4ReducedSumTableName))
reducedTotalSumSB.From(fmt.Sprintf("%s.%s", metricstelemetryschema.DBName, metricstelemetryschema.SamplesV4ReducedSumTableName))
reducedTotalSumSB.Where(reducedTotalSumSB.Between("unix_milli", req.Start, req.End))
reducedTotalSamplesSB := sqlbuilder.NewSelectBuilder()
@@ -1552,7 +1552,7 @@ func (m *module) getMetricDataPoints(ctx context.Context, metricName string) (ui
sb := sqlbuilder.NewSelectBuilder()
sb.Select("sum(count) AS data_points")
sb.From(fmt.Sprintf("%s.%s", telemetrymetrics.DBName, telemetrymetrics.SamplesV4Agg30mTableName))
sb.From(fmt.Sprintf("%s.%s", metricstelemetryschema.DBName, metricstelemetryschema.SamplesV4Agg30mTableName))
sb.Where(sb.E("metric_name", metricName))
query, args := sb.BuildWithFlavor(sqlbuilder.ClickHouse)
@@ -1574,7 +1574,7 @@ func (m *module) getMetricLastReceived(ctx context.Context, metricName string) (
sb := sqlbuilder.NewSelectBuilder()
sb.Select("MAX(last_reported_unix_milli) AS last_received_time")
sb.From(fmt.Sprintf("%s.%s", telemetrymetrics.DBName, telemetrymetrics.AttributesMetadataTableName))
sb.From(fmt.Sprintf("%s.%s", metricstelemetryschema.DBName, metricstelemetryschema.AttributesMetadataTableName))
sb.Where(sb.E("metric_name", metricName))
query, args := sb.BuildWithFlavor(sqlbuilder.ClickHouse)
@@ -1599,7 +1599,7 @@ func (m *module) getTotalTimeSeriesForMetricName(ctx context.Context, metricName
sb := sqlbuilder.NewSelectBuilder()
sb.Select("uniq(fingerprint) AS time_series_count")
sb.From(fmt.Sprintf("%s.%s", telemetrymetrics.DBName, telemetrymetrics.TimeseriesV41weekTableName))
sb.From(fmt.Sprintf("%s.%s", metricstelemetryschema.DBName, metricstelemetryschema.TimeseriesV41weekTableName))
sb.Where(sb.E("metric_name", metricName))
query, args := sb.BuildWithFlavor(sqlbuilder.ClickHouse)
@@ -1623,7 +1623,7 @@ func (m *module) getActiveTimeSeriesForMetricName(ctx context.Context, metricNam
sb := sqlbuilder.NewSelectBuilder()
sb.Select("uniq(fingerprint) AS active_time_series")
sb.From(fmt.Sprintf("%s.%s", telemetrymetrics.DBName, telemetrymetrics.TimeseriesV4TableName))
sb.From(fmt.Sprintf("%s.%s", metricstelemetryschema.DBName, metricstelemetryschema.TimeseriesV4TableName))
sb.Where(sb.E("metric_name", metricName))
sb.Where(sb.GTE("unix_milli", milli))
@@ -1649,7 +1649,7 @@ func (m *module) fetchMetricAttributes(ctx context.Context, metricName string, s
"groupUniqArray(1000)(attr_string_value) AS values",
"uniq(attr_string_value) AS valueCount",
)
sb.From(fmt.Sprintf("%s.%s", telemetrymetrics.DBName, telemetrymetrics.AttributesMetadataTableName))
sb.From(fmt.Sprintf("%s.%s", metricstelemetryschema.DBName, metricstelemetryschema.AttributesMetadataTableName))
sb.Where(sb.E("metric_name", metricName))
sb.Where("NOT startsWith(attr_name, '__')")

View File

@@ -9,7 +9,7 @@ import (
schemamigrator "github.com/SigNoz/signoz-otel-collector/cmd/signozschemamigrator/schema_migrator"
"github.com/SigNoz/signoz/pkg/errors"
"github.com/SigNoz/signoz/pkg/modules/promote"
"github.com/SigNoz/signoz/pkg/telemetrylogs"
"github.com/SigNoz/signoz/pkg/telemetryschema/logstelemetryschema"
"github.com/SigNoz/signoz/pkg/telemetrystore"
"github.com/SigNoz/signoz/pkg/types/ctxtypes"
"github.com/SigNoz/signoz/pkg/types/instrumentationtypes"
@@ -52,7 +52,7 @@ func (m *module) ListPromotedAndIndexedPaths(ctx context.Context) ([]promotetype
response := []promotetypes.PromotePath{}
for _, path := range promotedPaths {
fullPath := telemetrylogs.BodyPromotedColumnPrefix + path
fullPath := logstelemetryschema.BodyPromotedColumnPrefix + path
path = telemetrytypes.BodyJSONStringSearchPrefix + path
item := promotetypes.PromotePath{
Path: path,
@@ -68,7 +68,7 @@ func (m *module) ListPromotedAndIndexedPaths(ctx context.Context) ([]promotetype
// add the paths that are not promoted but have indexes
for path, indexes := range aggr {
path := strings.TrimPrefix(path, telemetrylogs.BodyV2ColumnPrefix)
path := strings.TrimPrefix(path, logstelemetryschema.BodyV2ColumnPrefix)
path = telemetrytypes.BodyJSONStringSearchPrefix + path
response = append(response, promotetypes.PromotePath{
Path: path,
@@ -108,8 +108,8 @@ func (m *module) createIndexes(ctx context.Context, indexes []schemamigrator.Ind
for _, index := range indexes {
alterStmt := schemamigrator.AlterTableAddIndex{
Database: telemetrylogs.DBName,
Table: telemetrylogs.LogsV2LocalTableName,
Database: logstelemetryschema.DBName,
Table: logstelemetryschema.LogsV2LocalTableName,
Index: index,
}
op := alterStmt.OnCluster(m.telemetryStore.Cluster())
@@ -153,10 +153,10 @@ func (m *module) PromoteAndIndexPaths(
}
}
if len(it.Indexes) > 0 {
parentColumn := telemetrylogs.LogsV2BodyV2Column
parentColumn := logstelemetryschema.LogsV2BodyV2Column
// if the path is already promoted or is being promoted, add it to the promoted column
if _, promoted := existingPromotedPaths[it.Path]; promoted || it.Promote {
parentColumn = telemetrylogs.LogsV2BodyPromotedColumn
parentColumn = logstelemetryschema.LogsV2BodyPromotedColumn
}
for _, index := range it.Indexes {

View File

@@ -11,8 +11,8 @@ import (
"github.com/SigNoz/signoz/pkg/errors"
"github.com/SigNoz/signoz/pkg/modules/services"
"github.com/SigNoz/signoz/pkg/querier"
"github.com/SigNoz/signoz/pkg/telemetryschema/tracestelemetryschema"
"github.com/SigNoz/signoz/pkg/telemetrystore"
"github.com/SigNoz/signoz/pkg/telemetrytraces"
"github.com/SigNoz/signoz/pkg/types/ctxtypes"
"github.com/SigNoz/signoz/pkg/types/instrumentationtypes"
qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
@@ -39,7 +39,7 @@ func (m *module) FetchTopLevelOperations(ctx context.Context, start time.Time, s
ctx = m.withServicesContext(ctx, "FetchTopLevelOperations")
db := m.TelemetryStore.ClickhouseDB()
query := fmt.Sprintf("SELECT name, serviceName, max(time) as ts FROM %s.%s WHERE time >= @start", telemetrytraces.DBName, telemetrytraces.TopLevelOperationsTableName)
query := fmt.Sprintf("SELECT name, serviceName, max(time) as ts FROM %s.%s WHERE time >= @start", tracestelemetryschema.DBName, tracestelemetryschema.TopLevelOperationsTableName)
args := []any{clickhouse.Named("start", start)}
if len(services) > 0 {
query += " AND serviceName IN @services"

View File

@@ -22,6 +22,13 @@ import (
"github.com/SigNoz/signoz/pkg/variables"
)
type Handler interface {
QueryRange(rw http.ResponseWriter, req *http.Request)
QueryRangePreview(rw http.ResponseWriter, req *http.Request)
QueryRawStream(rw http.ResponseWriter, req *http.Request)
ReplaceVariables(rw http.ResponseWriter, req *http.Request)
}
type handler struct {
set factory.ProviderSettings
analytics analytics.Analytics

View File

@@ -15,6 +15,14 @@ import (
"github.com/SigNoz/signoz/pkg/valuer"
)
// BucketCache is the interface for bucket-based caching.
type BucketCache interface {
// cached portion + list of gaps to fetch
GetMissRanges(ctx context.Context, orgID valuer.UUID, q qbtypes.Query, step qbtypes.Step) (cached *qbtypes.Result, missing []*qbtypes.TimeRange)
// store fresh buckets for future hits
Put(ctx context.Context, orgID valuer.UUID, q qbtypes.Query, step qbtypes.Step, fresh *qbtypes.Result)
}
// bucketCache implements the BucketCache interface.
type bucketCache struct {
cache cache.Cache

View File

@@ -11,8 +11,8 @@ import (
"github.com/ClickHouse/clickhouse-go/v2"
"github.com/SigNoz/signoz/pkg/errors"
"github.com/SigNoz/signoz/pkg/telemetryschema/tracestelemetryschema"
"github.com/SigNoz/signoz/pkg/telemetrystore"
"github.com/SigNoz/signoz/pkg/telemetrytraces"
"github.com/SigNoz/signoz/pkg/types/ctxtypes"
"github.com/SigNoz/signoz/pkg/types/instrumentationtypes"
qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
@@ -277,12 +277,12 @@ func (q *builderQuery[T]) narrowWindowByTraceID(ctx context.Context, fromMS, toM
return fromMS, toMS, true, ""
}
traceIDs, found := telemetrytraces.ExtractTraceIDsFromFilter(q.spec.Filter.Expression)
traceIDs, found := tracestelemetryschema.ExtractTraceIDsFromFilter(q.spec.Filter.Expression)
if !found || len(traceIDs) == 0 {
return fromMS, toMS, true, ""
}
finder := telemetrytraces.NewTraceTimeRangeFinder(q.telemetryStore)
finder := tracestelemetryschema.NewTraceTimeRangeFinder(q.telemetryStore)
traceStart, traceEnd, exists, err := finder.GetTraceTimeRangeMulti(ctx, traceIDs)
if err != nil {
return fromMS, toMS, true, ""

View File

@@ -6,18 +6,18 @@ import (
"time"
"github.com/SigNoz/signoz/pkg/errors"
"github.com/SigNoz/signoz/pkg/telemetrylogs"
"github.com/SigNoz/signoz/pkg/telemetrymetrics"
"github.com/SigNoz/signoz/pkg/telemetrytraces"
"github.com/SigNoz/signoz/pkg/telemetryschema/logstelemetryschema"
"github.com/SigNoz/signoz/pkg/telemetryschema/metricstelemetryschema"
"github.com/SigNoz/signoz/pkg/telemetryschema/tracestelemetryschema"
"github.com/SigNoz/signoz/pkg/valuer"
)
func (q *querier) Collect(ctx context.Context, _ valuer.UUID) (map[string]any, error) {
stats := make(map[string]any)
tracesTable := fmt.Sprintf("%s.%s", telemetrytraces.DBName, telemetrytraces.SpanIndexV3TableName)
logsTable := fmt.Sprintf("%s.%s", telemetrylogs.DBName, telemetrylogs.LogsV2TableName)
metricsTable := fmt.Sprintf("%s.%s", telemetrymetrics.DBName, telemetrymetrics.SamplesV4TableName)
tracesTable := fmt.Sprintf("%s.%s", tracestelemetryschema.DBName, tracestelemetryschema.SpanIndexV3TableName)
logsTable := fmt.Sprintf("%s.%s", logstelemetryschema.DBName, logstelemetryschema.LogsV2TableName)
metricsTable := fmt.Sprintf("%s.%s", metricstelemetryschema.DBName, metricstelemetryschema.SamplesV4TableName)
var (
traces uint64

View File

@@ -9,12 +9,6 @@ import (
const DefaultMaxConcurrentQueries = 8
type SkipResourceFingerprint struct {
Enabled bool `yaml:"enabled" mapstructure:"enabled"`
// If count of fingerprint is above threshold, skip the fingerprint subquery and filter on main table instead.
Threshold uint64 `yaml:"threshold" mapstructure:"threshold"`
}
// Config represents the configuration for the querier.
type Config struct {
// CacheTTL is the TTL for cached query results
@@ -23,8 +17,6 @@ type Config struct {
FluxInterval time.Duration `yaml:"flux_interval" mapstructure:"flux_interval"`
// MaxConcurrentQueries is the maximum number of concurrent queries for missing ranges
MaxConcurrentQueries int `yaml:"max_concurrent_queries" mapstructure:"max_concurrent_queries"`
// SkipResourceFingerprint configures when the resource fingerprint subquery is skipped in favor of main-table filtering.
SkipResourceFingerprint SkipResourceFingerprint `yaml:"skip_resource_fingerprint" mapstructure:"skip_resource_fingerprint"`
// LogTraceIDWindowPadding is the padding added to narrowed down timerange from trace summary to logs with trace_id filter.
LogTraceIDWindowPadding time.Duration `yaml:"log_trace_id_window_padding" mapstructure:"log_trace_id_window_padding"`
}
@@ -39,11 +31,7 @@ func newConfig() factory.Config {
// Default values
CacheTTL: 168 * time.Hour,
FluxInterval: 5 * time.Minute,
MaxConcurrentQueries: DefaultMaxConcurrentQueries,
SkipResourceFingerprint: SkipResourceFingerprint{
Enabled: false,
Threshold: 100000,
},
MaxConcurrentQueries: DefaultMaxConcurrentQueries,
LogTraceIDWindowPadding: 5 * time.Minute,
}
}
@@ -59,9 +47,6 @@ func (c Config) Validate() error {
if c.MaxConcurrentQueries <= 0 {
return errors.NewInvalidInputf(errors.CodeInvalidInput, "max_concurrent_queries must be positive, got %v", c.MaxConcurrentQueries)
}
if c.SkipResourceFingerprint.Enabled && c.SkipResourceFingerprint.Threshold == 0 {
return errors.NewInvalidInputf(errors.CodeInvalidInput, "skip_resource_fingerprint.threshold must be > 0 when enabled")
}
if c.LogTraceIDWindowPadding < 0 {
return errors.NewInvalidInputf(errors.CodeInvalidInput, "log_trace_id_window_padding must not be negative, got %v", c.LogTraceIDWindowPadding)
}

View File

@@ -1,35 +0,0 @@
package querier
import (
"context"
"net/http"
"github.com/SigNoz/signoz/pkg/statsreporter"
qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
"github.com/SigNoz/signoz/pkg/valuer"
)
// Querier interface defines the contract for querying data.
type Querier interface {
QueryRange(ctx context.Context, orgID valuer.UUID, req *qbtypes.QueryRangeRequest) (*qbtypes.QueryRangeResponse, error)
QueryRawStream(ctx context.Context, orgID valuer.UUID, req *qbtypes.QueryRangeRequest, client *qbtypes.RawStream)
statsreporter.StatsCollector
// QueryRangePreview validates and renders the queries without executing them.
QueryRangePreview(ctx context.Context, orgID valuer.UUID, req *qbtypes.QueryRangeRequest, opts qbtypes.QueryRangePreviewOptions) (*qbtypes.QueryRangePreviewResponse, error)
}
// BucketCache is the interface for bucket-based caching.
type BucketCache interface {
// cached portion + list of gaps to fetch
GetMissRanges(ctx context.Context, orgID valuer.UUID, q qbtypes.Query, step qbtypes.Step) (cached *qbtypes.Result, missing []*qbtypes.TimeRange)
// store fresh buckets for future hits
Put(ctx context.Context, orgID valuer.UUID, q qbtypes.Query, step qbtypes.Step, fresh *qbtypes.Result)
}
type Handler interface {
QueryRange(rw http.ResponseWriter, req *http.Request)
// QueryRangePreview is the dry-run endpoint: validate and render without executing.
QueryRangePreview(rw http.ResponseWriter, req *http.Request)
QueryRawStream(rw http.ResponseWriter, req *http.Request)
ReplaceVariables(rw http.ResponseWriter, req *http.Request)
}

View File

@@ -21,6 +21,8 @@ import (
"github.com/SigNoz/signoz/pkg/prometheus"
"github.com/SigNoz/signoz/pkg/query-service/utils"
"github.com/SigNoz/signoz/pkg/querybuilder"
"github.com/SigNoz/signoz/pkg/statementbuilder"
"github.com/SigNoz/signoz/pkg/statsreporter"
"github.com/SigNoz/signoz/pkg/telemetrystore"
"github.com/SigNoz/signoz/pkg/types/ctxtypes"
"github.com/SigNoz/signoz/pkg/types/instrumentationtypes"
@@ -35,6 +37,15 @@ var (
intervalWarn = "Query %s is requesting aggregation interval %v seconds, which is smaller than the minimum allowed interval of %v seconds for selected time range. Using the minimum instead"
)
// Querier interface defines the contract for querying data.
type Querier interface {
QueryRange(ctx context.Context, orgID valuer.UUID, req *qbtypes.QueryRangeRequest) (*qbtypes.QueryRangeResponse, error)
QueryRawStream(ctx context.Context, orgID valuer.UUID, req *qbtypes.QueryRangeRequest, client *qbtypes.RawStream)
statsreporter.StatsCollector
// QueryRangePreview validates and renders the queries without executing them.
QueryRangePreview(ctx context.Context, orgID valuer.UUID, req *qbtypes.QueryRangeRequest, opts qbtypes.QueryRangePreviewOptions) (*qbtypes.QueryRangePreviewResponse, error)
}
type querier struct {
logger *slog.Logger
fl flagger.Flagger
@@ -60,12 +71,7 @@ func New(
telemetryStore telemetrystore.TelemetryStore,
metadataStore telemetrytypes.MetadataStore,
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,
builders *statementbuilder.Builders,
bucketCache BucketCache,
flagger flagger.Flagger,
logTraceIDWindowPadding time.Duration,
@@ -75,18 +81,21 @@ func New(
if maxConcurrentQueries <= 0 {
maxConcurrentQueries = DefaultMaxConcurrentQueries
}
if builders == nil {
builders = &statementbuilder.Builders{}
}
return &querier{
logger: querierSettings.Logger(),
fl: flagger,
telemetryStore: telemetryStore,
metadataStore: metadataStore,
promEngine: promEngine,
traceStmtBuilder: traceStmtBuilder,
logStmtBuilder: logStmtBuilder,
auditStmtBuilder: auditStmtBuilder,
metricStmtBuilder: metricStmtBuilder,
meterStmtBuilder: meterStmtBuilder,
traceOperatorStmtBuilder: traceOperatorStmtBuilder,
traceStmtBuilder: builders.Trace,
logStmtBuilder: builders.Log,
auditStmtBuilder: builders.Audit,
metricStmtBuilder: builders.Metric,
meterStmtBuilder: builders.Meter,
traceOperatorStmtBuilder: builders.TraceOperator,
bucketCache: bucketCache,
liveDataRefresh: 5 * time.Second,
builderConfig: builderConfig{

View File

@@ -11,6 +11,7 @@ import (
"github.com/SigNoz/signoz/pkg/errors"
"github.com/SigNoz/signoz/pkg/flagger/flaggertest"
"github.com/SigNoz/signoz/pkg/instrumentation/instrumentationtest"
"github.com/SigNoz/signoz/pkg/statementbuilder"
"github.com/SigNoz/signoz/pkg/telemetrystore"
"github.com/SigNoz/signoz/pkg/telemetrystore/telemetrystoretest"
"github.com/SigNoz/signoz/pkg/types/metrictypes"
@@ -48,12 +49,7 @@ func TestQueryRange_MetricTypeMissing(t *testing.T) {
nil, // telemetryStore
metadataStore,
nil, // prometheus
nil, // traceStmtBuilder
nil, // logStmtBuilder
nil, // auditStmtBuilder
nil, // metricStmtBuilder
nil, // meterStmtBuilder
nil, // traceOperatorStmtBuilder
nil, // builders
nil, // bucketCache
flaggertest.New(t), // flagger
0, // logTraceIDWindowPadding
@@ -119,17 +115,12 @@ func TestQueryRange_MetricTypeFromStore(t *testing.T) {
providerSettings,
telemetryStore,
metadataStore,
nil, // prometheus
nil, // traceStmtBuilder
nil, // logStmtBuilder
nil, // auditStmtBuilder
&mockMetricStmtBuilder{}, // metricStmtBuilder
nil, // meterStmtBuilder
nil, // traceOperatorStmtBuilder
nil, // bucketCache
flaggertest.New(t), // flagger
0, // logTraceIDWindowPadding
0, // maxConcurrentQueries
nil, // prometheus
&statementbuilder.Builders{Metric: &mockMetricStmtBuilder{}},
nil, // bucketCache
flaggertest.New(t), // flagger
0, // logTraceIDWindowPadding
0, // maxConcurrentQueries
)
req := &qbtypes.QueryRangeRequest{

View File

@@ -0,0 +1,52 @@
package queriertest
import (
"context"
"github.com/SigNoz/signoz/pkg/querier"
qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
"github.com/SigNoz/signoz/pkg/valuer"
)
// MockQuerier implements querier.Querier for testing. Each behavior can be
// overridden via a function field; unset fields return empty, non-error values.
type MockQuerier struct {
QueryRangeFunc func(ctx context.Context, orgID valuer.UUID, req *qbtypes.QueryRangeRequest) (*qbtypes.QueryRangeResponse, error)
QueryRawStreamFunc func(ctx context.Context, orgID valuer.UUID, req *qbtypes.QueryRangeRequest, client *qbtypes.RawStream)
QueryRangePreviewFunc func(ctx context.Context, orgID valuer.UUID, req *qbtypes.QueryRangeRequest, opts qbtypes.QueryRangePreviewOptions) (*qbtypes.QueryRangePreviewResponse, error)
CollectFunc func(ctx context.Context, orgID valuer.UUID) (map[string]any, error)
}
var _ querier.Querier = (*MockQuerier)(nil)
// NewMockQuerier creates a new MockQuerier.
func NewMockQuerier() *MockQuerier {
return &MockQuerier{}
}
func (m *MockQuerier) QueryRange(ctx context.Context, orgID valuer.UUID, req *qbtypes.QueryRangeRequest) (*qbtypes.QueryRangeResponse, error) {
if m.QueryRangeFunc != nil {
return m.QueryRangeFunc(ctx, orgID, req)
}
return &qbtypes.QueryRangeResponse{}, nil
}
func (m *MockQuerier) QueryRawStream(ctx context.Context, orgID valuer.UUID, req *qbtypes.QueryRangeRequest, client *qbtypes.RawStream) {
if m.QueryRawStreamFunc != nil {
m.QueryRawStreamFunc(ctx, orgID, req, client)
}
}
func (m *MockQuerier) QueryRangePreview(ctx context.Context, orgID valuer.UUID, req *qbtypes.QueryRangeRequest, opts qbtypes.QueryRangePreviewOptions) (*qbtypes.QueryRangePreviewResponse, error) {
if m.QueryRangePreviewFunc != nil {
return m.QueryRangePreviewFunc(ctx, orgID, req, opts)
}
return &qbtypes.QueryRangePreviewResponse{}, nil
}
func (m *MockQuerier) Collect(ctx context.Context, orgID valuer.UUID) (map[string]any, error) {
if m.CollectFunc != nil {
return m.CollectFunc(ctx, orgID)
}
return map[string]any{}, nil
}

View File

@@ -3,192 +3,44 @@ package signozquerier
import (
"context"
"github.com/SigNoz/signoz/pkg/cache"
"github.com/SigNoz/signoz/pkg/factory"
"github.com/SigNoz/signoz/pkg/flagger"
"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"
"github.com/SigNoz/signoz/pkg/telemetrymetrics"
"github.com/SigNoz/signoz/pkg/statementbuilder"
"github.com/SigNoz/signoz/pkg/telemetrystore"
"github.com/SigNoz/signoz/pkg/telemetrytraces"
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
)
// NewFactory creates a new factory for the signoz querier provider.
// NewFactory creates a new factory for the signoz querier provider. The query
// stack (metadata store, statement builders, bucket cache) is assembled once in
// signoz.go and injected here.
func NewFactory(
telemetryStore telemetrystore.TelemetryStore,
prometheus prometheus.Prometheus,
cache cache.Cache,
metadataStore telemetrytypes.MetadataStore,
builders *statementbuilder.Builders,
bucketCache querier.BucketCache,
flagger flagger.Flagger,
) factory.ProviderFactory[querier.Querier, querier.Config] {
return factory.NewProviderFactory(
factory.MustNewName("signoz"),
func(
ctx context.Context,
_ context.Context,
settings factory.ProviderSettings,
cfg querier.Config,
) (querier.Querier, error) {
return newProvider(ctx, settings, cfg, telemetryStore, prometheus, cache, flagger)
return querier.New(
settings,
telemetryStore,
metadataStore,
prometheus,
builders,
bucketCache,
flagger,
cfg.LogTraceIDWindowPadding,
cfg.MaxConcurrentQueries,
), nil
},
)
}
func newProvider(
_ context.Context,
settings factory.ProviderSettings,
cfg querier.Config,
telemetryStore telemetrystore.TelemetryStore,
prometheus prometheus.Prometheus,
cache cache.Cache,
flagger flagger.Flagger,
) (querier.Querier, error) {
// Create telemetry metadata store
telemetryMetadataStore := telemetrymetadata.NewTelemetryMetaStore(
settings,
telemetryStore,
telemetrytraces.DBName,
telemetrytraces.TagAttributesV2TableName,
telemetrytraces.SpanAttributesKeysTblName,
telemetrytraces.SpanIndexV3TableName,
telemetrymetrics.DBName,
telemetrymetrics.AttributesMetadataTableName,
telemetrymeter.DBName,
telemetrymeter.SamplesAgg1dTableName,
telemetrylogs.DBName,
telemetrylogs.LogsV2TableName,
telemetrylogs.TagAttributesV2TableName,
telemetrylogs.LogAttributeKeysTblName,
telemetrylogs.LogResourceKeysTblName,
telemetryaudit.DBName,
telemetryaudit.AuditLogsTableName,
telemetryaudit.TagAttributesTableName,
telemetryaudit.LogAttributeKeysTblName,
telemetryaudit.LogResourceKeysTblName,
telemetrymetadata.DBName,
telemetrymetadata.AttributesMetadataTableName,
telemetrymetadata.ColumnEvolutionMetadataTableName,
flagger,
)
// Create trace statement builder
traceFieldMapper := telemetrytraces.NewFieldMapper()
traceConditionBuilder := telemetrytraces.NewConditionBuilder(traceFieldMapper)
traceAggExprRewriter := querybuilder.NewAggExprRewriter(settings, nil, traceFieldMapper, traceConditionBuilder, flagger)
traceStmtBuilder := telemetrytraces.NewTraceQueryStatementBuilder(
settings,
telemetryMetadataStore,
traceFieldMapper,
traceConditionBuilder,
traceAggExprRewriter,
telemetryStore,
flagger,
cfg.SkipResourceFingerprint.Enabled,
cfg.SkipResourceFingerprint.Threshold,
)
// Create trace operator statement builder
traceOperatorStmtBuilder := telemetrytraces.NewTraceOperatorStatementBuilder(
settings,
telemetryMetadataStore,
traceFieldMapper,
traceConditionBuilder,
traceStmtBuilder,
traceAggExprRewriter,
flagger,
)
// Create log statement builder
logFieldMapper := telemetrylogs.NewFieldMapper(flagger)
logConditionBuilder := telemetrylogs.NewConditionBuilder(logFieldMapper, flagger)
logAggExprRewriter := querybuilder.NewAggExprRewriter(
settings,
telemetrylogs.DefaultFullTextColumn,
logFieldMapper,
logConditionBuilder,
flagger,
)
logStmtBuilder := telemetrylogs.NewLogQueryStatementBuilder(
settings,
telemetryMetadataStore,
logFieldMapper,
logConditionBuilder,
logAggExprRewriter,
telemetrylogs.DefaultFullTextColumn,
flagger,
telemetryStore,
cfg.SkipResourceFingerprint.Enabled,
cfg.SkipResourceFingerprint.Threshold,
)
// Create audit statement builder
auditFieldMapper := telemetryaudit.NewFieldMapper()
auditConditionBuilder := telemetryaudit.NewConditionBuilder(auditFieldMapper)
auditAggExprRewriter := querybuilder.NewAggExprRewriter(
settings,
telemetryaudit.DefaultFullTextColumn,
auditFieldMapper,
auditConditionBuilder,
flagger,
)
auditStmtBuilder := telemetryaudit.NewAuditQueryStatementBuilder(
settings,
telemetryMetadataStore,
auditFieldMapper,
auditConditionBuilder,
auditAggExprRewriter,
telemetryaudit.DefaultFullTextColumn,
flagger,
)
// Create metric statement builder
metricFieldMapper := telemetrymetrics.NewFieldMapper()
metricConditionBuilder := telemetrymetrics.NewConditionBuilder(metricFieldMapper)
metricStmtBuilder := telemetrymetrics.NewMetricQueryStatementBuilder(
settings,
telemetryMetadataStore,
metricFieldMapper,
metricConditionBuilder,
flagger,
)
// Create meter statement builder
meterStmtBuilder := telemetrymeter.NewMeterQueryStatementBuilder(
settings,
telemetryMetadataStore,
metricFieldMapper,
metricConditionBuilder,
metricStmtBuilder,
)
// Create bucket cache
bucketCache := querier.NewBucketCache(
settings,
cache,
cfg.CacheTTL,
cfg.FluxInterval,
)
// Create and return the querier
return querier.New(
settings,
telemetryStore,
telemetryMetadataStore,
prometheus,
traceStmtBuilder,
logStmtBuilder,
auditStmtBuilder,
metricStmtBuilder,
meterStmtBuilder,
traceOperatorStmtBuilder,
bucketCache,
flagger,
cfg.LogTraceIDWindowPadding,
cfg.MaxConcurrentQueries,
), nil
}

View File

@@ -18,6 +18,8 @@ import (
"github.com/SigNoz/signoz/pkg/querier/signozquerier"
"github.com/SigNoz/signoz/pkg/sqlstore"
"github.com/SigNoz/signoz/pkg/sqlstore/sqlstoretest"
"github.com/SigNoz/signoz/pkg/statementbuilder"
"github.com/SigNoz/signoz/pkg/telemetrymetadata"
"github.com/SigNoz/signoz/pkg/telemetrystore"
"github.com/SigNoz/signoz/pkg/telemetrystore/telemetrystoretest"
"github.com/stretchr/testify/require"
@@ -105,7 +107,11 @@ func NewTestManager(t *testing.T, testOpts *TestManagerOptions) *Manager {
}
// Create querier with test values
providerFactory := signozquerier.NewFactory(telemetryStore, prometheus, cache, flagger)
metadataStore := telemetrymetadata.NewTelemetryMetaStore(providerSettings, telemetryStore, flagger)
builders, err := statementbuilder.NewFactory(telemetryStore, metadataStore, flagger).New(context.Background(), providerSettings, statementbuilder.Config{})
require.NoError(t, err)
bucketCache := querier.NewBucketCache(providerSettings, cache, 0, 0)
providerFactory := signozquerier.NewFactory(telemetryStore, prometheus, metadataStore, builders, bucketCache, flagger)
mockQuerier, err := providerFactory.New(context.Background(), providerSettings, querier.Config{})
require.NoError(t, err)

View File

@@ -8,11 +8,11 @@ import (
"github.com/SigNoz/signoz/pkg/flagger"
"github.com/SigNoz/signoz/pkg/instrumentation/instrumentationtest"
"github.com/SigNoz/signoz/pkg/querier"
"github.com/SigNoz/signoz/pkg/querybuilder"
"github.com/SigNoz/signoz/pkg/telemetrylogs"
"github.com/SigNoz/signoz/pkg/telemetrymetrics"
"github.com/SigNoz/signoz/pkg/statementbuilder"
"github.com/SigNoz/signoz/pkg/statementbuilder/logsstatementbuilder"
"github.com/SigNoz/signoz/pkg/statementbuilder/metricsstatementbuilder"
"github.com/SigNoz/signoz/pkg/statementbuilder/tracesstatementbuilder"
"github.com/SigNoz/signoz/pkg/telemetrystore"
"github.com/SigNoz/signoz/pkg/telemetrytraces"
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
"github.com/SigNoz/signoz/pkg/types/telemetrytypes/telemetrytypestest"
"github.com/stretchr/testify/require"
@@ -24,7 +24,7 @@ func prepareQuerierForMetrics(t *testing.T, telemetryStore telemetrystore.Teleme
providerSettings := instrumentationtest.New().ToProviderSettings()
metadataStore := telemetrytypestest.NewMockMetadataStore()
flagger, err := flagger.New(
fl, err := flagger.New(
context.Background(),
instrumentationtest.New().ToProviderSettings(),
flagger.Config{},
@@ -32,29 +32,16 @@ func prepareQuerierForMetrics(t *testing.T, telemetryStore telemetrystore.Teleme
)
require.NoError(t, err)
metricFieldMapper := telemetrymetrics.NewFieldMapper()
metricConditionBuilder := telemetrymetrics.NewConditionBuilder(metricFieldMapper)
metricStmtBuilder := telemetrymetrics.NewMetricQueryStatementBuilder(
providerSettings,
metadataStore,
metricFieldMapper,
metricConditionBuilder,
flagger,
)
metricStmtBuilder := metricsstatementbuilder.New(providerSettings, metadataStore, fl)
return querier.New(
providerSettings,
telemetryStore,
metadataStore,
nil, // prometheus
nil, // traceStmtBuilder
nil, // logStmtBuilder
nil, // auditStmtBuilder
metricStmtBuilder,
nil, // meterStmtBuilder
nil, // traceOperatorStmtBuilder
&statementbuilder.Builders{Metric: metricStmtBuilder},
nil, // bucketCache
flagger,
fl,
0,
0, // maxConcurrentQueries (0 means default)
), metadataStore
@@ -73,40 +60,15 @@ func prepareQuerierForLogs(t *testing.T, telemetryStore telemetrystore.Telemetry
metadataStore.KeysMap = keysMap
fl := flaggertest.New(t)
logFieldMapper := telemetrylogs.NewFieldMapper(fl)
logConditionBuilder := telemetrylogs.NewConditionBuilder(logFieldMapper, fl)
logAggExprRewriter := querybuilder.NewAggExprRewriter(
providerSettings,
telemetrylogs.DefaultFullTextColumn,
logFieldMapper,
logConditionBuilder,
fl,
)
logStmtBuilder := telemetrylogs.NewLogQueryStatementBuilder(
providerSettings,
metadataStore,
logFieldMapper,
logConditionBuilder,
logAggExprRewriter,
telemetrylogs.DefaultFullTextColumn,
fl,
nil,
false,
100000,
)
logStmtBuilder := logsstatementbuilder.New(providerSettings, metadataStore, nil, fl, false, 100000)
return querier.New(
providerSettings,
telemetryStore,
metadataStore,
nil, // prometheus
nil, // traceStmtBuilder
logStmtBuilder, // logStmtBuilder
nil, // auditStmtBuilder
nil, // metricStmtBuilder
nil, // meterStmtBuilder
nil, // traceOperatorStmtBuilder
nil, // bucketCache
nil, // prometheus
&statementbuilder.Builders{Log: logStmtBuilder},
nil, // bucketCache
fl,
5*time.Minute, // logTraceIDWindowPadding
0, // maxConcurrentQueries (0 means default)
@@ -126,36 +88,16 @@ func prepareQuerierForTraces(t *testing.T, telemetryStore telemetrystore.Telemet
}
metadataStore.KeysMap = keysMap
// Create trace statement builder
traceFieldMapper := telemetrytraces.NewFieldMapper()
traceConditionBuilder := telemetrytraces.NewConditionBuilder(traceFieldMapper)
fl := flaggertest.New(t)
traceAggExprRewriter := querybuilder.NewAggExprRewriter(providerSettings, nil, traceFieldMapper, traceConditionBuilder, fl)
traceStmtBuilder := telemetrytraces.NewTraceQueryStatementBuilder(
providerSettings,
metadataStore,
traceFieldMapper,
traceConditionBuilder,
traceAggExprRewriter,
telemetryStore,
fl,
false,
100000,
)
traceStmtBuilder, _ := tracesstatementbuilder.New(providerSettings, metadataStore, telemetryStore, fl, false, 100000)
return querier.New(
providerSettings,
telemetryStore,
metadataStore,
nil, // prometheus
traceStmtBuilder, // traceStmtBuilder
nil, // logStmtBuilder
nil, // auditStmtBuilder
nil, // metricStmtBuilder
nil, // meterStmtBuilder
nil, // traceOperatorStmtBuilder
nil, // bucketCache
nil, // prometheus
&statementbuilder.Builders{Trace: traceStmtBuilder},
nil, // bucketCache
fl,
0,
0, // maxConcurrentQueries (0 means default)

View File

@@ -96,9 +96,9 @@ func (r *aggExprRewriter) Rewrite(
}
if visitor.isRate {
return fmt.Sprintf("%s/%d", chparser.Format(sel.SelectItems[0]), rateInterval), visitor.chArgs, nil
return fmt.Sprintf("%s/%d", sel.SelectItems[0].String(), rateInterval), visitor.chArgs, nil
}
return chparser.Format(sel.SelectItems[0]), visitor.chArgs, nil
return sel.SelectItems[0].String(), visitor.chArgs, nil
}
// RewriteMulti rewrites a slice of expressions.
@@ -207,7 +207,7 @@ func (v *exprVisitor) VisitFunctionExpr(fn *chparser.FunctionExpr) error {
// Handle *If functions with predicate + values
if aggFunc.FuncCombinator {
// Map the predicate (last argument)
origPred := chparser.Format(args[len(args)-1])
origPred := args[len(args)-1].String()
whereClause, err := PrepareWhereClause(
origPred,
FilterExprVisitorOpts{
@@ -242,7 +242,7 @@ func (v *exprVisitor) VisitFunctionExpr(fn *chparser.FunctionExpr) error {
// Map each value column argument
for i := 0; i < len(args)-1; i++ {
origVal := chparser.Format(args[i])
origVal := args[i].String()
fieldKey := telemetrytypes.GetFieldKeyFromKeyText(origVal)
expr, err := v.fieldMapper.ColumnExpressionFor(v.ctx, v.orgID, v.startNs, v.endNs, &fieldKey, dataType, v.fieldKeys)
if err != nil {
@@ -259,7 +259,7 @@ func (v *exprVisitor) VisitFunctionExpr(fn *chparser.FunctionExpr) error {
} else {
// Non-If functions: map every argument as a column/value
for i, arg := range args {
orig := chparser.Format(arg)
orig := arg.String()
fieldKey := telemetrytypes.GetFieldKeyFromKeyText(orig)
expr, err := v.fieldMapper.ColumnExpressionFor(v.ctx, v.orgID, v.startNs, v.endNs, &fieldKey, dataType, v.fieldKeys)
if err != nil {

View File

@@ -316,17 +316,17 @@ func (e *ClickHouseFilterExtractor) extractColumnStrByExpr(expr clickhouse.Expr)
// FunctionExpr is a function call like "toDate(timestamp)"
case *clickhouse.FunctionExpr:
// For function expressions, return the complete function call string
return clickhouse.Format(ex)
return ex.String()
// ColumnExpr is a column expression like "m.region", "toDate(timestamp)"
case *clickhouse.ColumnExpr:
// ColumnExpr wraps another expression - extract the underlying expression
if ex.Expr != nil {
return e.extractColumnStrByExpr(ex.Expr)
}
return clickhouse.Format(ex)
return ex.String()
default:
// For other expression types, return the string representation
return clickhouse.Format(expr)
return expr.String()
}
}
@@ -524,7 +524,7 @@ func (e *ClickHouseFilterExtractor) extractFullExpression(expr clickhouse.Expr)
if expr == nil {
return ""
}
return clickhouse.Format(expr)
return expr.String()
}
// isSimpleColumnReference checks if an expression is just a simple column reference
@@ -657,7 +657,7 @@ func (e *ClickHouseFilterExtractor) extractCTEName(cte *clickhouse.CTEStmt) stri
case *clickhouse.Ident:
return name.Name
default:
return clickhouse.Format(cte.Expr)
return cte.Expr.String()
}
}

View File

@@ -39,6 +39,7 @@ import (
"github.com/SigNoz/signoz/pkg/sqlmigrator"
"github.com/SigNoz/signoz/pkg/sqlschema"
"github.com/SigNoz/signoz/pkg/sqlstore"
"github.com/SigNoz/signoz/pkg/statementbuilder"
"github.com/SigNoz/signoz/pkg/statsreporter"
"github.com/SigNoz/signoz/pkg/telemetrystore"
"github.com/SigNoz/signoz/pkg/tokenizer"
@@ -97,6 +98,9 @@ type Config struct {
// Querier config
Querier querier.Config `mapstructure:"querier"`
// StatementBuilder config
StatementBuilder statementbuilder.Config `mapstructure:"statementbuilder"`
// Ruler config
Ruler ruler.Config `mapstructure:"ruler"`
@@ -166,6 +170,7 @@ func NewConfig(ctx context.Context, logger *slog.Logger, resolverConfig config.R
prometheus.NewConfigFactory(),
alertmanager.NewConfigFactory(),
querier.NewConfigFactory(),
statementbuilder.NewConfigFactory(),
ruler.NewConfigFactory(),
emailing.NewConfigFactory(),
sharder.NewConfigFactory(),

View File

@@ -56,6 +56,7 @@ import (
"github.com/SigNoz/signoz/pkg/sqlstore"
"github.com/SigNoz/signoz/pkg/sqlstore/sqlitesqlstore"
"github.com/SigNoz/signoz/pkg/sqlstore/sqlstorehook"
"github.com/SigNoz/signoz/pkg/statementbuilder"
"github.com/SigNoz/signoz/pkg/statsreporter"
"github.com/SigNoz/signoz/pkg/statsreporter/analyticsstatsreporter"
"github.com/SigNoz/signoz/pkg/statsreporter/noopstatsreporter"
@@ -69,6 +70,7 @@ import (
"github.com/SigNoz/signoz/pkg/types/alertmanagertypes"
"github.com/SigNoz/signoz/pkg/types/dashboardtypes"
"github.com/SigNoz/signoz/pkg/types/featuretypes"
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
"github.com/SigNoz/signoz/pkg/version"
"github.com/SigNoz/signoz/pkg/web"
"github.com/SigNoz/signoz/pkg/web/noopweb"
@@ -285,9 +287,9 @@ func NewStatsReporterProviderFactories(aggregator statsreporter.Aggregator, orgG
)
}
func NewQuerierProviderFactories(telemetryStore telemetrystore.TelemetryStore, prometheus prometheus.Prometheus, cache cache.Cache, flagger flagger.Flagger) factory.NamedMap[factory.ProviderFactory[querier.Querier, querier.Config]] {
func NewQuerierProviderFactories(telemetryStore telemetrystore.TelemetryStore, prometheus prometheus.Prometheus, metadataStore telemetrytypes.MetadataStore, builders *statementbuilder.Builders, bucketCache querier.BucketCache, flagger flagger.Flagger) factory.NamedMap[factory.ProviderFactory[querier.Querier, querier.Config]] {
return factory.MustNewNamedMap(
signozquerier.NewFactory(telemetryStore, prometheus, cache, flagger),
signozquerier.NewFactory(telemetryStore, prometheus, metadataStore, builders, bucketCache, flagger),
)
}

View File

@@ -48,14 +48,10 @@ import (
"github.com/SigNoz/signoz/pkg/sqlmigrator"
"github.com/SigNoz/signoz/pkg/sqlschema"
"github.com/SigNoz/signoz/pkg/sqlstore"
"github.com/SigNoz/signoz/pkg/statementbuilder"
"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"
"github.com/SigNoz/signoz/pkg/telemetrymetrics"
"github.com/SigNoz/signoz/pkg/telemetrystore"
"github.com/SigNoz/signoz/pkg/telemetrytraces"
pkgtokenizer "github.com/SigNoz/signoz/pkg/tokenizer"
"github.com/SigNoz/signoz/pkg/types/authtypes"
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
@@ -96,6 +92,29 @@ type SigNoz struct {
MeterReporter meterreporter.Reporter
}
// newQueryStack assembles the query stack once: the shared telemetry metadata
// store, the per-signal statement builders, and the bucket cache. The metadata
// store is returned so callers can reuse the single instance downstream.
func newQueryStack(
ctx context.Context,
settings factory.ProviderSettings,
config Config,
telemetryStore telemetrystore.TelemetryStore,
cache cache.Cache,
fl flagger.Flagger,
) (telemetrytypes.MetadataStore, *statementbuilder.Builders, querier.BucketCache, error) {
metadataStore := telemetrymetadata.NewTelemetryMetaStore(settings, telemetryStore, fl)
builders, err := statementbuilder.NewFactory(telemetryStore, metadataStore, fl).New(ctx, settings, config.StatementBuilder)
if err != nil {
return nil, nil, nil, err
}
bucketCache := querier.NewBucketCache(settings, cache, config.Querier.CacheTTL, config.Querier.FluxInterval)
return metadataStore, builders, bucketCache, nil
}
func New(
ctx context.Context,
config Config,
@@ -254,12 +273,19 @@ func New(
return nil, err
}
// Assemble the query stack (metadata store, statement builders, bucket cache) once,
// and reuse the single metadata store everywhere downstream.
telemetryMetadataStore, builders, bucketCache, err := newQueryStack(ctx, providerSettings, config, telemetrystore, cache, flagger)
if err != nil {
return nil, err
}
// Initialize querier from the available querier provider factories
querier, err := factory.NewProviderFromNamedMap(
ctx,
providerSettings,
config.Querier,
NewQuerierProviderFactories(telemetrystore, prometheus, cache, flagger),
NewQuerierProviderFactories(telemetrystore, prometheus, telemetryMetadataStore, builders, bucketCache, flagger),
config.Querier.Provider(),
)
if err != nil {
@@ -427,35 +453,6 @@ func New(
return nil, err
}
// Initialize telemetry metadata store
// TODO: consolidate other telemetrymetadata.NewTelemetryMetaStore initializations to reuse this instance instead.
telemetryMetadataStore := telemetrymetadata.NewTelemetryMetaStore(
providerSettings,
telemetrystore,
telemetrytraces.DBName,
telemetrytraces.TagAttributesV2TableName,
telemetrytraces.SpanAttributesKeysTblName,
telemetrytraces.SpanIndexV3TableName,
telemetrymetrics.DBName,
telemetrymetrics.AttributesMetadataTableName,
telemetrymeter.DBName,
telemetrymeter.SamplesAgg1dTableName,
telemetrylogs.DBName,
telemetrylogs.LogsV2TableName,
telemetrylogs.TagAttributesV2TableName,
telemetrylogs.LogAttributeKeysTblName,
telemetrylogs.LogResourceKeysTblName,
telemetryaudit.DBName,
telemetryaudit.AuditLogsTableName,
telemetryaudit.TagAttributesTableName,
telemetryaudit.LogAttributeKeysTblName,
telemetryaudit.LogResourceKeysTblName,
telemetrymetadata.DBName,
telemetrymetadata.AttributesMetadataTableName,
telemetrymetadata.ColumnEvolutionMetadataTableName,
flagger,
)
global, err := factory.NewProviderFromNamedMap(
ctx,
providerSettings,

View File

@@ -0,0 +1,26 @@
package auditstatementbuilder
import (
"github.com/SigNoz/signoz/pkg/factory"
"github.com/SigNoz/signoz/pkg/flagger"
"github.com/SigNoz/signoz/pkg/querybuilder"
"github.com/SigNoz/signoz/pkg/telemetryschema/audittelemetryschema"
qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
)
// New builds the audit statement builder, internalizing the FieldMapper,
// ConditionBuilder, and AggExprRewriter.
func New(
settings factory.ProviderSettings,
metadataStore telemetrytypes.MetadataStore,
fl flagger.Flagger,
) qbtypes.StatementBuilder[qbtypes.LogAggregation] {
fm := audittelemetryschema.NewFieldMapper()
cb := audittelemetryschema.NewConditionBuilder(fm)
aggExprRewriter := querybuilder.NewAggExprRewriter(settings, audittelemetryschema.DefaultFullTextColumn, fm, cb, fl)
return NewAuditQueryStatementBuilder(
settings, metadataStore, fm, cb, aggExprRewriter, audittelemetryschema.DefaultFullTextColumn, fl,
)
}

View File

@@ -1,4 +1,4 @@
package telemetryaudit
package auditstatementbuilder
import (
"context"
@@ -10,7 +10,8 @@ import (
"github.com/SigNoz/signoz/pkg/factory"
"github.com/SigNoz/signoz/pkg/flagger"
"github.com/SigNoz/signoz/pkg/querybuilder"
"github.com/SigNoz/signoz/pkg/telemetryresourcefilter"
"github.com/SigNoz/signoz/pkg/statementbuilder/resourcefilter"
"github.com/SigNoz/signoz/pkg/telemetryschema/audittelemetryschema"
qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
"github.com/SigNoz/signoz/pkg/valuer"
@@ -38,12 +39,12 @@ func NewAuditQueryStatementBuilder(
fullTextColumn *telemetrytypes.TelemetryFieldKey,
flagger flagger.Flagger,
) *auditQueryStatementBuilder {
auditSettings := factory.NewScopedProviderSettings(settings, "github.com/SigNoz/signoz/pkg/telemetryaudit")
auditSettings := factory.NewScopedProviderSettings(settings, "github.com/SigNoz/signoz/pkg/telemetryschema/audittelemetryschema")
resourceFilterStmtBuilder := telemetryresourcefilter.New[qbtypes.LogAggregation](
resourceFilterStmtBuilder := resourcefilter.New[qbtypes.LogAggregation](
settings,
DBName,
LogsResourceTableName,
audittelemetryschema.DBName,
audittelemetryschema.LogsResourceTableName,
telemetrytypes.SignalLogs,
telemetrytypes.SourceAudit,
metadataStore,
@@ -191,8 +192,8 @@ func (b *auditQueryStatementBuilder) adjustKeys(ctx context.Context, keys map[st
}
func (b *auditQueryStatementBuilder) adjustKey(key *telemetrytypes.TelemetryFieldKey, keys map[string][]*telemetrytypes.TelemetryFieldKey) []string {
if _, ok := IntrinsicFields[key.Name]; ok {
intrinsicField := IntrinsicFields[key.Name]
if _, ok := audittelemetryschema.IntrinsicFields[key.Name]; ok {
intrinsicField := audittelemetryschema.IntrinsicFields[key.Name]
return querybuilder.AdjustKey(key, keys, &intrinsicField)
}
return querybuilder.AdjustKey(key, keys, nil)
@@ -219,26 +220,26 @@ func (b *auditQueryStatementBuilder) buildListQuery(
cteArgs = append(cteArgs, args)
}
sb.Select(TimestampColumn)
sb.SelectMore(IDColumn)
sb.Select(audittelemetryschema.TimestampColumn)
sb.SelectMore(audittelemetryschema.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)
sb.SelectMore(audittelemetryschema.TraceIDColumn)
sb.SelectMore(audittelemetryschema.SpanIDColumn)
sb.SelectMore(audittelemetryschema.TraceFlagsColumn)
sb.SelectMore(audittelemetryschema.SeverityTextColumn)
sb.SelectMore(audittelemetryschema.SeverityNumberColumn)
sb.SelectMore(audittelemetryschema.ScopeNameColumn)
sb.SelectMore(audittelemetryschema.ScopeVersionColumn)
sb.SelectMore(audittelemetryschema.BodyColumn)
sb.SelectMore(audittelemetryschema.EventNameColumn)
sb.SelectMore(audittelemetryschema.AttributesStringColumn)
sb.SelectMore(audittelemetryschema.AttributesNumberColumn)
sb.SelectMore(audittelemetryschema.AttributesBoolColumn)
sb.SelectMore(audittelemetryschema.ResourceColumn)
sb.SelectMore(audittelemetryschema.ScopeStringColumn)
} else {
for index := range query.SelectFields {
if query.SelectFields[index].Name == TimestampColumn || query.SelectFields[index].Name == IDColumn {
if query.SelectFields[index].Name == audittelemetryschema.TimestampColumn || query.SelectFields[index].Name == audittelemetryschema.IDColumn {
continue
}
@@ -250,7 +251,7 @@ func (b *auditQueryStatementBuilder) buildListQuery(
}
}
sb.From(fmt.Sprintf("%s.%s", DBName, AuditLogsTableName))
sb.From(fmt.Sprintf("%s.%s", audittelemetryschema.DBName, audittelemetryschema.AuditLogsTableName))
preparedWhereClause, err := b.addFilterCondition(ctx, orgID, sb, start, end, query, keys, variables)
if err != nil {
@@ -340,7 +341,7 @@ func (b *auditQueryStatementBuilder) buildTimeSeriesQuery(
sb.SelectMore(fmt.Sprintf("%s AS __result_%d", rewritten, i))
}
sb.From(fmt.Sprintf("%s.%s", DBName, AuditLogsTableName))
sb.From(fmt.Sprintf("%s.%s", audittelemetryschema.DBName, audittelemetryschema.AuditLogsTableName))
preparedWhereClause, err := b.addFilterCondition(ctx, orgID, sb, start, end, query, keys, variables)
if err != nil {
@@ -478,7 +479,7 @@ func (b *auditQueryStatementBuilder) buildScalarQuery(
}
}
sb.From(fmt.Sprintf("%s.%s", DBName, AuditLogsTableName))
sb.From(fmt.Sprintf("%s.%s", audittelemetryschema.DBName, audittelemetryschema.AuditLogsTableName))
preparedWhereClause, err := b.addFilterCondition(ctx, orgID, sb, start, end, query, keys, variables)
if err != nil {

View File

@@ -1,4 +1,4 @@
package telemetryaudit
package auditstatementbuilder
import (
"context"
@@ -8,6 +8,7 @@ import (
"github.com/SigNoz/signoz/pkg/flagger/flaggertest"
"github.com/SigNoz/signoz/pkg/instrumentation/instrumentationtest"
"github.com/SigNoz/signoz/pkg/querybuilder"
"github.com/SigNoz/signoz/pkg/telemetryschema/audittelemetryschema"
qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
"github.com/SigNoz/signoz/pkg/types/telemetrytypes/telemetrytypestest"
@@ -54,8 +55,8 @@ func newTestAuditStatementBuilder(t *testing.T) *auditQueryStatementBuilder {
mockMetadataStore := telemetrytypestest.NewMockMetadataStore()
mockMetadataStore.KeysMap = auditFieldKeyMap()
fm := NewFieldMapper()
cb := NewConditionBuilder(fm)
fm := audittelemetryschema.NewFieldMapper()
cb := audittelemetryschema.NewConditionBuilder(fm)
aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, fm, cb, fl)
return NewAuditQueryStatementBuilder(
@@ -64,7 +65,7 @@ func newTestAuditStatementBuilder(t *testing.T) *auditQueryStatementBuilder {
fm,
cb,
aggExprRewriter,
DefaultFullTextColumn,
audittelemetryschema.DefaultFullTextColumn,
fl,
)
}

View File

@@ -0,0 +1,41 @@
package statementbuilder
import (
"github.com/SigNoz/signoz/pkg/errors"
"github.com/SigNoz/signoz/pkg/factory"
)
// SkipResourceFingerprint configures when the resource fingerprint subquery is skipped in favor of main-table filtering.
type SkipResourceFingerprint struct {
Enabled bool `yaml:"enabled" mapstructure:"enabled"`
// If count of fingerprint is above threshold, skip the fingerprint subquery and filter on main table instead.
Threshold uint64 `yaml:"threshold" mapstructure:"threshold"`
}
// Config represents the configuration for the statement builders.
type Config struct {
// SkipResourceFingerprint configures when the resource fingerprint subquery is skipped in favor of main-table filtering.
SkipResourceFingerprint SkipResourceFingerprint `yaml:"skip_resource_fingerprint" mapstructure:"skip_resource_fingerprint"`
}
// NewConfigFactory creates a new config factory for the statement builders.
func NewConfigFactory() factory.ConfigFactory {
return factory.NewConfigFactory(factory.MustNewName("statementbuilder"), newConfig)
}
func newConfig() factory.Config {
return Config{
SkipResourceFingerprint: SkipResourceFingerprint{
Enabled: false,
Threshold: 100000,
},
}
}
// Validate validates the configuration.
func (c Config) Validate() error {
if c.SkipResourceFingerprint.Enabled && c.SkipResourceFingerprint.Threshold == 0 {
return errors.NewInvalidInputf(errors.CodeInvalidInput, "skip_resource_fingerprint.threshold must be > 0 when enabled")
}
return nil
}

View File

@@ -1,4 +1,4 @@
package telemetrylogs
package logsstatementbuilder
import (
"context"
@@ -8,6 +8,7 @@ import (
"github.com/SigNoz/signoz/pkg/flagger/flaggertest"
"github.com/SigNoz/signoz/pkg/instrumentation/instrumentationtest"
"github.com/SigNoz/signoz/pkg/querybuilder"
"github.com/SigNoz/signoz/pkg/telemetryschema/logstelemetryschema"
qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
"github.com/SigNoz/signoz/pkg/types/telemetrytypes/telemetrytypestest"
@@ -43,14 +44,14 @@ func TestStatementBuilderGroupByUnknownKey(t *testing.T) {
t.Run(c.name, func(t *testing.T) {
fl := flaggertest.WithUseJSONBody(t, c.useJSONBody)
mockMetadataStore := telemetrytypestest.NewMockMetadataStore()
mockMetadataStore.KeysMap = buildCompleteFieldKeyMap(releaseTime)
fm := NewFieldMapper(fl)
cb := NewConditionBuilder(fm, fl)
mockMetadataStore.KeysMap = logstelemetryschema.BuildCompleteFieldKeyMap(releaseTime)
fm := logstelemetryschema.NewFieldMapper(fl)
cb := logstelemetryschema.NewConditionBuilder(fm, fl)
aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, fm, cb, fl)
statementBuilder := NewLogQueryStatementBuilder(
instrumentationtest.New().ToProviderSettings(),
mockMetadataStore, fm, cb, aggExprRewriter,
DefaultFullTextColumn, fl, nil, false, 100000,
logstelemetryschema.DefaultFullTextColumn, fl, nil, false, 100000,
)
query := qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]{

View File

@@ -1,4 +1,4 @@
package telemetrylogs
package logsstatementbuilder
import (
"context"
@@ -11,6 +11,7 @@ import (
"github.com/SigNoz/signoz/pkg/flagger/flaggertest"
"github.com/SigNoz/signoz/pkg/instrumentation/instrumentationtest"
"github.com/SigNoz/signoz/pkg/querybuilder"
"github.com/SigNoz/signoz/pkg/telemetryschema/logstelemetryschema"
qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
"github.com/SigNoz/signoz/pkg/types/telemetrytypes/telemetrytypestest"
@@ -1225,7 +1226,7 @@ func TestJSONStmtBuilder_OrderBy(t *testing.T) {
func TestResourceAggrAndGroupBy_WithJSONEnabled(t *testing.T) {
statementBuilder, metadataStore := buildJSONTestStatementBuilder(t, false)
releaseTime := time.Date(2024, 1, 15, 10, 0, 0, 0, time.UTC)
keysMap := buildCompleteFieldKeyMap(releaseTime)
keysMap := logstelemetryschema.BuildCompleteFieldKeyMap(releaseTime)
for _, keys := range keysMap {
for _, key := range keys {
metadataStore.SetKey(key)
@@ -1287,7 +1288,7 @@ func TestResourceAggrAndGroupBy_WithJSONEnabled(t *testing.T) {
func buildTestTelemetryMetadataStore(t *testing.T, addIndexes bool) *telemetrytypestest.MockMetadataStore {
mockMetadataStore := telemetrytypestest.NewMockMetadataStore()
mockMetadataStore.SetStaticFields(IntrinsicFields)
mockMetadataStore.SetStaticFields(logstelemetryschema.IntrinsicFields)
types, _ := telemetrytypes.TestJSONTypeSet()
for path, fieldDataTypes := range types {
for _, fdt := range fieldDataTypes {
@@ -1307,15 +1308,15 @@ func buildTestTelemetryMetadataStore(t *testing.T, addIndexes bool) *telemetryty
Name: path,
FieldContext: telemetrytypes.FieldContextBody,
FieldDataType: fdt,
BaseColumn: LogsV2BodyV2Column,
IndexExpression: schemamigrator.JSONSubColumnIndexExpr(LogsV2BodyV2Column, path, jsonType.StringValue()),
BaseColumn: logstelemetryschema.LogsV2BodyV2Column,
IndexExpression: schemamigrator.JSONSubColumnIndexExpr(logstelemetryschema.LogsV2BodyV2Column, path, jsonType.StringValue()),
})
}
}
err := key.SetJSONAccessPlan(telemetrytypes.JSONColumnMetadata{
BaseColumn: LogsV2BodyV2Column,
PromotedColumn: LogsV2BodyPromotedColumn,
BaseColumn: logstelemetryschema.LogsV2BodyV2Column,
PromotedColumn: logstelemetryschema.LogsV2BodyPromotedColumn,
}, types)
require.NoError(t, err)
@@ -1331,8 +1332,8 @@ func buildJSONTestStatementBuilder(t *testing.T, addIndexes bool) (*logQueryStat
mockMetadataStore := buildTestTelemetryMetadataStore(t, addIndexes)
fl := flaggertest.WithUseJSONBody(t, true)
fm := NewFieldMapper(fl)
cb := NewConditionBuilder(fm, fl)
fm := logstelemetryschema.NewFieldMapper(fl)
cb := logstelemetryschema.NewConditionBuilder(fm, fl)
aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, fm, cb, fl)
@@ -1342,7 +1343,7 @@ func buildJSONTestStatementBuilder(t *testing.T, addIndexes bool) (*logQueryStat
fm,
cb,
aggExprRewriter,
DefaultFullTextColumn,
logstelemetryschema.DefaultFullTextColumn,
fl,
nil,
false,

View File

@@ -0,0 +1,31 @@
package logsstatementbuilder
import (
"github.com/SigNoz/signoz/pkg/factory"
"github.com/SigNoz/signoz/pkg/flagger"
"github.com/SigNoz/signoz/pkg/querybuilder"
"github.com/SigNoz/signoz/pkg/telemetryschema/logstelemetryschema"
"github.com/SigNoz/signoz/pkg/telemetrystore"
qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
)
// New builds the logs statement builder, internalizing the FieldMapper,
// ConditionBuilder, and AggExprRewriter.
func New(
settings factory.ProviderSettings,
metadataStore telemetrytypes.MetadataStore,
telemetryStore telemetrystore.TelemetryStore,
fl flagger.Flagger,
skipResourceFingerprintEnabled bool,
skipResourceFingerprintThreshold uint64,
) qbtypes.StatementBuilder[qbtypes.LogAggregation] {
fm := logstelemetryschema.NewFieldMapper(fl)
cb := logstelemetryschema.NewConditionBuilder(fm, fl)
aggExprRewriter := querybuilder.NewAggExprRewriter(settings, logstelemetryschema.DefaultFullTextColumn, fm, cb, fl)
return NewLogQueryStatementBuilder(
settings, metadataStore, fm, cb, aggExprRewriter, logstelemetryschema.DefaultFullTextColumn,
fl, telemetryStore, skipResourceFingerprintEnabled, skipResourceFingerprintThreshold,
)
}

View File

@@ -1,4 +1,4 @@
package telemetrylogs
package logsstatementbuilder
import (
"context"
@@ -10,7 +10,8 @@ import (
"github.com/SigNoz/signoz/pkg/factory"
"github.com/SigNoz/signoz/pkg/flagger"
"github.com/SigNoz/signoz/pkg/querybuilder"
"github.com/SigNoz/signoz/pkg/telemetryresourcefilter"
"github.com/SigNoz/signoz/pkg/statementbuilder/resourcefilter"
"github.com/SigNoz/signoz/pkg/telemetryschema/logstelemetryschema"
"github.com/SigNoz/signoz/pkg/telemetrystore"
"github.com/SigNoz/signoz/pkg/types/featuretypes"
qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
@@ -19,12 +20,22 @@ import (
"github.com/huandu/go-sqlbuilder"
)
const bodySearchDefaultWarning = "body searches default to `body.message:string`. Use `body.<key>` to search a different field inside body"
func bodyAliasExpression(bodyJSONEnabled bool) string {
if !bodyJSONEnabled {
return logstelemetryschema.LogsV2BodyColumn
}
return fmt.Sprintf("%s as body", logstelemetryschema.LogsV2BodyV2Column)
}
type logQueryStatementBuilder struct {
logger *slog.Logger
metadataStore telemetrytypes.MetadataStore
fm qbtypes.FieldMapper
cb qbtypes.ConditionBuilder
resourceFilterResolver *telemetryresourcefilter.ResourceFingerprintResolver[qbtypes.LogAggregation]
resourceFilterResolver *resourcefilter.ResourceFingerprintResolver[qbtypes.LogAggregation]
aggExprRewriter qbtypes.AggExprRewriter
fl flagger.Flagger
skipResourceFingerprintEnabled bool
@@ -46,12 +57,12 @@ func NewLogQueryStatementBuilder(
skipResourceFingerprintEnable bool,
skipResourceFingerprintThreshold uint64,
) *logQueryStatementBuilder {
logsSettings := factory.NewScopedProviderSettings(settings, "github.com/SigNoz/signoz/pkg/telemetrylogs")
logsSettings := factory.NewScopedProviderSettings(settings, "github.com/SigNoz/signoz/pkg/telemetryschema/logstelemetryschema")
resourceFilterResolver := telemetryresourcefilter.NewResolver[qbtypes.LogAggregation](
resourceFilterResolver := resourcefilter.NewResolver[qbtypes.LogAggregation](
settings,
DBName,
LogsResourceV2TableName,
logstelemetryschema.DBName,
logstelemetryschema.LogsResourceV2TableName,
telemetrytypes.SignalLogs,
telemetrytypes.SourceUnspecified,
metadataStore,
@@ -174,7 +185,7 @@ func getKeySelectors(query qbtypes.QueryBuilderQuery[qbtypes.LogAggregation], bo
// TODO(Piyush): Setup better for coming FTS support.
if bodyJSONEnabled {
for _, sel := range keySelectors {
if sel.Name == LogsV2BodyColumn {
if sel.Name == logstelemetryschema.LogsV2BodyColumn {
warnings = append(warnings, bodySearchDefaultWarning)
break
}
@@ -250,8 +261,8 @@ func (b *logQueryStatementBuilder) adjustKeys(ctx context.Context, keys map[stri
func (b *logQueryStatementBuilder) adjustKey(key *telemetrytypes.TelemetryFieldKey, keys map[string][]*telemetrytypes.TelemetryFieldKey) []string {
// First check if it matches with any intrinsic fields
var intrinsicOrCalculatedField telemetrytypes.TelemetryFieldKey
if _, ok := IntrinsicFields[key.Name]; ok {
intrinsicOrCalculatedField = IntrinsicFields[key.Name]
if _, ok := logstelemetryschema.IntrinsicFields[key.Name]; ok {
intrinsicOrCalculatedField = logstelemetryschema.IntrinsicFields[key.Name]
return querybuilder.AdjustKey(key, keys, &intrinsicOrCalculatedField)
}
@@ -284,28 +295,28 @@ func (b *logQueryStatementBuilder) buildListQuery(
}
// Select timestamp and id by default
sb.Select(LogsV2TimestampColumn)
sb.SelectMore(LogsV2IDColumn)
sb.Select(logstelemetryschema.LogsV2TimestampColumn)
sb.SelectMore(logstelemetryschema.LogsV2IDColumn)
if len(query.SelectFields) == 0 {
// Select all default columns
sb.SelectMore(LogsV2TraceIDColumn)
sb.SelectMore(LogsV2SpanIDColumn)
sb.SelectMore(LogsV2TraceFlagsColumn)
sb.SelectMore(LogsV2SeverityTextColumn)
sb.SelectMore(LogsV2SeverityNumberColumn)
sb.SelectMore(LogsV2ScopeNameColumn)
sb.SelectMore(LogsV2ScopeVersionColumn)
sb.SelectMore(logstelemetryschema.LogsV2TraceIDColumn)
sb.SelectMore(logstelemetryschema.LogsV2SpanIDColumn)
sb.SelectMore(logstelemetryschema.LogsV2TraceFlagsColumn)
sb.SelectMore(logstelemetryschema.LogsV2SeverityTextColumn)
sb.SelectMore(logstelemetryschema.LogsV2SeverityNumberColumn)
sb.SelectMore(logstelemetryschema.LogsV2ScopeNameColumn)
sb.SelectMore(logstelemetryschema.LogsV2ScopeVersionColumn)
sb.SelectMore(bodyAliasExpression(b.fl.BooleanOrEmpty(ctx, flagger.FeatureUseJSONBody, featuretypes.NewFlaggerEvaluationContext(orgID))))
sb.SelectMore(LogsV2AttributesStringColumn)
sb.SelectMore(LogsV2AttributesNumberColumn)
sb.SelectMore(LogsV2AttributesBoolColumn)
sb.SelectMore(LogsV2ResourcesStringColumn)
sb.SelectMore(LogsV2ScopeStringColumn)
sb.SelectMore(logstelemetryschema.LogsV2AttributesStringColumn)
sb.SelectMore(logstelemetryschema.LogsV2AttributesNumberColumn)
sb.SelectMore(logstelemetryschema.LogsV2AttributesBoolColumn)
sb.SelectMore(logstelemetryschema.LogsV2ResourcesStringColumn)
sb.SelectMore(logstelemetryschema.LogsV2ScopeStringColumn)
} else {
// Select specified columns
for index := range query.SelectFields {
if query.SelectFields[index].Name == LogsV2TimestampColumn || query.SelectFields[index].Name == LogsV2IDColumn {
if query.SelectFields[index].Name == logstelemetryschema.LogsV2TimestampColumn || query.SelectFields[index].Name == logstelemetryschema.LogsV2IDColumn {
continue
}
@@ -318,7 +329,7 @@ func (b *logQueryStatementBuilder) buildListQuery(
}
}
sb.From(fmt.Sprintf("%s.%s", DBName, LogsV2TableName))
sb.From(fmt.Sprintf("%s.%s", logstelemetryschema.DBName, logstelemetryschema.LogsV2TableName))
// Add filter conditions
preparedWhereClause, err := b.addFilterCondition(ctx, orgID, sb, start, end, query, keys, variables, skipResourceFilter)
@@ -424,7 +435,7 @@ func (b *logQueryStatementBuilder) buildTimeSeriesQuery(
}
// Add FROM clause
sb.From(fmt.Sprintf("%s.%s", DBName, LogsV2TableName))
sb.From(fmt.Sprintf("%s.%s", logstelemetryschema.DBName, logstelemetryschema.LogsV2TableName))
preparedWhereClause, err := b.addFilterCondition(ctx, orgID, sb, start, end, query, keys, variables, skipResourceFilter)
@@ -592,7 +603,7 @@ func (b *logQueryStatementBuilder) buildScalarQuery(
}
}
sb.From(fmt.Sprintf("%s.%s", DBName, LogsV2TableName))
sb.From(fmt.Sprintf("%s.%s", logstelemetryschema.DBName, logstelemetryschema.LogsV2TableName))
// Add filter conditions
preparedWhereClause, err := b.addFilterCondition(ctx, orgID, sb, start, end, query, keys, variables, skipResourceFilter)

View File

@@ -1,4 +1,4 @@
package telemetrylogs
package logsstatementbuilder
import (
"context"
@@ -11,6 +11,7 @@ import (
"github.com/SigNoz/signoz/pkg/flagger/flaggertest"
"github.com/SigNoz/signoz/pkg/instrumentation/instrumentationtest"
"github.com/SigNoz/signoz/pkg/querybuilder"
"github.com/SigNoz/signoz/pkg/telemetryschema/logstelemetryschema"
"github.com/SigNoz/signoz/pkg/telemetrystore"
"github.com/SigNoz/signoz/pkg/telemetrystore/telemetrystoretest"
qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
@@ -212,12 +213,12 @@ func TestStatementBuilderTimeSeries(t *testing.T) {
fl := flaggertest.New(t)
mockMetadataStore := telemetrytypestest.NewMockMetadataStore()
keysMap := buildCompleteFieldKeyMap(releaseTime)
keysMap := logstelemetryschema.BuildCompleteFieldKeyMap(releaseTime)
mockMetadataStore.KeysMap = keysMap
fm := NewFieldMapper(fl)
cb := NewConditionBuilder(fm, fl)
fm := logstelemetryschema.NewFieldMapper(fl)
cb := logstelemetryschema.NewConditionBuilder(fm, fl)
aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, fm, cb, fl)
@@ -227,7 +228,7 @@ func TestStatementBuilderTimeSeries(t *testing.T) {
fm,
cb,
aggExprRewriter,
DefaultFullTextColumn,
logstelemetryschema.DefaultFullTextColumn,
fl,
nil,
false,
@@ -355,12 +356,12 @@ func TestStatementBuilderListQuery(t *testing.T) {
ctx := context.Background()
fl := flaggertest.New(t)
mockMetadataStore := telemetrytypestest.NewMockMetadataStore()
fm := NewFieldMapper(fl)
fm := logstelemetryschema.NewFieldMapper(fl)
// Create a test release time
releaseTime := time.Date(2024, 1, 15, 10, 0, 0, 0, time.UTC)
mockMetadataStore.KeysMap = buildCompleteFieldKeyMap(releaseTime)
cb := NewConditionBuilder(fm, fl)
mockMetadataStore.KeysMap = logstelemetryschema.BuildCompleteFieldKeyMap(releaseTime)
cb := logstelemetryschema.NewConditionBuilder(fm, fl)
aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, fm, cb, fl)
@@ -370,7 +371,7 @@ func TestStatementBuilderListQuery(t *testing.T) {
fm,
cb,
aggExprRewriter,
DefaultFullTextColumn,
logstelemetryschema.DefaultFullTextColumn,
fl,
nil,
false,
@@ -507,11 +508,11 @@ func TestStatementBuilderListQueryResourceTests(t *testing.T) {
ctx := context.Background()
fl := flaggertest.New(t)
mockMetadataStore := telemetrytypestest.NewMockMetadataStore()
fm := NewFieldMapper(fl)
fm := logstelemetryschema.NewFieldMapper(fl)
// Create a test release time
releaseTime := time.Date(2024, 1, 15, 10, 0, 0, 0, time.UTC)
mockMetadataStore.KeysMap = buildCompleteFieldKeyMap(releaseTime)
cb := NewConditionBuilder(fm, fl)
mockMetadataStore.KeysMap = logstelemetryschema.BuildCompleteFieldKeyMap(releaseTime)
cb := logstelemetryschema.NewConditionBuilder(fm, fl)
aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, fm, cb, fl)
@@ -521,7 +522,7 @@ func TestStatementBuilderListQueryResourceTests(t *testing.T) {
fm,
cb,
aggExprRewriter,
DefaultFullTextColumn,
logstelemetryschema.DefaultFullTextColumn,
fl,
nil,
false,
@@ -585,11 +586,11 @@ func TestStatementBuilderTimeSeriesBodyGroupBy(t *testing.T) {
ctx := context.Background()
fl := flaggertest.New(t)
mockMetadataStore := telemetrytypestest.NewMockMetadataStore()
fm := NewFieldMapper(fl)
fm := logstelemetryschema.NewFieldMapper(fl)
// Create a test release time
releaseTime := time.Date(2024, 1, 15, 10, 0, 0, 0, time.UTC)
mockMetadataStore.KeysMap = buildCompleteFieldKeyMap(releaseTime)
cb := NewConditionBuilder(fm, fl)
mockMetadataStore.KeysMap = logstelemetryschema.BuildCompleteFieldKeyMap(releaseTime)
cb := logstelemetryschema.NewConditionBuilder(fm, fl)
aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, fm, cb, fl)
@@ -599,7 +600,7 @@ func TestStatementBuilderTimeSeriesBodyGroupBy(t *testing.T) {
fm,
cb,
aggExprRewriter,
DefaultFullTextColumn,
logstelemetryschema.DefaultFullTextColumn,
fl,
nil,
false,
@@ -684,9 +685,9 @@ func TestStatementBuilderListQueryServiceCollision(t *testing.T) {
ctx := context.Background()
mockMetadataStore := telemetrytypestest.NewMockMetadataStore()
fl := flaggertest.New(t)
fm := NewFieldMapper(fl)
mockMetadataStore.KeysMap = buildCompleteFieldKeyMapCollision()
cb := NewConditionBuilder(fm, fl)
fm := logstelemetryschema.NewFieldMapper(fl)
mockMetadataStore.KeysMap = logstelemetryschema.BuildCompleteFieldKeyMapCollision()
cb := logstelemetryschema.NewConditionBuilder(fm, fl)
aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, fm, cb, fl)
@@ -696,7 +697,7 @@ func TestStatementBuilderListQueryServiceCollision(t *testing.T) {
fm,
cb,
aggExprRewriter,
DefaultFullTextColumn,
logstelemetryschema.DefaultFullTextColumn,
fl,
nil,
false,
@@ -740,8 +741,8 @@ func TestAdjustKey(t *testing.T) {
FieldContext: telemetrytypes.FieldContextUnspecified,
FieldDataType: telemetrytypes.FieldDataTypeUnspecified,
},
keysMap: buildCompleteFieldKeyMap(releaseTime),
expectedKey: IntrinsicFields["severity_text"],
keysMap: logstelemetryschema.BuildCompleteFieldKeyMap(releaseTime),
expectedKey: logstelemetryschema.IntrinsicFields["severity_text"],
},
{
name: "intrinsic field with other key match - no override",
@@ -777,7 +778,7 @@ func TestAdjustKey(t *testing.T) {
FieldContext: telemetrytypes.FieldContextBody,
FieldDataType: telemetrytypes.FieldDataTypeUnspecified,
},
keysMap: buildCompleteFieldKeyMap(releaseTime),
keysMap: logstelemetryschema.BuildCompleteFieldKeyMap(releaseTime),
expectedKey: telemetrytypes.TelemetryFieldKey{
Name: "severity_number",
FieldContext: telemetrytypes.FieldContextBody,
@@ -791,8 +792,8 @@ func TestAdjustKey(t *testing.T) {
FieldContext: telemetrytypes.FieldContextUnspecified,
FieldDataType: telemetrytypes.FieldDataTypeUnspecified,
},
keysMap: buildCompleteFieldKeyMap(releaseTime),
expectedKey: *buildCompleteFieldKeyMap(releaseTime)["service.name"][0],
keysMap: logstelemetryschema.BuildCompleteFieldKeyMap(releaseTime),
expectedKey: *logstelemetryschema.BuildCompleteFieldKeyMap(releaseTime)["service.name"][0],
},
{
name: "single matching key with incorrect context specified - no override",
@@ -801,7 +802,7 @@ func TestAdjustKey(t *testing.T) {
FieldContext: telemetrytypes.FieldContextAttribute,
FieldDataType: telemetrytypes.FieldDataTypeUnspecified,
},
keysMap: buildCompleteFieldKeyMap(releaseTime),
keysMap: logstelemetryschema.BuildCompleteFieldKeyMap(releaseTime),
expectedKey: telemetrytypes.TelemetryFieldKey{
Name: "service.name",
FieldContext: telemetrytypes.FieldContextAttribute,
@@ -815,8 +816,8 @@ func TestAdjustKey(t *testing.T) {
FieldContext: telemetrytypes.FieldContextUnspecified,
FieldDataType: telemetrytypes.FieldDataTypeUnspecified,
},
keysMap: buildCompleteFieldKeyMap(releaseTime),
expectedKey: *buildCompleteFieldKeyMap(releaseTime)["service.name"][0],
keysMap: logstelemetryschema.BuildCompleteFieldKeyMap(releaseTime),
expectedKey: *logstelemetryschema.BuildCompleteFieldKeyMap(releaseTime)["service.name"][0],
},
{
name: "multiple matching keys - all materialized",
@@ -825,7 +826,7 @@ func TestAdjustKey(t *testing.T) {
FieldContext: telemetrytypes.FieldContextUnspecified,
FieldDataType: telemetrytypes.FieldDataTypeUnspecified,
},
keysMap: buildCompleteFieldKeyMap(releaseTime),
keysMap: logstelemetryschema.BuildCompleteFieldKeyMap(releaseTime),
expectedKey: telemetrytypes.TelemetryFieldKey{
Name: "multi.mat.key",
FieldDataType: telemetrytypes.FieldDataTypeString,
@@ -839,7 +840,7 @@ func TestAdjustKey(t *testing.T) {
FieldContext: telemetrytypes.FieldContextUnspecified,
FieldDataType: telemetrytypes.FieldDataTypeUnspecified,
},
keysMap: buildCompleteFieldKeyMap(releaseTime),
keysMap: logstelemetryschema.BuildCompleteFieldKeyMap(releaseTime),
expectedKey: telemetrytypes.TelemetryFieldKey{
Name: "mixed.materialization.key",
FieldDataType: telemetrytypes.FieldDataTypeString,
@@ -853,8 +854,8 @@ func TestAdjustKey(t *testing.T) {
FieldContext: telemetrytypes.FieldContextAttribute,
FieldDataType: telemetrytypes.FieldDataTypeUnspecified,
},
keysMap: buildCompleteFieldKeyMap(releaseTime),
expectedKey: *buildCompleteFieldKeyMap(releaseTime)["mixed.materialization.key"][0],
keysMap: logstelemetryschema.BuildCompleteFieldKeyMap(releaseTime),
expectedKey: *logstelemetryschema.BuildCompleteFieldKeyMap(releaseTime)["mixed.materialization.key"][0],
},
{
name: "no matching keys - unknown field",
@@ -863,7 +864,7 @@ func TestAdjustKey(t *testing.T) {
FieldContext: telemetrytypes.FieldContextUnspecified,
FieldDataType: telemetrytypes.FieldDataTypeUnspecified,
},
keysMap: buildCompleteFieldKeyMap(releaseTime),
keysMap: logstelemetryschema.BuildCompleteFieldKeyMap(releaseTime),
expectedKey: telemetrytypes.TelemetryFieldKey{
Name: "unknown.field",
FieldContext: telemetrytypes.FieldContextUnspecified,
@@ -878,7 +879,7 @@ func TestAdjustKey(t *testing.T) {
FieldContext: telemetrytypes.FieldContextAttribute,
FieldDataType: telemetrytypes.FieldDataTypeUnspecified,
},
keysMap: buildCompleteFieldKeyMap(releaseTime),
keysMap: logstelemetryschema.BuildCompleteFieldKeyMap(releaseTime),
expectedKey: telemetrytypes.TelemetryFieldKey{
Name: "unknown.field",
FieldContext: telemetrytypes.FieldContextAttribute,
@@ -893,8 +894,8 @@ func TestAdjustKey(t *testing.T) {
FieldContext: telemetrytypes.FieldContextUnspecified,
FieldDataType: telemetrytypes.FieldDataTypeUnspecified,
},
keysMap: buildCompleteFieldKeyMap(releaseTime),
expectedKey: *buildCompleteFieldKeyMap(releaseTime)["mat.key"][0],
keysMap: logstelemetryschema.BuildCompleteFieldKeyMap(releaseTime),
expectedKey: *logstelemetryschema.BuildCompleteFieldKeyMap(releaseTime)["mat.key"][0],
},
{
name: "non-materialized field",
@@ -903,16 +904,16 @@ func TestAdjustKey(t *testing.T) {
FieldContext: telemetrytypes.FieldContextUnspecified,
FieldDataType: telemetrytypes.FieldDataTypeUnspecified,
},
keysMap: buildCompleteFieldKeyMap(releaseTime),
expectedKey: *buildCompleteFieldKeyMap(releaseTime)["user.id"][0],
keysMap: logstelemetryschema.BuildCompleteFieldKeyMap(releaseTime),
expectedKey: *logstelemetryschema.BuildCompleteFieldKeyMap(releaseTime)["user.id"][0],
},
}
fl := flaggertest.New(t)
fm := NewFieldMapper(fl)
fm := logstelemetryschema.NewFieldMapper(fl)
mockMetadataStore := telemetrytypestest.NewMockMetadataStore()
mockMetadataStore.KeysMap = buildCompleteFieldKeyMapCollision()
cb := NewConditionBuilder(fm, fl)
mockMetadataStore.KeysMap = logstelemetryschema.BuildCompleteFieldKeyMapCollision()
cb := logstelemetryschema.NewConditionBuilder(fm, fl)
aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, fm, cb, fl)
@@ -922,7 +923,7 @@ func TestAdjustKey(t *testing.T) {
fm,
cb,
aggExprRewriter,
DefaultFullTextColumn,
logstelemetryschema.DefaultFullTextColumn,
fl,
nil,
false,
@@ -1054,11 +1055,11 @@ func TestStmtBuilderBodyField(t *testing.T) {
for _, c := range cases {
t.Run(c.name, func(t *testing.T) {
fl := flaggertest.WithUseJSONBody(t, c.enableUseJSONBody)
fm := NewFieldMapper(fl)
cb := NewConditionBuilder(fm, fl)
fm := logstelemetryschema.NewFieldMapper(fl)
cb := logstelemetryschema.NewConditionBuilder(fm, fl)
// build the key map
mockMetadataStore := telemetrytypestest.NewMockMetadataStore()
for _, field := range IntrinsicFields {
for _, field := range logstelemetryschema.IntrinsicFields {
f := field
mockMetadataStore.KeysMap[field.Name] = append(mockMetadataStore.KeysMap[field.Name], &f)
}
@@ -1069,7 +1070,7 @@ func TestStmtBuilderBodyField(t *testing.T) {
fm,
cb,
aggExprRewriter,
DefaultFullTextColumn,
logstelemetryschema.DefaultFullTextColumn,
fl,
nil,
false,
@@ -1155,11 +1156,11 @@ func TestStmtBuilderBodyFullTextSearch(t *testing.T) {
for _, c := range cases {
t.Run(c.name, func(t *testing.T) {
fl := flaggertest.WithUseJSONBody(t, c.enableUseJSONBody)
fm := NewFieldMapper(fl)
cb := NewConditionBuilder(fm, fl)
fm := logstelemetryschema.NewFieldMapper(fl)
cb := logstelemetryschema.NewConditionBuilder(fm, fl)
// build the key map
mockMetadataStore := telemetrytypestest.NewMockMetadataStore()
for _, field := range IntrinsicFields {
for _, field := range logstelemetryschema.IntrinsicFields {
f := field
mockMetadataStore.KeysMap[field.Name] = append(mockMetadataStore.KeysMap[field.Name], &f)
}
@@ -1170,7 +1171,7 @@ func TestStmtBuilderBodyFullTextSearch(t *testing.T) {
fm,
cb,
aggExprRewriter,
DefaultFullTextColumn,
logstelemetryschema.DefaultFullTextColumn,
fl,
nil,
false,
@@ -1273,14 +1274,14 @@ func newSkipResourceFingerprintLogsBuilder(
t.Helper()
fl := flaggertest.New(t)
fm := NewFieldMapper(fl)
cb := NewConditionBuilder(fm, fl)
fm := logstelemetryschema.NewFieldMapper(fl)
cb := logstelemetryschema.NewConditionBuilder(fm, fl)
mockMetadataStore := telemetrytypestest.NewMockMetadataStore()
mockMetadataStore.KeysMap = buildCompleteFieldKeyMap(time.Date(2024, 1, 15, 10, 0, 0, 0, time.UTC))
mockMetadataStore.KeysMap = logstelemetryschema.BuildCompleteFieldKeyMap(time.Date(2024, 1, 15, 10, 0, 0, 0, time.UTC))
aggExprRewriter := querybuilder.NewAggExprRewriter(
instrumentationtest.New().ToProviderSettings(),
DefaultFullTextColumn,
logstelemetryschema.DefaultFullTextColumn,
fm,
cb,
fl,
@@ -1292,7 +1293,7 @@ func newSkipResourceFingerprintLogsBuilder(
fm,
cb,
aggExprRewriter,
DefaultFullTextColumn,
logstelemetryschema.DefaultFullTextColumn,
fl,
telemetryStore,
skipEnable,

View File

@@ -0,0 +1,21 @@
package meterstatementbuilder
import (
"github.com/SigNoz/signoz/pkg/factory"
"github.com/SigNoz/signoz/pkg/statementbuilder/metricsstatementbuilder"
"github.com/SigNoz/signoz/pkg/telemetryschema/metricstelemetryschema"
qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
)
// New builds the meter statement builder. It reuses the metrics FieldMapper and
// ConditionBuilder, and delegates the final SELECT to the provided metrics statement builder.
func New(
settings factory.ProviderSettings,
metadataStore telemetrytypes.MetadataStore,
metricsStatementBuilder *metricsstatementbuilder.StatementBuilder,
) qbtypes.StatementBuilder[qbtypes.MetricAggregation] {
fm := metricstelemetryschema.NewFieldMapper()
cb := metricstelemetryschema.NewConditionBuilder(fm)
return NewMeterQueryStatementBuilder(settings, metadataStore, fm, cb, metricsStatementBuilder)
}

View File

@@ -1,4 +1,4 @@
package telemetrymeter
package meterstatementbuilder
import (
"context"
@@ -8,7 +8,8 @@ import (
"github.com/SigNoz/signoz/pkg/errors"
"github.com/SigNoz/signoz/pkg/factory"
"github.com/SigNoz/signoz/pkg/querybuilder"
"github.com/SigNoz/signoz/pkg/telemetrymetrics"
"github.com/SigNoz/signoz/pkg/statementbuilder/metricsstatementbuilder"
"github.com/SigNoz/signoz/pkg/telemetryschema/metertelemetryschema"
"github.com/SigNoz/signoz/pkg/types/metrictypes"
qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
@@ -21,7 +22,7 @@ type meterQueryStatementBuilder struct {
metadataStore telemetrytypes.MetadataStore
fm qbtypes.FieldMapper
cb qbtypes.ConditionBuilder
metricsStatementBuilder *telemetrymetrics.MetricQueryStatementBuilder
metricsStatementBuilder *metricsstatementbuilder.StatementBuilder
}
var _ qbtypes.StatementBuilder[qbtypes.MetricAggregation] = (*meterQueryStatementBuilder)(nil)
@@ -31,9 +32,9 @@ func NewMeterQueryStatementBuilder(
metadataStore telemetrytypes.MetadataStore,
fieldMapper qbtypes.FieldMapper,
conditionBuilder qbtypes.ConditionBuilder,
metricsStatementBuilder *telemetrymetrics.MetricQueryStatementBuilder,
metricsStatementBuilder *metricsstatementbuilder.StatementBuilder,
) *meterQueryStatementBuilder {
metricsSettings := factory.NewScopedProviderSettings(settings, "github.com/SigNoz/signoz/pkg/telemetrymeter")
metricsSettings := factory.NewScopedProviderSettings(settings, "github.com/SigNoz/signoz/pkg/telemetryschema/metertelemetryschema")
return &meterQueryStatementBuilder{
logger: metricsSettings.Logger(),
@@ -53,7 +54,7 @@ func (b *meterQueryStatementBuilder) Build(
query qbtypes.QueryBuilderQuery[qbtypes.MetricAggregation],
variables map[string]qbtypes.VariableItem,
) (*qbtypes.Statement, error) {
keySelectors := telemetrymetrics.GetKeySelectors(query)
keySelectors := metricsstatementbuilder.GetKeySelectors(query)
keys, _, err := b.metadataStore.GetKeysMulti(ctx, orgID, keySelectors)
if err != nil {
return nil, err
@@ -133,8 +134,8 @@ func (b *meterQueryStatementBuilder) buildTemporalAggDeltaFastPath(
sb.SelectMore(col)
}
tbl := WhichSamplesTableToUse(start, end, query.Aggregations[0].Type, query.Aggregations[0].TimeAggregation, query.Aggregations[0].TableHints)
aggCol, err := AggregationColumnForSamplesTable(start, end, query.Aggregations[0].Type, query.Aggregations[0].Temporality, query.Aggregations[0].TimeAggregation, query.Aggregations[0].TableHints)
tbl := metertelemetryschema.WhichSamplesTableToUse(start, end, query.Aggregations[0].Type, query.Aggregations[0].TimeAggregation, query.Aggregations[0].TableHints)
aggCol, err := metertelemetryschema.AggregationColumnForSamplesTable(start, end, query.Aggregations[0].Type, query.Aggregations[0].Temporality, query.Aggregations[0].TimeAggregation, query.Aggregations[0].TableHints)
if err != nil {
return "", nil, err
}
@@ -143,7 +144,7 @@ func (b *meterQueryStatementBuilder) buildTemporalAggDeltaFastPath(
}
sb.SelectMore(fmt.Sprintf("%s AS value", aggCol))
sb.From(fmt.Sprintf("%s.%s AS points", DBName, tbl))
sb.From(fmt.Sprintf("%s.%s AS points", metertelemetryschema.DBName, tbl))
sb.Where(
sb.In("metric_name", query.Aggregations[0].MetricName),
sb.GTE("unix_milli", start),
@@ -222,8 +223,8 @@ func (b *meterQueryStatementBuilder) buildTemporalAggDelta(
sb.SelectMore(col)
}
tbl := WhichSamplesTableToUse(start, end, query.Aggregations[0].Type, query.Aggregations[0].TimeAggregation, query.Aggregations[0].TableHints)
aggCol, err := AggregationColumnForSamplesTable(start, end, query.Aggregations[0].Type, query.Aggregations[0].Temporality,
tbl := metertelemetryschema.WhichSamplesTableToUse(start, end, query.Aggregations[0].Type, query.Aggregations[0].TimeAggregation, query.Aggregations[0].TableHints)
aggCol, err := metertelemetryschema.AggregationColumnForSamplesTable(start, end, query.Aggregations[0].Type, query.Aggregations[0].Temporality,
query.Aggregations[0].TimeAggregation, query.Aggregations[0].TableHints)
if err != nil {
return "", nil, err
@@ -234,7 +235,7 @@ func (b *meterQueryStatementBuilder) buildTemporalAggDelta(
sb.SelectMore(fmt.Sprintf("%s AS per_series_value", aggCol))
sb.From(fmt.Sprintf("%s.%s AS points", DBName, tbl))
sb.From(fmt.Sprintf("%s.%s AS points", metertelemetryschema.DBName, tbl))
sb.Where(
sb.In("metric_name", query.Aggregations[0].MetricName),
sb.GTE("unix_milli", start),
@@ -300,14 +301,14 @@ func (b *meterQueryStatementBuilder) buildTemporalAggCumulativeOrUnspecified(
baseSb.SelectMore(col)
}
tbl := WhichSamplesTableToUse(start, end, query.Aggregations[0].Type, query.Aggregations[0].TimeAggregation, query.Aggregations[0].TableHints)
aggCol, err := AggregationColumnForSamplesTable(start, end, query.Aggregations[0].Type, query.Aggregations[0].Temporality, query.Aggregations[0].TimeAggregation, query.Aggregations[0].TableHints)
tbl := metertelemetryschema.WhichSamplesTableToUse(start, end, query.Aggregations[0].Type, query.Aggregations[0].TimeAggregation, query.Aggregations[0].TableHints)
aggCol, err := metertelemetryschema.AggregationColumnForSamplesTable(start, end, query.Aggregations[0].Type, query.Aggregations[0].Temporality, query.Aggregations[0].TimeAggregation, query.Aggregations[0].TableHints)
if err != nil {
return "", nil, err
}
baseSb.SelectMore(fmt.Sprintf("%s AS per_series_value", aggCol))
baseSb.From(fmt.Sprintf("%s.%s AS points", DBName, tbl))
baseSb.From(fmt.Sprintf("%s.%s AS points", metertelemetryschema.DBName, tbl))
baseSb.Where(
baseSb.In("metric_name", query.Aggregations[0].MetricName),
baseSb.GTE("unix_milli", start),
@@ -350,7 +351,7 @@ func (b *meterQueryStatementBuilder) buildTemporalAggCumulativeOrUnspecified(
for _, g := range query.GroupBy {
wrapped.SelectMore(fmt.Sprintf("`%s`", g.Name))
}
wrapped.SelectMore(fmt.Sprintf("%s AS per_series_value", telemetrymetrics.RateTmpl))
wrapped.SelectMore(fmt.Sprintf("%s AS per_series_value", metricsstatementbuilder.RateTmpl))
wrapped.From(fmt.Sprintf("(%s) WINDOW rate_window AS (PARTITION BY fingerprint ORDER BY fingerprint, ts)", innerQuery))
q, args := wrapped.BuildWithFlavor(sqlbuilder.ClickHouse, innerArgs...)
return fmt.Sprintf("__temporal_aggregation_cte AS (%s)", q), args, nil
@@ -361,7 +362,7 @@ func (b *meterQueryStatementBuilder) buildTemporalAggCumulativeOrUnspecified(
for _, g := range query.GroupBy {
wrapped.SelectMore(fmt.Sprintf("`%s`", g.Name))
}
wrapped.SelectMore(fmt.Sprintf("%s AS per_series_value", telemetrymetrics.IncreaseTmpl))
wrapped.SelectMore(fmt.Sprintf("%s AS per_series_value", metricsstatementbuilder.IncreaseTmpl))
wrapped.From(fmt.Sprintf("(%s) WINDOW rate_window AS (PARTITION BY fingerprint ORDER BY fingerprint, ts)", innerQuery))
q, args := wrapped.BuildWithFlavor(sqlbuilder.ClickHouse, innerArgs...)
return fmt.Sprintf("__temporal_aggregation_cte AS (%s)", q), args, nil

View File

@@ -1,4 +1,4 @@
package telemetrymeter
package meterstatementbuilder
import (
"context"
@@ -7,7 +7,8 @@ import (
"github.com/SigNoz/signoz/pkg/flagger/flaggertest"
"github.com/SigNoz/signoz/pkg/instrumentation/instrumentationtest"
"github.com/SigNoz/signoz/pkg/telemetrymetrics"
"github.com/SigNoz/signoz/pkg/statementbuilder/metricsstatementbuilder"
"github.com/SigNoz/signoz/pkg/telemetryschema/metricstelemetryschema"
"github.com/SigNoz/signoz/pkg/types/metrictypes"
qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
@@ -158,8 +159,8 @@ func TestStatementBuilder(t *testing.T) {
},
}
fm := telemetrymetrics.NewFieldMapper()
cb := telemetrymetrics.NewConditionBuilder(fm)
fm := metricstelemetryschema.NewFieldMapper()
cb := metricstelemetryschema.NewConditionBuilder(fm)
mockMetadataStore := telemetrytypestest.NewMockMetadataStore()
keys, err := telemetrytypestest.LoadFieldKeysFromJSON("testdata/keys_map.json")
if err != nil {
@@ -169,7 +170,7 @@ func TestStatementBuilder(t *testing.T) {
flagger := flaggertest.New(t)
metricStmtBuilder := telemetrymetrics.NewMetricQueryStatementBuilder(instrumentationtest.New().ToProviderSettings(), mockMetadataStore, fm, cb, flagger)
metricStmtBuilder := metricsstatementbuilder.NewMetricQueryStatementBuilder(instrumentationtest.New().ToProviderSettings(), mockMetadataStore, fm, cb, flagger)
statementBuilder := NewMeterQueryStatementBuilder(
instrumentationtest.New().ToProviderSettings(),

View File

@@ -0,0 +1,20 @@
package metricsstatementbuilder
import (
"github.com/SigNoz/signoz/pkg/factory"
"github.com/SigNoz/signoz/pkg/flagger"
"github.com/SigNoz/signoz/pkg/telemetryschema/metricstelemetryschema"
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
)
// New builds the metrics statement builder, internalizing the FieldMapper and
// ConditionBuilder. It returns the concrete type so the meter builder can reuse it.
func New(
settings factory.ProviderSettings,
metadataStore telemetrytypes.MetadataStore,
fl flagger.Flagger,
) *StatementBuilder {
fm := metricstelemetryschema.NewFieldMapper()
cb := metricstelemetryschema.NewConditionBuilder(fm)
return NewMetricQueryStatementBuilder(settings, metadataStore, fm, cb, fl)
}

View File

@@ -1,4 +1,4 @@
package telemetrymetrics
package metricsstatementbuilder
import (
"context"
@@ -7,6 +7,7 @@ import (
"github.com/SigNoz/signoz/pkg/flagger"
"github.com/SigNoz/signoz/pkg/instrumentation/instrumentationtest"
"github.com/SigNoz/signoz/pkg/telemetryschema/metricstelemetryschema"
"github.com/SigNoz/signoz/pkg/types/metrictypes"
qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
@@ -118,8 +119,8 @@ func TestReducedStatementBuilder(t *testing.T) {
},
}
fm := NewFieldMapper()
cb := NewConditionBuilder(fm)
fm := metricstelemetryschema.NewFieldMapper()
cb := metricstelemetryschema.NewConditionBuilder(fm)
fl, err := flagger.New(context.Background(), instrumentationtest.New().ToProviderSettings(), flagger.Config{}, flagger.MustNewRegistry())
require.NoError(t, err)
sb := NewMetricQueryStatementBuilder(instrumentationtest.New().ToProviderSettings(), telemetrytypestest.NewMockMetadataStore(), fm, cb, fl)

View File

@@ -1,4 +1,4 @@
package telemetrymetrics
package metricsstatementbuilder
import (
"context"
@@ -9,6 +9,7 @@ import (
"github.com/SigNoz/signoz/pkg/factory"
"github.com/SigNoz/signoz/pkg/flagger"
"github.com/SigNoz/signoz/pkg/querybuilder"
"github.com/SigNoz/signoz/pkg/telemetryschema/metricstelemetryschema"
"github.com/SigNoz/signoz/pkg/types/metrictypes"
qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
@@ -29,7 +30,7 @@ const (
OthersMultiTemporality = `IF(LOWER(temporality) LIKE LOWER('delta'), %s, %s) AS per_series_value`
)
type MetricQueryStatementBuilder struct {
type StatementBuilder struct {
logger *slog.Logger
metadataStore telemetrytypes.MetadataStore
fm qbtypes.FieldMapper
@@ -37,7 +38,7 @@ type MetricQueryStatementBuilder struct {
flagger flagger.Flagger
}
var _ qbtypes.StatementBuilder[qbtypes.MetricAggregation] = (*MetricQueryStatementBuilder)(nil)
var _ qbtypes.StatementBuilder[qbtypes.MetricAggregation] = (*StatementBuilder)(nil)
func NewMetricQueryStatementBuilder(
settings factory.ProviderSettings,
@@ -45,9 +46,9 @@ func NewMetricQueryStatementBuilder(
fieldMapper qbtypes.FieldMapper,
conditionBuilder qbtypes.ConditionBuilder,
flagger flagger.Flagger,
) *MetricQueryStatementBuilder {
metricsSettings := factory.NewScopedProviderSettings(settings, "github.com/SigNoz/signoz/pkg/telemetrymetrics")
return &MetricQueryStatementBuilder{
) *StatementBuilder {
metricsSettings := factory.NewScopedProviderSettings(settings, "github.com/SigNoz/signoz/pkg/telemetryschema/metricstelemetryschema")
return &StatementBuilder{
logger: metricsSettings.Logger(),
metadataStore: metadataStore,
fm: fieldMapper,
@@ -89,7 +90,7 @@ func GetKeySelectors(query qbtypes.QueryBuilderQuery[qbtypes.MetricAggregation])
return keySelectors
}
func (b *MetricQueryStatementBuilder) Build(
func (b *StatementBuilder) Build(
ctx context.Context,
orgID valuer.UUID,
start uint64,
@@ -126,7 +127,7 @@ func (b *MetricQueryStatementBuilder) Build(
return b.buildPipelineStatement(ctx, orgID, start, end, query, keys, variables)
}
func (b *MetricQueryStatementBuilder) buildPipelineStatement(
func (b *StatementBuilder) buildPipelineStatement(
ctx context.Context,
orgID valuer.UUID,
start, end uint64,
@@ -188,11 +189,11 @@ func (b *MetricQueryStatementBuilder) buildPipelineStatement(
// shaped exactly like samples_v4 / time_series_v4, so once the table names are
// chosen the rest of the pipeline is unchanged.
useBuffer := agg.Reduced &&
end-start < oneDayInMilliseconds &&
start >= uint64(time.Now().UnixMilli())-oneDayInMilliseconds
end-start < metricstelemetryschema.OneDayInMilliseconds &&
start >= uint64(time.Now().UnixMilli())-metricstelemetryschema.OneDayInMilliseconds
samplesTable, _ := WhichSamplesTableToUse(start, end, agg.Type, agg.TimeAggregation, useBuffer, agg.TableHints)
tsStart, tsEnd, _, tsTable := WhichTSTableToUse(start, end, useBuffer, agg.TableHints)
samplesTable, _ := metricstelemetryschema.WhichSamplesTableToUse(start, end, agg.Type, agg.TimeAggregation, useBuffer, agg.TableHints)
tsStart, tsEnd, _, tsTable := metricstelemetryschema.WhichTSTableToUse(start, end, useBuffer, agg.TableHints)
var timeSeriesCTE string
var timeSeriesCTEArgs []any
@@ -232,7 +233,7 @@ func (b *MetricQueryStatementBuilder) buildPipelineStatement(
var tsCTE string
var tsArgs []any
// time series rows are written on hour boundaries
tsStart := start - (start % oneHourInMilliseconds)
tsStart := start - (start % metricstelemetryschema.OneHourInMilliseconds)
if tsCTE, tsArgs, err = b.buildReducedTimeSeriesCTE(ctx, orgID, tsStart, end, query, keys, variables); err != nil {
return nil, err
}
@@ -282,7 +283,7 @@ func unionStatements(main, reduced *qbtypes.Statement, query qbtypes.QueryBuilde
return &qbtypes.Statement{Query: q, Args: args, Warnings: warnings}, nil
}
func (b *MetricQueryStatementBuilder) buildReducedTimeSeriesCTE(
func (b *StatementBuilder) buildReducedTimeSeriesCTE(
ctx context.Context,
orgID valuer.UUID,
start, end uint64,
@@ -312,7 +313,7 @@ func (b *MetricQueryStatementBuilder) buildReducedTimeSeriesCTE(
}
}
sb.From(fmt.Sprintf("%s.%s", DBName, TimeseriesV4ReducedLocalTableName))
sb.From(fmt.Sprintf("%s.%s", metricstelemetryschema.DBName, metricstelemetryschema.TimeseriesV4ReducedLocalTableName))
sb.Select("fingerprint")
for _, g := range query.GroupBy {
col, err := b.fm.ColumnExpressionFor(ctx, orgID, start, end, &g.TelemetryFieldKey, telemetrytypes.FieldDataTypeString, keys)
@@ -343,7 +344,7 @@ func (b *MetricQueryStatementBuilder) buildReducedTimeSeriesCTE(
// spatial_aggregation_cte in one level with no per-series grouping, so shards
// send one state per (step, group) instead of per (series, step, group).
// FINAL still dedups recomputed 60s buckets at scan time.
func (b *MetricQueryStatementBuilder) buildReducedSpatialAggFastPath(
func (b *StatementBuilder) buildReducedSpatialAggFastPath(
start, end uint64,
query qbtypes.QueryBuilderQuery[qbtypes.MetricAggregation],
timeSeriesCTE string,
@@ -352,7 +353,7 @@ func (b *MetricQueryStatementBuilder) buildReducedSpatialAggFastPath(
agg := query.Aggregations[0]
stepSec := int64(query.StepInterval.Seconds())
value, _, ok := ReducedValueColumn(agg.Type, agg.SpaceAggregation)
value, _, ok := metricstelemetryschema.ReducedValueColumn(agg.Type, agg.SpaceAggregation)
if !ok {
return "", nil, false
}
@@ -362,8 +363,8 @@ func (b *MetricQueryStatementBuilder) buildReducedSpatialAggFastPath(
for _, g := range query.GroupBy {
sb.SelectMore(fmt.Sprintf("`%s`", g.Name))
}
sb.SelectMore(fmt.Sprintf("%s AS value", ReducedTimeAggregationColumn(agg.TimeAggregation, stepSec, value)))
sb.From(fmt.Sprintf("%s.%s AS points FINAL", DBName, WhichReducedSamplesTableToUse(agg.Type)))
sb.SelectMore(fmt.Sprintf("%s AS value", metricstelemetryschema.ReducedTimeAggregationColumn(agg.TimeAggregation, stepSec, value)))
sb.From(fmt.Sprintf("%s.%s AS points FINAL", metricstelemetryschema.DBName, metricstelemetryschema.WhichReducedSamplesTableToUse(agg.Type)))
sb.JoinWithOption(sqlbuilder.InnerJoin, timeSeriesCTE, "points.reduced_fingerprint = filtered_time_series.fingerprint")
sb.Where(
sb.In("metric_name", agg.MetricName),
@@ -377,7 +378,7 @@ func (b *MetricQueryStatementBuilder) buildReducedSpatialAggFastPath(
return fmt.Sprintf("__spatial_aggregation_cte AS (%s)", q), args, true
}
func (b *MetricQueryStatementBuilder) buildReducedTemporalAggregationCTE(
func (b *StatementBuilder) buildReducedTemporalAggregationCTE(
start, end uint64,
query qbtypes.QueryBuilderQuery[qbtypes.MetricAggregation],
timeSeriesCTE string,
@@ -386,7 +387,7 @@ func (b *MetricQueryStatementBuilder) buildReducedTemporalAggregationCTE(
agg := query.Aggregations[0]
stepSec := int64(query.StepInterval.Seconds())
value, weight, ok := ReducedValueColumn(agg.Type, agg.SpaceAggregation)
value, weight, ok := metricstelemetryschema.ReducedValueColumn(agg.Type, agg.SpaceAggregation)
if !ok {
return "", nil, false
}
@@ -399,13 +400,13 @@ func (b *MetricQueryStatementBuilder) buildReducedTemporalAggregationCTE(
for _, g := range query.GroupBy {
sb.SelectMore(fmt.Sprintf("`%s`", g.Name))
}
sb.SelectMore(fmt.Sprintf("%s AS per_series_value", ReducedTimeAggregationColumn(agg.TimeAggregation, stepSec, value)))
sb.SelectMore(fmt.Sprintf("%s AS per_series_value", metricstelemetryschema.ReducedTimeAggregationColumn(agg.TimeAggregation, stepSec, value)))
if weight != "" {
// count_series is a series count, not additive over time, so the avg
// denominator is reduced with avg
sb.SelectMore(fmt.Sprintf("avg(%s) AS per_series_weight", weight))
}
sb.From(fmt.Sprintf("%s.%s AS points FINAL", DBName, WhichReducedSamplesTableToUse(agg.Type)))
sb.From(fmt.Sprintf("%s.%s AS points FINAL", metricstelemetryschema.DBName, metricstelemetryschema.WhichReducedSamplesTableToUse(agg.Type)))
sb.JoinWithOption(sqlbuilder.InnerJoin, timeSeriesCTE, "points.reduced_fingerprint = filtered_time_series.fingerprint")
sb.Where(
sb.In("metric_name", agg.MetricName),
@@ -419,7 +420,7 @@ func (b *MetricQueryStatementBuilder) buildReducedTemporalAggregationCTE(
return fmt.Sprintf("__temporal_aggregation_cte AS (%s)", q), args, true
}
func (b *MetricQueryStatementBuilder) buildReducedSpatialAggregationCTE(
func (b *StatementBuilder) buildReducedSpatialAggregationCTE(
query qbtypes.QueryBuilderQuery[qbtypes.MetricAggregation],
) (string, []any) {
spatial := "sum(per_series_value)"
@@ -446,7 +447,7 @@ func (b *MetricQueryStatementBuilder) buildReducedSpatialAggregationCTE(
return fmt.Sprintf("__spatial_aggregation_cte AS (%s)", q), args
}
func (b *MetricQueryStatementBuilder) buildTemporalAggDeltaFastPath(
func (b *StatementBuilder) buildTemporalAggDeltaFastPath(
start, end uint64,
query qbtypes.QueryBuilderQuery[qbtypes.MetricAggregation],
samplesTable string,
@@ -465,7 +466,7 @@ func (b *MetricQueryStatementBuilder) buildTemporalAggDeltaFastPath(
sb.SelectMore(fmt.Sprintf("`%s`", g.Name))
}
aggCol, err := AggregationColumnForSamplesTable(
aggCol, err := metricstelemetryschema.AggregationColumnForSamplesTable(
samplesTable, query.Aggregations[0].Temporality, query.Aggregations[0].TimeAggregation,
)
if err != nil {
@@ -483,7 +484,7 @@ func (b *MetricQueryStatementBuilder) buildTemporalAggDeltaFastPath(
sb.SelectMore(fmt.Sprintf("%s AS value", aggCol))
sb.From(fmt.Sprintf("%s.%s AS points", DBName, samplesTable))
sb.From(fmt.Sprintf("%s.%s AS points", metricstelemetryschema.DBName, samplesTable))
sb.JoinWithOption(sqlbuilder.InnerJoin, timeSeriesCTE, "points.fingerprint = filtered_time_series.fingerprint")
sb.Where(
sb.In("metric_name", query.Aggregations[0].MetricName),
@@ -497,7 +498,7 @@ func (b *MetricQueryStatementBuilder) buildTemporalAggDeltaFastPath(
return fmt.Sprintf("__spatial_aggregation_cte AS (%s)", q), args, nil
}
func (b *MetricQueryStatementBuilder) buildTimeSeriesCTE(
func (b *StatementBuilder) buildTimeSeriesCTE(
ctx context.Context,
orgID valuer.UUID,
start, end uint64,
@@ -529,7 +530,7 @@ func (b *MetricQueryStatementBuilder) buildTimeSeriesCTE(
}
}
sb.From(fmt.Sprintf("%s.%s", DBName, tsTable))
sb.From(fmt.Sprintf("%s.%s", metricstelemetryschema.DBName, tsTable))
sb.Select("fingerprint")
for _, g := range query.GroupBy {
@@ -552,7 +553,7 @@ func (b *MetricQueryStatementBuilder) buildTimeSeriesCTE(
// the buffer holds both raw rows and the reduced catalog rows; the raw read
// only wants the original series
if tsTable == TimeseriesV4BufferLocalTableName {
if tsTable == metricstelemetryschema.TimeseriesV4BufferLocalTableName {
sb.Where(sb.EQ("is_reduced", false))
}
@@ -567,7 +568,7 @@ func (b *MetricQueryStatementBuilder) buildTimeSeriesCTE(
return fmt.Sprintf("(%s) AS filtered_time_series", q), args, nil
}
func (b *MetricQueryStatementBuilder) buildTemporalAggregationCTE(
func (b *StatementBuilder) buildTemporalAggregationCTE(
ctx context.Context,
start, end uint64,
query qbtypes.QueryBuilderQuery[qbtypes.MetricAggregation],
@@ -584,7 +585,7 @@ func (b *MetricQueryStatementBuilder) buildTemporalAggregationCTE(
return b.buildTemporalAggForMultipleTemporalities(ctx, start, end, query, samplesTable, timeSeriesCTE, timeSeriesCTEArgs)
}
func (b *MetricQueryStatementBuilder) buildTemporalAggDelta(
func (b *StatementBuilder) buildTemporalAggDelta(
_ context.Context,
start, end uint64,
query qbtypes.QueryBuilderQuery[qbtypes.MetricAggregation],
@@ -605,7 +606,7 @@ func (b *MetricQueryStatementBuilder) buildTemporalAggDelta(
sb.SelectMore(fmt.Sprintf("`%s`", g.Name))
}
aggCol, err := AggregationColumnForSamplesTable(samplesTable, query.Aggregations[0].Temporality, query.Aggregations[0].TimeAggregation)
aggCol, err := metricstelemetryschema.AggregationColumnForSamplesTable(samplesTable, query.Aggregations[0].Temporality, query.Aggregations[0].TimeAggregation)
if err != nil {
return "", nil, err
}
@@ -616,7 +617,7 @@ func (b *MetricQueryStatementBuilder) buildTemporalAggDelta(
sb.SelectMore(fmt.Sprintf("%s AS per_series_value", aggCol))
sb.From(fmt.Sprintf("%s.%s AS points", DBName, samplesTable))
sb.From(fmt.Sprintf("%s.%s AS points", metricstelemetryschema.DBName, samplesTable))
sb.JoinWithOption(sqlbuilder.InnerJoin, timeSeriesCTE, "points.fingerprint = filtered_time_series.fingerprint")
sb.Where(
sb.In("metric_name", query.Aggregations[0].MetricName),
@@ -631,7 +632,7 @@ func (b *MetricQueryStatementBuilder) buildTemporalAggDelta(
return fmt.Sprintf("__temporal_aggregation_cte AS (%s)", q), args, nil
}
func (b *MetricQueryStatementBuilder) buildTemporalAggCumulativeOrUnspecified(
func (b *StatementBuilder) buildTemporalAggCumulativeOrUnspecified(
_ context.Context,
start, end uint64,
query qbtypes.QueryBuilderQuery[qbtypes.MetricAggregation],
@@ -651,13 +652,13 @@ func (b *MetricQueryStatementBuilder) buildTemporalAggCumulativeOrUnspecified(
baseSb.SelectMore(fmt.Sprintf("`%s`", g.Name))
}
aggCol, err := AggregationColumnForSamplesTable(samplesTable, query.Aggregations[0].Temporality, query.Aggregations[0].TimeAggregation)
aggCol, err := metricstelemetryschema.AggregationColumnForSamplesTable(samplesTable, query.Aggregations[0].Temporality, query.Aggregations[0].TimeAggregation)
if err != nil {
return "", nil, err
}
baseSb.SelectMore(fmt.Sprintf("%s AS per_series_value", aggCol))
baseSb.From(fmt.Sprintf("%s.%s AS points", DBName, samplesTable))
baseSb.From(fmt.Sprintf("%s.%s AS points", metricstelemetryschema.DBName, samplesTable))
baseSb.JoinWithOption(sqlbuilder.InnerJoin, timeSeriesCTE, "points.fingerprint = filtered_time_series.fingerprint")
baseSb.Where(
baseSb.In("metric_name", query.Aggregations[0].MetricName),
@@ -697,7 +698,7 @@ func (b *MetricQueryStatementBuilder) buildTemporalAggCumulativeOrUnspecified(
}
}
func (b *MetricQueryStatementBuilder) buildTemporalAggForMultipleTemporalities(
func (b *StatementBuilder) buildTemporalAggForMultipleTemporalities(
_ context.Context,
start, end uint64,
query qbtypes.QueryBuilderQuery[qbtypes.MetricAggregation],
@@ -716,11 +717,11 @@ func (b *MetricQueryStatementBuilder) buildTemporalAggForMultipleTemporalities(
sb.SelectMore(fmt.Sprintf("`%s`", g.Name))
}
aggForDeltaTemporality, err := AggregationColumnForSamplesTable(samplesTable, metrictypes.Delta, query.Aggregations[0].TimeAggregation)
aggForDeltaTemporality, err := metricstelemetryschema.AggregationColumnForSamplesTable(samplesTable, metrictypes.Delta, query.Aggregations[0].TimeAggregation)
if err != nil {
return "", nil, err
}
aggForCumulativeTemporality, err := AggregationColumnForSamplesTable(samplesTable, metrictypes.Cumulative, query.Aggregations[0].TimeAggregation)
aggForCumulativeTemporality, err := metricstelemetryschema.AggregationColumnForSamplesTable(samplesTable, metrictypes.Cumulative, query.Aggregations[0].TimeAggregation)
if err != nil {
return "", nil, err
}
@@ -748,7 +749,7 @@ func (b *MetricQueryStatementBuilder) buildTemporalAggForMultipleTemporalities(
sb.SelectMore(expr)
}
sb.From(fmt.Sprintf("%s.%s AS points", DBName, samplesTable))
sb.From(fmt.Sprintf("%s.%s AS points", metricstelemetryschema.DBName, samplesTable))
sb.JoinWithOption(sqlbuilder.InnerJoin, timeSeriesCTE, "points.fingerprint = filtered_time_series.fingerprint")
sb.Where(
sb.In("metric_name", query.Aggregations[0].MetricName),
@@ -762,7 +763,7 @@ func (b *MetricQueryStatementBuilder) buildTemporalAggForMultipleTemporalities(
return fmt.Sprintf("__temporal_aggregation_cte AS (%s)", queryWithWindowAndOrder), args, nil
}
func (b *MetricQueryStatementBuilder) buildSpatialAggregationCTE(
func (b *StatementBuilder) buildSpatialAggregationCTE(
_ context.Context,
_ uint64,
_ uint64,
@@ -788,7 +789,7 @@ func (b *MetricQueryStatementBuilder) buildSpatialAggregationCTE(
return fmt.Sprintf("__spatial_aggregation_cte AS (%s)", q), args
}
func (b *MetricQueryStatementBuilder) BuildFinalSelect(
func (b *StatementBuilder) BuildFinalSelect(
cteFragments []string,
cteArgs [][]any,
query qbtypes.QueryBuilderQuery[qbtypes.MetricAggregation],
@@ -833,7 +834,7 @@ func (b *MetricQueryStatementBuilder) BuildFinalSelect(
sb.SelectMore(fmt.Sprintf("`%s`", g.Name))
}
aggQuery, err := AggregationQueryForHistogramCountWithParams(query.Aggregations[0].ComparisonSpaceAggregationParam)
aggQuery, err := metricstelemetryschema.AggregationQueryForHistogramCountWithParams(query.Aggregations[0].ComparisonSpaceAggregationParam)
if err != nil {
return nil, err
}

View File

@@ -1,4 +1,4 @@
package telemetrymetrics
package metricsstatementbuilder
import (
"context"
@@ -7,6 +7,7 @@ import (
"github.com/SigNoz/signoz/pkg/flagger"
"github.com/SigNoz/signoz/pkg/instrumentation/instrumentationtest"
"github.com/SigNoz/signoz/pkg/telemetryschema/metricstelemetryschema"
"github.com/SigNoz/signoz/pkg/types/metrictypes"
qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
@@ -252,8 +253,8 @@ func TestStatementBuilder(t *testing.T) {
},
}
fm := NewFieldMapper()
cb := NewConditionBuilder(fm)
fm := metricstelemetryschema.NewFieldMapper()
cb := metricstelemetryschema.NewConditionBuilder(fm)
mockMetadataStore := telemetrytypestest.NewMockMetadataStore()
keys, err := telemetrytypestest.LoadFieldKeysFromJSON("testdata/keys_map.json")
if err != nil {

View File

@@ -1,4 +1,4 @@
package telemetryresourcefilter
package resourcefilter
import (
"context"

View File

@@ -1,4 +1,4 @@
package telemetryresourcefilter
package resourcefilter
import (
"context"

View File

@@ -1,4 +1,4 @@
package telemetryresourcefilter
package resourcefilter
import (
"context"

View File

@@ -1,4 +1,4 @@
package telemetryresourcefilter
package resourcefilter
import (
"context"

View File

@@ -1,4 +1,4 @@
package telemetryresourcefilter
package resourcefilter
import (
"context"
@@ -45,7 +45,7 @@ func New[T any](
fullTextColumn *telemetrytypes.TelemetryFieldKey,
fl flagger.Flagger,
) *resourceFilterStatementBuilder[T] {
set := factory.NewScopedProviderSettings(settings, "github.com/SigNoz/signoz/pkg/telemetryresourcefilter")
set := factory.NewScopedProviderSettings(settings, "github.com/SigNoz/signoz/pkg/statementbuilder/resourcefilter")
fm := NewFieldMapper()
cb := NewConditionBuilder(fm)
return &resourceFilterStatementBuilder[T]{

View File

@@ -1,4 +1,4 @@
package telemetryresourcefilter
package resourcefilter
import (
"context"

View File

@@ -0,0 +1,59 @@
package statementbuilder
import (
"context"
"github.com/SigNoz/signoz/pkg/factory"
"github.com/SigNoz/signoz/pkg/flagger"
"github.com/SigNoz/signoz/pkg/statementbuilder/auditstatementbuilder"
"github.com/SigNoz/signoz/pkg/statementbuilder/logsstatementbuilder"
"github.com/SigNoz/signoz/pkg/statementbuilder/meterstatementbuilder"
"github.com/SigNoz/signoz/pkg/statementbuilder/metricsstatementbuilder"
"github.com/SigNoz/signoz/pkg/statementbuilder/tracesstatementbuilder"
"github.com/SigNoz/signoz/pkg/telemetrystore"
qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
)
// Builders bundles the per-signal statement builders that make up the query stack.
type Builders struct {
Trace qbtypes.StatementBuilder[qbtypes.TraceAggregation]
TraceOperator qbtypes.TraceOperatorStatementBuilder
Log qbtypes.StatementBuilder[qbtypes.LogAggregation]
Audit qbtypes.StatementBuilder[qbtypes.LogAggregation]
Metric qbtypes.StatementBuilder[qbtypes.MetricAggregation]
Meter qbtypes.StatementBuilder[qbtypes.MetricAggregation]
}
// NewFactory returns a provider factory that builds the per-signal statement builders once.
func NewFactory(
telemetryStore telemetrystore.TelemetryStore,
metadataStore telemetrytypes.MetadataStore,
fl flagger.Flagger,
) factory.ProviderFactory[*Builders, Config] {
return factory.NewProviderFactory(
factory.MustNewName("signoz"),
func(_ context.Context, settings factory.ProviderSettings, cfg Config) (*Builders, error) {
traceStmtBuilder, traceOperatorStmtBuilder := tracesstatementbuilder.New(
settings, metadataStore, telemetryStore, fl,
cfg.SkipResourceFingerprint.Enabled, cfg.SkipResourceFingerprint.Threshold,
)
logStmtBuilder := logsstatementbuilder.New(
settings, metadataStore, telemetryStore, fl,
cfg.SkipResourceFingerprint.Enabled, cfg.SkipResourceFingerprint.Threshold,
)
auditStmtBuilder := auditstatementbuilder.New(settings, metadataStore, fl)
metricStmtBuilder := metricsstatementbuilder.New(settings, metadataStore, fl)
meterStmtBuilder := meterstatementbuilder.New(settings, metadataStore, metricStmtBuilder)
return &Builders{
Trace: traceStmtBuilder,
TraceOperator: traceOperatorStmtBuilder,
Log: logStmtBuilder,
Audit: auditStmtBuilder,
Metric: metricStmtBuilder,
Meter: meterStmtBuilder,
}, nil
},
)
}

View File

@@ -0,0 +1,35 @@
package tracesstatementbuilder
import (
"github.com/SigNoz/signoz/pkg/factory"
"github.com/SigNoz/signoz/pkg/flagger"
"github.com/SigNoz/signoz/pkg/querybuilder"
"github.com/SigNoz/signoz/pkg/telemetryschema/tracestelemetryschema"
"github.com/SigNoz/signoz/pkg/telemetrystore"
qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
)
// New builds the trace and trace-operator statement builders, internalizing the
// FieldMapper, ConditionBuilder, and AggExprRewriter.
func New(
settings factory.ProviderSettings,
metadataStore telemetrytypes.MetadataStore,
telemetryStore telemetrystore.TelemetryStore,
fl flagger.Flagger,
skipResourceFingerprintEnabled bool,
skipResourceFingerprintThreshold uint64,
) (qbtypes.StatementBuilder[qbtypes.TraceAggregation], qbtypes.TraceOperatorStatementBuilder) {
fm := tracestelemetryschema.NewFieldMapper()
cb := tracestelemetryschema.NewConditionBuilder(fm)
aggExprRewriter := querybuilder.NewAggExprRewriter(settings, nil, fm, cb, fl)
traceStmtBuilder := NewTraceQueryStatementBuilder(
settings, metadataStore, fm, cb, aggExprRewriter, telemetryStore, fl,
skipResourceFingerprintEnabled, skipResourceFingerprintThreshold,
)
traceOperatorStmtBuilder := NewTraceOperatorStatementBuilder(
settings, metadataStore, fm, cb, traceStmtBuilder, aggExprRewriter, fl,
)
return traceStmtBuilder, traceOperatorStmtBuilder
}

View File

@@ -1,4 +1,4 @@
package telemetrytraces
package tracesstatementbuilder
import (
"context"
@@ -10,7 +10,8 @@ import (
"github.com/SigNoz/signoz/pkg/factory"
"github.com/SigNoz/signoz/pkg/flagger"
"github.com/SigNoz/signoz/pkg/querybuilder"
"github.com/SigNoz/signoz/pkg/telemetryresourcefilter"
"github.com/SigNoz/signoz/pkg/statementbuilder/resourcefilter"
"github.com/SigNoz/signoz/pkg/telemetryschema/tracestelemetryschema"
"github.com/SigNoz/signoz/pkg/telemetrystore"
qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
@@ -27,7 +28,7 @@ type traceQueryStatementBuilder struct {
metadataStore telemetrytypes.MetadataStore
fm qbtypes.FieldMapper
cb qbtypes.ConditionBuilder
resourceFilterResolver *telemetryresourcefilter.ResourceFingerprintResolver[qbtypes.TraceAggregation]
resourceFilterResolver *resourcefilter.ResourceFingerprintResolver[qbtypes.TraceAggregation]
aggExprRewriter qbtypes.AggExprRewriter
skipResourceFingerprintEnabled bool
}
@@ -45,12 +46,12 @@ func NewTraceQueryStatementBuilder(
skipResourceFingerprintEnable bool,
skipResourceFingerprintThreshold uint64,
) *traceQueryStatementBuilder {
tracesSettings := factory.NewScopedProviderSettings(settings, "github.com/SigNoz/signoz/pkg/telemetrytraces")
tracesSettings := factory.NewScopedProviderSettings(settings, "github.com/SigNoz/signoz/pkg/telemetryschema/tracestelemetryschema")
resourceFilterResolver := telemetryresourcefilter.NewResolver[qbtypes.TraceAggregation](
resourceFilterResolver := resourcefilter.NewResolver[qbtypes.TraceAggregation](
settings,
DBName,
TracesResourceV3TableName,
tracestelemetryschema.DBName,
tracestelemetryschema.TracesResourceV3TableName,
telemetrytypes.SignalTraces,
telemetrytypes.SourceUnspecified,
metadataStore,
@@ -183,10 +184,10 @@ func getKeySelectors(query qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation])
// 3. We prepend (not append) so the intrinsic/calculated entry wins ordering
// in the multi_if SQL expression.
func mergeDeprecatedTraceKeys(keys map[string][]*telemetrytypes.TelemetryFieldKey) {
for fieldKeyName, fieldKey := range IntrinsicFieldsDeprecated {
for fieldKeyName, fieldKey := range tracestelemetryschema.IntrinsicFieldsDeprecated {
keys[fieldKeyName] = append([]*telemetrytypes.TelemetryFieldKey{&fieldKey}, keys[fieldKeyName]...)
}
for fieldKeyName, fieldKey := range CalculatedFieldsDeprecated {
for fieldKeyName, fieldKey := range tracestelemetryschema.CalculatedFieldsDeprecated {
keys[fieldKeyName] = append([]*telemetrytypes.TelemetryFieldKey{&fieldKey}, keys[fieldKeyName]...)
}
}
@@ -246,18 +247,18 @@ func adjustTraceKey(key *telemetrytypes.TelemetryFieldKey, keys map[string][]*te
*/
var isIntrinsicOrCalculatedField bool
var intrinsicOrCalculatedField telemetrytypes.TelemetryFieldKey
if _, ok := IntrinsicFields[key.Name]; ok {
if _, ok := tracestelemetryschema.IntrinsicFields[key.Name]; ok {
isIntrinsicOrCalculatedField = true
intrinsicOrCalculatedField = IntrinsicFields[key.Name]
} else if _, ok := CalculatedFields[key.Name]; ok {
intrinsicOrCalculatedField = tracestelemetryschema.IntrinsicFields[key.Name]
} else if _, ok := tracestelemetryschema.CalculatedFields[key.Name]; ok {
isIntrinsicOrCalculatedField = true
intrinsicOrCalculatedField = CalculatedFields[key.Name]
} else if _, ok := IntrinsicFieldsDeprecated[key.Name]; ok {
intrinsicOrCalculatedField = tracestelemetryschema.CalculatedFields[key.Name]
} else if _, ok := tracestelemetryschema.IntrinsicFieldsDeprecated[key.Name]; ok {
isIntrinsicOrCalculatedField = true
intrinsicOrCalculatedField = IntrinsicFieldsDeprecated[key.Name]
} else if _, ok := CalculatedFieldsDeprecated[key.Name]; ok {
intrinsicOrCalculatedField = tracestelemetryschema.IntrinsicFieldsDeprecated[key.Name]
} else if _, ok := tracestelemetryschema.CalculatedFieldsDeprecated[key.Name]; ok {
isIntrinsicOrCalculatedField = true
intrinsicOrCalculatedField = CalculatedFieldsDeprecated[key.Name]
intrinsicOrCalculatedField = tracestelemetryschema.CalculatedFieldsDeprecated[key.Name]
}
if isIntrinsicOrCalculatedField {
@@ -304,13 +305,13 @@ func (b *traceQueryStatementBuilder) buildListQuery(
}
if isSelectFieldsEmpty {
for _, col := range ContextualSpanColumns {
for _, col := range tracestelemetryschema.ContextualSpanColumns {
sb.SelectMore(col)
}
}
// From table
sb.From(fmt.Sprintf("%s.%s", DBName, SpanIndexV3TableName))
sb.From(fmt.Sprintf("%s.%s", tracestelemetryschema.DBName, tracestelemetryschema.SpanIndexV3TableName))
// Add filter conditions
preparedWhereClause, err := b.addFilterCondition(ctx, orgID, sb, start, end, query, keys, variables, skipResourceFilter)
@@ -368,7 +369,7 @@ func (b *traceQueryStatementBuilder) buildTraceQuery(
distSB := sqlbuilder.NewSelectBuilder()
distSB.Select("trace_id")
distSB.From(fmt.Sprintf("%s.%s", DBName, SpanIndexV3TableName))
distSB.From(fmt.Sprintf("%s.%s", tracestelemetryschema.DBName, tracestelemetryschema.SpanIndexV3TableName))
var (
cteFragments []string
@@ -398,7 +399,7 @@ func (b *traceQueryStatementBuilder) buildTraceQuery(
// Build the inner subquery for root spans
innerSB := sqlbuilder.NewSelectBuilder()
innerSB.Select("trace_id", "duration_nano", sqlbuilder.Escape("resource_string_service$$name as `service.name`"), "name")
innerSB.From(fmt.Sprintf("%s.%s", DBName, SpanIndexV3TableName))
innerSB.From(fmt.Sprintf("%s.%s", tracestelemetryschema.DBName, tracestelemetryschema.SpanIndexV3TableName))
innerSB.Where("parent_span_id = ''")
// this only helps when there is a filter
@@ -528,7 +529,7 @@ func (b *traceQueryStatementBuilder) buildTimeSeriesQuery(
sb.SelectMore(fmt.Sprintf("%s AS __result_%d", rewritten, i))
}
sb.From(fmt.Sprintf("%s.%s", DBName, SpanIndexV3TableName))
sb.From(fmt.Sprintf("%s.%s", tracestelemetryschema.DBName, tracestelemetryschema.SpanIndexV3TableName))
preparedWhereClause, err := b.addFilterCondition(ctx, orgID, sb, start, end, query, keys, variables, skipResourceFilter)
if err != nil {
return nil, err
@@ -690,7 +691,7 @@ func (b *traceQueryStatementBuilder) buildScalarQuery(
}
// From table
sb.From(fmt.Sprintf("%s.%s", DBName, SpanIndexV3TableName))
sb.From(fmt.Sprintf("%s.%s", tracestelemetryschema.DBName, tracestelemetryschema.SpanIndexV3TableName))
// Add filter conditions
preparedWhereClause, err := b.addFilterCondition(ctx, orgID, sb, start, end, query, keys, variables, skipResourceFilter)
@@ -873,21 +874,21 @@ func (b *traceQueryStatementBuilder) maybeAttachResourceFilter(
// It must be called before adjustKeys so that normalization runs over the full set.
func (b *traceQueryStatementBuilder) expandRawSelectFields(query qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]) qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation] {
if len(query.SelectFields) == 0 {
selectFields := make([]telemetrytypes.TelemetryFieldKey, 0, len(IntrinsicSpanFields)+len(CalculatedSpanFields))
selectFields = append(selectFields, IntrinsicSpanFields...)
selectFields = append(selectFields, CalculatedSpanFields...)
selectFields := make([]telemetrytypes.TelemetryFieldKey, 0, len(tracestelemetryschema.IntrinsicSpanFields)+len(tracestelemetryschema.CalculatedSpanFields))
selectFields = append(selectFields, tracestelemetryschema.IntrinsicSpanFields...)
selectFields = append(selectFields, tracestelemetryschema.CalculatedSpanFields...)
query.SelectFields = selectFields
return query
}
selectFields := []telemetrytypes.TelemetryFieldKey{
{Name: SpanTimestampColumn, FieldContext: telemetrytypes.FieldContextSpan},
{Name: SpanTraceIDColumn, FieldContext: telemetrytypes.FieldContextSpan},
{Name: SpanSpanIDColumn, FieldContext: telemetrytypes.FieldContextSpan},
{Name: tracestelemetryschema.SpanTimestampColumn, FieldContext: telemetrytypes.FieldContextSpan},
{Name: tracestelemetryschema.SpanTraceIDColumn, FieldContext: telemetrytypes.FieldContextSpan},
{Name: tracestelemetryschema.SpanSpanIDColumn, FieldContext: telemetrytypes.FieldContextSpan},
}
for _, field := range query.SelectFields {
// TODO(tvats): If a user specifies attribute.timestamp in the select fields, this loop will basically ignore it, as we already added a field by default. This can be fixed once we close https://github.com/SigNoz/engineering-pod/issues/3693
if field.Name == SpanTimestampColumn || field.Name == SpanTraceIDColumn || field.Name == SpanSpanIDColumn {
if field.Name == tracestelemetryschema.SpanTimestampColumn || field.Name == tracestelemetryschema.SpanTraceIDColumn || field.Name == tracestelemetryschema.SpanSpanIDColumn {
continue
}
selectFields = append(selectFields, field)

View File

@@ -1,4 +1,4 @@
package telemetrytraces
package tracesstatementbuilder
import (
"context"
@@ -11,6 +11,7 @@ import (
"github.com/SigNoz/signoz/pkg/flagger/flaggertest"
"github.com/SigNoz/signoz/pkg/instrumentation/instrumentationtest"
"github.com/SigNoz/signoz/pkg/querybuilder"
"github.com/SigNoz/signoz/pkg/telemetryschema/tracestelemetryschema"
"github.com/SigNoz/signoz/pkg/telemetrystore"
"github.com/SigNoz/signoz/pkg/telemetrystore/telemetrystoretest"
qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
@@ -375,10 +376,10 @@ func TestStatementBuilder(t *testing.T) {
}
fl := flaggertest.New(t)
fm := NewFieldMapper()
cb := NewConditionBuilder(fm)
fm := tracestelemetryschema.NewFieldMapper()
cb := tracestelemetryschema.NewConditionBuilder(fm)
mockMetadataStore := telemetrytypestest.NewMockMetadataStore()
mockMetadataStore.KeysMap = buildCompleteFieldKeyMap(releaseTime)
mockMetadataStore.KeysMap = tracestelemetryschema.BuildCompleteFieldKeyMap(releaseTime)
aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, fm, cb, fl)
statementBuilder := NewTraceQueryStatementBuilder(
@@ -676,10 +677,10 @@ func TestStatementBuilderListQuery(t *testing.T) {
}
fl := flaggertest.New(t)
fm := NewFieldMapper()
cb := NewConditionBuilder(fm)
fm := tracestelemetryschema.NewFieldMapper()
cb := tracestelemetryschema.NewConditionBuilder(fm)
mockMetadataStore := telemetrytypestest.NewMockMetadataStore()
mockMetadataStore.KeysMap = buildCompleteFieldKeyMap(releaseTime)
mockMetadataStore.KeysMap = tracestelemetryschema.BuildCompleteFieldKeyMap(releaseTime)
aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, fm, cb, fl)
statementBuilder := NewTraceQueryStatementBuilder(
@@ -740,7 +741,7 @@ func TestStatementBuilderListQueryWithCorruptData(t *testing.T) {
Signal: telemetrytypes.SignalTraces,
FieldContext: telemetrytypes.FieldContextResource,
FieldDataType: telemetrytypes.FieldDataTypeString,
Evolutions: mockEvolutionData(time.Date(2025, 5, 22, 22, 0, 0, 0, time.UTC)),
Evolutions: tracestelemetryschema.MockEvolutionData(time.Date(2025, 5, 22, 22, 0, 0, 0, time.UTC)),
},
},
},
@@ -774,7 +775,7 @@ func TestStatementBuilderListQueryWithCorruptData(t *testing.T) {
Signal: telemetrytypes.SignalTraces,
FieldContext: telemetrytypes.FieldContextResource,
FieldDataType: telemetrytypes.FieldDataTypeString,
Evolutions: mockEvolutionData(time.Date(2025, 5, 22, 22, 0, 0, 0, time.UTC)),
Evolutions: tracestelemetryschema.MockEvolutionData(time.Date(2025, 5, 22, 22, 0, 0, 0, time.UTC)),
},
},
},
@@ -803,12 +804,12 @@ func TestStatementBuilderListQueryWithCorruptData(t *testing.T) {
for _, c := range cases {
t.Run(c.name, func(t *testing.T) {
fl := flaggertest.New(t)
fm := NewFieldMapper()
cb := NewConditionBuilder(fm)
fm := tracestelemetryschema.NewFieldMapper()
cb := tracestelemetryschema.NewConditionBuilder(fm)
mockMetadataStore := telemetrytypestest.NewMockMetadataStore()
mockMetadataStore.KeysMap = c.keysMap
if mockMetadataStore.KeysMap == nil {
mockMetadataStore.KeysMap = buildCompleteFieldKeyMap(releaseTime)
mockMetadataStore.KeysMap = tracestelemetryschema.BuildCompleteFieldKeyMap(releaseTime)
}
aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, fm, cb, fl)
@@ -878,10 +879,10 @@ func TestStatementBuilderGroupByResourceEvolution(t *testing.T) {
}
fl := flaggertest.New(t)
fm := NewFieldMapper()
cb := NewConditionBuilder(fm)
fm := tracestelemetryschema.NewFieldMapper()
cb := tracestelemetryschema.NewConditionBuilder(fm)
mockMetadataStore := telemetrytypestest.NewMockMetadataStore()
mockMetadataStore.KeysMap = buildCompleteFieldKeyMap(releaseTime)
mockMetadataStore.KeysMap = tracestelemetryschema.BuildCompleteFieldKeyMap(releaseTime)
aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, fm, cb, fl)
statementBuilder := NewTraceQueryStatementBuilder(
@@ -1045,10 +1046,10 @@ func TestStatementBuilderTraceQuery(t *testing.T) {
}
fl := flaggertest.New(t)
fm := NewFieldMapper()
cb := NewConditionBuilder(fm)
fm := tracestelemetryschema.NewFieldMapper()
cb := tracestelemetryschema.NewConditionBuilder(fm)
mockMetadataStore := telemetrytypestest.NewMockMetadataStore()
mockMetadataStore.KeysMap = buildCompleteFieldKeyMap(releaseTime)
mockMetadataStore.KeysMap = tracestelemetryschema.BuildCompleteFieldKeyMap(releaseTime)
aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, fm, cb, fl)
statementBuilder := NewTraceQueryStatementBuilder(
@@ -1096,8 +1097,8 @@ func TestAdjustKey(t *testing.T) {
FieldContext: telemetrytypes.FieldContextUnspecified,
FieldDataType: telemetrytypes.FieldDataTypeUnspecified,
},
keysMap: buildCompleteFieldKeyMap(releaseTime),
expectedKey: IntrinsicFields["trace_id"],
keysMap: tracestelemetryschema.BuildCompleteFieldKeyMap(releaseTime),
expectedKey: tracestelemetryschema.IntrinsicFields["trace_id"],
},
{
name: "intrinsic field with metadata match with incorrect context - override",
@@ -1106,7 +1107,7 @@ func TestAdjustKey(t *testing.T) {
FieldContext: telemetrytypes.FieldContextBody, // incorrect context
FieldDataType: telemetrytypes.FieldDataTypeInt64,
},
keysMap: buildCompleteFieldKeyMap(releaseTime),
keysMap: tracestelemetryschema.BuildCompleteFieldKeyMap(releaseTime),
expectedKey: telemetrytypes.TelemetryFieldKey{
Name: "duration_nano",
FieldContext: telemetrytypes.FieldContextSpan, // should be corrected
@@ -1120,7 +1121,7 @@ func TestAdjustKey(t *testing.T) {
FieldContext: telemetrytypes.FieldContextSpan, // correct context
FieldDataType: telemetrytypes.FieldDataTypeInt64,
},
keysMap: buildCompleteFieldKeyMap(releaseTime),
keysMap: tracestelemetryschema.BuildCompleteFieldKeyMap(releaseTime),
expectedKey: telemetrytypes.TelemetryFieldKey{
Name: "duration_nano",
FieldContext: telemetrytypes.FieldContextSpan, // should be corrected
@@ -1134,8 +1135,8 @@ func TestAdjustKey(t *testing.T) {
FieldContext: telemetrytypes.FieldContextUnspecified,
FieldDataType: telemetrytypes.FieldDataTypeUnspecified,
},
keysMap: buildCompleteFieldKeyMap(releaseTime),
expectedKey: *buildCompleteFieldKeyMap(releaseTime)["service.name"][0],
keysMap: tracestelemetryschema.BuildCompleteFieldKeyMap(releaseTime),
expectedKey: *tracestelemetryschema.BuildCompleteFieldKeyMap(releaseTime)["service.name"][0],
},
{
name: "single matching key with context specified - override",
@@ -1144,8 +1145,8 @@ func TestAdjustKey(t *testing.T) {
FieldContext: telemetrytypes.FieldContextAttribute,
FieldDataType: telemetrytypes.FieldDataTypeUnspecified,
},
keysMap: buildCompleteFieldKeyMap(releaseTime),
expectedKey: *buildCompleteFieldKeyMap(releaseTime)["cart.items_count"][0],
keysMap: tracestelemetryschema.BuildCompleteFieldKeyMap(releaseTime),
expectedKey: *tracestelemetryschema.BuildCompleteFieldKeyMap(releaseTime)["cart.items_count"][0],
},
{
name: "multiple matching keys - all materialized",
@@ -1182,7 +1183,7 @@ func TestAdjustKey(t *testing.T) {
FieldContext: telemetrytypes.FieldContextUnspecified,
FieldDataType: telemetrytypes.FieldDataTypeUnspecified,
},
keysMap: buildCompleteFieldKeyMap(releaseTime),
keysMap: tracestelemetryschema.BuildCompleteFieldKeyMap(releaseTime),
expectedKey: telemetrytypes.TelemetryFieldKey{
Name: "mixed.materialization.key",
FieldDataType: telemetrytypes.FieldDataTypeString,
@@ -1196,7 +1197,7 @@ func TestAdjustKey(t *testing.T) {
FieldContext: telemetrytypes.FieldContextAttribute,
FieldDataType: telemetrytypes.FieldDataTypeUnspecified,
},
keysMap: buildCompleteFieldKeyMap(releaseTime),
keysMap: tracestelemetryschema.BuildCompleteFieldKeyMap(releaseTime),
expectedKey: telemetrytypes.TelemetryFieldKey{
Name: "mixed.materialization.key",
FieldContext: telemetrytypes.FieldContextAttribute,
@@ -1211,7 +1212,7 @@ func TestAdjustKey(t *testing.T) {
FieldContext: telemetrytypes.FieldContextUnspecified,
FieldDataType: telemetrytypes.FieldDataTypeUnspecified,
},
keysMap: buildCompleteFieldKeyMap(releaseTime),
keysMap: tracestelemetryschema.BuildCompleteFieldKeyMap(releaseTime),
expectedKey: telemetrytypes.TelemetryFieldKey{
Name: "unknown.field",
Materialized: false,
@@ -1224,7 +1225,7 @@ func TestAdjustKey(t *testing.T) {
FieldContext: telemetrytypes.FieldContextAttribute,
FieldDataType: telemetrytypes.FieldDataTypeUnspecified,
},
keysMap: buildCompleteFieldKeyMap(releaseTime),
keysMap: tracestelemetryschema.BuildCompleteFieldKeyMap(releaseTime),
expectedKey: telemetrytypes.TelemetryFieldKey{
Name: "service.name",
FieldContext: telemetrytypes.FieldContextAttribute,
@@ -1239,7 +1240,7 @@ func TestAdjustKey(t *testing.T) {
FieldContext: telemetrytypes.FieldContextUnspecified,
FieldDataType: telemetrytypes.FieldDataTypeUnspecified,
},
keysMap: buildCompleteFieldKeyMap(releaseTime),
keysMap: tracestelemetryschema.BuildCompleteFieldKeyMap(releaseTime),
expectedKey: telemetrytypes.TelemetryFieldKey{
Name: "cart.items_count",
FieldContext: telemetrytypes.FieldContextAttribute,
@@ -1254,7 +1255,7 @@ func TestAdjustKey(t *testing.T) {
FieldContext: telemetrytypes.FieldContextUnspecified,
FieldDataType: telemetrytypes.FieldDataTypeUnspecified,
},
keysMap: buildCompleteFieldKeyMap(releaseTime),
keysMap: tracestelemetryschema.BuildCompleteFieldKeyMap(releaseTime),
expectedKey: telemetrytypes.TelemetryFieldKey{
Name: "user.id",
FieldContext: telemetrytypes.FieldContextAttribute,
@@ -1308,7 +1309,7 @@ func TestAdjustKeys(t *testing.T) {
},
},
},
keysMap: buildCompleteFieldKeyMap(releaseTime),
keysMap: tracestelemetryschema.BuildCompleteFieldKeyMap(releaseTime),
expectedSelectFields: []telemetrytypes.TelemetryFieldKey{
{
Name: "service.name",
@@ -1345,7 +1346,7 @@ func TestAdjustKeys(t *testing.T) {
},
},
},
keysMap: buildCompleteFieldKeyMap(releaseTime),
keysMap: tracestelemetryschema.BuildCompleteFieldKeyMap(releaseTime),
expectedGroupBy: []qbtypes.GroupByKey{
{
TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{
@@ -1392,7 +1393,7 @@ func TestAdjustKeys(t *testing.T) {
},
},
},
keysMap: buildCompleteFieldKeyMap(releaseTime),
keysMap: tracestelemetryschema.BuildCompleteFieldKeyMap(releaseTime),
expectedOrder: []qbtypes.OrderBy{
{
Key: qbtypes.OrderByKey{
@@ -1451,7 +1452,7 @@ func TestAdjustKeys(t *testing.T) {
},
},
},
keysMap: buildCompleteFieldKeyMap(releaseTime),
keysMap: tracestelemetryschema.BuildCompleteFieldKeyMap(releaseTime),
expectedSelectFields: []telemetrytypes.TelemetryFieldKey{
{
Name: "trace_id",
@@ -1506,7 +1507,7 @@ func TestAdjustKeys(t *testing.T) {
},
},
},
keysMap: buildCompleteFieldKeyMap(releaseTime),
keysMap: tracestelemetryschema.BuildCompleteFieldKeyMap(releaseTime),
// After alias adjustment, name becomes "span.duration" with FieldContextUnspecified
// "span.duration" is not in keysMap, so context stays unspecified
expectedOrder: []qbtypes.OrderBy{
@@ -1574,14 +1575,14 @@ func TestAdjustKeys(t *testing.T) {
if c.expectDeprecatedFieldsAdd {
// Check that at least some deprecated fields were added
foundDeprecatedField := false
for fieldName := range IntrinsicFieldsDeprecated {
for fieldName := range tracestelemetryschema.IntrinsicFieldsDeprecated {
if _, ok := keysMapCopy[fieldName]; ok {
foundDeprecatedField = true
break
}
}
if !foundDeprecatedField {
for fieldName := range CalculatedFieldsDeprecated {
for fieldName := range tracestelemetryschema.CalculatedFieldsDeprecated {
if _, ok := keysMapCopy[fieldName]; ok {
foundDeprecatedField = true
break
@@ -1683,11 +1684,11 @@ func newSkipResourceFingerprintBuilder(
t.Helper()
fl := flaggertest.New(t)
fm := NewFieldMapper()
cb := NewConditionBuilder(fm)
fm := tracestelemetryschema.NewFieldMapper()
cb := tracestelemetryschema.NewConditionBuilder(fm)
mockMetadataStore := telemetrytypestest.NewMockMetadataStore()
releaseTime := time.Date(2025, 5, 22, 22, 0, 0, 0, time.UTC)
mockMetadataStore.KeysMap = buildCompleteFieldKeyMap(releaseTime)
mockMetadataStore.KeysMap = tracestelemetryschema.BuildCompleteFieldKeyMap(releaseTime)
aggExprRewriter := querybuilder.NewAggExprRewriter(
instrumentationtest.New().ToProviderSettings(), nil, fm, cb, fl,
@@ -1713,10 +1714,10 @@ func TestStatementBuilderGroupByUnseenKey(t *testing.T) {
releaseTime := time.Date(2025, 5, 22, 22, 0, 0, 0, time.UTC)
fl := flaggertest.New(t)
fm := NewFieldMapper()
cb := NewConditionBuilder(fm)
fm := tracestelemetryschema.NewFieldMapper()
cb := tracestelemetryschema.NewConditionBuilder(fm)
mockMetadataStore := telemetrytypestest.NewMockMetadataStore()
mockMetadataStore.KeysMap = buildCompleteFieldKeyMap(releaseTime)
mockMetadataStore.KeysMap = tracestelemetryschema.BuildCompleteFieldKeyMap(releaseTime)
aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, fm, cb, fl)
statementBuilder := NewTraceQueryStatementBuilder(
@@ -1755,10 +1756,10 @@ func TestStatementBuilderAggregationUnseenKey(t *testing.T) {
releaseTime := time.Date(2025, 5, 22, 22, 0, 0, 0, time.UTC)
fl := flaggertest.New(t)
fm := NewFieldMapper()
cb := NewConditionBuilder(fm)
fm := tracestelemetryschema.NewFieldMapper()
cb := tracestelemetryschema.NewConditionBuilder(fm)
mockMetadataStore := telemetrytypestest.NewMockMetadataStore()
mockMetadataStore.KeysMap = buildCompleteFieldKeyMap(releaseTime)
mockMetadataStore.KeysMap = tracestelemetryschema.BuildCompleteFieldKeyMap(releaseTime)
aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, fm, cb, fl)
statementBuilder := NewTraceQueryStatementBuilder(

View File

@@ -1,4 +1,4 @@
package telemetrytraces
package tracesstatementbuilder
import (
"context"
@@ -10,6 +10,7 @@ import (
"github.com/SigNoz/signoz/pkg/errors"
"github.com/SigNoz/signoz/pkg/querybuilder"
"github.com/SigNoz/signoz/pkg/telemetryschema/tracestelemetryschema"
qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
"github.com/SigNoz/signoz/pkg/valuer"
@@ -136,7 +137,7 @@ func (b *traceOperatorCTEBuilder) buildAllSpansCTE(ctx context.Context) {
sb.Select("*")
sb.SelectMore(sqlbuilder.Escape("resource_string_service$$name") + " AS `service.name`")
sb.From(fmt.Sprintf("%s.%s", DBName, SpanIndexV3TableName))
sb.From(fmt.Sprintf("%s.%s", tracestelemetryschema.DBName, tracestelemetryschema.SpanIndexV3TableName))
startBucket := b.start/querybuilder.NsToSeconds - querybuilder.BucketAdjustment
endBucket := b.end / querybuilder.NsToSeconds
sb.Where(
@@ -245,7 +246,7 @@ func (b *traceOperatorCTEBuilder) buildQueryCTE(ctx context.Context, queryName s
sb := sqlbuilder.NewSelectBuilder()
sb.Select("*")
sb.From(fmt.Sprintf("%s.%s", DBName, SpanIndexV3TableName))
sb.From(fmt.Sprintf("%s.%s", tracestelemetryschema.DBName, tracestelemetryschema.SpanIndexV3TableName))
if resourceFilterCTEName != "" {
sb.Where(fmt.Sprintf("resource_fingerprint GLOBAL IN (SELECT fingerprint FROM %s)", resourceFilterCTEName))
}
@@ -436,9 +437,9 @@ func (b *traceOperatorCTEBuilder) buildFinalQuery(ctx context.Context, selectFro
if requestType == qbtypes.RequestTypeRaw {
isSelectFieldsEmpty = len(b.operator.SelectFields) == 0
if isSelectFieldsEmpty {
b.operator.SelectFields = make([]telemetrytypes.TelemetryFieldKey, 0, len(IntrinsicSpanFields)+len(CalculatedSpanFields))
b.operator.SelectFields = append(b.operator.SelectFields, IntrinsicSpanFields...)
b.operator.SelectFields = append(b.operator.SelectFields, CalculatedSpanFields...)
b.operator.SelectFields = make([]telemetrytypes.TelemetryFieldKey, 0, len(tracestelemetryschema.IntrinsicSpanFields)+len(tracestelemetryschema.CalculatedSpanFields))
b.operator.SelectFields = append(b.operator.SelectFields, tracestelemetryschema.IntrinsicSpanFields...)
b.operator.SelectFields = append(b.operator.SelectFields, tracestelemetryschema.CalculatedSpanFields...)
}
}
@@ -502,7 +503,7 @@ func (b *traceOperatorCTEBuilder) buildListQuery(ctx context.Context, selectFrom
}
if isSelectFieldsEmpty {
for _, col := range ContextualSpanColumns {
for _, col := range tracestelemetryschema.ContextualSpanColumns {
sb.SelectMore(col)
}
}

View File

@@ -1,4 +1,4 @@
package telemetrytraces
package tracesstatementbuilder
import (
"context"
@@ -8,6 +8,7 @@ import (
"github.com/SigNoz/signoz/pkg/flagger/flaggertest"
"github.com/SigNoz/signoz/pkg/instrumentation/instrumentationtest"
"github.com/SigNoz/signoz/pkg/querybuilder"
"github.com/SigNoz/signoz/pkg/telemetryschema/tracestelemetryschema"
qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
"github.com/SigNoz/signoz/pkg/types/telemetrytypes/telemetrytypestest"
@@ -19,10 +20,10 @@ func newTestTraceOperatorStatementBuilder(t *testing.T) *traceOperatorStatementB
t.Helper()
releaseTime := time.Date(2025, 5, 22, 22, 0, 0, 0, time.UTC)
fl := flaggertest.New(t)
fm := NewFieldMapper()
cb := NewConditionBuilder(fm)
fm := tracestelemetryschema.NewFieldMapper()
cb := tracestelemetryschema.NewConditionBuilder(fm)
mockMetadataStore := telemetrytypestest.NewMockMetadataStore()
mockMetadataStore.KeysMap = buildCompleteFieldKeyMap(releaseTime)
mockMetadataStore.KeysMap = tracestelemetryschema.BuildCompleteFieldKeyMap(releaseTime)
aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, fm, cb, fl)
traceStmtBuilder := NewTraceQueryStatementBuilder(
instrumentationtest.New().ToProviderSettings(),

View File

@@ -1,4 +1,4 @@
package telemetrytraces
package tracesstatementbuilder
import (
"context"
@@ -8,7 +8,8 @@ import (
"github.com/SigNoz/signoz/pkg/factory"
"github.com/SigNoz/signoz/pkg/flagger"
"github.com/SigNoz/signoz/pkg/querybuilder"
"github.com/SigNoz/signoz/pkg/telemetryresourcefilter"
"github.com/SigNoz/signoz/pkg/statementbuilder/resourcefilter"
"github.com/SigNoz/signoz/pkg/telemetryschema/tracestelemetryschema"
qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
"github.com/SigNoz/signoz/pkg/valuer"
@@ -35,12 +36,12 @@ func NewTraceOperatorStatementBuilder(
aggExprRewriter qbtypes.AggExprRewriter,
flagger flagger.Flagger,
) *traceOperatorStatementBuilder {
tracesSettings := factory.NewScopedProviderSettings(settings, "github.com/SigNoz/signoz/pkg/telemetrytraces")
tracesSettings := factory.NewScopedProviderSettings(settings, "github.com/SigNoz/signoz/pkg/telemetryschema/tracestelemetryschema")
resourceFilterStmtBuilder := telemetryresourcefilter.New[qbtypes.TraceAggregation](
resourceFilterStmtBuilder := resourcefilter.New[qbtypes.TraceAggregation](
settings,
DBName,
TracesResourceV3TableName,
tracestelemetryschema.DBName,
tracestelemetryschema.TracesResourceV3TableName,
telemetrytypes.SignalTraces,
telemetrytypes.SourceUnspecified,
metadataStore,

View File

@@ -1,4 +1,4 @@
package telemetrytraces
package tracesstatementbuilder
import (
"context"
@@ -9,6 +9,7 @@ import (
"github.com/SigNoz/signoz/pkg/flagger/flaggertest"
"github.com/SigNoz/signoz/pkg/instrumentation/instrumentationtest"
"github.com/SigNoz/signoz/pkg/querybuilder"
"github.com/SigNoz/signoz/pkg/telemetryschema/tracestelemetryschema"
qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
"github.com/SigNoz/signoz/pkg/types/telemetrytypes/telemetrytypestest"
@@ -20,11 +21,11 @@ import (
func TestTraceTimeRangeOptimization(t *testing.T) {
releaseTime := time.Date(2025, 5, 22, 22, 0, 0, 0, time.UTC)
fm := NewFieldMapper()
cb := NewConditionBuilder(fm)
fm := tracestelemetryschema.NewFieldMapper()
cb := tracestelemetryschema.NewConditionBuilder(fm)
mockMetadataStore := telemetrytypestest.NewMockMetadataStore()
mockMetadataStore.KeysMap = buildCompleteFieldKeyMap(releaseTime)
mockMetadataStore.KeysMap = tracestelemetryschema.BuildCompleteFieldKeyMap(releaseTime)
mockMetadataStore.KeysMap["trace_id"] = []*telemetrytypes.TelemetryFieldKey{{
Name: "trace_id",
FieldContext: telemetrytypes.FieldContextSpan,
@@ -129,7 +130,7 @@ func TestTraceTimeRangeOptimization(t *testing.T) {
assert.NotEmpty(t, stmt.Query)
if tt.query.Filter != nil && tt.query.Filter.Expression != "" {
traceIDs, found := ExtractTraceIDsFromFilter(tt.query.Filter.Expression)
traceIDs, found := tracestelemetryschema.ExtractTraceIDsFromFilter(tt.query.Filter.Expression)
assert.Equal(t, tt.expectTimeOptimization, found && len(traceIDs) > 0)
}
})

View File

@@ -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"
)

View File

@@ -1,15 +0,0 @@
package telemetrylogs
const (
DBName = "signoz_logs"
LogsV2TableName = "distributed_logs_v2"
LogsV2LocalTableName = "logs_v2"
TagAttributesV2TableName = "distributed_tag_attributes_v2"
TagAttributesV2LocalTableName = "tag_attributes_v2"
LogAttributeKeysTblName = "distributed_logs_attribute_keys"
LogResourceKeysTblName = "distributed_logs_resource_keys"
LogsResourceV2TableName = "distributed_logs_v2_resource"
PathTypesTableName = "distributed_json_path_types"
PromotedPathsTableName = "distributed_json_promoted_paths"
SkipIndexTableName = "system.data_skipping_indices"
)

View File

@@ -13,7 +13,7 @@ import (
"github.com/SigNoz/signoz-otel-collector/constants"
"github.com/SigNoz/signoz/pkg/errors"
"github.com/SigNoz/signoz/pkg/querybuilder"
"github.com/SigNoz/signoz/pkg/telemetrylogs"
"github.com/SigNoz/signoz/pkg/telemetryschema/logstelemetryschema"
"github.com/SigNoz/signoz/pkg/types/ctxtypes"
"github.com/SigNoz/signoz/pkg/types/instrumentationtypes"
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
@@ -139,8 +139,8 @@ func buildListLogsJSONIndexesQuery(cluster string, filters ...string) (string, [
"name", "type_full", "expr", "granularity",
).From(fmt.Sprintf("clusterAllReplicas('%s', %s)", cluster, SkipIndexTableName))
sb.Where(sb.Equal("database", telemetrylogs.DBName))
sb.Where(sb.Equal("table", telemetrylogs.LogsV2LocalTableName))
sb.Where(sb.Equal("database", logstelemetryschema.DBName))
sb.Where(sb.Equal("table", logstelemetryschema.LogsV2LocalTableName))
sb.Where(sb.Or(
sb.ILike("expr", fmt.Sprintf("%%%s%%", querybuilder.FormatValueForContains(constants.BodyV2ColumnPrefix))),
sb.ILike("expr", fmt.Sprintf("%%%s%%", querybuilder.FormatValueForContains(constants.BodyPromotedColumnPrefix))),
@@ -189,12 +189,12 @@ func (t *telemetryMetaStore) ListLogsJSONIndexes(ctx context.Context, filters ..
baseColumn := ""
fieldName := ""
switch {
case strings.HasPrefix(columnExpr, telemetrylogs.BodyV2ColumnPrefix):
baseColumn = telemetrylogs.BodyV2ColumnPrefix
fieldName = strings.TrimPrefix(columnExpr, telemetrylogs.BodyV2ColumnPrefix)
case strings.HasPrefix(columnExpr, telemetrylogs.BodyPromotedColumnPrefix):
baseColumn = telemetrylogs.BodyPromotedColumnPrefix
fieldName = strings.TrimPrefix(columnExpr, telemetrylogs.BodyPromotedColumnPrefix)
case strings.HasPrefix(columnExpr, logstelemetryschema.BodyV2ColumnPrefix):
baseColumn = logstelemetryschema.BodyV2ColumnPrefix
fieldName = strings.TrimPrefix(columnExpr, logstelemetryschema.BodyV2ColumnPrefix)
case strings.HasPrefix(columnExpr, logstelemetryschema.BodyPromotedColumnPrefix):
baseColumn = logstelemetryschema.BodyPromotedColumnPrefix
fieldName = strings.TrimPrefix(columnExpr, logstelemetryschema.BodyPromotedColumnPrefix)
}
fieldName = strings.ReplaceAll(fieldName, "`", "")
@@ -228,12 +228,12 @@ func (t *telemetryMetaStore) ListJSONValues(ctx context.Context, path string, li
}
if promoted {
path = telemetrylogs.BodyPromotedColumnPrefix + path
path = logstelemetryschema.BodyPromotedColumnPrefix + path
} else {
path = telemetrylogs.BodyV2ColumnPrefix + path
path = logstelemetryschema.BodyV2ColumnPrefix + path
}
from := fmt.Sprintf("%s.%s", telemetrylogs.DBName, telemetrylogs.LogsV2TableName)
from := fmt.Sprintf("%s.%s", logstelemetryschema.DBName, logstelemetryschema.LogsV2TableName)
colExpr := func(typ telemetrytypes.JSONDataType) string {
return fmt.Sprintf("dynamicElement(%s, '%s')", path, typ.StringValue())
}
@@ -382,7 +382,7 @@ func (t *telemetryMetaStore) IsPathPromoted(ctx context.Context, path string) (b
split := strings.Split(path, telemetrytypes.ArraySep)
pathSegment := split[0]
query := fmt.Sprintf("SELECT 1 FROM %s.%s WHERE signal = ? AND column_name = ? AND field_context = ? AND field_name = ? LIMIT 1", DBName, PromotedPathsTableName)
rows, err := t.telemetrystore.ClickhouseDB().Query(ctx, query, telemetrytypes.SignalLogs, telemetrylogs.LogsV2BodyPromotedColumn, telemetrytypes.FieldContextBody, pathSegment)
rows, err := t.telemetrystore.ClickhouseDB().Query(ctx, query, telemetrytypes.SignalLogs, logstelemetryschema.LogsV2BodyPromotedColumn, telemetrytypes.FieldContextBody, pathSegment)
if err != nil {
return false, errors.WrapInternalf(err, CodeFailCheckPathPromoted, "failed to check if path %s is promoted", path)
}
@@ -397,7 +397,7 @@ func (t *telemetryMetaStore) GetPromotedPaths(ctx context.Context, paths ...stri
sb := sqlbuilder.Select("field_name").From(fmt.Sprintf("%s.%s", DBName, PromotedPathsTableName))
conditions := []string{
sb.Equal("signal", telemetrytypes.SignalLogs),
sb.Equal("column_name", telemetrylogs.LogsV2BodyPromotedColumn),
sb.Equal("column_name", logstelemetryschema.LogsV2BodyPromotedColumn),
sb.Equal("field_context", telemetrytypes.FieldContextBody),
sb.NotEqual("field_name", "__all__"),
}
@@ -433,8 +433,8 @@ func (t *telemetryMetaStore) GetPromotedPaths(ctx context.Context, paths ...stri
// TODO(Piyush): Remove this function.
func CleanPathPrefixes(path string) string {
path = strings.TrimPrefix(path, telemetrytypes.BodyJSONStringSearchPrefix)
path = strings.TrimPrefix(path, telemetrylogs.BodyV2ColumnPrefix)
path = strings.TrimPrefix(path, telemetrylogs.BodyPromotedColumnPrefix)
path = strings.TrimPrefix(path, logstelemetryschema.BodyV2ColumnPrefix)
path = strings.TrimPrefix(path, logstelemetryschema.BodyPromotedColumnPrefix)
return path
}
@@ -454,7 +454,7 @@ func (t *telemetryMetaStore) PromotePaths(ctx context.Context, paths ...string)
if trimmed == "" {
continue
}
if err := batch.Append(telemetrytypes.SignalLogs, telemetrylogs.LogsV2BodyPromotedColumn, "JSON()", telemetrytypes.FieldContextBody, trimmed, 0, releaseTime); err != nil {
if err := batch.Append(telemetrytypes.SignalLogs, logstelemetryschema.LogsV2BodyPromotedColumn, "JSON()", telemetrytypes.FieldContextBody, trimmed, 0, releaseTime); err != nil {
_ = batch.Abort()
return errors.WrapInternalf(err, CodeFailedToAppendPath, "failed to append path")
}

View File

@@ -6,7 +6,7 @@ import (
"github.com/SigNoz/signoz-otel-collector/constants"
"github.com/SigNoz/signoz/pkg/querybuilder"
"github.com/SigNoz/signoz/pkg/telemetrylogs"
"github.com/SigNoz/signoz/pkg/telemetryschema/logstelemetryschema"
"github.com/stretchr/testify/require"
)
@@ -25,8 +25,8 @@ func TestBuildListLogsJSONIndexesQuery(t *testing.T) {
expectedSQL: "SELECT name, type_full, expr, granularity FROM clusterAllReplicas('test-cluster', system.data_skipping_indices) " +
"WHERE database = ? AND table = ? AND (LOWER(expr) LIKE LOWER(?) OR LOWER(expr) LIKE LOWER(?))",
expectedArgs: []any{
telemetrylogs.DBName,
telemetrylogs.LogsV2LocalTableName,
logstelemetryschema.DBName,
logstelemetryschema.LogsV2LocalTableName,
fmt.Sprintf("%%%s%%", querybuilder.FormatValueForContains(constants.BodyV2ColumnPrefix)),
fmt.Sprintf("%%%s%%", querybuilder.FormatValueForContains(constants.BodyPromotedColumnPrefix)),
},
@@ -38,8 +38,8 @@ func TestBuildListLogsJSONIndexesQuery(t *testing.T) {
expectedSQL: "SELECT name, type_full, expr, granularity FROM clusterAllReplicas('test-cluster', system.data_skipping_indices) " +
"WHERE database = ? AND table = ? AND (LOWER(expr) LIKE LOWER(?) OR LOWER(expr) LIKE LOWER(?)) AND (LOWER(replaceAll(expr, '`', '')) LIKE LOWER(?) OR LOWER(replaceAll(expr, '`', '')) LIKE LOWER(?))",
expectedArgs: []any{
telemetrylogs.DBName,
telemetrylogs.LogsV2LocalTableName,
logstelemetryschema.DBName,
logstelemetryschema.LogsV2LocalTableName,
fmt.Sprintf("%%%s%%", querybuilder.FormatValueForContains(constants.BodyV2ColumnPrefix)),
fmt.Sprintf("%%%s%%", querybuilder.FormatValueForContains(constants.BodyPromotedColumnPrefix)),
fmt.Sprintf("%%%s%%", querybuilder.FormatValueForContains("foo")),

View File

@@ -14,11 +14,12 @@ import (
"github.com/SigNoz/signoz/pkg/factory"
"github.com/SigNoz/signoz/pkg/flagger"
"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/telemetryschema/audittelemetryschema"
"github.com/SigNoz/signoz/pkg/telemetryschema/logstelemetryschema"
"github.com/SigNoz/signoz/pkg/telemetryschema/metertelemetryschema"
"github.com/SigNoz/signoz/pkg/telemetryschema/metricstelemetryschema"
"github.com/SigNoz/signoz/pkg/telemetryschema/tracestelemetryschema"
"github.com/SigNoz/signoz/pkg/telemetrystore"
"github.com/SigNoz/signoz/pkg/telemetrytraces"
"github.com/SigNoz/signoz/pkg/types/ctxtypes"
"github.com/SigNoz/signoz/pkg/types/featuretypes"
"github.com/SigNoz/signoz/pkg/types/instrumentationtypes"
@@ -77,27 +78,6 @@ func escapeForLike(s string) string {
func NewTelemetryMetaStore(
settings factory.ProviderSettings,
telemetrystore telemetrystore.TelemetryStore,
tracesDBName string,
tracesFieldsTblName string,
spanAttributesKeysTblName string,
indexV3TblName string,
metricsDBName string,
metricsFieldsTblName string,
meterDBName string,
meterFieldsTblName string,
logsDBName string,
logsV2TblName string,
logsFieldsTblName string,
logAttributeKeysTblName string,
logResourceKeysTblName string,
auditDBName string,
auditLogsTblName string,
auditFieldsTblName string,
auditAttributeKeysTblName string,
auditResourceKeysTblName string,
relatedMetadataDBName string,
relatedMetadataTblName string,
columnEvolutionMetadataTblName string,
fl flagger.Flagger,
) telemetrytypes.MetadataStore {
metadataSettings := factory.NewScopedProviderSettings(settings, "github.com/SigNoz/signoz/pkg/telemetrymetadata")
@@ -108,32 +88,32 @@ func NewTelemetryMetaStore(
t := &telemetryMetaStore{
logger: metadataSettings.Logger(),
telemetrystore: telemetrystore,
tracesDBName: tracesDBName,
tracesFieldsTblName: tracesFieldsTblName,
spanAttributesKeysTblName: spanAttributesKeysTblName,
indexV3TblName: indexV3TblName,
metricsDBName: metricsDBName,
metricsFieldsTblName: metricsFieldsTblName,
meterDBName: meterDBName,
meterFieldsTblName: meterFieldsTblName,
logsDBName: logsDBName,
logsV2TblName: logsV2TblName,
logsFieldsTblName: logsFieldsTblName,
logAttributeKeysTblName: logAttributeKeysTblName,
logResourceKeysTblName: logResourceKeysTblName,
auditDBName: auditDBName,
auditLogsTblName: auditLogsTblName,
auditFieldsTblName: auditFieldsTblName,
auditAttributeKeysTblName: auditAttributeKeysTblName,
auditResourceKeysTblName: auditResourceKeysTblName,
relatedMetadataDBName: relatedMetadataDBName,
relatedMetadataTblName: relatedMetadataTblName,
columnEvolutionMetadataTblName: columnEvolutionMetadataTblName,
tracesDBName: tracestelemetryschema.DBName,
tracesFieldsTblName: tracestelemetryschema.TagAttributesV2TableName,
spanAttributesKeysTblName: tracestelemetryschema.SpanAttributesKeysTblName,
indexV3TblName: tracestelemetryschema.SpanIndexV3TableName,
metricsDBName: metricstelemetryschema.DBName,
metricsFieldsTblName: metricstelemetryschema.AttributesMetadataTableName,
meterDBName: metertelemetryschema.DBName,
meterFieldsTblName: metertelemetryschema.SamplesAgg1dTableName,
logsDBName: logstelemetryschema.DBName,
logsV2TblName: logstelemetryschema.LogsV2TableName,
logsFieldsTblName: logstelemetryschema.TagAttributesV2TableName,
logAttributeKeysTblName: logstelemetryschema.LogAttributeKeysTblName,
logResourceKeysTblName: logstelemetryschema.LogResourceKeysTblName,
auditDBName: audittelemetryschema.DBName,
auditLogsTblName: audittelemetryschema.AuditLogsTableName,
auditFieldsTblName: audittelemetryschema.TagAttributesTableName,
auditAttributeKeysTblName: audittelemetryschema.LogAttributeKeysTblName,
auditResourceKeysTblName: audittelemetryschema.LogResourceKeysTblName,
relatedMetadataDBName: DBName,
relatedMetadataTblName: AttributesMetadataTableName,
columnEvolutionMetadataTblName: ColumnEvolutionMetadataTableName,
jsonColumnMetadata: map[telemetrytypes.Signal]map[telemetrytypes.FieldContext]telemetrytypes.JSONColumnMetadata{
telemetrytypes.SignalLogs: {
telemetrytypes.FieldContextBody: telemetrytypes.JSONColumnMetadata{
BaseColumn: telemetrylogs.LogsV2BodyV2Column,
PromotedColumn: telemetrylogs.LogsV2BodyPromotedColumn,
BaseColumn: logstelemetryschema.LogsV2BodyV2Column,
PromotedColumn: logstelemetryschema.LogsV2BodyPromotedColumn,
},
},
},
@@ -309,8 +289,8 @@ func (t *telemetryMetaStore) getTracesKeys(ctx context.Context, fieldKeySelector
complete := rowCount <= limit
staticKeys := []string{"isRoot", "isEntryPoint"}
staticKeys = append(staticKeys, maps.Keys(telemetrytraces.IntrinsicFields)...)
staticKeys = append(staticKeys, maps.Keys(telemetrytraces.CalculatedFields)...)
staticKeys = append(staticKeys, maps.Keys(tracestelemetryschema.IntrinsicFields)...)
staticKeys = append(staticKeys, maps.Keys(tracestelemetryschema.CalculatedFields)...)
// Add matching intrinsic and matching calculated fields
// These don't count towards the limit
@@ -324,14 +304,14 @@ func (t *telemetryMetaStore) getTracesKeys(ctx context.Context, fieldKeySelector
}
if found {
if field, exists := telemetrytraces.IntrinsicFields[key]; exists {
if field, exists := tracestelemetryschema.IntrinsicFields[key]; exists {
if _, added := mapOfKeys[field.Name+";"+field.FieldContext.StringValue()+";"+field.FieldDataType.StringValue()]; !added {
keys = append(keys, &field)
}
continue
}
if field, exists := telemetrytraces.CalculatedFields[key]; exists {
if field, exists := tracestelemetryschema.CalculatedFields[key]; exists {
if _, added := mapOfKeys[field.Name+";"+field.FieldContext.StringValue()+";"+field.FieldDataType.StringValue()]; !added {
keys = append(keys, &field)
}
@@ -656,7 +636,7 @@ func (t *telemetryMetaStore) getLogsKeys(ctx context.Context, orgID valuer.UUID,
complete := rowCount <= limit
staticKeys := []string{}
staticKeys = append(staticKeys, maps.Keys(telemetrylogs.IntrinsicFields)...)
staticKeys = append(staticKeys, maps.Keys(logstelemetryschema.IntrinsicFields)...)
// Add matching intrinsic and matching calculated fields
// These don't count towards the limit
@@ -670,7 +650,7 @@ func (t *telemetryMetaStore) getLogsKeys(ctx context.Context, orgID valuer.UUID,
}
if found {
if field, exists := telemetrylogs.IntrinsicFields[key]; exists {
if field, exists := logstelemetryschema.IntrinsicFields[key]; exists {
if _, added := mapOfKeys[field.Name+";"+field.FieldContext.StringValue()+";"+field.FieldDataType.StringValue()]; !added {
keys = append(keys, &field)
}
@@ -898,7 +878,7 @@ func (t *telemetryMetaStore) getAuditKeys(ctx context.Context, fieldKeySelectors
complete := rowCount <= limit
// Add intrinsic audit fields (same as logs intrinsics: body, severity_text, etc.)
staticKeys := maps.Keys(telemetryaudit.IntrinsicFields)
staticKeys := maps.Keys(audittelemetryschema.IntrinsicFields)
for _, key := range staticKeys {
found := false
for _, v := range searchTexts {
@@ -909,7 +889,7 @@ func (t *telemetryMetaStore) getAuditKeys(ctx context.Context, fieldKeySelectors
}
if found {
if field, exists := telemetryaudit.IntrinsicFields[key]; exists {
if field, exists := audittelemetryschema.IntrinsicFields[key]; exists {
if _, added := mapOfKeys[field.Name+";"+field.FieldContext.StringValue()+";"+field.FieldDataType.StringValue()]; !added {
keys = append(keys, &field)
}
@@ -1176,7 +1156,7 @@ func enrichWithIntrinsicMetricKeys(keys map[string][]*telemetrytypes.TelemetryFi
continue
}
for name, key := range telemetrymetrics.IntrinsicMetricFieldDefinitions {
for name, key := range metricstelemetryschema.IntrinsicMetricFieldDefinitions {
if !selectorMatchesIntrinsicField(selector, key) {
continue
}
@@ -1892,7 +1872,7 @@ func (t *telemetryMetaStore) getIntrinsicMetricFieldValues(ctx context.Context,
instrumentationtypes.CodeNamespace: "metadata",
instrumentationtypes.CodeFunctionName: "getIntrinsicMetricFieldValues",
})
key, ok := telemetrymetrics.IntrinsicMetricFieldDefinitions[fieldValueSelector.Name]
key, ok := metricstelemetryschema.IntrinsicMetricFieldDefinitions[fieldValueSelector.Name]
if !ok {
return &telemetrytypes.TelemetryFieldValues{}, nil
}
@@ -1922,18 +1902,18 @@ func (t *telemetryMetaStore) getIntrinsicMetricFieldValues(ctx context.Context,
tableName string
extraConds []string
}{
{telemetrymetrics.TimeseriesV41weekTableName, nil},
{metricstelemetryschema.TimeseriesV41weekTableName, nil},
}
if reductionEnabled {
sources = append(sources,
struct {
tableName string
extraConds []string
}{telemetrymetrics.TimeseriesV4BufferTableName, []string{"is_reduced = false"}},
}{metricstelemetryschema.TimeseriesV4BufferTableName, []string{"is_reduced = false"}},
struct {
tableName string
extraConds []string
}{telemetrymetrics.TimeseriesV4ReducedTableName, nil},
}{metricstelemetryschema.TimeseriesV4ReducedTableName, nil},
)
}
@@ -2262,7 +2242,7 @@ func (t *telemetryMetaStore) fetchMetricsTemporalityAndType(ctx context.Context,
instrumentationtypes.CodeNamespace: "metadata",
instrumentationtypes.CodeFunctionName: "fetchMetricsTemporalityAndType",
})
adjustedStartTs, adjustedEndTs, tsTableName, _ := telemetrymetrics.WhichTSTableToUse(queryTimeRangeStartTs, queryTimeRangeEndTs, false, nil)
adjustedStartTs, adjustedEndTs, tsTableName, _ := metricstelemetryschema.WhichTSTableToUse(queryTimeRangeStartTs, queryTimeRangeEndTs, false, nil)
temporalities, types, err := t.fetchTemporalityTypeForTable(ctx, tsTableName, adjustedStartTs, adjustedEndTs, metricNames)
if err != nil {
@@ -2271,7 +2251,7 @@ func (t *telemetryMetaStore) fetchMetricsTemporalityAndType(ctx context.Context,
reduced := make(map[string]bool)
if reductionEnabled {
bufferTemporalities, bufferTypes, err := t.fetchTemporalityTypeForTable(ctx, telemetrymetrics.TimeseriesV4BufferTableName, adjustedStartTs, adjustedEndTs, metricNames, "is_reduced = false")
bufferTemporalities, bufferTypes, err := t.fetchTemporalityTypeForTable(ctx, metricstelemetryschema.TimeseriesV4BufferTableName, adjustedStartTs, adjustedEndTs, metricNames, "is_reduced = false")
if err != nil {
return nil, nil, nil, err
}
@@ -2284,7 +2264,7 @@ func (t *telemetryMetaStore) fetchMetricsTemporalityAndType(ctx context.Context,
}
}
reducedTemporalities, reducedTypes, err := t.fetchTemporalityTypeForTable(ctx, telemetrymetrics.TimeseriesV4ReducedTableName, adjustedStartTs, adjustedEndTs, metricNames)
reducedTemporalities, reducedTypes, err := t.fetchTemporalityTypeForTable(ctx, metricstelemetryschema.TimeseriesV4ReducedTableName, adjustedStartTs, adjustedEndTs, metricNames)
if err != nil {
return nil, nil, nil, err
}
@@ -2616,13 +2596,13 @@ func (t *telemetryMetaStore) GetFirstSeenFromMetricMetadata(ctx context.Context,
}
func (t *telemetryMetaStore) FetchLastSeenInfoMulti(ctx context.Context, orgID valuer.UUID, metricNames ...string) (map[string]int64, error) {
lastSeenInfo, err := t.fetchLastSeenInfoForTable(ctx, telemetrymetrics.TimeseriesV4TableName, metricNames)
lastSeenInfo, err := t.fetchLastSeenInfoForTable(ctx, metricstelemetryschema.TimeseriesV4TableName, metricNames)
if err != nil {
return nil, err
}
if t.fl.BooleanOrEmpty(ctx, flagger.FeatureEnableMetricsReduction, featuretypes.NewFlaggerEvaluationContext(orgID)) {
reducedLastSeen, err := t.fetchLastSeenInfoForTable(ctx, telemetrymetrics.TimeseriesV4ReducedTableName, metricNames)
reducedLastSeen, err := t.fetchLastSeenInfoForTable(ctx, metricstelemetryschema.TimeseriesV4ReducedTableName, metricNames)
if err != nil {
return nil, err
}

View File

@@ -9,13 +9,8 @@ import (
"github.com/SigNoz/signoz/pkg/errors"
"github.com/SigNoz/signoz/pkg/flagger/flaggertest"
"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"
"github.com/SigNoz/signoz/pkg/telemetrystore"
"github.com/SigNoz/signoz/pkg/telemetrystore/telemetrystoretest"
"github.com/SigNoz/signoz/pkg/telemetrytraces"
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
@@ -42,27 +37,6 @@ func TestGetFirstSeenFromMetricMetadata(t *testing.T) {
metadata := NewTelemetryMetaStore(
instrumentationtest.New().ToProviderSettings(),
mockTelemetryStore,
telemetrytraces.DBName,
telemetrytraces.TagAttributesV2TableName,
telemetrytraces.SpanAttributesKeysTblName,
telemetrytraces.SpanIndexV3TableName,
telemetrymetrics.DBName,
telemetrymetrics.AttributesMetadataTableName,
telemetrymeter.DBName,
telemetrymeter.SamplesAgg1dTableName,
telemetrylogs.DBName,
telemetrylogs.LogsV2TableName,
telemetrylogs.TagAttributesV2TableName,
telemetrylogs.LogAttributeKeysTblName,
telemetrylogs.LogResourceKeysTblName,
telemetryaudit.DBName,
telemetryaudit.AuditLogsTableName,
telemetryaudit.TagAttributesTableName,
telemetryaudit.LogAttributeKeysTblName,
telemetryaudit.LogResourceKeysTblName,
DBName,
AttributesMetadataTableName,
ColumnEvolutionMetadataTableName,
flaggertest.New(t),
)

View File

@@ -27,7 +27,7 @@ func (v *TelemetryFieldVisitor) VisitColumnDef(expr *parser.ColumnDef) error {
}
// Parse column name to extract context and data type
columnName := parser.Format(expr.Name)
columnName := expr.Name.String()
// Remove backticks if present
columnName = strings.TrimPrefix(columnName, "`")
@@ -75,7 +75,7 @@ func (v *TelemetryFieldVisitor) VisitColumnDef(expr *parser.ColumnDef) error {
// Extract field name from the DEFAULT expression
// The DEFAULT expression should be something like: resources_string['k8s.cluster.name']
// We need to extract the key inside the square brackets
defaultExprStr := parser.Format(expr.DefaultExpr)
defaultExprStr := expr.DefaultExpr.String()
// Look for the pattern: map['key']
startIdx := strings.Index(defaultExprStr, "['")

View File

@@ -1,37 +0,0 @@
package telemetrymetrics
import "github.com/SigNoz/signoz/pkg/types/telemetrytypes"
var IntrinsicFields = []string{
"temporality",
"metric_name",
"type",
"is_monotonic",
}
var IntrinsicMetricFieldDefinitions = map[string]telemetrytypes.TelemetryFieldKey{
"metric_name": {
Name: "metric_name",
Signal: telemetrytypes.SignalMetrics,
FieldContext: telemetrytypes.FieldContextMetric,
FieldDataType: telemetrytypes.FieldDataTypeString,
},
// "type": {
// Name: "type",
// Signal: telemetrytypes.SignalMetrics,
// FieldContext: telemetrytypes.FieldContextMetric,
// FieldDataType: telemetrytypes.FieldDataTypeString,
// },
// "temporality": {
// Name: "temporality",
// Signal: telemetrytypes.SignalMetrics,
// FieldContext: telemetrytypes.FieldContextMetric,
// FieldDataType: telemetrytypes.FieldDataTypeString,
// },
// "is_monotonic": {
// Name: "is_monotonic",
// Signal: telemetrytypes.SignalMetrics,
// FieldContext: telemetrytypes.FieldContextMetric,
// FieldDataType: telemetrytypes.FieldDataTypeBool,
// },
}

View File

@@ -1,4 +1,4 @@
package telemetryaudit
package audittelemetryschema
import (
"context"

View File

@@ -1,4 +1,4 @@
package telemetryaudit
package audittelemetryschema
import (
schema "github.com/SigNoz/signoz-otel-collector/cmd/signozschemamigrator/schema_migrator"
@@ -127,3 +127,14 @@ var auditLogColumns = map[string]*schema.Column{
"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}},
}
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"
)

View File

@@ -1,4 +1,4 @@
package telemetryaudit
package audittelemetryschema
import (
"context"

View File

@@ -1,4 +1,4 @@
package telemetrylogs
package logstelemetryschema
import (
"context"

View File

@@ -1,4 +1,4 @@
package telemetrylogs
package logstelemetryschema
import (
"context"
@@ -548,7 +548,7 @@ func TestConditionForSynthesizedPrefixedKeys(t *testing.T) {
t.Run("bare intrinsic column resolves to the column, not synthesized attributes", func(t *testing.T) {
sb := sqlbuilder.NewSelectBuilder()
key := telemetrytypes.TelemetryFieldKey{Name: "severity_text"}
conds, _, err := cb.ConditionFor(ctx, valuer.UUID{}, 0, 0, &key, nil, qbtypes.ConditionBuilderOptions{},qbtypes.FilterOperatorEqual, "ERROR", sb)
conds, _, err := cb.ConditionFor(ctx, valuer.UUID{}, 0, 0, &key, nil, qbtypes.ConditionBuilderOptions{}, qbtypes.FilterOperatorEqual, "ERROR", sb)
require.NoError(t, err)
require.Len(t, conds, 1)
sb.Where(conds...)
@@ -560,7 +560,7 @@ func TestConditionForSynthesizedPrefixedKeys(t *testing.T) {
t.Run("attribute context", func(t *testing.T) {
sb := sqlbuilder.NewSelectBuilder()
key := telemetrytypes.TelemetryFieldKey{Name: "custom.key", FieldContext: telemetrytypes.FieldContextAttribute}
conds, warnings, err := cb.ConditionFor(ctx, valuer.UUID{}, 0, 0, &key, nil, qbtypes.ConditionBuilderOptions{},qbtypes.FilterOperatorEqual, "v", sb)
conds, warnings, err := cb.ConditionFor(ctx, valuer.UUID{}, 0, 0, &key, nil, qbtypes.ConditionBuilderOptions{}, qbtypes.FilterOperatorEqual, "v", sb)
require.NoError(t, err)
assert.NotEmpty(t, warnings)
require.Len(t, conds, 2)
@@ -571,7 +571,7 @@ func TestConditionForSynthesizedPrefixedKeys(t *testing.T) {
t.Run("resource context", func(t *testing.T) {
sb := sqlbuilder.NewSelectBuilder()
key := telemetrytypes.TelemetryFieldKey{Name: "custom.key", FieldContext: telemetrytypes.FieldContextResource}
conds, warnings, err := cb.ConditionFor(ctx, valuer.UUID{}, 0, 0, &key, nil, qbtypes.ConditionBuilderOptions{},qbtypes.FilterOperatorEqual, "v", sb)
conds, warnings, err := cb.ConditionFor(ctx, valuer.UUID{}, 0, 0, &key, nil, qbtypes.ConditionBuilderOptions{}, qbtypes.FilterOperatorEqual, "v", sb)
require.NoError(t, err)
assert.NotEmpty(t, warnings)
require.Len(t, conds, 2)
@@ -582,7 +582,7 @@ func TestConditionForSynthesizedPrefixedKeys(t *testing.T) {
t.Run("log context folds to attributes then body", func(t *testing.T) {
sb := sqlbuilder.NewSelectBuilder()
key := telemetrytypes.TelemetryFieldKey{Name: "custom.key", FieldContext: telemetrytypes.FieldContextLog}
conds, warnings, err := cb.ConditionFor(ctx, valuer.UUID{}, 0, 0, &key, nil, qbtypes.ConditionBuilderOptions{},qbtypes.FilterOperatorEqual, "v", sb)
conds, warnings, err := cb.ConditionFor(ctx, valuer.UUID{}, 0, 0, &key, nil, qbtypes.ConditionBuilderOptions{}, qbtypes.FilterOperatorEqual, "v", sb)
require.NoError(t, err)
assert.NotEmpty(t, warnings)
require.Len(t, conds, 4)

View File

@@ -1,8 +1,6 @@
package telemetrylogs
package logstelemetryschema
import (
"fmt"
"github.com/SigNoz/signoz-otel-collector/constants"
qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
@@ -41,9 +39,8 @@ const (
// messageSubColumn is the ClickHouse sub-column that body searches map to
// when use_json_body feature flag is true.
messageSubField = "message"
messageSubColumn = "body_v2.message"
bodySearchDefaultWarning = "body searches default to `body.message:string`. Use `body.<key>` to search a different field inside body"
messageSubField = "message"
messageSubColumn = "body_v2.message"
// bodyMessageField is the field name addressing the message sub-field of the
// body when use_json_body is enabled (i.e. `body` + `.` + `message`). hasToken
@@ -137,10 +134,16 @@ var (
}
)
func bodyAliasExpression(bodyJSONEnabled bool) string {
if !bodyJSONEnabled {
return LogsV2BodyColumn
}
return fmt.Sprintf("%s as body", LogsV2BodyV2Column)
}
const (
DBName = "signoz_logs"
LogsV2TableName = "distributed_logs_v2"
LogsV2LocalTableName = "logs_v2"
TagAttributesV2TableName = "distributed_tag_attributes_v2"
TagAttributesV2LocalTableName = "tag_attributes_v2"
LogAttributeKeysTblName = "distributed_logs_attribute_keys"
LogResourceKeysTblName = "distributed_logs_resource_keys"
LogsResourceV2TableName = "distributed_logs_v2_resource"
PathTypesTableName = "distributed_json_path_types"
PromotedPathsTableName = "distributed_json_promoted_paths"
SkipIndexTableName = "system.data_skipping_indices"
)

View File

@@ -1,4 +1,4 @@
package telemetrylogs
package logstelemetryschema
import (
"context"

View File

@@ -1,4 +1,4 @@
package telemetrylogs
package logstelemetryschema
import (
"context"

View File

@@ -1,4 +1,4 @@
package telemetrylogs
package logstelemetryschema
import (
"context"
@@ -19,7 +19,7 @@ func TestLikeAndILikeWithoutWildcards_Warns(t *testing.T) {
cb := NewConditionBuilder(fm, fl)
releaseTime := time.Date(2024, 1, 15, 10, 0, 0, 0, time.UTC)
keys := buildCompleteFieldKeyMap(releaseTime)
keys := BuildCompleteFieldKeyMap(releaseTime)
opts := querybuilder.FilterExprVisitorOpts{
Context: ctx,
@@ -57,7 +57,7 @@ func TestLikeAndILikeWithWildcards_NoWarn(t *testing.T) {
cb := NewConditionBuilder(fm, fl)
releaseTime := time.Date(2024, 1, 15, 10, 0, 0, 0, time.UTC)
keys := buildCompleteFieldKeyMap(releaseTime)
keys := BuildCompleteFieldKeyMap(releaseTime)
opts := querybuilder.FilterExprVisitorOpts{
Context: context.Background(),

View File

@@ -1,4 +1,4 @@
package telemetrylogs
package logstelemetryschema
import (
"context"
@@ -21,7 +21,7 @@ func TestFilterExprLogsBodyJSON(t *testing.T) {
cb := NewConditionBuilder(fm, fl)
// Define a comprehensive set of field keys to support all test cases
releaseTime := time.Date(2024, 1, 15, 10, 0, 0, 0, time.UTC)
keys := buildCompleteFieldKeyMap(releaseTime)
keys := BuildCompleteFieldKeyMap(releaseTime)
opts := querybuilder.FilterExprVisitorOpts{
Context: context.Background(),

View File

@@ -1,4 +1,4 @@
package telemetrylogs
package logstelemetryschema
import (
"context"
@@ -41,7 +41,7 @@ func TestFilterExprLogs(t *testing.T) {
cb := NewConditionBuilder(fm, fl)
// Define a comprehensive set of field keys to support all test cases
keys := buildCompleteFieldKeyMap(releaseTime)
keys := BuildCompleteFieldKeyMap(releaseTime)
opts := querybuilder.FilterExprVisitorOpts{
Context: ctx,
@@ -2476,7 +2476,7 @@ func TestFilterExprLogsConflictNegation(t *testing.T) {
// Define a comprehensive set of field keys to support all test cases
releaseTime := time.Date(2024, 1, 15, 10, 0, 0, 0, time.UTC)
keys := buildCompleteFieldKeyMap(releaseTime)
keys := BuildCompleteFieldKeyMap(releaseTime)
keys["body"] = []*telemetrytypes.TelemetryFieldKey{
{

View File

@@ -1,4 +1,4 @@
package telemetrylogs
package logstelemetryschema
import (
"fmt"

View File

@@ -1,4 +1,4 @@
package telemetrylogs
package logstelemetryschema
import (
"strings"

View File

@@ -1,4 +1,4 @@
package telemetrylogs
package logstelemetryschema
import (
"context"

View File

@@ -1,4 +1,4 @@
package telemetrylogs
package logstelemetryschema
import (
"testing"

View File

@@ -1,4 +1,4 @@
package telemetrylogs
package logstelemetryschema
import (
"strings"
@@ -19,7 +19,7 @@ func limitString(s string, maxLen int) string {
}
// Function to build a complete field key map for testing all scenarios.
func buildCompleteFieldKeyMap(releaseTime time.Time) map[string][]*telemetrytypes.TelemetryFieldKey {
func BuildCompleteFieldKeyMap(releaseTime time.Time) map[string][]*telemetrytypes.TelemetryFieldKey {
keysMap := map[string][]*telemetrytypes.TelemetryFieldKey{
"service.name": {
{
@@ -953,7 +953,7 @@ func buildCompleteFieldKeyMap(releaseTime time.Time) map[string][]*telemetrytype
return keysMap
}
func buildCompleteFieldKeyMapCollision() map[string][]*telemetrytypes.TelemetryFieldKey {
func BuildCompleteFieldKeyMapCollision() map[string][]*telemetrytypes.TelemetryFieldKey {
keysMap := map[string][]*telemetrytypes.TelemetryFieldKey{
"service.name": {
{

View File

@@ -1,4 +1,4 @@
package telemetrymeter
package metertelemetryschema
import (
"time"

View File

@@ -1,4 +1,4 @@
package telemetrymetrics
package metricstelemetryschema
import (
"context"

View File

@@ -1,4 +1,4 @@
package telemetrymetrics
package metricstelemetryschema
import (
"context"

View File

@@ -1,4 +1,4 @@
package telemetrymetrics
package metricstelemetryschema
import (
"fmt"
@@ -6,6 +6,7 @@ import (
"github.com/SigNoz/signoz/pkg/errors"
"github.com/SigNoz/signoz/pkg/types/metrictypes"
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
)
const (
@@ -46,10 +47,10 @@ const (
)
var (
oneHourInMilliseconds = uint64(time.Hour.Milliseconds() * 1)
OneHourInMilliseconds = uint64(time.Hour.Milliseconds() * 1)
sixHoursInMilliseconds = uint64(time.Hour.Milliseconds() * 6)
oneDayInMilliseconds = uint64(time.Hour.Milliseconds() * 24)
oneWeekInMilliseconds = uint64(oneDayInMilliseconds * 7)
OneDayInMilliseconds = uint64(time.Hour.Milliseconds() * 24)
oneWeekInMilliseconds = uint64(OneDayInMilliseconds * 7)
// when the query requests for almost 1 day, but not exactly 1 day, we need to add an offset to the end time
// to make sure that we are using the correct table
@@ -68,7 +69,7 @@ func WhichTSTableToUse(
// the buffer holds the recent raw window for reduced metrics and has the same
// shape as time_series_v4; round the start to the hour like the v4 table.
if useBuffer {
start = start - (start % (oneHourInMilliseconds))
start = start - (start % (OneHourInMilliseconds))
return start, end, TimeseriesV4BufferTableName, TimeseriesV4BufferLocalTableName
}
@@ -80,7 +81,7 @@ func WhichTSTableToUse(
switch tableHints.TimeSeriesTableName {
case TimeseriesV4LocalTableName:
// adjust the start time to nearest 1 hour
start = start - (start % (oneHourInMilliseconds))
start = start - (start % (OneHourInMilliseconds))
distributedTableName = TimeseriesV4TableName
case TimeseriesV46hrsLocalTableName:
// adjust the start time to nearest 6 hours
@@ -88,7 +89,7 @@ func WhichTSTableToUse(
distributedTableName = TimeseriesV46hrsTableName
case TimeseriesV41dayLocalTableName:
// adjust the start time to nearest 1 day
start = start - (start % (oneDayInMilliseconds))
start = start - (start % (OneDayInMilliseconds))
distributedTableName = TimeseriesV41dayTableName
case TimeseriesV41weekLocalTableName:
// adjust the start time to nearest 1 week
@@ -107,17 +108,17 @@ func WhichTSTableToUse(
var localTableName string
if end-start < sixHoursInMilliseconds {
// adjust the start time to nearest 1 hour
start = start - (start % (oneHourInMilliseconds))
start = start - (start % (OneHourInMilliseconds))
distributedTableName = TimeseriesV4TableName
localTableName = TimeseriesV4LocalTableName
} else if end-start < oneDayInMilliseconds {
} else if end-start < OneDayInMilliseconds {
// adjust the start time to nearest 6 hours
start = start - (start % (sixHoursInMilliseconds))
distributedTableName = TimeseriesV46hrsTableName
localTableName = TimeseriesV46hrsLocalTableName
} else if end-start < oneWeekInMilliseconds {
// adjust the start time to nearest 1 day
start = start - (start % (oneDayInMilliseconds))
start = start - (start % (OneDayInMilliseconds))
distributedTableName = TimeseriesV41dayTableName
localTableName = TimeseriesV41dayLocalTableName
} else {
@@ -206,7 +207,7 @@ func WhichSamplesTableToUse(
return SamplesV4TableName, SamplesV4LocalTableName
}
if end-start < oneDayInMilliseconds+offsetBucket {
if end-start < OneDayInMilliseconds+offsetBucket {
return SamplesV4TableName, SamplesV4LocalTableName
} else if end-start < oneWeekInMilliseconds+offsetBucket {
return SamplesV4Agg5mTableName, SamplesV4Agg5mLocalTableName
@@ -430,3 +431,37 @@ func AggregationQueryForHistogramCountWithParams(param *metrictypes.ComparisonSp
}
}
var IntrinsicFields = []string{
"temporality",
"metric_name",
"type",
"is_monotonic",
}
var IntrinsicMetricFieldDefinitions = map[string]telemetrytypes.TelemetryFieldKey{
"metric_name": {
Name: "metric_name",
Signal: telemetrytypes.SignalMetrics,
FieldContext: telemetrytypes.FieldContextMetric,
FieldDataType: telemetrytypes.FieldDataTypeString,
},
// "type": {
// Name: "type",
// Signal: telemetrytypes.SignalMetrics,
// FieldContext: telemetrytypes.FieldContextMetric,
// FieldDataType: telemetrytypes.FieldDataTypeString,
// },
// "temporality": {
// Name: "temporality",
// Signal: telemetrytypes.SignalMetrics,
// FieldContext: telemetrytypes.FieldContextMetric,
// FieldDataType: telemetrytypes.FieldDataTypeString,
// },
// "is_monotonic": {
// Name: "is_monotonic",
// Signal: telemetrytypes.SignalMetrics,
// FieldContext: telemetrytypes.FieldContextMetric,
// FieldDataType: telemetrytypes.FieldDataTypeBool,
// },
}

View File

@@ -1,4 +1,4 @@
package telemetrymetrics
package metricstelemetryschema
import (
"context"

View File

@@ -1,4 +1,4 @@
package telemetrymetrics
package metricstelemetryschema
import (
"context"

View File

@@ -1,4 +1,4 @@
package telemetrytraces
package tracestelemetryschema
import (
"context"

View File

@@ -1,4 +1,4 @@
package telemetrytraces
package tracestelemetryschema
import (
"context"
@@ -16,7 +16,7 @@ import (
func TestConditionFor(t *testing.T) {
ctx := context.Background()
mockEvolution := mockEvolutionData(time.Date(2025, 10, 26, 0, 10, 0, 0, time.UTC))
mockEvolution := MockEvolutionData(time.Date(2025, 10, 26, 0, 10, 0, 0, time.UTC))
testCases := []struct {
name string
key telemetrytypes.TelemetryFieldKey
@@ -311,7 +311,7 @@ func TestConditionFor(t *testing.T) {
func TestConditionForResourceWithEvolution(t *testing.T) {
ctx := context.Background()
releaseTime := time.Date(2025, 1, 1, 0, 0, 0, 0, time.UTC)
evolutions := mockEvolutionData(releaseTime)
evolutions := MockEvolutionData(releaseTime)
testCases := []struct {
name string
@@ -404,7 +404,7 @@ func TestConditionForSynthesizedKeys(t *testing.T) {
t.Run("bare key with string operand -> attribute string", func(t *testing.T) {
sb := sqlbuilder.NewSelectBuilder()
key := telemetrytypes.TelemetryFieldKey{Name: "error.type"}
conds, warnings, err := cb.ConditionFor(ctx, valuer.UUID{}, 0, 0, &key, noMatches, qbtypes.ConditionBuilderOptions{},qbtypes.FilterOperatorEqual, "timeout", sb)
conds, warnings, err := cb.ConditionFor(ctx, valuer.UUID{}, 0, 0, &key, noMatches, qbtypes.ConditionBuilderOptions{}, qbtypes.FilterOperatorEqual, "timeout", sb)
assert.NoError(t, err)
assert.NotEmpty(t, warnings, "a not-found warning should be emitted")
sb.Where(conds...)
@@ -416,7 +416,7 @@ func TestConditionForSynthesizedKeys(t *testing.T) {
t.Run("bare key with number operand -> attribute number", func(t *testing.T) {
sb := sqlbuilder.NewSelectBuilder()
key := telemetrytypes.TelemetryFieldKey{Name: "http.status"}
conds, warnings, err := cb.ConditionFor(ctx, valuer.UUID{}, 0, 0, &key, noMatches, qbtypes.ConditionBuilderOptions{},qbtypes.FilterOperatorGreaterThan, float64(5), sb)
conds, warnings, err := cb.ConditionFor(ctx, valuer.UUID{}, 0, 0, &key, noMatches, qbtypes.ConditionBuilderOptions{}, qbtypes.FilterOperatorGreaterThan, float64(5), sb)
assert.NoError(t, err)
assert.NotEmpty(t, warnings)
sb.Where(conds...)
@@ -427,7 +427,7 @@ func TestConditionForSynthesizedKeys(t *testing.T) {
t.Run("bare key with bool operand -> attribute bool", func(t *testing.T) {
sb := sqlbuilder.NewSelectBuilder()
key := telemetrytypes.TelemetryFieldKey{Name: "sampled"}
conds, _, err := cb.ConditionFor(ctx, valuer.UUID{}, 0, 0, &key, noMatches, qbtypes.ConditionBuilderOptions{},qbtypes.FilterOperatorEqual, true, sb)
conds, _, err := cb.ConditionFor(ctx, valuer.UUID{}, 0, 0, &key, noMatches, qbtypes.ConditionBuilderOptions{}, qbtypes.FilterOperatorEqual, true, sb)
assert.NoError(t, err)
sb.Where(conds...)
sql, _ := sb.BuildWithFlavor(sqlbuilder.ClickHouse)
@@ -437,7 +437,7 @@ func TestConditionForSynthesizedKeys(t *testing.T) {
t.Run("exists with no operand fans out across type variants", func(t *testing.T) {
sb := sqlbuilder.NewSelectBuilder()
key := telemetrytypes.TelemetryFieldKey{Name: "exception.type"}
conds, warnings, err := cb.ConditionFor(ctx, valuer.UUID{}, 0, 0, &key, noMatches, qbtypes.ConditionBuilderOptions{},qbtypes.FilterOperatorExists, nil, sb)
conds, warnings, err := cb.ConditionFor(ctx, valuer.UUID{}, 0, 0, &key, noMatches, qbtypes.ConditionBuilderOptions{}, qbtypes.FilterOperatorExists, nil, sb)
assert.NoError(t, err)
assert.NotEmpty(t, warnings)
assert.Len(t, conds, 3, "exists should fan out to string/number/bool")
@@ -451,7 +451,7 @@ func TestConditionForSynthesizedKeys(t *testing.T) {
t.Run("qualified data type honored without fanout", func(t *testing.T) {
sb := sqlbuilder.NewSelectBuilder()
key := telemetrytypes.TelemetryFieldKey{Name: "custom.key", FieldDataType: telemetrytypes.FieldDataTypeString}
conds, _, err := cb.ConditionFor(ctx, valuer.UUID{}, 0, 0, &key, noMatches, qbtypes.ConditionBuilderOptions{},qbtypes.FilterOperatorEqual, "v", sb)
conds, _, err := cb.ConditionFor(ctx, valuer.UUID{}, 0, 0, &key, noMatches, qbtypes.ConditionBuilderOptions{}, qbtypes.FilterOperatorEqual, "v", sb)
assert.NoError(t, err)
assert.Len(t, conds, 1)
sb.Where(conds...)
@@ -462,7 +462,7 @@ func TestConditionForSynthesizedKeys(t *testing.T) {
t.Run("bare intrinsic column resolves to the column, not synthesized attributes", func(t *testing.T) {
sb := sqlbuilder.NewSelectBuilder()
key := telemetrytypes.TelemetryFieldKey{Name: "duration_nano"}
conds, _, err := cb.ConditionFor(ctx, valuer.UUID{}, 0, 0, &key, noMatches, qbtypes.ConditionBuilderOptions{},qbtypes.FilterOperatorGreaterThan, float64(100), sb)
conds, _, err := cb.ConditionFor(ctx, valuer.UUID{}, 0, 0, &key, noMatches, qbtypes.ConditionBuilderOptions{}, qbtypes.FilterOperatorGreaterThan, float64(100), sb)
require.NoError(t, err)
require.Len(t, conds, 1)
sb.Where(conds...)
@@ -475,7 +475,7 @@ func TestConditionForSynthesizedKeys(t *testing.T) {
t.Run("bare deprecated alias resolves to the calculated column", func(t *testing.T) {
sb := sqlbuilder.NewSelectBuilder()
key := telemetrytypes.TelemetryFieldKey{Name: "httpMethod"}
conds, _, err := cb.ConditionFor(ctx, valuer.UUID{}, 0, 0, &key, noMatches, qbtypes.ConditionBuilderOptions{},qbtypes.FilterOperatorEqual, "GET", sb)
conds, _, err := cb.ConditionFor(ctx, valuer.UUID{}, 0, 0, &key, noMatches, qbtypes.ConditionBuilderOptions{}, qbtypes.FilterOperatorEqual, "GET", sb)
require.NoError(t, err)
require.Len(t, conds, 1)
sb.Where(conds...)
@@ -487,7 +487,7 @@ func TestConditionForSynthesizedKeys(t *testing.T) {
t.Run("span-context key that is a real column stays the column", func(t *testing.T) {
sb := sqlbuilder.NewSelectBuilder()
key := telemetrytypes.TelemetryFieldKey{Name: "duration_nano", FieldContext: telemetrytypes.FieldContextSpan}
conds, _, err := cb.ConditionFor(ctx, valuer.UUID{}, 0, 0, &key, noMatches, qbtypes.ConditionBuilderOptions{},qbtypes.FilterOperatorGreaterThan, float64(100), sb)
conds, _, err := cb.ConditionFor(ctx, valuer.UUID{}, 0, 0, &key, noMatches, qbtypes.ConditionBuilderOptions{}, qbtypes.FilterOperatorGreaterThan, float64(100), sb)
require.NoError(t, err)
require.Len(t, conds, 1)
sb.Where(conds...)
@@ -515,7 +515,7 @@ func TestConditionForSynthesizedKeys(t *testing.T) {
t.Run("span-context key absent from metadata synthesizes the stripped name", func(t *testing.T) {
sb := sqlbuilder.NewSelectBuilder()
key := telemetrytypes.TelemetryFieldKey{Name: "custom.attr", FieldContext: telemetrytypes.FieldContextSpan}
conds, warnings, err := cb.ConditionFor(ctx, valuer.UUID{}, 0, 0, &key, noMatches, qbtypes.ConditionBuilderOptions{},qbtypes.FilterOperatorEqual, "v", sb)
conds, warnings, err := cb.ConditionFor(ctx, valuer.UUID{}, 0, 0, &key, noMatches, qbtypes.ConditionBuilderOptions{}, qbtypes.FilterOperatorEqual, "v", sb)
require.NoError(t, err)
assert.NotEmpty(t, warnings)
require.Len(t, conds, 1)
@@ -528,7 +528,7 @@ func TestConditionForSynthesizedKeys(t *testing.T) {
t.Run("qualified resource context honored with literal spelling second", func(t *testing.T) {
sb := sqlbuilder.NewSelectBuilder()
key := telemetrytypes.TelemetryFieldKey{Name: "k8s.cluster.name", FieldContext: telemetrytypes.FieldContextResource}
conds, _, err := cb.ConditionFor(ctx, valuer.UUID{}, 0, 0, &key, noMatches, qbtypes.ConditionBuilderOptions{},qbtypes.FilterOperatorEqual, "prod", sb)
conds, _, err := cb.ConditionFor(ctx, valuer.UUID{}, 0, 0, &key, noMatches, qbtypes.ConditionBuilderOptions{}, qbtypes.FilterOperatorEqual, "prod", sb)
assert.NoError(t, err)
assert.Len(t, conds, 2, "stripped interpretation first, literal `resource.` spelling second")
sb.Where(conds...)
@@ -565,7 +565,7 @@ func TestConditionForSynthesizedKeys(t *testing.T) {
t.Run("synthesized resource key coerces numeric operand", func(t *testing.T) {
sb := sqlbuilder.NewSelectBuilder()
key := telemetrytypes.TelemetryFieldKey{Name: "replica.count", FieldContext: telemetrytypes.FieldContextResource}
conds, _, err := cb.ConditionFor(ctx, valuer.UUID{}, 0, 0, &key, noMatches, qbtypes.ConditionBuilderOptions{},qbtypes.FilterOperatorEqual, float64(3), sb)
conds, _, err := cb.ConditionFor(ctx, valuer.UUID{}, 0, 0, &key, noMatches, qbtypes.ConditionBuilderOptions{}, qbtypes.FilterOperatorEqual, float64(3), sb)
assert.NoError(t, err)
sb.Where(conds...)
sql, _ := sb.BuildWithFlavor(sqlbuilder.ClickHouse)
@@ -576,7 +576,7 @@ func TestConditionForSynthesizedKeys(t *testing.T) {
t.Run("negative operator builds without exists guard (matches-everything semantics)", func(t *testing.T) {
sb := sqlbuilder.NewSelectBuilder()
key := telemetrytypes.TelemetryFieldKey{Name: "error.type"}
conds, _, err := cb.ConditionFor(ctx, valuer.UUID{}, 0, 0, &key, noMatches, qbtypes.ConditionBuilderOptions{},qbtypes.FilterOperatorNotEqual, "fatal", sb)
conds, _, err := cb.ConditionFor(ctx, valuer.UUID{}, 0, 0, &key, noMatches, qbtypes.ConditionBuilderOptions{}, qbtypes.FilterOperatorNotEqual, "fatal", sb)
assert.NoError(t, err)
sb.Where(conds...)
sql, _ := sb.BuildWithFlavor(sqlbuilder.ClickHouse)

View File

@@ -1,4 +1,4 @@
package telemetrytraces
package tracestelemetryschema
import (
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
@@ -469,3 +469,15 @@ var (
},
}
)
const (
DBName = "signoz_traces"
SpanIndexV3TableName = "distributed_signoz_index_v3"
SpanIndexV3LocalTableName = "signoz_index_v3"
TagAttributesV2TableName = "distributed_tag_attributes_v2"
TagAttributesV2LocalTableName = "tag_attributes_v2"
TopLevelOperationsTableName = "distributed_top_level_operations"
TraceSummaryTableName = "distributed_trace_summary"
SpanAttributesKeysTblName = "distributed_span_attributes_keys"
TracesResourceV3TableName = "distributed_traces_v3_resource"
)

View File

@@ -1,4 +1,4 @@
package telemetrytraces
package tracestelemetryschema
import (
"context"
@@ -159,15 +159,15 @@ var (
}
)
type defaultFieldMapper struct{}
type fieldMapper struct{}
var _ qbtypes.FieldMapper = (*defaultFieldMapper)(nil)
var _ qbtypes.FieldMapper = (*fieldMapper)(nil)
func NewFieldMapper() *defaultFieldMapper {
return &defaultFieldMapper{}
func NewFieldMapper() *fieldMapper {
return &fieldMapper{}
}
func (m *defaultFieldMapper) getColumn(
func (m *fieldMapper) getColumn(
_ context.Context,
_, _ uint64,
key *telemetrytypes.TelemetryFieldKey,
@@ -214,7 +214,7 @@ func (m *defaultFieldMapper) getColumn(
return nil, qbtypes.ErrColumnNotFound
}
func (m *defaultFieldMapper) ColumnFor(
func (m *fieldMapper) ColumnFor(
ctx context.Context,
_ valuer.UUID,
startNs, endNs uint64,
@@ -225,7 +225,7 @@ func (m *defaultFieldMapper) ColumnFor(
// FieldFor returns the table field name for the given key if it exists
// otherwise it returns qbtypes.ErrColumnNotFound.
func (m *defaultFieldMapper) FieldFor(
func (m *fieldMapper) FieldFor(
ctx context.Context,
_ valuer.UUID,
startNs, endNs uint64,
@@ -263,7 +263,7 @@ func (m *defaultFieldMapper) FieldFor(
// resolveColumnExprs resolves key to its per-column value expressions and existence guards
// (after evolution selection); existExprs only carries guards for guardable column types.
func (m *defaultFieldMapper) resolveColumnExprs(
func (m *fieldMapper) resolveColumnExprs(
ctx context.Context,
startNs, endNs uint64,
key *telemetrytypes.TelemetryFieldKey,
@@ -339,7 +339,7 @@ func (m *defaultFieldMapper) resolveColumnExprs(
// ColumnExpressionFor returns the bare (unaliased) SQL expression for the field, resolving
// unknown keys via CandidateKeys and wrapping guardable columns with exists-guard multiIfs
// so an absent key yields NULL.
func (m *defaultFieldMapper) ColumnExpressionFor(
func (m *fieldMapper) ColumnExpressionFor(
ctx context.Context,
orgID valuer.UUID,
startNs, endNs uint64,
@@ -448,7 +448,7 @@ func columnMatchesDataType(col *schema.Column, dt telemetrytypes.FieldDataType)
// metadata keys; a forgiving span/trace context is honored as-is, correcting to a
// metadata match for the stripped name when present; strict attribute/resource contexts
// are honored as-is. Falls back to synthesized type-variant keys.
func (m *defaultFieldMapper) CandidateKeys(ctx context.Context, _ valuer.UUID, field *telemetrytypes.TelemetryFieldKey, value any, keys map[string][]*telemetrytypes.TelemetryFieldKey) []*telemetrytypes.TelemetryFieldKey {
func (m *fieldMapper) CandidateKeys(ctx context.Context, _ valuer.UUID, field *telemetrytypes.TelemetryFieldKey, value any, keys map[string][]*telemetrytypes.TelemetryFieldKey) []*telemetrytypes.TelemetryFieldKey {
// A real column is considered before metadata for bare and forgiving contexts, so a
// same-named corrupt attribute can't shadow the intrinsic/calculated column.
switch field.FieldContext {
@@ -500,7 +500,7 @@ func (m *defaultFieldMapper) CandidateKeys(ctx context.Context, _ valuer.UUID, f
return nil
}
func (m *defaultFieldMapper) existsExpressionFor(
func (m *fieldMapper) existsExpressionFor(
ctx context.Context,
orgID valuer.UUID,
tsStart, tsEnd uint64,

View File

@@ -1,4 +1,4 @@
package telemetrytraces
package tracestelemetryschema
import (
"context"
@@ -15,7 +15,7 @@ import (
func TestGetFieldKeyName(t *testing.T) {
ctx := context.Background()
mockEvolution := mockEvolutionData(time.Date(2024, 6, 2, 0, 0, 0, 0, time.UTC))
mockEvolution := MockEvolutionData(time.Date(2024, 6, 2, 0, 0, 0, 0, time.UTC))
testCases := []struct {
name string
key telemetrytypes.TelemetryFieldKey
@@ -123,7 +123,7 @@ func TestGetFieldKeyName(t *testing.T) {
func TestFieldForResourceWithEvolution(t *testing.T) {
ctx := context.Background()
releaseTime := time.Date(2025, 1, 1, 0, 0, 0, 0, time.UTC)
evolutions := mockEvolutionData(releaseTime)
evolutions := MockEvolutionData(releaseTime)
testCases := []struct {
name string

View File

@@ -1,4 +1,4 @@
package telemetrytraces
package tracestelemetryschema
import (
"context"

Some files were not shown because too many files have changed in this diff Show More