mirror of
https://github.com/SigNoz/signoz.git
synced 2026-10-30 04:50:45 +00:00
Compare commits
6 Commits
main
...
ns/trace-a
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
571edf8dcf | ||
|
|
12abe12f83 | ||
|
|
2c8ac0c054 | ||
|
|
00d8d31b6d | ||
|
|
f30998462d | ||
|
|
593863cea5 |
@@ -16,6 +16,14 @@ packages:
|
||||
filename: "maintenance.go"
|
||||
structname: 'Mock{{.InterfaceName}}'
|
||||
pkgname: '{{.SrcPackageName}}test'
|
||||
github.com/SigNoz/signoz/pkg/types/telemetrytypes:
|
||||
interfaces:
|
||||
SchemaMetadataStore:
|
||||
config:
|
||||
dir: '{{.InterfaceDir}}/telemetrytypestest'
|
||||
filename: "schema_metadata_store.go"
|
||||
structname: 'Mock{{.InterfaceName}}'
|
||||
pkgname: '{{.SrcPackageName}}test'
|
||||
github.com/SigNoz/signoz/pkg/tokenizer:
|
||||
config:
|
||||
all: true
|
||||
|
||||
@@ -538,7 +538,7 @@ func TestTracesNumberIndexExpr_MatchesQuerierFilters(t *testing.T) {
|
||||
for _, testCase := range testCases {
|
||||
t.Run(testCase.name, func(t *testing.T) {
|
||||
sb := sqlbuilder.NewSelectBuilder()
|
||||
conditions, _, err := querybuilder.Conditions(context.Background(), queryInfo, tracestelemetryschema.NewStorage(), &key, testCase.operator, testCase.value, map[string][]*telemetrytypes.TelemetryFieldKey{key.Name: {&key}}, false, sb)
|
||||
conditions, _, err := querybuilder.Conditions(context.Background(), queryInfo, tracestelemetryschema.NewStorage(nil), &key, testCase.operator, testCase.value, map[string][]*telemetrytypes.TelemetryFieldKey{key.Name: {&key}}, false, sb)
|
||||
require.NoError(t, err)
|
||||
sb.Where(conditions...)
|
||||
sql, _ := sb.BuildWithFlavor(sqlbuilder.ClickHouse)
|
||||
|
||||
@@ -13,6 +13,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/telemetrymetadata"
|
||||
"github.com/SigNoz/signoz/pkg/telemetryschema/tracestelemetryschema"
|
||||
"github.com/SigNoz/signoz/pkg/telemetrystore"
|
||||
"github.com/SigNoz/signoz/pkg/types/aiobservabilitytypes"
|
||||
@@ -68,7 +69,7 @@ func NewTraceStore(ts telemetrystore.TelemetryStore, metadataStore telemetrytype
|
||||
return &traceStore{
|
||||
telemetryStore: ts,
|
||||
metadataStore: metadataStore,
|
||||
storage: tracestelemetryschema.NewStorage(),
|
||||
storage: tracestelemetryschema.NewStorage(telemetrymetadata.NewSchemaMetadataStore(ts)),
|
||||
flagger: fl,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -15,6 +15,7 @@ import (
|
||||
"github.com/SigNoz/signoz/pkg/statementbuilder"
|
||||
"github.com/SigNoz/signoz/pkg/statementbuilder/resourcefilter"
|
||||
"github.com/SigNoz/signoz/pkg/statementbuilder/tracesstatementbuilder"
|
||||
"github.com/SigNoz/signoz/pkg/telemetrymetadata"
|
||||
"github.com/SigNoz/signoz/pkg/telemetryschema/tracestelemetryschema"
|
||||
"github.com/SigNoz/signoz/pkg/telemetrystore"
|
||||
qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
|
||||
@@ -59,7 +60,7 @@ func NewFactory(
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return NewScopedTraceStatementBuilder(settings, metadataStore, tracestelemetryschema.NewStorage(), scope, traceStmtBuilder, fl), nil
|
||||
return NewScopedTraceStatementBuilder(settings, metadataStore, tracestelemetryschema.NewStorage(telemetrymetadata.NewSchemaMetadataStore(telemetryStore)), scope, traceStmtBuilder, fl), nil
|
||||
},
|
||||
)
|
||||
}
|
||||
|
||||
@@ -17,6 +17,7 @@ import (
|
||||
"github.com/SigNoz/signoz/pkg/types/telemetrytypes/telemetrytypestest"
|
||||
"github.com/SigNoz/signoz/pkg/valuer"
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/mock"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
@@ -26,7 +27,9 @@ var jsonAttrColRe = regexp.MustCompile(`,\s*attributes\s*(,| FROM )`)
|
||||
func newBulkTestBuilder(t *testing.T, releaseTime time.Time) *traceQueryStatementBuilder {
|
||||
t.Helper()
|
||||
fl := flaggertest.WithBooleanFlags(t, map[string]bool{flagger.FeatureUseTraceAttributesJSON.String(): true})
|
||||
storage := tracestelemetryschema.NewStorage()
|
||||
schemaMetadataStore := telemetrytypestest.NewMockSchemaMetadataStore(t)
|
||||
schemaMetadataStore.EXPECT().GetBaseColumnEvolutions(mock.Anything, telemetrytypes.SignalTraces, telemetrytypes.FieldContextAttribute).Return(tracestelemetryschema.MockAttributeEvolutionData(releaseTime), nil).Maybe()
|
||||
storage := tracestelemetryschema.NewStorage(schemaMetadataStore)
|
||||
store := telemetrytypestest.NewMockMetadataStore()
|
||||
store.KeysMap = tracestelemetryschema.BuildCompleteFieldKeyMap(releaseTime)
|
||||
store.KeysMap["http.route"] = []*telemetrytypes.TelemetryFieldKey{{
|
||||
@@ -148,3 +151,153 @@ func TestGroupByAttributeHomeAcrossRollout(t *testing.T) {
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestUnknownAttributeFollowsRollout(t *testing.T) {
|
||||
releaseTime := time.Date(2025, 5, 22, 22, 0, 0, 0, time.UTC)
|
||||
rel := releaseTime.UnixMilli()
|
||||
day := int64(24 * time.Hour / time.Millisecond)
|
||||
|
||||
b := newBulkTestBuilder(t, releaseTime)
|
||||
|
||||
testCases := []struct {
|
||||
name string
|
||||
startMs uint64
|
||||
endMs uint64
|
||||
wantContains []string
|
||||
wantNotContains []string
|
||||
}{
|
||||
{
|
||||
name: "BeforeRollout_ReadsLegacyMap",
|
||||
startMs: uint64(rel - 2*day),
|
||||
endMs: uint64(rel - day),
|
||||
wantContains: []string{"attributes_string['unknown.attr'] = ?", "mapContains(attributes_string, 'unknown.attr')"},
|
||||
wantNotContains: []string{"attributes.`unknown.attr`"},
|
||||
},
|
||||
{
|
||||
name: "StraddlingRollout_JSONThenLegacyFallback",
|
||||
startMs: uint64(rel - day),
|
||||
endMs: uint64(rel + day),
|
||||
wantContains: []string{"attributes.`unknown.attr` IS NOT NULL", "attributes.`unknown.attr`::String", "attributes_string['unknown.attr']"},
|
||||
wantNotContains: nil,
|
||||
},
|
||||
{
|
||||
name: "AfterRollout_ReadsJSONOnly",
|
||||
startMs: uint64(rel + day),
|
||||
endMs: uint64(rel + 2*day),
|
||||
wantContains: []string{"attributes.`unknown.attr`::String = ?"},
|
||||
wantNotContains: []string{"attributes_string"},
|
||||
},
|
||||
}
|
||||
|
||||
for _, testCase := range testCases {
|
||||
t.Run(testCase.name, func(t *testing.T) {
|
||||
stmt, err := b.Build(
|
||||
context.Background(), valuer.UUID{}, testCase.startMs, testCase.endMs,
|
||||
qbtypes.RequestTypeTimeSeries,
|
||||
qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
|
||||
Signal: telemetrytypes.SignalTraces,
|
||||
StepInterval: qbtypes.Step{Duration: 30 * time.Second},
|
||||
Aggregations: []qbtypes.TraceAggregation{{Expression: "count()"}},
|
||||
Filter: &qbtypes.Filter{Expression: "unknown.attr = 'x'"},
|
||||
GroupBy: []qbtypes.GroupByKey{{TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{
|
||||
Name: "unknown.attr",
|
||||
}}},
|
||||
Limit: 10,
|
||||
},
|
||||
nil,
|
||||
)
|
||||
require.NoError(t, err)
|
||||
for _, want := range testCase.wantContains {
|
||||
assert.Contains(t, stmt.Query, want)
|
||||
}
|
||||
for _, unwanted := range testCase.wantNotContains {
|
||||
assert.NotContains(t, stmt.Query, unwanted)
|
||||
}
|
||||
assert.Contains(t, strings.Join(stmt.Warnings, "\n"), "not found in metadata")
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestUnknownAttributeFollowsRolloutAfterRelease(t *testing.T) {
|
||||
releaseTime := time.Date(2025, 5, 22, 22, 0, 0, 0, time.UTC)
|
||||
startMs := uint64(releaseTime.Add(24 * time.Hour).UnixMilli())
|
||||
endMs := uint64(releaseTime.Add(48 * time.Hour).UnixMilli())
|
||||
|
||||
b := newBulkTestBuilder(t, releaseTime)
|
||||
|
||||
testCases := []struct {
|
||||
name string
|
||||
requestType qbtypes.RequestType
|
||||
query qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]
|
||||
wantContains []string
|
||||
wantNotContains []string
|
||||
}{
|
||||
{
|
||||
name: "AggregationArgument_ReadsJSONOnly",
|
||||
requestType: qbtypes.RequestTypeTimeSeries,
|
||||
query: qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
|
||||
Signal: telemetrytypes.SignalTraces,
|
||||
StepInterval: qbtypes.Step{Duration: 30 * time.Second},
|
||||
Aggregations: []qbtypes.TraceAggregation{{Expression: "sum(unknown.num)"}},
|
||||
},
|
||||
wantContains: []string{"attributes.`unknown.num`"},
|
||||
wantNotContains: []string{"attributes_number['unknown.num']"},
|
||||
},
|
||||
{
|
||||
name: "RawSelectAndOrder_ReadsJSONOnly",
|
||||
requestType: qbtypes.RequestTypeRaw,
|
||||
query: qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
|
||||
Signal: telemetrytypes.SignalTraces,
|
||||
SelectFields: []telemetrytypes.TelemetryFieldKey{{Name: "unknown.attr"}},
|
||||
Order: []qbtypes.OrderBy{{Key: qbtypes.OrderByKey{TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{Name: "unknown.attr"}}, Direction: qbtypes.OrderDirectionAsc}},
|
||||
Limit: 10,
|
||||
},
|
||||
wantContains: []string{"attributes.`unknown.attr`"},
|
||||
wantNotContains: []string{"attributes_string['unknown.attr']"},
|
||||
},
|
||||
}
|
||||
|
||||
for _, testCase := range testCases {
|
||||
t.Run(testCase.name, func(t *testing.T) {
|
||||
stmt, err := b.Build(context.Background(), valuer.UUID{}, startMs, endMs, testCase.requestType, testCase.query, nil)
|
||||
require.NoError(t, err)
|
||||
for _, want := range testCase.wantContains {
|
||||
assert.Contains(t, stmt.Query, want)
|
||||
}
|
||||
for _, unwanted := range testCase.wantNotContains {
|
||||
assert.NotContains(t, stmt.Query, unwanted)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestTraceOperatorUnknownAttributeFollowsRollout(t *testing.T) {
|
||||
releaseTime := time.Date(2025, 5, 22, 22, 0, 0, 0, time.UTC)
|
||||
startMs := uint64(releaseTime.Add(24 * time.Hour).UnixMilli())
|
||||
endMs := uint64(releaseTime.Add(48 * time.Hour).UnixMilli())
|
||||
|
||||
fl := flaggertest.WithBooleanFlags(t, map[string]bool{flagger.FeatureUseTraceAttributesJSON.String(): true})
|
||||
schemaMetadataStore := telemetrytypestest.NewMockSchemaMetadataStore(t)
|
||||
schemaMetadataStore.EXPECT().GetBaseColumnEvolutions(mock.Anything, telemetrytypes.SignalTraces, telemetrytypes.FieldContextAttribute).Return(tracestelemetryschema.MockAttributeEvolutionData(releaseTime), nil).Maybe()
|
||||
storage := tracestelemetryschema.NewStorage(schemaMetadataStore)
|
||||
store := telemetrytypestest.NewMockMetadataStore()
|
||||
store.KeysMap = tracestelemetryschema.BuildCompleteFieldKeyMap(releaseTime)
|
||||
store.ColumnEvolutionMetadataMap["traces:attribute:__all__"] = tracestelemetryschema.MockAttributeEvolutionData(releaseTime)
|
||||
aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, storage, fl, telemetrytypes.SignalTraces)
|
||||
traceStmtBuilder := NewTraceQueryStatementBuilder(instrumentationtest.New().ToProviderSettings(), store, storage, aggExprRewriter, nil, fl, false, 100000)
|
||||
b := NewTraceOperatorStatementBuilder(instrumentationtest.New().ToProviderSettings(), store, storage, traceStmtBuilder, aggExprRewriter, fl)
|
||||
|
||||
operator := qbtypes.QueryBuilderTraceOperator{Expression: "A => B", Limit: 10}
|
||||
require.NoError(t, operator.ParseExpression())
|
||||
compositeQuery := &qbtypes.CompositeQuery{Queries: []qbtypes.QueryEnvelope{
|
||||
{Type: qbtypes.QueryTypeBuilder, Spec: qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{Name: "A", Signal: telemetrytypes.SignalTraces, Filter: &qbtypes.Filter{Expression: "unknown.attr = 'x'"}}},
|
||||
{Type: qbtypes.QueryTypeBuilder, Spec: qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{Name: "B", Signal: telemetrytypes.SignalTraces, Filter: &qbtypes.Filter{Expression: "unknown.other = 'y'"}}},
|
||||
}}
|
||||
|
||||
stmt, err := b.Build(context.Background(), valuer.UUID{}, startMs, endMs, qbtypes.RequestTypeRaw, operator, compositeQuery)
|
||||
require.NoError(t, err)
|
||||
assert.Contains(t, stmt.Query, "attributes.`unknown.attr`::String = ?")
|
||||
assert.Contains(t, stmt.Query, "attributes.`unknown.other`::String = ?")
|
||||
assert.NotContains(t, stmt.Query, "attributes_string['unknown.attr']")
|
||||
assert.NotContains(t, stmt.Query, "attributes_string['unknown.other']")
|
||||
}
|
||||
|
||||
@@ -13,6 +13,7 @@ import (
|
||||
"github.com/SigNoz/signoz/pkg/querybuilder"
|
||||
"github.com/SigNoz/signoz/pkg/statementbuilder"
|
||||
"github.com/SigNoz/signoz/pkg/statementbuilder/resourcefilter"
|
||||
"github.com/SigNoz/signoz/pkg/telemetrymetadata"
|
||||
"github.com/SigNoz/signoz/pkg/telemetryschema/tracestelemetryschema"
|
||||
"github.com/SigNoz/signoz/pkg/telemetrystore"
|
||||
qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
|
||||
@@ -54,7 +55,7 @@ func NewFactory(
|
||||
return factory.NewProviderFactory(
|
||||
factory.MustNewName("traces"),
|
||||
func(_ context.Context, settings factory.ProviderSettings, cfg statementbuilder.Config) (qbtypes.StatementBuilder[qbtypes.TraceAggregation], error) {
|
||||
storage := tracestelemetryschema.NewStorage()
|
||||
storage := tracestelemetryschema.NewStorage(telemetrymetadata.NewSchemaMetadataStore(telemetryStore))
|
||||
aggExprRewriter := querybuilder.NewAggExprRewriter(settings, nil, storage, fl, telemetrytypes.SignalTraces)
|
||||
return NewTraceQueryStatementBuilder(
|
||||
settings, metadataStore, storage, aggExprRewriter, telemetryStore, fl,
|
||||
|
||||
@@ -465,7 +465,7 @@ func TestStatementBuilder(t *testing.T) {
|
||||
}
|
||||
|
||||
fl := flaggertest.New(t)
|
||||
storage := tracestelemetryschema.NewStorage()
|
||||
storage := tracestelemetryschema.NewStorage(nil)
|
||||
mockMetadataStore := telemetrytypestest.NewMockMetadataStore()
|
||||
mockMetadataStore.KeysMap = tracestelemetryschema.BuildCompleteFieldKeyMap(releaseTime)
|
||||
aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, storage, fl, telemetrytypes.SignalTraces)
|
||||
@@ -764,7 +764,7 @@ func TestStatementBuilderListQuery(t *testing.T) {
|
||||
}
|
||||
|
||||
fl := flaggertest.New(t)
|
||||
storage := tracestelemetryschema.NewStorage()
|
||||
storage := tracestelemetryschema.NewStorage(nil)
|
||||
mockMetadataStore := telemetrytypestest.NewMockMetadataStore()
|
||||
mockMetadataStore.KeysMap = tracestelemetryschema.BuildCompleteFieldKeyMap(releaseTime)
|
||||
aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, storage, fl, telemetrytypes.SignalTraces)
|
||||
@@ -1026,7 +1026,7 @@ func TestStatementBuilderListQueryWithCorruptData(t *testing.T) {
|
||||
for _, c := range cases {
|
||||
t.Run(c.name, func(t *testing.T) {
|
||||
fl := flaggertest.New(t)
|
||||
storage := tracestelemetryschema.NewStorage()
|
||||
storage := tracestelemetryschema.NewStorage(nil)
|
||||
mockMetadataStore := telemetrytypestest.NewMockMetadataStore()
|
||||
mockMetadataStore.KeysMap = c.keysMap
|
||||
if mockMetadataStore.KeysMap == nil {
|
||||
@@ -1099,7 +1099,7 @@ func TestStatementBuilderGroupByResourceEvolution(t *testing.T) {
|
||||
}
|
||||
|
||||
fl := flaggertest.New(t)
|
||||
storage := tracestelemetryschema.NewStorage()
|
||||
storage := tracestelemetryschema.NewStorage(nil)
|
||||
mockMetadataStore := telemetrytypestest.NewMockMetadataStore()
|
||||
mockMetadataStore.KeysMap = tracestelemetryschema.BuildCompleteFieldKeyMap(releaseTime)
|
||||
aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, storage, fl, telemetrytypes.SignalTraces)
|
||||
@@ -1264,7 +1264,7 @@ func TestStatementBuilderTraceQuery(t *testing.T) {
|
||||
}
|
||||
|
||||
fl := flaggertest.New(t)
|
||||
storage := tracestelemetryschema.NewStorage()
|
||||
storage := tracestelemetryschema.NewStorage(nil)
|
||||
mockMetadataStore := telemetrytypestest.NewMockMetadataStore()
|
||||
mockMetadataStore.KeysMap = tracestelemetryschema.BuildCompleteFieldKeyMap(releaseTime)
|
||||
aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, storage, fl, telemetrytypes.SignalTraces)
|
||||
@@ -1900,7 +1900,7 @@ func newSkipResourceFingerprintBuilder(
|
||||
t.Helper()
|
||||
|
||||
fl := flaggertest.New(t)
|
||||
storage := tracestelemetryschema.NewStorage()
|
||||
storage := tracestelemetryschema.NewStorage(nil)
|
||||
mockMetadataStore := telemetrytypestest.NewMockMetadataStore()
|
||||
releaseTime := time.Date(2025, 5, 22, 22, 0, 0, 0, time.UTC)
|
||||
mockMetadataStore.KeysMap = tracestelemetryschema.BuildCompleteFieldKeyMap(releaseTime)
|
||||
@@ -1926,7 +1926,7 @@ func TestStatementBuilderGroupByUnseenKey(t *testing.T) {
|
||||
releaseTime := time.Date(2025, 5, 22, 22, 0, 0, 0, time.UTC)
|
||||
|
||||
fl := flaggertest.New(t)
|
||||
storage := tracestelemetryschema.NewStorage()
|
||||
storage := tracestelemetryschema.NewStorage(nil)
|
||||
mockMetadataStore := telemetrytypestest.NewMockMetadataStore()
|
||||
mockMetadataStore.KeysMap = tracestelemetryschema.BuildCompleteFieldKeyMap(releaseTime)
|
||||
aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, storage, fl, telemetrytypes.SignalTraces)
|
||||
@@ -1966,7 +1966,7 @@ func TestStatementBuilderAggregationUnseenKey(t *testing.T) {
|
||||
releaseTime := time.Date(2025, 5, 22, 22, 0, 0, 0, time.UTC)
|
||||
|
||||
fl := flaggertest.New(t)
|
||||
storage := tracestelemetryschema.NewStorage()
|
||||
storage := tracestelemetryschema.NewStorage(nil)
|
||||
mockMetadataStore := telemetrytypestest.NewMockMetadataStore()
|
||||
mockMetadataStore.KeysMap = tracestelemetryschema.BuildCompleteFieldKeyMap(releaseTime)
|
||||
aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, storage, fl, telemetrytypes.SignalTraces)
|
||||
@@ -2039,7 +2039,7 @@ func TestStatementBuilderSemconvFamilies(t *testing.T) {
|
||||
fl := flaggertest.WithBooleanFlags(t, map[string]bool{
|
||||
flagger.FeatureResolveSemconvFamilies.String(): c.flag,
|
||||
})
|
||||
storage := tracestelemetryschema.NewStorage()
|
||||
storage := tracestelemetryschema.NewStorage(nil)
|
||||
mockMetadataStore := telemetrytypestest.NewMockMetadataStore()
|
||||
mockMetadataStore.KeysMap = tracestelemetryschema.BuildCompleteFieldKeyMap(releaseTime)
|
||||
aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, storage, fl, telemetrytypes.SignalTraces)
|
||||
@@ -2071,7 +2071,7 @@ func TestStatementBuilderSemconvSingleSpelling(t *testing.T) {
|
||||
fl := flaggertest.WithBooleanFlags(t, map[string]bool{
|
||||
flagger.FeatureResolveSemconvFamilies.String(): true,
|
||||
})
|
||||
storage := tracestelemetryschema.NewStorage()
|
||||
storage := tracestelemetryschema.NewStorage(nil)
|
||||
mockMetadataStore := telemetrytypestest.NewMockMetadataStore()
|
||||
mockMetadataStore.KeysMap = map[string][]*telemetrytypes.TelemetryFieldKey{
|
||||
"deployment.environment": {{
|
||||
|
||||
@@ -20,7 +20,7 @@ func newTestTraceOperatorStatementBuilder(t *testing.T) *traceOperatorStatementB
|
||||
t.Helper()
|
||||
releaseTime := time.Date(2025, 5, 22, 22, 0, 0, 0, time.UTC)
|
||||
fl := flaggertest.New(t)
|
||||
storage := tracestelemetryschema.NewStorage()
|
||||
storage := tracestelemetryschema.NewStorage(nil)
|
||||
mockMetadataStore := telemetrytypestest.NewMockMetadataStore()
|
||||
mockMetadataStore.KeysMap = tracestelemetryschema.BuildCompleteFieldKeyMap(releaseTime)
|
||||
aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, storage, fl, telemetrytypes.SignalTraces)
|
||||
|
||||
@@ -10,6 +10,7 @@ import (
|
||||
"github.com/SigNoz/signoz/pkg/querybuilder"
|
||||
"github.com/SigNoz/signoz/pkg/statementbuilder"
|
||||
"github.com/SigNoz/signoz/pkg/statementbuilder/resourcefilter"
|
||||
"github.com/SigNoz/signoz/pkg/telemetrymetadata"
|
||||
"github.com/SigNoz/signoz/pkg/telemetryschema/tracestelemetryschema"
|
||||
"github.com/SigNoz/signoz/pkg/telemetrystore"
|
||||
qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
|
||||
@@ -41,7 +42,7 @@ func NewOperatorFactory(
|
||||
return factory.NewProviderFactory(
|
||||
factory.MustNewName("traceoperator"),
|
||||
func(_ context.Context, settings factory.ProviderSettings, cfg statementbuilder.Config) (qbtypes.TraceOperatorStatementBuilder, error) {
|
||||
storage := tracestelemetryschema.NewStorage()
|
||||
storage := tracestelemetryschema.NewStorage(telemetrymetadata.NewSchemaMetadataStore(telemetryStore))
|
||||
aggExprRewriter := querybuilder.NewAggExprRewriter(settings, nil, storage, fl, telemetrytypes.SignalTraces)
|
||||
traceStmtBuilder := NewTraceQueryStatementBuilder(
|
||||
settings, metadataStore, storage, aggExprRewriter, telemetryStore, fl,
|
||||
|
||||
@@ -22,7 +22,7 @@ func TestTraceTimeRangeOptimization(t *testing.T) {
|
||||
releaseTime := time.Date(2025, 5, 22, 22, 0, 0, 0, time.UTC)
|
||||
|
||||
fl := flaggertest.New(t)
|
||||
storage := tracestelemetryschema.NewStorage()
|
||||
storage := tracestelemetryschema.NewStorage(nil)
|
||||
mockMetadataStore := telemetrytypestest.NewMockMetadataStore()
|
||||
|
||||
mockMetadataStore.KeysMap = tracestelemetryschema.BuildCompleteFieldKeyMap(releaseTime)
|
||||
|
||||
@@ -2442,10 +2442,10 @@ func (t *telemetryMetaStore) fetchMeterSourceMetricsTemporalityAndType(ctx conte
|
||||
return temporalities, types, nil
|
||||
}
|
||||
|
||||
func (k *telemetryMetaStore) fetchEvolutionEntryFromClickHouse(ctx context.Context, selectors []*telemetrytypes.EvolutionSelector) ([]*telemetrytypes.EvolutionEntry, error) {
|
||||
func fetchEvolutionEntryFromClickHouse(ctx context.Context, telemetryStore telemetrystore.TelemetryStore, table string, selectors []*telemetrytypes.EvolutionSelector) ([]*telemetrytypes.EvolutionEntry, error) {
|
||||
sb := sqlbuilder.NewSelectBuilder()
|
||||
sb.Select("signal", "column_name", "column_type", "field_context", "field_name", "version", "release_time")
|
||||
sb.From(fmt.Sprintf("%s.%s", k.relatedMetadataDBName, k.columnEvolutionMetadataTblName))
|
||||
sb.From(table)
|
||||
sb.OrderBy("release_time ASC")
|
||||
|
||||
var clauses []string
|
||||
@@ -2467,7 +2467,7 @@ func (k *telemetryMetaStore) fetchEvolutionEntryFromClickHouse(ctx context.Conte
|
||||
query, args := sb.BuildWithFlavor(sqlbuilder.ClickHouse)
|
||||
|
||||
var entries []*telemetrytypes.EvolutionEntry
|
||||
rows, err := k.telemetrystore.ClickhouseDB().Query(ctx, query, args...)
|
||||
rows, err := telemetryStore.ClickhouseDB().Query(ctx, query, args...)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
@@ -2517,7 +2517,7 @@ func (k *telemetryMetaStore) updateColumnEvolutionMetadataForKeys(ctx context.Co
|
||||
metadataKeySelectors = append(metadataKeySelectors, selector)
|
||||
}
|
||||
|
||||
evolutions, err := k.fetchEvolutionEntryFromClickHouse(ctx, metadataKeySelectors)
|
||||
evolutions, err := fetchEvolutionEntryFromClickHouse(ctx, k.telemetrystore, fmt.Sprintf("%s.%s", k.relatedMetadataDBName, k.columnEvolutionMetadataTblName), metadataKeySelectors)
|
||||
if err != nil {
|
||||
return errors.Newf(errors.TypeInternal, errors.CodeInternal, "failed to fetch evolution from clickhouse %s", err.Error())
|
||||
}
|
||||
|
||||
39
pkg/telemetrymetadata/schema_metadata_store.go
Normal file
39
pkg/telemetrymetadata/schema_metadata_store.go
Normal file
@@ -0,0 +1,39 @@
|
||||
package telemetrymetadata
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"sync"
|
||||
|
||||
"github.com/SigNoz/signoz/pkg/errors"
|
||||
"github.com/SigNoz/signoz/pkg/telemetrystore"
|
||||
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
|
||||
)
|
||||
|
||||
var CodeFailLoadBaseColumnEvolutions = errors.MustNewCode("fail_load_base_column_evolutions")
|
||||
|
||||
type schemaMetadataStore struct {
|
||||
telemetryStore telemetrystore.TelemetryStore
|
||||
baseColumnEvolutions sync.Map
|
||||
}
|
||||
|
||||
func NewSchemaMetadataStore(telemetryStore telemetrystore.TelemetryStore) telemetrytypes.SchemaMetadataStore {
|
||||
return &schemaMetadataStore{telemetryStore: telemetryStore}
|
||||
}
|
||||
|
||||
// An empty result is not cached: the migrator can add the entries at runtime.
|
||||
func (s *schemaMetadataStore) GetBaseColumnEvolutions(ctx context.Context, signal telemetrytypes.Signal, fieldContext telemetrytypes.FieldContext) ([]*telemetrytypes.EvolutionEntry, error) {
|
||||
selector := &telemetrytypes.EvolutionSelector{Signal: signal, FieldContext: fieldContext, FieldName: telemetrytypes.EvolutionFieldNameAll}
|
||||
if cached, ok := s.baseColumnEvolutions.Load(selector.QualifiedName()); ok {
|
||||
return cached.([]*telemetrytypes.EvolutionEntry), nil
|
||||
}
|
||||
|
||||
entries, err := fetchEvolutionEntryFromClickHouse(ctx, s.telemetryStore, fmt.Sprintf("%s.%s", DBName, ColumnEvolutionMetadataTableName), []*telemetrytypes.EvolutionSelector{selector})
|
||||
if err != nil {
|
||||
return nil, errors.WrapInternalf(err, CodeFailLoadBaseColumnEvolutions, "failed to fetch the base column evolutions of %s", selector.QualifiedName())
|
||||
}
|
||||
if len(entries) > 0 {
|
||||
s.baseColumnEvolutions.Store(selector.QualifiedName(), entries)
|
||||
}
|
||||
return entries, nil
|
||||
}
|
||||
72
pkg/telemetrymetadata/schema_metadata_store_test.go
Normal file
72
pkg/telemetrymetadata/schema_metadata_store_test.go
Normal file
@@ -0,0 +1,72 @@
|
||||
package telemetrymetadata
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
cmock "github.com/SigNoz/clickhouse-go-mock"
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
|
||||
"github.com/SigNoz/signoz/pkg/telemetrystore"
|
||||
"github.com/SigNoz/signoz/pkg/telemetrystore/telemetrystoretest"
|
||||
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
|
||||
)
|
||||
|
||||
var evolutionColumns = []cmock.ColumnType{
|
||||
{Name: "signal", Type: "String"},
|
||||
{Name: "column_name", Type: "String"},
|
||||
{Name: "column_type", Type: "String"},
|
||||
{Name: "field_context", Type: "String"},
|
||||
{Name: "field_name", Type: "String"},
|
||||
{Name: "version", Type: "UInt32"},
|
||||
{Name: "release_time", Type: "Float64"},
|
||||
}
|
||||
|
||||
func expectBaseColumnEvolutionsQuery(mock cmock.ClickConnMockCommon, rows [][]any) {
|
||||
mock.ExpectQuery(`SELECT signal, column_name, column_type, field_context, field_name, version, release_time FROM signoz_metadata\.distributed_column_evolution_metadata`).
|
||||
WithArgs(telemetrytypes.SignalTraces, telemetrytypes.FieldContextAttribute, "__all__", "__all__").
|
||||
WillReturnRows(cmock.NewRows(evolutionColumns, rows))
|
||||
}
|
||||
|
||||
func TestGetBaseColumnEvolutionsReadsCache(t *testing.T) {
|
||||
mockTelemetryStore := telemetrystoretest.New(telemetrystore.Config{}, ®exMatcher{})
|
||||
mock := mockTelemetryStore.Mock()
|
||||
store := NewSchemaMetadataStore(mockTelemetryStore)
|
||||
|
||||
releaseTime := time.Date(2026, 9, 1, 0, 0, 0, 0, time.UTC)
|
||||
expectBaseColumnEvolutionsQuery(mock, [][]any{
|
||||
{"traces", "attributes", "JSON()", "attribute", "__all__", uint32(0), float64(releaseTime.UnixNano())},
|
||||
})
|
||||
|
||||
for range 2 {
|
||||
entries, err := store.GetBaseColumnEvolutions(context.Background(), telemetrytypes.SignalTraces, telemetrytypes.FieldContextAttribute)
|
||||
require.NoError(t, err)
|
||||
require.Len(t, entries, 1)
|
||||
assert.Equal(t, "attributes", entries[0].ColumnName)
|
||||
assert.Equal(t, releaseTime, entries[0].ReleaseTime.UTC())
|
||||
}
|
||||
require.NoError(t, mock.ExpectationsWereMet())
|
||||
}
|
||||
|
||||
func TestGetBaseColumnEvolutionsRefetchesEmptyResult(t *testing.T) {
|
||||
mockTelemetryStore := telemetrystoretest.New(telemetrystore.Config{}, ®exMatcher{})
|
||||
mock := mockTelemetryStore.Mock()
|
||||
store := NewSchemaMetadataStore(mockTelemetryStore)
|
||||
get := func() []*telemetrytypes.EvolutionEntry {
|
||||
entries, err := store.GetBaseColumnEvolutions(context.Background(), telemetrytypes.SignalTraces, telemetrytypes.FieldContextAttribute)
|
||||
require.NoError(t, err)
|
||||
return entries
|
||||
}
|
||||
|
||||
expectBaseColumnEvolutionsQuery(mock, nil)
|
||||
assert.Empty(t, get())
|
||||
|
||||
expectBaseColumnEvolutionsQuery(mock, [][]any{
|
||||
{"traces", "attributes", "JSON()", "attribute", "__all__", uint32(0), float64(time.Now().UnixNano())},
|
||||
})
|
||||
assert.Len(t, get(), 1)
|
||||
assert.Len(t, get(), 1, "a found result is cached")
|
||||
require.NoError(t, mock.ExpectationsWereMet())
|
||||
}
|
||||
@@ -288,7 +288,7 @@ func TestConditionFor(t *testing.T) {
|
||||
},
|
||||
}
|
||||
|
||||
storage := NewStorage()
|
||||
storage := NewStorage(nil)
|
||||
|
||||
for _, tc := range testCases {
|
||||
sb := sqlbuilder.NewSelectBuilder()
|
||||
@@ -374,7 +374,7 @@ func TestConditionForResourceWithEvolution(t *testing.T) {
|
||||
},
|
||||
}
|
||||
|
||||
storage := NewStorage()
|
||||
storage := NewStorage(nil)
|
||||
|
||||
for _, tc := range testCases {
|
||||
sb := sqlbuilder.NewSelectBuilder()
|
||||
@@ -396,7 +396,7 @@ func TestConditionForResourceWithEvolution(t *testing.T) {
|
||||
// undeclared (Dynamic) paths and genuinely NULL when absent.
|
||||
func TestConditionForScopeIntrinsicFields(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
storage := NewStorage()
|
||||
storage := NewStorage(nil)
|
||||
|
||||
testCases := []struct {
|
||||
name string
|
||||
@@ -482,7 +482,7 @@ func TestConditionForScopeIntrinsicFields(t *testing.T) {
|
||||
// user input and queries anyway, emitting a warning instead of failing.
|
||||
func TestConditionForSynthesizedKeys(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
storage := NewStorage()
|
||||
storage := NewStorage(nil)
|
||||
|
||||
// no metadata matches -> the builder must synthesize from user input
|
||||
var noMatches map[string][]*telemetrytypes.TelemetryFieldKey
|
||||
@@ -692,7 +692,7 @@ func TestConditionForSynthesizedKeys(t *testing.T) {
|
||||
// attribute, with or without a metadata entry for the column.
|
||||
func TestIntrinsicColumnFirst(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
storage := NewStorage()
|
||||
storage := NewStorage(nil)
|
||||
q := qbtypes.QueryInfo{StartNs: 1761437108000000000, EndNs: 1761458708000000000}
|
||||
attr := func(name string, dataType telemetrytypes.FieldDataType) *telemetrytypes.TelemetryFieldKey {
|
||||
return &telemetrytypes.TelemetryFieldKey{Name: name, Signal: telemetrytypes.SignalTraces, FieldContext: telemetrytypes.FieldContextAttribute, FieldDataType: dataType}
|
||||
|
||||
@@ -51,7 +51,7 @@ func TestConditionForFamilyMergesMembersCurrentFirst(t *testing.T) {
|
||||
}},
|
||||
}
|
||||
fl := familyFlagOn(t)
|
||||
storage := NewStorage()
|
||||
storage := NewStorage(nil)
|
||||
|
||||
// The requested spelling is the old name; precedence must still be
|
||||
// current-first.
|
||||
@@ -87,7 +87,7 @@ func TestConditionForFamilyNegativeKeepsKeylessRows(t *testing.T) {
|
||||
}},
|
||||
}
|
||||
fl := familyFlagOn(t)
|
||||
storage := NewStorage()
|
||||
storage := NewStorage(nil)
|
||||
|
||||
sb := sqlbuilder.NewSelectBuilder()
|
||||
conds, _, err := querybuilder.Conditions(context.Background(), querybuilder.NewQueryInfo(context.Background(), valuer.UUID{}, fl, telemetrytypes.SignalTraces, nil, startNs, endNs), storage, &telemetrytypes.TelemetryFieldKey{Name: "deployment.environment.name"}, qbtypes.FilterOperatorNotEqual, "production", fieldKeys, false, sb)
|
||||
@@ -122,7 +122,7 @@ func TestConditionForFamilyExists(t *testing.T) {
|
||||
}},
|
||||
}
|
||||
fl := familyFlagOn(t)
|
||||
storage := NewStorage()
|
||||
storage := NewStorage(nil)
|
||||
|
||||
sb := sqlbuilder.NewSelectBuilder()
|
||||
conds, _, err := querybuilder.Conditions(context.Background(), querybuilder.NewQueryInfo(context.Background(), valuer.UUID{}, fl, telemetrytypes.SignalTraces, nil, startNs, endNs), storage, &telemetrytypes.TelemetryFieldKey{Name: "deployment.environment.name"}, qbtypes.FilterOperatorNotExists, nil, fieldKeys, false, sb)
|
||||
@@ -157,7 +157,7 @@ func TestConditionForFamilyOffByDefault(t *testing.T) {
|
||||
}},
|
||||
}
|
||||
fl := flaggertest.New(t)
|
||||
storage := NewStorage()
|
||||
storage := NewStorage(nil)
|
||||
|
||||
sb := sqlbuilder.NewSelectBuilder()
|
||||
conds, _, err := querybuilder.Conditions(context.Background(), querybuilder.NewQueryInfo(context.Background(), valuer.UUID{}, fl, telemetrytypes.SignalTraces, nil, startNs, endNs), storage, &telemetrytypes.TelemetryFieldKey{Name: "deployment.environment.name"}, qbtypes.FilterOperatorEqual, "production", fieldKeys, false, sb)
|
||||
@@ -193,7 +193,7 @@ func TestConditionForSingleMemberIsUnchanged(t *testing.T) {
|
||||
}
|
||||
delete(fieldKeys, "deployment.environment")
|
||||
fl := familyFlagOn(t)
|
||||
storage := NewStorage()
|
||||
storage := NewStorage(nil)
|
||||
|
||||
sb := sqlbuilder.NewSelectBuilder()
|
||||
conds, _, err := querybuilder.Conditions(context.Background(), querybuilder.NewQueryInfo(context.Background(), valuer.UUID{}, fl, telemetrytypes.SignalTraces, nil, startNs, endNs), storage, &telemetrytypes.TelemetryFieldKey{Name: "deployment.environment.name"}, qbtypes.FilterOperatorEqual, "production", fieldKeys, false, sb)
|
||||
@@ -226,7 +226,7 @@ func TestColumnExpressionForFamilyGroupBy(t *testing.T) {
|
||||
}},
|
||||
}
|
||||
fl := familyFlagOn(t)
|
||||
storage := NewStorage()
|
||||
storage := NewStorage(nil)
|
||||
|
||||
expr, err := querybuilder.ResolveColumn(context.Background(), querybuilder.NewQueryInfo(context.Background(), valuer.UUID{}, fl, telemetrytypes.SignalTraces, nil, startNs, endNs), storage, &telemetrytypes.TelemetryFieldKey{Name: "deployment.environment.name"}, telemetrytypes.FieldDataTypeString, fieldKeys)
|
||||
require.NoError(t, err)
|
||||
|
||||
@@ -15,7 +15,7 @@ import (
|
||||
|
||||
func TestSpanScopeFilterExpression(t *testing.T) {
|
||||
// Test that span scope fields work in filter expressions
|
||||
storage := NewStorage()
|
||||
storage := NewStorage(nil)
|
||||
|
||||
tests := []struct {
|
||||
name string
|
||||
@@ -125,7 +125,7 @@ func TestSpanScopeWithResourceFilter(t *testing.T) {
|
||||
// For now, just verify the expression parses correctly
|
||||
// In a real implementation, we'd need to check that the resource filter
|
||||
// is properly skipped when span scope fields are present
|
||||
storage := NewStorage()
|
||||
storage := NewStorage(nil)
|
||||
|
||||
// Prepare field keys for the test
|
||||
fieldKeys := make(map[string][]*telemetrytypes.TelemetryFieldKey)
|
||||
|
||||
@@ -162,12 +162,14 @@ var (
|
||||
}
|
||||
)
|
||||
|
||||
type storage struct{}
|
||||
type storage struct {
|
||||
schemaMetadataStore telemetrytypes.SchemaMetadataStore
|
||||
}
|
||||
|
||||
var _ qbtypes.Storage = (*storage)(nil)
|
||||
|
||||
func NewStorage() qbtypes.Storage {
|
||||
return &storage{}
|
||||
func NewStorage(schemaMetadataStore telemetrytypes.SchemaMetadataStore) qbtypes.Storage {
|
||||
return &storage{schemaMetadataStore: schemaMetadataStore}
|
||||
}
|
||||
|
||||
func (m *storage) getColumn(
|
||||
@@ -562,9 +564,35 @@ func (m *storage) Fallback(ctx context.Context, q qbtypes.QueryInfo, key *teleme
|
||||
literal := telemetrytypes.NewTelemetryFieldKey(key.FieldContext.StringValue()+"."+key.Name, key.FieldContext, key.FieldDataType)
|
||||
keys = append(querybuilder.SynthesizeKeys(key, value), querybuilder.SynthesizeKeys(literal, value)...)
|
||||
}
|
||||
if err := m.addBaseColumnEvolutions(ctx, q, keys); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return querybuilder.WrapAsLogicalFields(key.Name, keys), nil
|
||||
}
|
||||
|
||||
func (m *storage) addBaseColumnEvolutions(ctx context.Context, q qbtypes.QueryInfo, keys []*telemetrytypes.TelemetryFieldKey) error {
|
||||
if !q.TraceAttrsJSONOn || m.schemaMetadataStore == nil {
|
||||
return nil
|
||||
}
|
||||
var attributeKeys []*telemetrytypes.TelemetryFieldKey
|
||||
for _, key := range keys {
|
||||
if key.FieldContext == telemetrytypes.FieldContextAttribute && len(key.Evolutions) == 0 {
|
||||
attributeKeys = append(attributeKeys, key)
|
||||
}
|
||||
}
|
||||
if len(attributeKeys) == 0 {
|
||||
return nil
|
||||
}
|
||||
evolutions, err := m.schemaMetadataStore.GetBaseColumnEvolutions(ctx, telemetrytypes.SignalTraces, telemetrytypes.FieldContextAttribute)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
for _, key := range attributeKeys {
|
||||
key.Evolutions = evolutions
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// stampColumnType gives an untyped column key the data type its column reads
|
||||
// as, so the intrinsic-column step can compare it with same-named metadata.
|
||||
func stampColumnType(key *telemetrytypes.TelemetryFieldKey, columns []*schema.Column) *telemetrytypes.TelemetryFieldKey {
|
||||
|
||||
@@ -8,8 +8,10 @@ import (
|
||||
"github.com/SigNoz/signoz/pkg/querybuilder"
|
||||
qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
|
||||
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
|
||||
"github.com/SigNoz/signoz/pkg/types/telemetrytypes/telemetrytypestest"
|
||||
"github.com/huandu/go-sqlbuilder"
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/mock"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
@@ -44,7 +46,7 @@ func attrKey(name string, dt telemetrytypes.FieldDataType, evo []*telemetrytypes
|
||||
// cast, straddling a dual-read multiIf with the JSON column first.
|
||||
func TestFieldForAttributeJSONEvolution(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
storage := NewStorage()
|
||||
storage := NewStorage(nil)
|
||||
evo := MockAttributeEvolutionData(attrJSONRelease)
|
||||
|
||||
testCases := []struct {
|
||||
@@ -79,7 +81,7 @@ func TestFieldForAttributeJSONEvolution(t *testing.T) {
|
||||
// entry is registered: a key with no evolutions resolves to the Map column for every window.
|
||||
func TestFieldForAttributeNoEvolutionParity(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
storage := NewStorage()
|
||||
storage := NewStorage(nil)
|
||||
|
||||
for _, dt := range []struct {
|
||||
dataType telemetrytypes.FieldDataType
|
||||
@@ -101,7 +103,7 @@ func TestFieldForAttributeNoEvolutionParity(t *testing.T) {
|
||||
// every window.
|
||||
func TestAttributeJSONFlagOffParity(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
storage := NewStorage()
|
||||
storage := NewStorage(nil)
|
||||
evo := MockAttributeEvolutionData(attrJSONRelease)
|
||||
windows := map[string][2]uint64{
|
||||
"BeforeRelease": attrWindowBefore,
|
||||
@@ -154,7 +156,7 @@ func TestAttributeJSONFlagOffParity(t *testing.T) {
|
||||
// numeric comparisons keep numeric semantics; existence never tests the ::String cast.
|
||||
func TestConditionForAttributeJSON(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
storage := NewStorage()
|
||||
storage := NewStorage(nil)
|
||||
evo := MockAttributeEvolutionData(attrJSONRelease)
|
||||
|
||||
testCases := []struct {
|
||||
@@ -261,7 +263,7 @@ func TestConditionForAttributeJSON(t *testing.T) {
|
||||
// home is excluded (De Morgan), including rows that predate the JSON column.
|
||||
func TestConditionForAttributeJSONNotExistsDualRead(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
storage := NewStorage()
|
||||
storage := NewStorage(nil)
|
||||
evo := MockAttributeEvolutionData(attrJSONRelease)
|
||||
|
||||
key := attrKey("user.id", telemetrytypes.FieldDataTypeString, evo)
|
||||
@@ -281,7 +283,7 @@ func TestConditionForAttributeJSONNotExistsDualRead(t *testing.T) {
|
||||
// path is NULL rather than a spurious ”/0, and the numeric branch keeps its toFloat64 coercion.
|
||||
func TestColumnExpressionForAttributeJSON(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
storage := NewStorage()
|
||||
storage := NewStorage(nil)
|
||||
evo := MockAttributeEvolutionData(attrJSONRelease)
|
||||
|
||||
t.Run("group by string", func(t *testing.T) {
|
||||
@@ -304,7 +306,7 @@ func TestColumnExpressionForAttributeJSON(t *testing.T) {
|
||||
// attribute filter must not emit the "ambiguous key" warning.
|
||||
func TestAttributeJSONNoAmbiguityWarning(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
storage := NewStorage()
|
||||
storage := NewStorage(nil)
|
||||
evo := MockAttributeEvolutionData(attrJSONRelease)
|
||||
|
||||
key := attrKey("user.id", telemetrytypes.FieldDataTypeString, evo)
|
||||
@@ -320,7 +322,7 @@ func TestAttributeJSONNoAmbiguityWarning(t *testing.T) {
|
||||
// warning. In the JSON column the two branches share the raw path; only the cast differs.
|
||||
func TestConditionForAttributeJSONTypeCollision(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
storage := NewStorage()
|
||||
storage := NewStorage(nil)
|
||||
evo := MockAttributeEvolutionData(attrJSONRelease)
|
||||
|
||||
strKey := attrKey("http.status_code", telemetrytypes.FieldDataTypeString, evo)
|
||||
@@ -350,7 +352,7 @@ func TestConditionForAttributeJSONTypeCollision(t *testing.T) {
|
||||
// as its actual stored type.
|
||||
func TestColumnExpressionForAttributeJSONTypeCollision(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
storage := NewStorage()
|
||||
storage := NewStorage(nil)
|
||||
evo := MockAttributeEvolutionData(attrJSONRelease)
|
||||
|
||||
strKey := attrKey("http.status_code", telemetrytypes.FieldDataTypeString, evo)
|
||||
@@ -374,7 +376,7 @@ func TestColumnExpressionForAttributeJSONTypeCollision(t *testing.T) {
|
||||
// coverage.
|
||||
func TestColumnExpressionForAttributeJSONTypeCollisionNumericAgg(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
storage := NewStorage()
|
||||
storage := NewStorage(nil)
|
||||
evo := MockAttributeEvolutionData(attrJSONRelease)
|
||||
|
||||
numKey := attrKey("http.status_code", telemetrytypes.FieldDataTypeNumber, evo)
|
||||
@@ -396,7 +398,7 @@ func TestColumnExpressionForAttributeJSONTypeCollisionNumericAgg(t *testing.T) {
|
||||
// columns (attributes_string / attributes_number), each with its own mapContains guard.
|
||||
func TestConditionForAttributeMapTypeCollisionParity(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
storage := NewStorage()
|
||||
storage := NewStorage(nil)
|
||||
evo := MockAttributeEvolutionData(attrJSONRelease)
|
||||
|
||||
strKey := attrKey("http.status_code", telemetrytypes.FieldDataTypeString, evo)
|
||||
@@ -440,7 +442,7 @@ func TestColumnForUnspecifiedAttributeNoBranchFlip(t *testing.T) {
|
||||
// evolution: a key without it (the pre-rollout system) is byte-identical to today.
|
||||
func TestConditionForAttributeJSONNegativeOperatorParity(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
storage := NewStorage()
|
||||
storage := NewStorage(nil)
|
||||
evo := MockAttributeEvolutionData(attrJSONRelease)
|
||||
|
||||
build := func(t *testing.T, key telemetrytypes.TelemetryFieldKey, window [2]uint64, op qbtypes.FilterOperator, value any) string {
|
||||
@@ -520,7 +522,7 @@ func TestConditionForAttributeJSONNegativeOperatorParity(t *testing.T) {
|
||||
// EXISTS/NOT EXISTS must still exclude a key absent from every home, rather than matching it.
|
||||
func TestConditionForAttributeJSONStraddleAbsentKeyExclusion(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
storage := NewStorage()
|
||||
storage := NewStorage(nil)
|
||||
evo := MockAttributeEvolutionData(attrJSONRelease)
|
||||
|
||||
build := func(t *testing.T, key telemetrytypes.TelemetryFieldKey, op qbtypes.FilterOperator, value any) string {
|
||||
@@ -551,3 +553,139 @@ func TestConditionForAttributeJSONStraddleAbsentKeyExclusion(t *testing.T) {
|
||||
assert.Contains(t, sql, ", NULL) IS NULL")
|
||||
})
|
||||
}
|
||||
|
||||
func TestConditionForAttributeJSONUnknownKey(t *testing.T) {
|
||||
evolutions := MockAttributeEvolutionData(attrJSONRelease)
|
||||
bare := telemetrytypes.TelemetryFieldKey{Name: "unknown.attr"}
|
||||
|
||||
testCases := []struct {
|
||||
name string
|
||||
window [2]uint64
|
||||
flagOff bool
|
||||
evolutions []*telemetrytypes.EvolutionEntry
|
||||
key telemetrytypes.TelemetryFieldKey
|
||||
operator qbtypes.FilterOperator
|
||||
value any
|
||||
wantContains []string
|
||||
wantNotContains []string
|
||||
}{
|
||||
{
|
||||
name: "BeforeRollout_ReadsMap",
|
||||
window: attrWindowBefore,
|
||||
evolutions: evolutions,
|
||||
key: bare,
|
||||
operator: qbtypes.FilterOperatorEqual,
|
||||
value: "x",
|
||||
wantContains: []string{"attributes_string['unknown.attr'] = ?"},
|
||||
wantNotContains: []string{"attributes.`unknown.attr`"},
|
||||
},
|
||||
{
|
||||
name: "AfterRollout_ReadsJSON",
|
||||
window: attrWindowAfter,
|
||||
evolutions: evolutions,
|
||||
key: bare,
|
||||
operator: qbtypes.FilterOperatorEqual,
|
||||
value: "x",
|
||||
wantContains: []string{"attributes.`unknown.attr`::String = ?"},
|
||||
wantNotContains: []string{"attributes_string"},
|
||||
},
|
||||
{
|
||||
name: "StraddlingRollout_ReadsJSONThenMap",
|
||||
window: attrWindowStraddle,
|
||||
evolutions: evolutions,
|
||||
key: bare,
|
||||
operator: qbtypes.FilterOperatorEqual,
|
||||
value: "x",
|
||||
wantContains: []string{"multiIf(attributes.`unknown.attr` IS NOT NULL, attributes.`unknown.attr`::String, mapContains(attributes_string, 'unknown.attr'), attributes_string['unknown.attr'], NULL) = ?"},
|
||||
},
|
||||
{
|
||||
name: "AfterRollout_NumberOperand_ReadsJSON",
|
||||
window: attrWindowAfter,
|
||||
evolutions: evolutions,
|
||||
key: bare,
|
||||
operator: qbtypes.FilterOperatorEqual,
|
||||
value: float64(1),
|
||||
wantContains: []string{"toFloat64OrNull(attributes.`unknown.attr`::String) = ?"},
|
||||
wantNotContains: []string{"attributes_number"},
|
||||
},
|
||||
{
|
||||
name: "AfterRollout_Exists_ReadsJSONOnly",
|
||||
window: attrWindowAfter,
|
||||
evolutions: evolutions,
|
||||
key: bare,
|
||||
operator: qbtypes.FilterOperatorExists,
|
||||
wantContains: []string{"attributes.`unknown.attr` IS NOT NULL"},
|
||||
wantNotContains: []string{"mapContains"},
|
||||
},
|
||||
{
|
||||
name: "AfterRollout_AttributeContext_ReadsJSONForBothSpellings",
|
||||
window: attrWindowAfter,
|
||||
evolutions: evolutions,
|
||||
key: telemetrytypes.TelemetryFieldKey{Name: "unknown.attr", FieldContext: telemetrytypes.FieldContextAttribute},
|
||||
operator: qbtypes.FilterOperatorEqual,
|
||||
value: "x",
|
||||
wantContains: []string{"attributes.`unknown.attr`::String = ?", "attributes.`attribute.unknown.attr`::String = ?"},
|
||||
wantNotContains: []string{"attributes_string"},
|
||||
},
|
||||
{
|
||||
name: "AfterRollout_SpanContext_ReadsJSONAttribute",
|
||||
window: attrWindowAfter,
|
||||
evolutions: evolutions,
|
||||
key: telemetrytypes.TelemetryFieldKey{Name: "unknown.attr", FieldContext: telemetrytypes.FieldContextSpan},
|
||||
operator: qbtypes.FilterOperatorEqual,
|
||||
value: "x",
|
||||
wantContains: []string{"attributes.`unknown.attr`::String = ?"},
|
||||
wantNotContains: []string{"attributes_string"},
|
||||
},
|
||||
{
|
||||
name: "NoEvolutions_ReadsMap",
|
||||
window: attrWindowAfter,
|
||||
key: bare,
|
||||
operator: qbtypes.FilterOperatorEqual,
|
||||
value: "x",
|
||||
wantContains: []string{"attributes_string['unknown.attr'] = ?"},
|
||||
wantNotContains: []string{"attributes.`unknown.attr`"},
|
||||
},
|
||||
{
|
||||
name: "FlagOff_ReadsMap",
|
||||
window: attrWindowAfter,
|
||||
flagOff: true,
|
||||
evolutions: evolutions,
|
||||
key: bare,
|
||||
operator: qbtypes.FilterOperatorEqual,
|
||||
value: "x",
|
||||
wantContains: []string{"attributes_string['unknown.attr'] = ?"},
|
||||
wantNotContains: []string{"attributes.`unknown.attr`"},
|
||||
},
|
||||
{
|
||||
name: "ResourceContext_IgnoresAttributeEvolutions",
|
||||
window: attrWindowAfter,
|
||||
evolutions: evolutions,
|
||||
key: telemetrytypes.TelemetryFieldKey{Name: "unknown.attr", FieldContext: telemetrytypes.FieldContextResource},
|
||||
operator: qbtypes.FilterOperatorEqual,
|
||||
value: "x",
|
||||
wantContains: []string{"resource.`unknown.attr`::String", "resources_string['unknown.attr']"},
|
||||
},
|
||||
}
|
||||
|
||||
for _, testCase := range testCases {
|
||||
t.Run(testCase.name, func(t *testing.T) {
|
||||
schemaMetadataStore := telemetrytypestest.NewMockSchemaMetadataStore(t)
|
||||
schemaMetadataStore.EXPECT().GetBaseColumnEvolutions(mock.Anything, telemetrytypes.SignalTraces, telemetrytypes.FieldContextAttribute).Return(testCase.evolutions, nil).Maybe()
|
||||
|
||||
q := qbtypes.QueryInfo{StartNs: testCase.window[0], EndNs: testCase.window[1], TraceAttrsJSONOn: !testCase.flagOff}
|
||||
sb := sqlbuilder.NewSelectBuilder()
|
||||
conds, _, err := querybuilder.Conditions(context.Background(), q, NewStorage(schemaMetadataStore), &testCase.key, testCase.operator, testCase.value, map[string][]*telemetrytypes.TelemetryFieldKey{}, false, sb)
|
||||
require.NoError(t, err)
|
||||
sb.Where(conds...)
|
||||
sql, _ := sb.BuildWithFlavor(sqlbuilder.ClickHouse)
|
||||
|
||||
for _, want := range testCase.wantContains {
|
||||
assert.Contains(t, sql, want)
|
||||
}
|
||||
for _, unwanted := range testCase.wantNotContains {
|
||||
assert.NotContains(t, sql, unwanted)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
@@ -24,7 +24,7 @@ var (
|
||||
// `attributes_promoted` alone after promotion — fanning out only across an evolution boundary.
|
||||
func TestFieldForAttributePromotedEvolution(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
storage := NewStorage()
|
||||
storage := NewStorage(nil)
|
||||
evo := MockPromotedAttributeEvolutionData("span.operation", promoJSONRelease, promoPromoRelease)
|
||||
|
||||
win := func(from, to string) [2]uint64 {
|
||||
@@ -65,7 +65,7 @@ func TestFieldForAttributePromotedEvolution(t *testing.T) {
|
||||
// attributes_promoted_paths_tokenbf) — not the attributes column.
|
||||
func TestConditionForAttributePromoted(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
storage := NewStorage()
|
||||
storage := NewStorage(nil)
|
||||
evo := MockPromotedAttributeEvolutionData("span.operation", promoJSONRelease, promoPromoRelease)
|
||||
afterPromo := [2]uint64{
|
||||
uint64(time.Date(2025, 7, 1, 0, 0, 0, 0, time.UTC).UnixNano()),
|
||||
|
||||
@@ -146,7 +146,7 @@ func TestGetFieldKeyName(t *testing.T) {
|
||||
|
||||
for _, tc := range testCases {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
storage := NewStorage()
|
||||
storage := NewStorage(nil)
|
||||
read, err := storage.Read(ctx, qbtypes.QueryInfo{StartNs: uint64(time.Date(2024, 6, 1, 0, 0, 0, 0, time.UTC).UnixNano()), EndNs: uint64(time.Date(2024, 6, 5, 0, 0, 0, 0, time.UTC).UnixNano())}, &tc.key)
|
||||
|
||||
if tc.expectedError != nil {
|
||||
@@ -234,7 +234,7 @@ func TestFieldForResourceWithEvolution(t *testing.T) {
|
||||
|
||||
for _, tc := range testCases {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
storage := NewStorage()
|
||||
storage := NewStorage(nil)
|
||||
read, err := storage.Read(ctx, qbtypes.QueryInfo{StartNs: tc.tsStart, EndNs: tc.tsEnd}, &tc.key)
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, tc.expectedResult, read.SQL)
|
||||
@@ -294,7 +294,7 @@ func TestColumnExpressionForTemporalColumn(t *testing.T) {
|
||||
|
||||
for _, tc := range testCases {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
storage := NewStorage()
|
||||
storage := NewStorage(nil)
|
||||
result, err := querybuilder.ResolveColumn(ctx, qbtypes.QueryInfo{StartNs: tsStart, EndNs: tsEnd}, storage, &tc.key, tc.requiredDataType, nil)
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, tc.expectedResult, result)
|
||||
@@ -322,7 +322,7 @@ func TestColumnExpressionForTimestampAttributeCollision(t *testing.T) {
|
||||
},
|
||||
}
|
||||
|
||||
storage := NewStorage()
|
||||
storage := NewStorage(nil)
|
||||
|
||||
t.Run("bare timestamp resolves to the intrinsic column alone", func(t *testing.T) {
|
||||
bare := telemetrytypes.TelemetryFieldKey{Name: "timestamp"}
|
||||
@@ -431,7 +431,7 @@ func TestColumnExpressionForScopeDeclaredPath(t *testing.T) {
|
||||
|
||||
for _, tc := range testCases {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
storage := NewStorage()
|
||||
storage := NewStorage(nil)
|
||||
result, err := querybuilder.ResolveColumn(ctx, qbtypes.QueryInfo{}, storage, &tc.key, telemetrytypes.FieldDataTypeUnspecified, tc.keys)
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, tc.expectedResult, result)
|
||||
|
||||
@@ -4,6 +4,8 @@ import (
|
||||
"time"
|
||||
)
|
||||
|
||||
const EvolutionFieldNameAll = "__all__"
|
||||
|
||||
type EvolutionEntry struct {
|
||||
Signal Signal `json:"signal"`
|
||||
ColumnName string `json:"column_name"`
|
||||
|
||||
@@ -7,6 +7,10 @@ import (
|
||||
"github.com/SigNoz/signoz/pkg/valuer"
|
||||
)
|
||||
|
||||
type SchemaMetadataStore interface {
|
||||
GetBaseColumnEvolutions(ctx context.Context, signal Signal, fieldContext FieldContext) ([]*EvolutionEntry, error)
|
||||
}
|
||||
|
||||
// MetadataStore is the interface for the telemetry metadata store.
|
||||
type MetadataStore interface {
|
||||
// GetKeys returns a map of field keys types.TelemetryFieldKey by name, there can be multiple keys with the same name
|
||||
|
||||
@@ -0,0 +1,122 @@
|
||||
// Code generated by mockery; DO NOT EDIT.
|
||||
// github.com/vektra/mockery
|
||||
// template: testify
|
||||
|
||||
package telemetrytypestest
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
|
||||
mock "github.com/stretchr/testify/mock"
|
||||
)
|
||||
|
||||
// NewMockSchemaMetadataStore creates a new instance of MockSchemaMetadataStore. It also registers a testing interface on the mock and a cleanup function to assert the mocks expectations.
|
||||
// The first argument is typically a *testing.T value.
|
||||
func NewMockSchemaMetadataStore(t interface {
|
||||
mock.TestingT
|
||||
Cleanup(func())
|
||||
}) *MockSchemaMetadataStore {
|
||||
if helper, ok := t.(interface{ Helper() }); ok {
|
||||
helper.Helper()
|
||||
}
|
||||
|
||||
mock := &MockSchemaMetadataStore{}
|
||||
mock.Mock.Test(t)
|
||||
|
||||
t.Cleanup(func() {
|
||||
if helper, ok := t.(interface{ Helper() }); ok {
|
||||
helper.Helper()
|
||||
}
|
||||
mock.AssertExpectations(t)
|
||||
})
|
||||
|
||||
return mock
|
||||
}
|
||||
|
||||
// MockSchemaMetadataStore is an autogenerated mock type for the SchemaMetadataStore type
|
||||
type MockSchemaMetadataStore struct {
|
||||
mock.Mock
|
||||
}
|
||||
|
||||
type MockSchemaMetadataStore_Expecter struct {
|
||||
mock *mock.Mock
|
||||
}
|
||||
|
||||
func (_m *MockSchemaMetadataStore) EXPECT() *MockSchemaMetadataStore_Expecter {
|
||||
return &MockSchemaMetadataStore_Expecter{mock: &_m.Mock}
|
||||
}
|
||||
|
||||
// GetBaseColumnEvolutions provides a mock function for the type MockSchemaMetadataStore
|
||||
func (_mock *MockSchemaMetadataStore) GetBaseColumnEvolutions(ctx context.Context, signal telemetrytypes.Signal, fieldContext telemetrytypes.FieldContext) ([]*telemetrytypes.EvolutionEntry, error) {
|
||||
ret := _mock.Called(ctx, signal, fieldContext)
|
||||
|
||||
if len(ret) == 0 {
|
||||
panic("no return value specified for GetBaseColumnEvolutions")
|
||||
}
|
||||
|
||||
var r0 []*telemetrytypes.EvolutionEntry
|
||||
var r1 error
|
||||
if returnFunc, ok := ret.Get(0).(func(context.Context, telemetrytypes.Signal, telemetrytypes.FieldContext) ([]*telemetrytypes.EvolutionEntry, error)); ok {
|
||||
return returnFunc(ctx, signal, fieldContext)
|
||||
}
|
||||
if returnFunc, ok := ret.Get(0).(func(context.Context, telemetrytypes.Signal, telemetrytypes.FieldContext) []*telemetrytypes.EvolutionEntry); ok {
|
||||
r0 = returnFunc(ctx, signal, fieldContext)
|
||||
} else {
|
||||
if ret.Get(0) != nil {
|
||||
r0 = ret.Get(0).([]*telemetrytypes.EvolutionEntry)
|
||||
}
|
||||
}
|
||||
if returnFunc, ok := ret.Get(1).(func(context.Context, telemetrytypes.Signal, telemetrytypes.FieldContext) error); ok {
|
||||
r1 = returnFunc(ctx, signal, fieldContext)
|
||||
} else {
|
||||
r1 = ret.Error(1)
|
||||
}
|
||||
return r0, r1
|
||||
}
|
||||
|
||||
// MockSchemaMetadataStore_GetBaseColumnEvolutions_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'GetBaseColumnEvolutions'
|
||||
type MockSchemaMetadataStore_GetBaseColumnEvolutions_Call struct {
|
||||
*mock.Call
|
||||
}
|
||||
|
||||
// GetBaseColumnEvolutions is a helper method to define mock.On call
|
||||
// - ctx context.Context
|
||||
// - signal telemetrytypes.Signal
|
||||
// - fieldContext telemetrytypes.FieldContext
|
||||
func (_e *MockSchemaMetadataStore_Expecter) GetBaseColumnEvolutions(ctx any, signal any, fieldContext any) *MockSchemaMetadataStore_GetBaseColumnEvolutions_Call {
|
||||
return &MockSchemaMetadataStore_GetBaseColumnEvolutions_Call{Call: _e.mock.On("GetBaseColumnEvolutions", ctx, signal, fieldContext)}
|
||||
}
|
||||
|
||||
func (_c *MockSchemaMetadataStore_GetBaseColumnEvolutions_Call) Run(run func(ctx context.Context, signal telemetrytypes.Signal, fieldContext telemetrytypes.FieldContext)) *MockSchemaMetadataStore_GetBaseColumnEvolutions_Call {
|
||||
_c.Call.Run(func(args mock.Arguments) {
|
||||
var arg0 context.Context
|
||||
if args[0] != nil {
|
||||
arg0 = args[0].(context.Context)
|
||||
}
|
||||
var arg1 telemetrytypes.Signal
|
||||
if args[1] != nil {
|
||||
arg1 = args[1].(telemetrytypes.Signal)
|
||||
}
|
||||
var arg2 telemetrytypes.FieldContext
|
||||
if args[2] != nil {
|
||||
arg2 = args[2].(telemetrytypes.FieldContext)
|
||||
}
|
||||
run(
|
||||
arg0,
|
||||
arg1,
|
||||
arg2,
|
||||
)
|
||||
})
|
||||
return _c
|
||||
}
|
||||
|
||||
func (_c *MockSchemaMetadataStore_GetBaseColumnEvolutions_Call) Return(evolutionEntrys []*telemetrytypes.EvolutionEntry, err error) *MockSchemaMetadataStore_GetBaseColumnEvolutions_Call {
|
||||
_c.Call.Return(evolutionEntrys, err)
|
||||
return _c
|
||||
}
|
||||
|
||||
func (_c *MockSchemaMetadataStore_GetBaseColumnEvolutions_Call) RunAndReturn(run func(ctx context.Context, signal telemetrytypes.Signal, fieldContext telemetrytypes.FieldContext) ([]*telemetrytypes.EvolutionEntry, error)) *MockSchemaMetadataStore_GetBaseColumnEvolutions_Call {
|
||||
_c.Call.Return(run)
|
||||
return _c
|
||||
}
|
||||
@@ -13,8 +13,11 @@ from fixtures.querier import (
|
||||
build_aggregation,
|
||||
build_group_by_field,
|
||||
build_traces_scalar_query,
|
||||
get_all_warnings,
|
||||
get_preview_sql,
|
||||
get_rows,
|
||||
index_series_by_label,
|
||||
make_preview_query_request,
|
||||
make_query_request,
|
||||
)
|
||||
from fixtures.traces import TraceIdGenerator, Traces
|
||||
@@ -467,3 +470,79 @@ def test_traces_attributes_json_list_view(
|
||||
assert row["data"]["name"] == name
|
||||
# int/float compare equal in Python, so map-sourced float64 and JSON-sourced numbers both match.
|
||||
assert row["data"]["attributes"] == attributes, name
|
||||
|
||||
|
||||
def test_traces_attributes_json_unreported_key(
|
||||
signoz: types.SigNoz,
|
||||
create_user_admin: None, # pylint: disable=unused-argument
|
||||
get_token: Callable[[str, str], str],
|
||||
insert_traces: Callable[[list[Traces]], None],
|
||||
seed_attribute_evolution: Callable[[str, datetime], None],
|
||||
) -> None:
|
||||
"""An attribute metadata does not report follows the rollout like a reported one: the legacy
|
||||
map before it, the attributes JSON column after it."""
|
||||
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
|
||||
|
||||
evolution_time = datetime.now(tz=UTC).replace(second=0, microsecond=0) - timedelta(minutes=30)
|
||||
seed_attribute_evolution("traces", evolution_time)
|
||||
|
||||
before = evolution_time - timedelta(minutes=5)
|
||||
after = evolution_time + timedelta(minutes=5)
|
||||
spans = [
|
||||
Traces(
|
||||
timestamp=before,
|
||||
trace_id=TraceIdGenerator.trace_id(),
|
||||
span_id=TraceIdGenerator.span_id(),
|
||||
name="before",
|
||||
attributes={"unreported.key": "value"},
|
||||
attribute_write_mode="legacy_only",
|
||||
),
|
||||
Traces(
|
||||
timestamp=after,
|
||||
trace_id=TraceIdGenerator.trace_id(),
|
||||
span_id=TraceIdGenerator.span_id(),
|
||||
name="after",
|
||||
attributes={"unreported.key": "value"},
|
||||
attribute_write_mode="json_only",
|
||||
),
|
||||
]
|
||||
# no metadata rows, so the querier does not know the key
|
||||
for span in spans:
|
||||
span.attribute_keys = []
|
||||
span.tag_attributes = []
|
||||
insert_traces(spans)
|
||||
|
||||
queries = [
|
||||
build_traces_scalar_query(
|
||||
aggregations=[build_aggregation("count()")],
|
||||
filter_expression="unreported.key = 'value'",
|
||||
)
|
||||
]
|
||||
|
||||
# before the rollout -> legacy map
|
||||
start_ms = int((before - timedelta(minutes=1)).timestamp() * 1000)
|
||||
end_ms = int((before + timedelta(minutes=1)).timestamp() * 1000)
|
||||
|
||||
sql = get_preview_sql(make_preview_query_request(signoz, token, start_ms=start_ms, end_ms=end_ms, queries=queries), "A")
|
||||
assert "attributes_string['unreported.key']" in sql
|
||||
assert "attributes.`unreported.key`" not in sql
|
||||
|
||||
response = make_query_request(signoz, token, start_ms=start_ms, end_ms=end_ms, queries=queries)
|
||||
assert response.status_code == HTTPStatus.OK
|
||||
series = response.json()["data"]["data"]["results"][0]["aggregations"][0]["series"]
|
||||
assert [value["value"] for value in series[0]["values"]] == [1]
|
||||
assert any("not found in metadata" in warning.get("message", "") for warning in get_all_warnings(response.json()))
|
||||
|
||||
# after the rollout -> attributes JSON column
|
||||
start_ms = int((after - timedelta(minutes=1)).timestamp() * 1000)
|
||||
end_ms = int((after + timedelta(minutes=1)).timestamp() * 1000)
|
||||
|
||||
sql = get_preview_sql(make_preview_query_request(signoz, token, start_ms=start_ms, end_ms=end_ms, queries=queries), "A")
|
||||
assert "attributes.`unreported.key`" in sql
|
||||
assert "attributes_string" not in sql
|
||||
|
||||
response = make_query_request(signoz, token, start_ms=start_ms, end_ms=end_ms, queries=queries)
|
||||
assert response.status_code == HTTPStatus.OK
|
||||
series = response.json()["data"]["data"]["results"][0]["aggregations"][0]["series"]
|
||||
assert [value["value"] for value in series[0]["values"]] == [1]
|
||||
assert any("not found in metadata" in warning.get("message", "") for warning in get_all_warnings(response.json()))
|
||||
|
||||
Reference in New Issue
Block a user