mirror of
https://github.com/SigNoz/signoz.git
synced 2026-08-12 16:00:46 +01:00
Compare commits
4 Commits
fix/uplot-
...
feat/semco
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
a0e5c7c126 | ||
|
|
78c4f5a7e4 | ||
|
|
09bdda1280 | ||
|
|
7b9e9207ea |
8
Makefile
8
Makefile
@@ -220,6 +220,14 @@ py-test-teardown: ## Tear down the shared SigNoz backend
|
||||
py-test: ## Runs integration tests
|
||||
@cd tests && uv run pytest --basetemp=./tmp/ -vv --capture=no integration/tests/
|
||||
|
||||
.PHONY: py-test-semconv-phase1
|
||||
py-test-semconv-phase1: py-test-setup ## Rebuild the shared stack and run the semantic-convention Phase 1 matrix
|
||||
@cd tests && uv run pytest --basetemp=./tmp/ -vv --reuse --capture=no integration/tests/queriertraces/13_semconv_evolution.py
|
||||
|
||||
.PHONY: py-test-semconv-phase2
|
||||
py-test-semconv-phase2: py-test-setup ## Rebuild the shared stack and run the Phase 1-2 cross-signal matrices
|
||||
@cd tests && uv run pytest --basetemp=./tmp/ -vv --reuse --capture=no integration/tests/queriertraces/13_semconv_evolution.py integration/tests/queriersemconv/02_cross_signal.py
|
||||
|
||||
.PHONY: py-clean
|
||||
py-clean: ## Clear all pycache and pytest cache from tests directory recursively
|
||||
@echo ">> cleaning python cache files from tests directory"
|
||||
|
||||
@@ -11,12 +11,21 @@ export type SemconvFamily = {
|
||||
};
|
||||
|
||||
export const SEMCONV_FAMILIES: readonly SemconvFamily[] = [
|
||||
{
|
||||
current: 'container.cpu.usage',
|
||||
old: ['container.cpu.utilization'],
|
||||
kind: 'metric',
|
||||
contexts: ['metric'],
|
||||
signals: ['metrics'],
|
||||
applyToMetrics: [],
|
||||
valueMap: {},
|
||||
},
|
||||
{
|
||||
current: 'db.system.name',
|
||||
old: ['db.system'],
|
||||
kind: 'attribute',
|
||||
contexts: [],
|
||||
signals: [],
|
||||
contexts: ['attribute', 'resource'],
|
||||
signals: ['logs', 'metrics', 'traces'],
|
||||
applyToMetrics: [],
|
||||
valueMap: {},
|
||||
},
|
||||
@@ -24,8 +33,26 @@ export const SEMCONV_FAMILIES: readonly SemconvFamily[] = [
|
||||
current: 'deployment.environment.name',
|
||||
old: ['deployment.environment'],
|
||||
kind: 'attribute',
|
||||
contexts: [],
|
||||
signals: [],
|
||||
contexts: ['attribute', 'resource'],
|
||||
signals: ['logs', 'metrics', 'traces'],
|
||||
applyToMetrics: [],
|
||||
valueMap: {},
|
||||
},
|
||||
{
|
||||
current: 'k8s.node.cpu.usage',
|
||||
old: ['k8s.node.cpu.utilization'],
|
||||
kind: 'metric',
|
||||
contexts: ['metric'],
|
||||
signals: ['metrics'],
|
||||
applyToMetrics: [],
|
||||
valueMap: {},
|
||||
},
|
||||
{
|
||||
current: 'k8s.pod.cpu.usage',
|
||||
old: ['k8s.pod.cpu.utilization'],
|
||||
kind: 'metric',
|
||||
contexts: ['metric'],
|
||||
signals: ['metrics'],
|
||||
applyToMetrics: [],
|
||||
valueMap: {},
|
||||
},
|
||||
|
||||
@@ -52,10 +52,10 @@ import (
|
||||
"github.com/SigNoz/signoz/pkg/query-service/constants"
|
||||
|
||||
chErrors "github.com/SigNoz/signoz/pkg/query-service/errors"
|
||||
"github.com/SigNoz/signoz/pkg/query-service/metrics"
|
||||
"github.com/SigNoz/signoz/pkg/query-service/model"
|
||||
v3 "github.com/SigNoz/signoz/pkg/query-service/model/v3"
|
||||
"github.com/SigNoz/signoz/pkg/query-service/utils"
|
||||
"github.com/SigNoz/signoz/pkg/semconv"
|
||||
)
|
||||
|
||||
const (
|
||||
@@ -3190,11 +3190,6 @@ func (r *ClickHouseReader) GetMetricAttributeValues(ctx context.Context, orgID v
|
||||
var rows driver.Rows
|
||||
var attributeValues v3.FilterAttributeValueResponse
|
||||
|
||||
normalized := true
|
||||
if constants.IsDotMetricsEnabled {
|
||||
normalized = false
|
||||
}
|
||||
|
||||
reductionEnabled := r.fl.BooleanOrEmpty(ctx, flagger.FeatureEnableMetricsReduction, featuretypes.NewFlaggerEvaluationContext(orgID))
|
||||
|
||||
if reductionEnabled {
|
||||
@@ -3205,8 +3200,7 @@ func (r *ClickHouseReader) GetMetricAttributeValues(ctx context.Context, orgID v
|
||||
if req.Limit != 0 {
|
||||
query = query + fmt.Sprintf(" LIMIT %d;", req.Limit)
|
||||
}
|
||||
names := []string{req.AggregateAttribute}
|
||||
names = append(names, metrics.GetTransitionedMetric(req.AggregateAttribute, normalized))
|
||||
names := semconv.MetricNames(req.AggregateAttribute)
|
||||
|
||||
rows, err = r.db.Query(ctx, query, req.FilterAttributeKey, names, req.FilterAttributeKey, fmt.Sprintf("%%%s%%", req.SearchText), common.PastDayRoundOff())
|
||||
|
||||
|
||||
@@ -6,6 +6,8 @@ import (
|
||||
|
||||
v3 "github.com/SigNoz/signoz/pkg/query-service/model/v3"
|
||||
"github.com/SigNoz/signoz/pkg/query-service/utils"
|
||||
"github.com/SigNoz/signoz/pkg/semconv"
|
||||
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
|
||||
)
|
||||
|
||||
var resourceLogOperators = map[v3.FilterOperator]string{
|
||||
@@ -29,13 +31,61 @@ var resourceLogOperators = map[v3.FilterOperator]string{
|
||||
v3.FilterOperatorNotILike: "NOT ILIKE",
|
||||
}
|
||||
|
||||
func resourceSemconvMembers(key string) []string {
|
||||
return semconv.Members(semconv.KindAttribute, telemetrytypes.FieldKeySelector{
|
||||
Name: key,
|
||||
Signal: telemetrytypes.SignalTraces,
|
||||
FieldContext: telemetrytypes.FieldContextResource,
|
||||
})
|
||||
}
|
||||
|
||||
func resourceValueExpression(key string) string {
|
||||
members := resourceSemconvMembers(key)
|
||||
if len(members) == 1 {
|
||||
return fmt.Sprintf("simpleJSONExtractString(labels, '%s')", key)
|
||||
}
|
||||
|
||||
values := make([]string, 0, len(members))
|
||||
for _, member := range members {
|
||||
values = append(values, fmt.Sprintf("NULLIF(simpleJSONExtractString(labels, '%s'), '')", member))
|
||||
}
|
||||
return "COALESCE(" + strings.Join(values, ", ") + ")"
|
||||
}
|
||||
|
||||
func resourcePresenceExpression(key string, exists bool) string {
|
||||
members := resourceSemconvMembers(key)
|
||||
if len(members) == 1 {
|
||||
if exists {
|
||||
return fmt.Sprintf("simpleJSONHas(labels, '%s')", key)
|
||||
}
|
||||
return fmt.Sprintf("not simpleJSONHas(labels, '%s')", key)
|
||||
}
|
||||
|
||||
conditions := make([]string, 0, len(members))
|
||||
for _, member := range members {
|
||||
if exists {
|
||||
conditions = append(conditions, fmt.Sprintf("simpleJSONHas(labels, '%s')", member))
|
||||
} else {
|
||||
conditions = append(conditions, fmt.Sprintf("not simpleJSONHas(labels, '%s')", member))
|
||||
}
|
||||
}
|
||||
separator := " OR "
|
||||
if !exists {
|
||||
separator = " AND "
|
||||
}
|
||||
return "(" + strings.Join(conditions, separator) + ")"
|
||||
}
|
||||
|
||||
// buildResourceFilter builds a clickhouse filter string for resource labels
|
||||
func buildResourceFilter(logsOp string, key string, op v3.FilterOperator, value interface{}) string {
|
||||
// for all operators except contains and like
|
||||
searchKey := fmt.Sprintf("simpleJSONExtractString(labels, '%s')", key)
|
||||
searchKey := resourceValueExpression(key)
|
||||
|
||||
// for contains and like it will be case insensitive
|
||||
lowerSearchKey := fmt.Sprintf("simpleJSONExtractString(lower(labels), '%s')", key)
|
||||
if len(resourceSemconvMembers(key)) > 1 {
|
||||
lowerSearchKey = "lower(" + searchKey + ")"
|
||||
}
|
||||
|
||||
chFmtVal := utils.ClickHouseFormattedValue(value)
|
||||
|
||||
@@ -43,9 +93,9 @@ func buildResourceFilter(logsOp string, key string, op v3.FilterOperator, value
|
||||
|
||||
switch op {
|
||||
case v3.FilterOperatorExists:
|
||||
return fmt.Sprintf("simpleJSONHas(labels, '%s')", key)
|
||||
return resourcePresenceExpression(key, true)
|
||||
case v3.FilterOperatorNotExists:
|
||||
return fmt.Sprintf("not simpleJSONHas(labels, '%s')", key)
|
||||
return resourcePresenceExpression(key, false)
|
||||
case v3.FilterOperatorRegex, v3.FilterOperatorNotRegex:
|
||||
return fmt.Sprintf(logsOp, searchKey, chFmtVal)
|
||||
case v3.FilterOperatorContains, v3.FilterOperatorNotContains:
|
||||
@@ -110,6 +160,38 @@ func buildIndexFilterForInOperator(key string, op v3.FilterOperator, value inter
|
||||
// we can use lower index for =, in etc but it's difficult to do it for !=, NIN etc
|
||||
// if as x != "ABC" we cannot predict something like "not lower(labels) like '%%x%%abc%%'". It has it be "not lower(labels) like '%%x%%ABC%%'"
|
||||
func buildResourceIndexFilter(key string, op v3.FilterOperator, value interface{}) string {
|
||||
return buildResourceIndexFilterForKey(key, op, value, true)
|
||||
}
|
||||
|
||||
func buildResourceIndexFilterForKey(key string, op v3.FilterOperator, value interface{}, resolveFamily bool) string {
|
||||
members := []string{key}
|
||||
if resolveFamily {
|
||||
members = resourceSemconvMembers(key)
|
||||
}
|
||||
if len(members) > 1 {
|
||||
switch op {
|
||||
case v3.FilterOperatorNotEqual,
|
||||
v3.FilterOperatorNotLike,
|
||||
v3.FilterOperatorNotILike,
|
||||
v3.FilterOperatorNotContains,
|
||||
v3.FilterOperatorNotExists,
|
||||
v3.FilterOperatorNotRegex,
|
||||
v3.FilterOperatorNotIn:
|
||||
return ""
|
||||
}
|
||||
|
||||
conditions := make([]string, 0, len(members))
|
||||
for _, member := range members {
|
||||
if condition := buildResourceIndexFilterForKey(member, op, value, false); condition != "" {
|
||||
conditions = append(conditions, condition)
|
||||
}
|
||||
}
|
||||
if len(conditions) == 0 {
|
||||
return ""
|
||||
}
|
||||
return "(" + strings.Join(conditions, " OR ") + ")"
|
||||
}
|
||||
|
||||
// not using clickhouseFormattedValue as we don't wan't the quotes
|
||||
strVal := fmt.Sprintf("%s", value)
|
||||
fmtValEscapedForContains := utils.QuoteEscapedStringForContains(strVal, true)
|
||||
@@ -206,14 +288,31 @@ func buildResourceFiltersFromGroupBy(groupBy []v3.AttributeKey) []string {
|
||||
if attr.Type != v3.AttributeKeyTypeResource {
|
||||
continue
|
||||
}
|
||||
conditions = append(conditions, fmt.Sprintf("(simpleJSONHas(labels, '%s') AND labels like '%%%s%%')", attr.Key, attr.Key))
|
||||
members := resourceSemconvMembers(attr.Key)
|
||||
if len(members) == 1 {
|
||||
conditions = append(conditions, fmt.Sprintf("(simpleJSONHas(labels, '%s') AND labels like '%%%s%%')", attr.Key, attr.Key))
|
||||
continue
|
||||
}
|
||||
indexConditions := make([]string, 0, len(members))
|
||||
for _, member := range members {
|
||||
indexConditions = append(indexConditions, fmt.Sprintf("labels like '%%%s%%'", member))
|
||||
}
|
||||
conditions = append(conditions, fmt.Sprintf("(%s AND (%s))", resourcePresenceExpression(attr.Key, true), strings.Join(indexConditions, " OR ")))
|
||||
}
|
||||
return conditions
|
||||
}
|
||||
|
||||
func buildResourceFiltersFromAggregateAttribute(aggregateAttribute v3.AttributeKey) string {
|
||||
if aggregateAttribute.Key != "" && aggregateAttribute.Type == v3.AttributeKeyTypeResource {
|
||||
return fmt.Sprintf("(simpleJSONHas(labels, '%s') AND labels like '%%%s%%')", aggregateAttribute.Key, aggregateAttribute.Key)
|
||||
members := resourceSemconvMembers(aggregateAttribute.Key)
|
||||
if len(members) == 1 {
|
||||
return fmt.Sprintf("(simpleJSONHas(labels, '%s') AND labels like '%%%s%%')", aggregateAttribute.Key, aggregateAttribute.Key)
|
||||
}
|
||||
indexConditions := make([]string, 0, len(members))
|
||||
for _, member := range members {
|
||||
indexConditions = append(indexConditions, fmt.Sprintf("labels like '%%%s%%'", member))
|
||||
}
|
||||
return fmt.Sprintf("(%s AND (%s))", resourcePresenceExpression(aggregateAttribute.Key, true), strings.Join(indexConditions, " OR "))
|
||||
}
|
||||
|
||||
return ""
|
||||
|
||||
@@ -5,6 +5,8 @@ import (
|
||||
"testing"
|
||||
|
||||
v3 "github.com/SigNoz/signoz/pkg/query-service/model/v3"
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
func Test_buildResourceFilter(t *testing.T) {
|
||||
@@ -552,3 +554,38 @@ func Test_buildResourceSubQuery(t *testing.T) {
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestSemanticConventionResourceFamily(t *testing.T) {
|
||||
const resolvedValue = "COALESCE(NULLIF(simpleJSONExtractString(labels, 'deployment.environment.name'), ''), NULLIF(simpleJSONExtractString(labels, 'deployment.environment'), ''))"
|
||||
|
||||
for _, requestedName := range []string{"deployment.environment.name", "deployment.environment"} {
|
||||
t.Run(requestedName, func(t *testing.T) {
|
||||
assert.Equal(t, resolvedValue+" = 'production'", buildResourceFilter("=", requestedName, v3.FilterOperatorEqual, "production"))
|
||||
assert.Equal(t, "(simpleJSONHas(labels, 'deployment.environment.name') OR simpleJSONHas(labels, 'deployment.environment'))", buildResourceFilter("", requestedName, v3.FilterOperatorExists, nil))
|
||||
assert.Equal(t, "(not simpleJSONHas(labels, 'deployment.environment.name') AND not simpleJSONHas(labels, 'deployment.environment'))", buildResourceFilter("", requestedName, v3.FilterOperatorNotExists, nil))
|
||||
assert.Equal(t, "(labels like '%deployment.environment.name\":\"production%' OR labels like '%deployment.environment\":\"production%')", buildResourceIndexFilter(requestedName, v3.FilterOperatorEqual, "production"))
|
||||
assert.Empty(t, buildResourceIndexFilter(requestedName, v3.FilterOperatorNotEqual, "production"), "negative family filter must not use a rejecting index hint")
|
||||
})
|
||||
}
|
||||
|
||||
filters, err := buildResourceFiltersFromFilterItems(&v3.FilterSet{Items: []v3.FilterItem{{
|
||||
Key: v3.AttributeKey{
|
||||
Key: "deployment.environment.name",
|
||||
DataType: v3.AttributeKeyDataTypeString,
|
||||
Type: v3.AttributeKeyTypeResource,
|
||||
},
|
||||
Operator: v3.FilterOperatorEqual,
|
||||
Value: "production",
|
||||
}}})
|
||||
require.NoError(t, err, "family filter items must build before their output is inspected")
|
||||
wantFilters := []string{
|
||||
resolvedValue + " = 'production'",
|
||||
"(labels like '%deployment.environment.name\":\"production%' OR labels like '%deployment.environment\":\"production%')",
|
||||
}
|
||||
assert.Equal(t, wantFilters, filters)
|
||||
|
||||
wantPresence := "((simpleJSONHas(labels, 'deployment.environment.name') OR simpleJSONHas(labels, 'deployment.environment')) AND (labels like '%deployment.environment.name%' OR labels like '%deployment.environment%'))"
|
||||
groupBy := buildResourceFiltersFromGroupBy([]v3.AttributeKey{{Key: "deployment.environment", Type: v3.AttributeKeyTypeResource}})
|
||||
assert.Equal(t, []string{wantPresence}, groupBy)
|
||||
assert.Equal(t, wantPresence, buildResourceFiltersFromAggregateAttribute(v3.AttributeKey{Key: "deployment.environment.name", Type: v3.AttributeKeyTypeResource}))
|
||||
}
|
||||
|
||||
@@ -6,16 +6,32 @@ import (
|
||||
|
||||
"github.com/ClickHouse/clickhouse-go/v2"
|
||||
"github.com/SigNoz/signoz/pkg/query-service/model"
|
||||
"github.com/SigNoz/signoz/pkg/semconv"
|
||||
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
|
||||
)
|
||||
|
||||
var (
|
||||
columns = map[string]struct{}{
|
||||
"deployment_environment": {},
|
||||
"k8s_cluster_name": {},
|
||||
"k8s_namespace_name": {},
|
||||
}
|
||||
columns = serviceMapColumns()
|
||||
)
|
||||
|
||||
func serviceMapColumns() map[string]string {
|
||||
columns := map[string]string{
|
||||
"k8s_cluster_name": "k8s_cluster_name",
|
||||
"k8s_namespace_name": "k8s_namespace_name",
|
||||
}
|
||||
|
||||
// Dependency-graph rows keep their historical physical column name. Both
|
||||
// semantic-convention request spellings target that same derived column.
|
||||
for _, member := range semconv.Members(semconv.KindAttribute, telemetrytypes.FieldKeySelector{
|
||||
Name: "deployment.environment.name",
|
||||
Signal: telemetrytypes.SignalTraces,
|
||||
FieldContext: telemetrytypes.FieldContextResource,
|
||||
}) {
|
||||
columns[strings.ReplaceAll(member, ".", "_")] = "deployment_environment"
|
||||
}
|
||||
return columns
|
||||
}
|
||||
|
||||
func BuildServiceMapQuery(tags []model.TagQuery) (string, []interface{}) {
|
||||
var filterQuery string
|
||||
var namedArgs []interface{}
|
||||
@@ -24,39 +40,40 @@ func BuildServiceMapQuery(tags []model.TagQuery) (string, []interface{}) {
|
||||
operator := tag.GetOperator()
|
||||
value := tag.GetValues()
|
||||
|
||||
if _, ok := columns[key]; !ok {
|
||||
column, ok := columns[key]
|
||||
if !ok {
|
||||
continue
|
||||
}
|
||||
|
||||
switch operator {
|
||||
case model.InOperator:
|
||||
filterQuery += fmt.Sprintf(" AND %s IN @%s", key, key)
|
||||
filterQuery += fmt.Sprintf(" AND %s IN @%s", column, key)
|
||||
namedArgs = append(namedArgs, clickhouse.Named(key, value))
|
||||
case model.NotInOperator:
|
||||
filterQuery += fmt.Sprintf(" AND %s NOT IN @%s", key, key)
|
||||
filterQuery += fmt.Sprintf(" AND %s NOT IN @%s", column, key)
|
||||
namedArgs = append(namedArgs, clickhouse.Named(key, value))
|
||||
case model.EqualOperator:
|
||||
filterQuery += fmt.Sprintf(" AND %s = @%s", key, key)
|
||||
filterQuery += fmt.Sprintf(" AND %s = @%s", column, key)
|
||||
namedArgs = append(namedArgs, clickhouse.Named(key, value))
|
||||
case model.NotEqualOperator:
|
||||
filterQuery += fmt.Sprintf(" AND %s != @%s", key, key)
|
||||
filterQuery += fmt.Sprintf(" AND %s != @%s", column, key)
|
||||
namedArgs = append(namedArgs, clickhouse.Named(key, value))
|
||||
case model.ContainsOperator:
|
||||
filterQuery += fmt.Sprintf(" AND %s LIKE @%s", key, key)
|
||||
filterQuery += fmt.Sprintf(" AND %s LIKE @%s", column, key)
|
||||
namedArgs = append(namedArgs, clickhouse.Named(key, fmt.Sprintf("%%%s%%", value)))
|
||||
case model.NotContainsOperator:
|
||||
filterQuery += fmt.Sprintf(" AND %s NOT LIKE @%s", key, key)
|
||||
filterQuery += fmt.Sprintf(" AND %s NOT LIKE @%s", column, key)
|
||||
namedArgs = append(namedArgs, clickhouse.Named(key, fmt.Sprintf("%%%s%%", value)))
|
||||
case model.StartsWithOperator:
|
||||
filterQuery += fmt.Sprintf(" AND %s LIKE @%s", key, key)
|
||||
filterQuery += fmt.Sprintf(" AND %s LIKE @%s", column, key)
|
||||
namedArgs = append(namedArgs, clickhouse.Named(key, fmt.Sprintf("%s%%", value)))
|
||||
case model.NotStartsWithOperator:
|
||||
filterQuery += fmt.Sprintf(" AND %s NOT LIKE @%s", key, key)
|
||||
filterQuery += fmt.Sprintf(" AND %s NOT LIKE @%s", column, key)
|
||||
namedArgs = append(namedArgs, clickhouse.Named(key, fmt.Sprintf("%s%%", value)))
|
||||
case model.ExistsOperator:
|
||||
filterQuery += fmt.Sprintf(" AND %s IS NOT NULL", key)
|
||||
filterQuery += fmt.Sprintf(" AND %s IS NOT NULL", column)
|
||||
case model.NotExistsOperator:
|
||||
filterQuery += fmt.Sprintf(" AND %s IS NULL", key)
|
||||
filterQuery += fmt.Sprintf(" AND %s IS NULL", column)
|
||||
}
|
||||
}
|
||||
return filterQuery, namedArgs
|
||||
|
||||
35
pkg/query-service/app/services/map_test.go
Normal file
35
pkg/query-service/app/services/map_test.go
Normal file
@@ -0,0 +1,35 @@
|
||||
package services
|
||||
|
||||
import (
|
||||
"testing"
|
||||
|
||||
"github.com/ClickHouse/clickhouse-go/v2/lib/driver"
|
||||
"github.com/SigNoz/signoz/pkg/query-service/model"
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
func TestBuildServiceMapQueryAcceptsEnvironmentFamily(t *testing.T) {
|
||||
for _, requestedName := range []string{"deployment.environment.name", "deployment.environment"} {
|
||||
t.Run(requestedName, func(t *testing.T) {
|
||||
tags := []model.TagQuery{model.NewTagQueryString(model.TagQueryParam{
|
||||
Key: requestedName,
|
||||
StringValues: []string{"production"},
|
||||
Operator: model.InOperator,
|
||||
TagType: model.ResourceAttributeTagType,
|
||||
})}
|
||||
|
||||
query, args := BuildServiceMapQuery(tags)
|
||||
argName := "deployment_environment"
|
||||
if requestedName == "deployment.environment.name" {
|
||||
argName = "deployment_environment_name"
|
||||
}
|
||||
assert.Equal(t, " AND deployment_environment IN @"+argName, query)
|
||||
require.Len(t, args, 1)
|
||||
named, ok := args[0].(driver.NamedValue)
|
||||
require.True(t, ok)
|
||||
assert.Equal(t, argName, named.Name)
|
||||
assert.Equal(t, []interface{}{"production"}, named.Value)
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -1,27 +0,0 @@
|
||||
package metrics
|
||||
|
||||
var MetricsUnderTransition = map[string]string{
|
||||
"k8s_pod_cpu_utilization": "k8s_pod_cpu_usage",
|
||||
"k8s_node_cpu_utilization": "k8s_node_cpu_usage",
|
||||
"container_cpu_utilization": "container_cpu_usage",
|
||||
}
|
||||
|
||||
var DotMetricsUnderTransition = map[string]string{
|
||||
"k8s.pod.cpu.utilization": "k8s.pod.cpu.usage",
|
||||
"k8s.node.cpu.utilization": "k8s.node.cpu.usage",
|
||||
"container.cpu.utilization": "container.cpu.usage",
|
||||
}
|
||||
|
||||
func GetTransitionedMetric(metric string, normalized bool) string {
|
||||
if normalized {
|
||||
if _, ok := MetricsUnderTransition[metric]; ok {
|
||||
return MetricsUnderTransition[metric]
|
||||
}
|
||||
return metric
|
||||
} else {
|
||||
if _, ok := DotMetricsUnderTransition[metric]; ok {
|
||||
return DotMetricsUnderTransition[metric]
|
||||
}
|
||||
return metric
|
||||
}
|
||||
}
|
||||
@@ -10,8 +10,8 @@ import (
|
||||
"log/slog"
|
||||
|
||||
"github.com/SigNoz/signoz/pkg/query-service/constants"
|
||||
"github.com/SigNoz/signoz/pkg/query-service/metrics"
|
||||
v3 "github.com/SigNoz/signoz/pkg/query-service/model/v3"
|
||||
"github.com/SigNoz/signoz/pkg/semconv"
|
||||
)
|
||||
|
||||
// ValidateAndCastValue validates and casts the value of a key to the corresponding data type of the key
|
||||
@@ -234,12 +234,12 @@ func ClickHouseFormattedValue(v interface{}) string {
|
||||
|
||||
func ClickHouseFormattedMetricNames(v interface{}) string {
|
||||
if name, ok := v.(string); ok {
|
||||
transitionedMetrics := metrics.GetTransitionedMetric(name, !constants.IsDotMetricsEnabled)
|
||||
if transitionedMetrics != name {
|
||||
return ClickHouseFormattedValue([]interface{}{transitionedMetrics})
|
||||
} else {
|
||||
return ClickHouseFormattedValue([]interface{}{name})
|
||||
members := semconv.MetricNames(name)
|
||||
values := make([]interface{}, 0, len(members))
|
||||
for _, member := range members {
|
||||
values = append(values, member)
|
||||
}
|
||||
return ClickHouseFormattedValue(values)
|
||||
}
|
||||
|
||||
return ClickHouseFormattedValue(v)
|
||||
|
||||
@@ -2,13 +2,29 @@ package querybuilder
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"strings"
|
||||
|
||||
schema "github.com/SigNoz/signoz-otel-collector/cmd/signozschemamigrator/schema_migrator"
|
||||
"github.com/SigNoz/signoz/pkg/errors"
|
||||
"github.com/SigNoz/signoz/pkg/semconv"
|
||||
qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
|
||||
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
|
||||
)
|
||||
|
||||
func physicalSemconvMembers(key *telemetrytypes.TelemetryFieldKey) []string {
|
||||
if len(key.SemconvMembers) > 0 {
|
||||
return key.SemconvMembers
|
||||
}
|
||||
if key.FieldContext != telemetrytypes.FieldContextResource && key.FieldContext != telemetrytypes.FieldContextAttribute {
|
||||
return []string{key.Name}
|
||||
}
|
||||
return semconv.AttributeMembers(telemetrytypes.FieldKeySelector{
|
||||
Name: key.Name,
|
||||
Signal: key.Signal,
|
||||
FieldContext: key.FieldContext,
|
||||
})
|
||||
}
|
||||
|
||||
// ExistsExpression renders the existence predicate for a key resolved to the given
|
||||
// columns (negated when exists is false). Comparisons are against constants rendered
|
||||
// as literals, so the expression carries no bind args and can guard column expressions
|
||||
@@ -43,11 +59,26 @@ func ExistsExpression(columns []*schema.Column, key *telemetrytypes.TelemetryFie
|
||||
if len(evolutionsEntries) > 0 && evolutionsEntries[0] != nil {
|
||||
columnName = evolutionsEntries[0].ColumnName
|
||||
}
|
||||
rawPath := fmt.Sprintf("%s.`%s`", columnName, key.Name)
|
||||
if exists {
|
||||
return rawPath + " IS NOT NULL", nil
|
||||
members := physicalSemconvMembers(key)
|
||||
paths := make([]string, 0, len(members))
|
||||
for _, member := range members {
|
||||
paths = append(paths, fmt.Sprintf("%s.`%s`", columnName, member))
|
||||
}
|
||||
return rawPath + " IS NULL", nil
|
||||
if len(paths) == 1 {
|
||||
if exists {
|
||||
return paths[0] + " IS NOT NULL", nil
|
||||
}
|
||||
return paths[0] + " IS NULL", nil
|
||||
}
|
||||
guards := make([]string, 0, len(paths))
|
||||
for _, path := range paths {
|
||||
guards = append(guards, path+" IS NOT NULL")
|
||||
}
|
||||
rawPath := "(" + strings.Join(guards, " OR ") + ")"
|
||||
if exists {
|
||||
return rawPath, nil
|
||||
}
|
||||
return "NOT " + rawPath, nil
|
||||
case schema.ColumnTypeEnumString,
|
||||
schema.ColumnTypeEnumFixedString:
|
||||
if exists {
|
||||
@@ -88,8 +119,16 @@ func ExistsExpression(columns []*schema.Column, key *telemetrytypes.TelemetryFie
|
||||
|
||||
switch valueType := column.Type.(schema.MapColumnType).ValueType; valueType.GetType() {
|
||||
case schema.ColumnTypeEnumString, schema.ColumnTypeEnumBool, schema.ColumnTypeEnumFloat64:
|
||||
leftOperand := fmt.Sprintf("mapContains(%s, '%s')", column.Name, key.Name)
|
||||
if key.Materialized {
|
||||
members := physicalSemconvMembers(key)
|
||||
operands := make([]string, 0, len(members))
|
||||
for _, member := range members {
|
||||
operands = append(operands, fmt.Sprintf("mapContains(%s, '%s')", column.Name, member))
|
||||
}
|
||||
leftOperand := strings.Join(operands, " OR ")
|
||||
if len(operands) > 1 {
|
||||
leftOperand = "(" + leftOperand + ")"
|
||||
}
|
||||
if key.Materialized && (len(members) == 1 || key.MaterializedSemconv) {
|
||||
leftOperand = telemetrytypes.FieldKeyToMaterializedColumnNameForExists(key)
|
||||
}
|
||||
if exists {
|
||||
|
||||
82
pkg/querybuilder/semconv_materialization_test.go
Normal file
82
pkg/querybuilder/semconv_materialization_test.go
Normal file
@@ -0,0 +1,82 @@
|
||||
package querybuilder_test
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
|
||||
"github.com/SigNoz/signoz/pkg/querybuilder"
|
||||
"github.com/SigNoz/signoz/pkg/telemetryschema/tracestelemetryschema"
|
||||
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
|
||||
"github.com/SigNoz/signoz/pkg/valuer"
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
func TestTraceFamilyUsesMaterializedHistoricalMember(t *testing.T) {
|
||||
current := &telemetrytypes.TelemetryFieldKey{
|
||||
Name: "deployment.environment.name",
|
||||
Signal: telemetrytypes.SignalTraces,
|
||||
FieldContext: telemetrytypes.FieldContextAttribute,
|
||||
FieldDataType: telemetrytypes.FieldDataTypeString,
|
||||
}
|
||||
historical := &telemetrytypes.TelemetryFieldKey{
|
||||
Name: "deployment.environment",
|
||||
Signal: telemetrytypes.SignalTraces,
|
||||
FieldContext: telemetrytypes.FieldContextAttribute,
|
||||
FieldDataType: telemetrytypes.FieldDataTypeString,
|
||||
Materialized: true,
|
||||
}
|
||||
requested := telemetrytypes.NewTelemetryFieldKey(
|
||||
current.Name,
|
||||
telemetrytypes.FieldContextAttribute,
|
||||
telemetrytypes.FieldDataTypeString,
|
||||
)
|
||||
|
||||
matches := querybuilder.MatchingFieldKeys(requested, map[string][]*telemetrytypes.TelemetryFieldKey{
|
||||
current.Name: {current},
|
||||
historical.Name: {historical},
|
||||
})
|
||||
require.Len(t, matches, 1, "family metadata should resolve to one logical field")
|
||||
expression, err := tracestelemetryschema.NewFieldMapper().FieldFor(context.Background(), valuer.UUID{}, 0, 0, matches[0])
|
||||
require.NoError(t, err, "resolved trace family should map to a value expression")
|
||||
|
||||
assert.Equal(
|
||||
t,
|
||||
"COALESCE(NULLIF(attributes_string['deployment.environment.name'], ''), NULLIF(`attribute_string_deployment$$environment`, ''), '')",
|
||||
expression,
|
||||
"family expression should retain the promoted historical member",
|
||||
)
|
||||
}
|
||||
|
||||
func TestTraceFamilyUsesFamilyAwareMaterializedColumn(t *testing.T) {
|
||||
current := &telemetrytypes.TelemetryFieldKey{
|
||||
Name: "db.system.name",
|
||||
Signal: telemetrytypes.SignalTraces,
|
||||
FieldContext: telemetrytypes.FieldContextAttribute,
|
||||
FieldDataType: telemetrytypes.FieldDataTypeString,
|
||||
Materialized: true,
|
||||
MaterializedColumnName: "attribute_string_db$$system",
|
||||
MaterializedSemconv: true,
|
||||
}
|
||||
historical := &telemetrytypes.TelemetryFieldKey{
|
||||
Name: "db.system",
|
||||
Signal: telemetrytypes.SignalTraces,
|
||||
FieldContext: telemetrytypes.FieldContextAttribute,
|
||||
FieldDataType: telemetrytypes.FieldDataTypeString,
|
||||
}
|
||||
requested := telemetrytypes.NewTelemetryFieldKey(
|
||||
current.Name,
|
||||
telemetrytypes.FieldContextAttribute,
|
||||
telemetrytypes.FieldDataTypeString,
|
||||
)
|
||||
|
||||
matches := querybuilder.MatchingFieldKeys(requested, map[string][]*telemetrytypes.TelemetryFieldKey{
|
||||
current.Name: {current},
|
||||
historical.Name: {historical},
|
||||
})
|
||||
require.Len(t, matches, 1, "family metadata should resolve to one logical field")
|
||||
expression, err := tracestelemetryschema.NewFieldMapper().FieldFor(context.Background(), valuer.UUID{}, 0, 0, matches[0])
|
||||
require.NoError(t, err, "resolved trace family should map to a value expression")
|
||||
|
||||
assert.Equal(t, "`attribute_string_db$$system`", expression, "family-aware promoted column should serve the complete family")
|
||||
}
|
||||
@@ -4,12 +4,14 @@ import (
|
||||
"context"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"maps"
|
||||
"slices"
|
||||
"strconv"
|
||||
"strings"
|
||||
|
||||
"github.com/SigNoz/signoz/pkg/errors"
|
||||
grammar "github.com/SigNoz/signoz/pkg/parser/filterquery/grammar"
|
||||
"github.com/SigNoz/signoz/pkg/semconv"
|
||||
qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
|
||||
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
|
||||
"github.com/SigNoz/signoz/pkg/valuer"
|
||||
@@ -38,6 +40,7 @@ type filterExpressionVisitor struct {
|
||||
fullTextColumn *telemetrytypes.TelemetryFieldKey
|
||||
skipResourceFilter bool
|
||||
skipFullTextFilter bool
|
||||
exactSemconv bool
|
||||
variables map[string]qbtypes.VariableItem
|
||||
|
||||
keysWithWarnings map[string]bool
|
||||
@@ -58,6 +61,7 @@ type FilterExprVisitorOpts struct {
|
||||
FullTextColumn *telemetrytypes.TelemetryFieldKey
|
||||
SkipResourceFilter bool
|
||||
SkipFullTextFilter bool
|
||||
ExactSemconv bool
|
||||
Variables map[string]qbtypes.VariableItem
|
||||
StartNs uint64
|
||||
EndNs uint64
|
||||
@@ -75,6 +79,7 @@ func newFilterExpressionVisitor(opts FilterExprVisitorOpts) *filterExpressionVis
|
||||
fullTextColumn: opts.FullTextColumn,
|
||||
skipResourceFilter: opts.SkipResourceFilter,
|
||||
skipFullTextFilter: opts.SkipFullTextFilter,
|
||||
exactSemconv: opts.ExactSemconv,
|
||||
variables: opts.Variables,
|
||||
keysWithWarnings: make(map[string]bool),
|
||||
startNs: opts.StartNs,
|
||||
@@ -379,7 +384,7 @@ func (v *filterExpressionVisitor) VisitPrimary(ctx *grammar.PrimaryContext) any
|
||||
// VisitComparison handles all comparison operators.
|
||||
func (v *filterExpressionVisitor) VisitComparison(ctx *grammar.ComparisonContext) any {
|
||||
key := v.Visit(ctx.Key()).(*telemetrytypes.TelemetryFieldKey)
|
||||
matching := MatchingFieldKeys(key, v.fieldKeys)
|
||||
matching := v.matchingFieldKeys(key)
|
||||
|
||||
// Handle EXISTS specially
|
||||
if ctx.EXISTS() != nil {
|
||||
@@ -730,7 +735,7 @@ func (v *filterExpressionVisitor) VisitFunctionCall(ctx *grammar.FunctionCallCon
|
||||
return ErrorConditionLiteral
|
||||
}
|
||||
|
||||
conds, ok := v.buildConditions(key, MatchingFieldKeys(key, v.fieldKeys), operator, value)
|
||||
conds, ok := v.buildConditions(key, v.matchingFieldKeys(key), operator, value)
|
||||
if !ok {
|
||||
return ErrorConditionLiteral
|
||||
}
|
||||
@@ -923,7 +928,7 @@ func (v *filterExpressionVisitor) VisitKey(ctx *grammar.KeyContext) any {
|
||||
// buildConditions invokes the condition builder for a filter term, folding its
|
||||
// warnings/errors into visitor state; returns false if an error was recorded.
|
||||
func (v *filterExpressionVisitor) buildConditions(key *telemetrytypes.TelemetryFieldKey, matching []*telemetrytypes.TelemetryFieldKey, op qbtypes.FilterOperator, value any) ([]string, bool) {
|
||||
conds, warns, err := v.conditionBuilder.ConditionFor(v.context, v.orgID, v.startNs, v.endNs, key, v.fieldKeys, qbtypes.ConditionBuilderOptions{SkipResourceFilter: v.skipResourceFilter}, op, value, v.builder)
|
||||
conds, warns, err := v.conditionBuilder.ConditionFor(v.context, v.orgID, v.startNs, v.endNs, key, v.fieldKeys, qbtypes.ConditionBuilderOptions{SkipResourceFilter: v.skipResourceFilter, ExactSemconv: v.exactSemconv}, op, value, v.builder)
|
||||
if err != nil {
|
||||
_, _, _, _, errURL, _ := errors.Unwrapb(err)
|
||||
assignIfEmpty(&v.mainErrorURL, errURL)
|
||||
@@ -982,27 +987,159 @@ func assignIfEmpty(s *string, value string) {
|
||||
// MatchingFieldKeys returns the field keys from the map that match the given key,
|
||||
// honoring any context/data type the user specified.
|
||||
func MatchingFieldKeys(field *telemetrytypes.TelemetryFieldKey, fieldKeys map[string][]*telemetrytypes.TelemetryFieldKey) []*telemetrytypes.TelemetryFieldKey {
|
||||
fieldKeysForName := []*telemetrytypes.TelemetryFieldKey{}
|
||||
return matchingFieldKeys(field, fieldKeys, true)
|
||||
}
|
||||
|
||||
// match by name; keep items whose context and data type match (unspecified matches any)
|
||||
for _, item := range fieldKeys[field.Name] {
|
||||
if (field.FieldContext == telemetrytypes.FieldContextUnspecified || field.FieldContext == item.FieldContext) &&
|
||||
(field.FieldDataType == telemetrytypes.FieldDataTypeUnspecified || field.FieldDataType == item.FieldDataType) {
|
||||
fieldKeysForName = append(fieldKeysForName, item)
|
||||
// MatchingFieldKeysExact matches only the requested physical spelling.
|
||||
func MatchingFieldKeysExact(field *telemetrytypes.TelemetryFieldKey, fieldKeys map[string][]*telemetrytypes.TelemetryFieldKey) []*telemetrytypes.TelemetryFieldKey {
|
||||
return matchingFieldKeys(field, fieldKeys, false)
|
||||
}
|
||||
|
||||
func matchingFieldKeys(field *telemetrytypes.TelemetryFieldKey, fieldKeys map[string][]*telemetrytypes.TelemetryFieldKey, resolveSemconv bool) []*telemetrytypes.TelemetryFieldKey {
|
||||
members := []string{field.Name}
|
||||
if resolveSemconv {
|
||||
signals := []telemetrytypes.Signal{field.Signal}
|
||||
if field.Signal == telemetrytypes.SignalUnspecified {
|
||||
signals = []telemetrytypes.Signal{
|
||||
telemetrytypes.SignalTraces,
|
||||
telemetrytypes.SignalLogs,
|
||||
telemetrytypes.SignalMetrics,
|
||||
}
|
||||
}
|
||||
members = nil
|
||||
for _, signal := range signals {
|
||||
selector := telemetrytypes.FieldKeySelector{
|
||||
Name: field.Name,
|
||||
Signal: signal,
|
||||
FieldContext: field.FieldContext,
|
||||
}
|
||||
for _, member := range semconv.AttributeMembers(selector) {
|
||||
if !slices.Contains(members, member) {
|
||||
members = append(members, member)
|
||||
}
|
||||
}
|
||||
}
|
||||
if len(members) == 0 {
|
||||
members = []string{field.Name}
|
||||
}
|
||||
}
|
||||
fieldKeysForName := make([]*telemetrytypes.TelemetryFieldKey, 0)
|
||||
indexByIdentity := make(map[string]int)
|
||||
|
||||
appendMatches := func(lookupName string, memberName string, contextAlreadyMatched bool) {
|
||||
for _, item := range fieldKeys[lookupName] {
|
||||
if !contextAlreadyMatched && field.FieldContext != telemetrytypes.FieldContextUnspecified && field.FieldContext != item.FieldContext {
|
||||
continue
|
||||
}
|
||||
if field.FieldDataType != telemetrytypes.FieldDataTypeUnspecified && field.FieldDataType != item.FieldDataType {
|
||||
continue
|
||||
}
|
||||
|
||||
itemMembers := []string{field.Name}
|
||||
if resolveSemconv {
|
||||
itemMembers = semconv.AttributeMembers(telemetrytypes.FieldKeySelector{
|
||||
Name: field.Name,
|
||||
Signal: item.Signal,
|
||||
FieldContext: item.FieldContext,
|
||||
})
|
||||
}
|
||||
familyMatch := resolveSemconv && len(itemMembers) > 1
|
||||
// A wildcard lookup may find the same name in a signal or context where
|
||||
// the family does not apply. Exact names remain valid there; aliases do not.
|
||||
if memberName != field.Name && (!familyMatch || !slices.Contains(itemMembers, memberName)) {
|
||||
continue
|
||||
}
|
||||
|
||||
physicalMembers := item.SemconvMembers
|
||||
if len(physicalMembers) == 0 {
|
||||
physicalMembers = []string{memberName}
|
||||
}
|
||||
materializedColumns := maps.Clone(item.SemconvMaterializedColumns)
|
||||
if item.Materialized {
|
||||
if materializedColumns == nil {
|
||||
materializedColumns = make(map[string]string)
|
||||
}
|
||||
physicalKey := *item
|
||||
physicalKey.Name = memberName
|
||||
materializedColumns[memberName] = strings.Trim(telemetrytypes.FieldKeyToMaterializedColumnName(&physicalKey), "`")
|
||||
}
|
||||
identity := item.Signal.StringValue() + ";" + item.FieldContext.StringValue() + ";" + item.FieldDataType.StringValue()
|
||||
if familyMatch {
|
||||
if index, found := indexByIdentity[identity]; found {
|
||||
for _, physicalMember := range physicalMembers {
|
||||
if !slices.Contains(fieldKeysForName[index].SemconvMembers, physicalMember) {
|
||||
fieldKeysForName[index].SemconvMembers = append(fieldKeysForName[index].SemconvMembers, physicalMember)
|
||||
}
|
||||
}
|
||||
if len(materializedColumns) > 0 {
|
||||
if fieldKeysForName[index].SemconvMaterializedColumns == nil {
|
||||
fieldKeysForName[index].SemconvMaterializedColumns = make(map[string]string)
|
||||
}
|
||||
maps.Copy(fieldKeysForName[index].SemconvMaterializedColumns, materializedColumns)
|
||||
}
|
||||
if item.Materialized && item.MaterializedSemconv {
|
||||
fieldKeysForName[index].Materialized = true
|
||||
fieldKeysForName[index].MaterializedColumnName = item.MaterializedColumnName
|
||||
fieldKeysForName[index].MaterializedSemconv = true
|
||||
}
|
||||
continue
|
||||
}
|
||||
indexByIdentity[identity] = len(fieldKeysForName)
|
||||
}
|
||||
resolved := *item
|
||||
// The requested spelling is the response identity. Field mappers use
|
||||
// it to resolve the available family members current-first.
|
||||
if familyMatch {
|
||||
resolved.Name = field.Name
|
||||
resolved.SemconvMembers = slices.Clone(physicalMembers)
|
||||
resolved.SemconvMaterializedColumns = materializedColumns
|
||||
if !resolved.MaterializedSemconv {
|
||||
// Materialization is member-specific after family keys are merged.
|
||||
resolved.Materialized = false
|
||||
}
|
||||
} else if !resolveSemconv {
|
||||
resolved.SemconvMembers = []string{field.Name}
|
||||
}
|
||||
fieldKeysForName = append(fieldKeysForName, &resolved)
|
||||
}
|
||||
}
|
||||
|
||||
// A context may have been split off a name that legitimately contained it (e.g.
|
||||
// `attribute.key`); also look up the context-prefixed name so both readings resolve.
|
||||
// Members are current-first, so metadata from the current key wins when
|
||||
// both spellings describe the same signal/context/type.
|
||||
for _, member := range members {
|
||||
appendMatches(member, member, false)
|
||||
}
|
||||
|
||||
// A context may have been split off a name that legitimately contained it
|
||||
// (e.g. `attribute.key`); preserve that historical alternate reading for
|
||||
// every family member.
|
||||
if field.FieldContext != telemetrytypes.FieldContextUnspecified {
|
||||
contextPrefixedFieldName := fmt.Sprintf("%s.%s", field.FieldContext.StringValue(), field.Name)
|
||||
for _, item := range fieldKeys[contextPrefixedFieldName] {
|
||||
// Context already matched via the lookup key; only data type needs checking.
|
||||
if field.FieldDataType == telemetrytypes.FieldDataTypeUnspecified || item.FieldDataType == field.FieldDataType {
|
||||
fieldKeysForName = append(fieldKeysForName, item)
|
||||
}
|
||||
for _, member := range members {
|
||||
appendMatches(fmt.Sprintf("%s.%s", field.FieldContext.StringValue(), member), member, true)
|
||||
}
|
||||
}
|
||||
|
||||
return fieldKeysForName
|
||||
}
|
||||
|
||||
func (v *filterExpressionVisitor) matchingFieldKeys(field *telemetrytypes.TelemetryFieldKey) []*telemetrytypes.TelemetryFieldKey {
|
||||
if v.exactSemconv {
|
||||
return MatchingFieldKeysExact(field, v.fieldKeys)
|
||||
}
|
||||
return MatchingFieldKeys(field, v.fieldKeys)
|
||||
}
|
||||
|
||||
// ExactSemconvKeys returns copies pinned to their physical names, preventing a
|
||||
// field mapper from expanding a synthesized or metadata-free key into a family.
|
||||
func ExactSemconvKeys(keys []*telemetrytypes.TelemetryFieldKey) []*telemetrytypes.TelemetryFieldKey {
|
||||
result := make([]*telemetrytypes.TelemetryFieldKey, 0, len(keys))
|
||||
for _, key := range keys {
|
||||
if key == nil {
|
||||
continue
|
||||
}
|
||||
resolved := *key
|
||||
resolved.SemconvMembers = []string{key.Name}
|
||||
result = append(result, &resolved)
|
||||
}
|
||||
return result
|
||||
}
|
||||
|
||||
@@ -14,6 +14,7 @@ import (
|
||||
"github.com/antlr4-go/antlr/v4"
|
||||
sqlbuilder "github.com/huandu/go-sqlbuilder"
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
// TestPrepareWhereClause_EmptyVariableList ensures PrepareWhereClause errors when a variable has an empty list value.
|
||||
@@ -685,6 +686,147 @@ func TestVisitKey(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestMatchingFieldKeysResolvesCurrentTraceNameFromOldMetadata(t *testing.T) {
|
||||
old := &telemetrytypes.TelemetryFieldKey{
|
||||
Name: "deployment.environment",
|
||||
Description: "old metadata",
|
||||
Signal: telemetrytypes.SignalTraces,
|
||||
FieldContext: telemetrytypes.FieldContextResource,
|
||||
FieldDataType: telemetrytypes.FieldDataTypeString,
|
||||
}
|
||||
requested := telemetrytypes.NewTelemetryFieldKey(
|
||||
"deployment.environment.name",
|
||||
telemetrytypes.FieldContextResource,
|
||||
telemetrytypes.FieldDataTypeString,
|
||||
)
|
||||
|
||||
matches := MatchingFieldKeys(requested, map[string][]*telemetrytypes.TelemetryFieldKey{old.Name: {old}})
|
||||
|
||||
require.Len(t, matches, 1, "trace family lookup must resolve before inspecting metadata")
|
||||
assert.Equal(t, "deployment.environment.name", matches[0].Name)
|
||||
assert.Equal(t, "old metadata", matches[0].Description)
|
||||
assert.Equal(t, []string{"deployment.environment"}, matches[0].SemconvMembers)
|
||||
}
|
||||
|
||||
func TestMatchingFieldKeysUsesCurrentTraceMetadataForOldName(t *testing.T) {
|
||||
current := &telemetrytypes.TelemetryFieldKey{
|
||||
Name: "deployment.environment.name",
|
||||
Description: "current metadata",
|
||||
Signal: telemetrytypes.SignalTraces,
|
||||
FieldContext: telemetrytypes.FieldContextResource,
|
||||
FieldDataType: telemetrytypes.FieldDataTypeString,
|
||||
}
|
||||
old := &telemetrytypes.TelemetryFieldKey{
|
||||
Name: "deployment.environment",
|
||||
Description: "old metadata",
|
||||
Signal: telemetrytypes.SignalTraces,
|
||||
FieldContext: telemetrytypes.FieldContextResource,
|
||||
FieldDataType: telemetrytypes.FieldDataTypeString,
|
||||
}
|
||||
requested := telemetrytypes.NewTelemetryFieldKey(
|
||||
old.Name,
|
||||
telemetrytypes.FieldContextResource,
|
||||
telemetrytypes.FieldDataTypeString,
|
||||
)
|
||||
|
||||
matches := MatchingFieldKeys(requested, map[string][]*telemetrytypes.TelemetryFieldKey{
|
||||
current.Name: {current},
|
||||
old.Name: {old},
|
||||
})
|
||||
|
||||
require.Len(t, matches, 1, "trace family lookup must resolve before inspecting metadata")
|
||||
assert.Equal(t, old.Name, matches[0].Name)
|
||||
assert.Equal(t, "current metadata", matches[0].Description)
|
||||
assert.Equal(t, []string{current.Name, old.Name}, matches[0].SemconvMembers)
|
||||
}
|
||||
|
||||
func TestMatchingFieldKeysResolvesLogSemconvFamily(t *testing.T) {
|
||||
current := &telemetrytypes.TelemetryFieldKey{
|
||||
Name: "deployment.environment.name",
|
||||
Signal: telemetrytypes.SignalLogs,
|
||||
FieldContext: telemetrytypes.FieldContextResource,
|
||||
FieldDataType: telemetrytypes.FieldDataTypeString,
|
||||
}
|
||||
old := &telemetrytypes.TelemetryFieldKey{
|
||||
Name: "deployment.environment",
|
||||
Signal: telemetrytypes.SignalLogs,
|
||||
FieldContext: telemetrytypes.FieldContextResource,
|
||||
FieldDataType: telemetrytypes.FieldDataTypeString,
|
||||
}
|
||||
requested := telemetrytypes.NewTelemetryFieldKey(
|
||||
current.Name,
|
||||
telemetrytypes.FieldContextResource,
|
||||
telemetrytypes.FieldDataTypeString,
|
||||
)
|
||||
|
||||
matches := MatchingFieldKeys(requested, map[string][]*telemetrytypes.TelemetryFieldKey{
|
||||
current.Name: {current},
|
||||
old.Name: {old},
|
||||
})
|
||||
|
||||
require.Len(t, matches, 1, "log family lookup must resolve before inspecting metadata")
|
||||
assert.Equal(t, current.Name, matches[0].Name)
|
||||
assert.Equal(t, []string{current.Name, old.Name}, matches[0].SemconvMembers)
|
||||
}
|
||||
|
||||
func TestMatchingFieldKeysResolvesMetricStorageSpellings(t *testing.T) {
|
||||
current := &telemetrytypes.TelemetryFieldKey{
|
||||
Name: "resource_deployment_environment_name",
|
||||
Signal: telemetrytypes.SignalMetrics,
|
||||
FieldContext: telemetrytypes.FieldContextResource,
|
||||
FieldDataType: telemetrytypes.FieldDataTypeString,
|
||||
}
|
||||
old := &telemetrytypes.TelemetryFieldKey{
|
||||
Name: "resource_deployment_environment",
|
||||
Signal: telemetrytypes.SignalMetrics,
|
||||
FieldContext: telemetrytypes.FieldContextResource,
|
||||
FieldDataType: telemetrytypes.FieldDataTypeString,
|
||||
}
|
||||
requested := telemetrytypes.NewTelemetryFieldKey(
|
||||
"deployment.environment.name",
|
||||
telemetrytypes.FieldContextResource,
|
||||
telemetrytypes.FieldDataTypeString,
|
||||
)
|
||||
|
||||
matches := MatchingFieldKeys(requested, map[string][]*telemetrytypes.TelemetryFieldKey{
|
||||
current.Name: {current},
|
||||
old.Name: {old},
|
||||
})
|
||||
|
||||
require.Len(t, matches, 1, "metric family lookup must resolve storage spellings")
|
||||
assert.Equal(t, requested.Name, matches[0].Name)
|
||||
assert.Equal(t, []string{current.Name, old.Name}, matches[0].SemconvMembers)
|
||||
}
|
||||
|
||||
func TestMatchingFieldKeysExactKeepsRequestedPhysicalName(t *testing.T) {
|
||||
current := &telemetrytypes.TelemetryFieldKey{
|
||||
Name: "deployment.environment.name",
|
||||
Signal: telemetrytypes.SignalTraces,
|
||||
FieldContext: telemetrytypes.FieldContextResource,
|
||||
FieldDataType: telemetrytypes.FieldDataTypeString,
|
||||
}
|
||||
old := &telemetrytypes.TelemetryFieldKey{
|
||||
Name: "deployment.environment",
|
||||
Signal: telemetrytypes.SignalTraces,
|
||||
FieldContext: telemetrytypes.FieldContextResource,
|
||||
FieldDataType: telemetrytypes.FieldDataTypeString,
|
||||
}
|
||||
requested := telemetrytypes.NewTelemetryFieldKey(
|
||||
current.Name,
|
||||
telemetrytypes.FieldContextResource,
|
||||
telemetrytypes.FieldDataTypeString,
|
||||
)
|
||||
|
||||
matches := MatchingFieldKeysExact(requested, map[string][]*telemetrytypes.TelemetryFieldKey{
|
||||
current.Name: {current},
|
||||
old.Name: {old},
|
||||
})
|
||||
|
||||
require.Len(t, matches, 1, "exact lookup must return the requested physical field")
|
||||
assert.Equal(t, current.Name, matches[0].Name)
|
||||
assert.Equal(t, []string{current.Name}, matches[0].SemconvMembers)
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// TestVisitComparison
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
@@ -2,21 +2,47 @@
|
||||
|
||||
package semconv
|
||||
|
||||
import "github.com/SigNoz/signoz/pkg/types/telemetrytypes"
|
||||
|
||||
var families = []Family{
|
||||
{
|
||||
Current: "container.cpu.usage",
|
||||
Old: []string{"container.cpu.utilization"},
|
||||
Kind: KindMetric,
|
||||
Contexts: []telemetrytypes.FieldContext{telemetrytypes.FieldContextMetric},
|
||||
Signals: []telemetrytypes.Signal{telemetrytypes.SignalMetrics},
|
||||
ApplyToMetrics: nil,
|
||||
},
|
||||
{
|
||||
Current: "db.system.name",
|
||||
Old: []string{"db.system"},
|
||||
Kind: KindAttribute,
|
||||
Contexts: nil,
|
||||
Signals: nil,
|
||||
Contexts: []telemetrytypes.FieldContext{telemetrytypes.FieldContextAttribute, telemetrytypes.FieldContextResource},
|
||||
Signals: []telemetrytypes.Signal{telemetrytypes.SignalLogs, telemetrytypes.SignalMetrics, telemetrytypes.SignalTraces},
|
||||
ApplyToMetrics: nil,
|
||||
},
|
||||
{
|
||||
Current: "deployment.environment.name",
|
||||
Old: []string{"deployment.environment"},
|
||||
Kind: KindAttribute,
|
||||
Contexts: nil,
|
||||
Signals: nil,
|
||||
Contexts: []telemetrytypes.FieldContext{telemetrytypes.FieldContextAttribute, telemetrytypes.FieldContextResource},
|
||||
Signals: []telemetrytypes.Signal{telemetrytypes.SignalLogs, telemetrytypes.SignalMetrics, telemetrytypes.SignalTraces},
|
||||
ApplyToMetrics: nil,
|
||||
},
|
||||
{
|
||||
Current: "k8s.node.cpu.usage",
|
||||
Old: []string{"k8s.node.cpu.utilization"},
|
||||
Kind: KindMetric,
|
||||
Contexts: []telemetrytypes.FieldContext{telemetrytypes.FieldContextMetric},
|
||||
Signals: []telemetrytypes.Signal{telemetrytypes.SignalMetrics},
|
||||
ApplyToMetrics: nil,
|
||||
},
|
||||
{
|
||||
Current: "k8s.pod.cpu.usage",
|
||||
Old: []string{"k8s.pod.cpu.utilization"},
|
||||
Kind: KindMetric,
|
||||
Contexts: []telemetrytypes.FieldContext{telemetrytypes.FieldContextMetric},
|
||||
Signals: []telemetrytypes.Signal{telemetrytypes.SignalMetrics},
|
||||
ApplyToMetrics: nil,
|
||||
},
|
||||
}
|
||||
|
||||
@@ -2,6 +2,7 @@ package semconv
|
||||
|
||||
import (
|
||||
"slices"
|
||||
"strings"
|
||||
|
||||
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
|
||||
"github.com/SigNoz/signoz/pkg/valuer"
|
||||
@@ -26,6 +27,13 @@ type Family struct {
|
||||
ValueMap map[string]string
|
||||
}
|
||||
|
||||
type metricSpelling uint8
|
||||
|
||||
const (
|
||||
metricSpellingDotted metricSpelling = iota
|
||||
metricSpellingNormalized
|
||||
)
|
||||
|
||||
var (
|
||||
KindAttribute = Kind{String: valuer.NewString("attribute")}
|
||||
KindMetric = Kind{String: valuer.NewString("metric")}
|
||||
@@ -59,6 +67,88 @@ func Members(kind Kind, selector telemetrytypes.FieldKeySelector) []string {
|
||||
return familyMembers[idx]
|
||||
}
|
||||
|
||||
// AttributeMembers returns the physical attribute spellings that may represent
|
||||
// selector.Name. Metrics have used both dotted and normalized label layouts;
|
||||
// resource labels have additionally used a resource_ prefix. Keeping that
|
||||
// storage detail here prevents metrics readers from maintaining local
|
||||
// transition tables.
|
||||
func AttributeMembers(selector telemetrytypes.FieldKeySelector) []string {
|
||||
if selector.Signal != telemetrytypes.SignalMetrics {
|
||||
return Members(KindAttribute, selector)
|
||||
}
|
||||
|
||||
lookupSelector := selector
|
||||
lookupSelector.Name = strings.TrimPrefix(selector.Name, "resource_")
|
||||
family, style, ok := lookupMetricSpelling(KindAttribute, lookupSelector)
|
||||
if !ok {
|
||||
return []string{selector.Name}
|
||||
}
|
||||
|
||||
logicalMembers := familyMembers[family]
|
||||
result := make([]string, 0, len(logicalMembers)*4)
|
||||
for _, member := range logicalMembers {
|
||||
dotted := member
|
||||
normalized := normalizeMetricSpelling(member)
|
||||
variants := []string{dotted, normalized}
|
||||
if style == metricSpellingNormalized {
|
||||
variants[0], variants[1] = variants[1], variants[0]
|
||||
}
|
||||
|
||||
if selector.FieldContext == telemetrytypes.FieldContextResource ||
|
||||
selector.FieldContext == telemetrytypes.FieldContextUnspecified ||
|
||||
strings.HasPrefix(selector.Name, "resource_") {
|
||||
for _, variant := range variants {
|
||||
result = appendUniqueString(result, "resource_"+variant)
|
||||
}
|
||||
}
|
||||
for _, variant := range variants {
|
||||
result = appendUniqueString(result, variant)
|
||||
}
|
||||
}
|
||||
return result
|
||||
}
|
||||
|
||||
// MetricNames returns the current and historical storage names for a metric.
|
||||
// The input's dotted or normalized style is preserved because both layouts are
|
||||
// valid metric identities and must not be mixed in one query.
|
||||
func MetricNames(name string) []string {
|
||||
selector := telemetrytypes.FieldKeySelector{
|
||||
Name: name,
|
||||
Signal: telemetrytypes.SignalMetrics,
|
||||
FieldContext: telemetrytypes.FieldContextMetric,
|
||||
}
|
||||
family, style, ok := lookupMetricSpelling(KindMetric, selector)
|
||||
if !ok {
|
||||
return []string{name}
|
||||
}
|
||||
|
||||
logicalMembers := familyMembers[family]
|
||||
result := make([]string, 0, len(logicalMembers))
|
||||
for _, member := range logicalMembers {
|
||||
if style == metricSpellingNormalized {
|
||||
member = normalizeMetricSpelling(member)
|
||||
}
|
||||
result = appendUniqueString(result, member)
|
||||
}
|
||||
return result
|
||||
}
|
||||
|
||||
// CurrentAttribute returns the canonical dotted name for an attribute
|
||||
// spelling, or selector.Name if no enabled family matches.
|
||||
func CurrentAttribute(selector telemetrytypes.FieldKeySelector) string {
|
||||
if selector.Signal != telemetrytypes.SignalMetrics {
|
||||
return Current(KindAttribute, selector)
|
||||
}
|
||||
|
||||
lookupSelector := selector
|
||||
lookupSelector.Name = strings.TrimPrefix(selector.Name, "resource_")
|
||||
family, _, ok := lookupMetricSpelling(KindAttribute, lookupSelector)
|
||||
if !ok {
|
||||
return selector.Name
|
||||
}
|
||||
return families[family].Current
|
||||
}
|
||||
|
||||
// Current returns the current name for selector.Name, or the input name when
|
||||
// it does not belong to an enabled family.
|
||||
func Current(kind Kind, selector telemetrytypes.FieldKeySelector) string {
|
||||
@@ -99,6 +189,35 @@ func lookupIndex(kind Kind, selector telemetrytypes.FieldKeySelector) (int, bool
|
||||
return 0, false
|
||||
}
|
||||
|
||||
func lookupMetricSpelling(kind Kind, selector telemetrytypes.FieldKeySelector) (int, metricSpelling, bool) {
|
||||
if idx, ok := lookupIndex(kind, selector); ok {
|
||||
return idx, metricSpellingDotted, true
|
||||
}
|
||||
|
||||
for idx, family := range families {
|
||||
if !matchesSelector(family, kind, selector) {
|
||||
continue
|
||||
}
|
||||
for _, member := range familyMembers[idx] {
|
||||
if normalizeMetricSpelling(member) == selector.Name {
|
||||
return idx, metricSpellingNormalized, true
|
||||
}
|
||||
}
|
||||
}
|
||||
return 0, metricSpellingDotted, false
|
||||
}
|
||||
|
||||
func normalizeMetricSpelling(name string) string {
|
||||
return strings.ReplaceAll(name, ".", "_")
|
||||
}
|
||||
|
||||
func appendUniqueString(values []string, value string) []string {
|
||||
if value == "" || slices.Contains(values, value) {
|
||||
return values
|
||||
}
|
||||
return append(values, value)
|
||||
}
|
||||
|
||||
func matchesSelector(family Family, kind Kind, selector telemetrytypes.FieldKeySelector) bool {
|
||||
if family.Kind != kind {
|
||||
return false
|
||||
|
||||
@@ -78,3 +78,57 @@ func TestMembersReturnsInputWhenKindDoesNotMatch(t *testing.T) {
|
||||
"an attribute family must not match a metric-name lookup",
|
||||
)
|
||||
}
|
||||
func TestAttributeMembersIncludesMetricResourceStorageSpellings(t *testing.T) {
|
||||
selector := telemetrytypes.FieldKeySelector{
|
||||
Name: "db.system.name",
|
||||
Signal: telemetrytypes.SignalMetrics,
|
||||
FieldContext: telemetrytypes.FieldContextResource,
|
||||
}
|
||||
|
||||
assert.Equal(t, []string{
|
||||
"resource_db.system.name", "resource_db_system_name", "db.system.name", "db_system_name",
|
||||
"resource_db.system", "resource_db_system", "db.system", "db_system",
|
||||
}, AttributeMembers(selector), "resource metric attributes should cover every historical storage layout")
|
||||
}
|
||||
|
||||
func TestCurrentAttributeResolvesNormalizedMetricResourceSpelling(t *testing.T) {
|
||||
selector := telemetrytypes.FieldKeySelector{
|
||||
Name: "resource_db_system",
|
||||
Signal: telemetrytypes.SignalMetrics,
|
||||
FieldContext: telemetrytypes.FieldContextResource,
|
||||
}
|
||||
|
||||
assert.Equal(t, "db.system.name", CurrentAttribute(selector), "normalized resource spelling should resolve to the dotted current name")
|
||||
}
|
||||
|
||||
func TestAttributeMembersPreservesNormalizedMetricPointStyle(t *testing.T) {
|
||||
selector := telemetrytypes.FieldKeySelector{
|
||||
Name: "db_system",
|
||||
Signal: telemetrytypes.SignalMetrics,
|
||||
FieldContext: telemetrytypes.FieldContextAttribute,
|
||||
}
|
||||
|
||||
assert.Equal(t, []string{
|
||||
"db_system_name", "db.system.name", "db_system", "db.system",
|
||||
}, AttributeMembers(selector), "normalized point attribute should remain the preferred storage spelling")
|
||||
}
|
||||
|
||||
func TestMetricNamesPreservesDottedStyle(t *testing.T) {
|
||||
assert.Equal(t,
|
||||
[]string{"k8s.pod.cpu.usage", "k8s.pod.cpu.utilization"},
|
||||
MetricNames("k8s.pod.cpu.usage"),
|
||||
"dotted metric input should produce dotted family names",
|
||||
)
|
||||
}
|
||||
|
||||
func TestMetricNamesPreservesNormalizedStyle(t *testing.T) {
|
||||
assert.Equal(t,
|
||||
[]string{"k8s_pod_cpu_usage", "k8s_pod_cpu_utilization"},
|
||||
MetricNames("k8s_pod_cpu_utilization"),
|
||||
"normalized metric input should produce normalized family names",
|
||||
)
|
||||
}
|
||||
|
||||
func TestMetricNamesReturnsUnknownNameUnchanged(t *testing.T) {
|
||||
assert.Equal(t, []string{"custom_metric"}, MetricNames("custom_metric"), "unknown metrics should not be expanded")
|
||||
}
|
||||
|
||||
@@ -237,6 +237,7 @@ func NewSQLMigrationProviderFactories(
|
||||
sqlmigration.NewAddDashboardTuplesFactory(sqlstore),
|
||||
sqlmigration.NewRestructureSavedViewSpecFactory(sqlstore, sqlschema),
|
||||
sqlmigration.NewAddSavedViewTuplesFactory(sqlstore),
|
||||
sqlmigration.NewMigrateDeploymentEnvironmentQuickFilterFactory(),
|
||||
)
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,127 @@
|
||||
package sqlmigration
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"log/slog"
|
||||
"time"
|
||||
|
||||
"github.com/SigNoz/signoz/pkg/factory"
|
||||
"github.com/SigNoz/signoz/pkg/semconv"
|
||||
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
|
||||
"github.com/uptrace/bun"
|
||||
"github.com/uptrace/bun/migrate"
|
||||
)
|
||||
|
||||
const deploymentEnvironmentCurrent = "deployment.environment.name"
|
||||
|
||||
type migrateDeploymentEnvironmentQuickFilter struct {
|
||||
logger *slog.Logger
|
||||
}
|
||||
|
||||
type semconvQuickFilterRow struct {
|
||||
bun.BaseModel `bun:"table:quick_filter"`
|
||||
|
||||
ID string `bun:"id"`
|
||||
Filter string `bun:"filter"`
|
||||
}
|
||||
|
||||
func NewMigrateDeploymentEnvironmentQuickFilterFactory() factory.ProviderFactory[SQLMigration, Config] {
|
||||
return factory.NewProviderFactory(
|
||||
factory.MustNewName("migrate_semconv_quick_filter"),
|
||||
func(_ context.Context, settings factory.ProviderSettings, _ Config) (SQLMigration, error) {
|
||||
return &migrateDeploymentEnvironmentQuickFilter{logger: settings.Logger}, nil
|
||||
},
|
||||
)
|
||||
}
|
||||
|
||||
func (migration *migrateDeploymentEnvironmentQuickFilter) Register(migrations *migrate.Migrations) error {
|
||||
return migrations.Register(migration.Up, migration.Down)
|
||||
}
|
||||
|
||||
func deploymentEnvironmentOld() string {
|
||||
members := semconv.Members(semconv.KindAttribute, telemetrytypes.FieldKeySelector{
|
||||
Name: deploymentEnvironmentCurrent,
|
||||
Signal: telemetrytypes.SignalTraces,
|
||||
FieldContext: telemetrytypes.FieldContextResource,
|
||||
})
|
||||
if len(members) < 2 {
|
||||
return deploymentEnvironmentCurrent
|
||||
}
|
||||
return members[1]
|
||||
}
|
||||
|
||||
func rewriteQuickFilterSemconv(filterJSON, from, to string) (string, bool, error) {
|
||||
var filters []map[string]any
|
||||
if err := json.Unmarshal([]byte(filterJSON), &filters); err != nil {
|
||||
return "", false, err
|
||||
}
|
||||
|
||||
changed := false
|
||||
for _, filter := range filters {
|
||||
if key, ok := filter["key"].(string); ok && key == from {
|
||||
filter["key"] = to
|
||||
changed = true
|
||||
}
|
||||
}
|
||||
if !changed {
|
||||
return filterJSON, false, nil
|
||||
}
|
||||
|
||||
rewritten, err := json.Marshal(filters)
|
||||
if err != nil {
|
||||
return "", false, err
|
||||
}
|
||||
return string(rewritten), true, nil
|
||||
}
|
||||
|
||||
func (migration *migrateDeploymentEnvironmentQuickFilter) migrate(ctx context.Context, db *bun.DB, from, to string) error {
|
||||
tx, err := db.BeginTx(ctx, nil)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer func() { _ = tx.Rollback() }()
|
||||
|
||||
rows := make([]*semconvQuickFilterRow, 0)
|
||||
if err := tx.NewSelect().
|
||||
Model(&rows).
|
||||
Where("signal IN (?)", bun.In([]string{"traces", "api_monitoring", "exceptions"})).
|
||||
Scan(ctx); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
for _, row := range rows {
|
||||
rewritten, changed, err := rewriteQuickFilterSemconv(row.Filter, from, to)
|
||||
if err != nil {
|
||||
// Quick filters are user-editable. One malformed legacy row must not
|
||||
// prevent the application from starting or block every other org's
|
||||
// migration.
|
||||
if migration.logger != nil {
|
||||
migration.logger.WarnContext(ctx, "skipping quick filter with unreadable filter JSON",
|
||||
slog.String("quick_filter_id", row.ID), slog.Any("error", err))
|
||||
}
|
||||
continue
|
||||
}
|
||||
if !changed {
|
||||
continue
|
||||
}
|
||||
if _, err := tx.NewUpdate().
|
||||
Model((*semconvQuickFilterRow)(nil)).
|
||||
Set("filter = ?", rewritten).
|
||||
Set("updated_at = ?", time.Now()).
|
||||
Where("id = ?", row.ID).
|
||||
Exec(ctx); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
return tx.Commit()
|
||||
}
|
||||
|
||||
func (migration *migrateDeploymentEnvironmentQuickFilter) Up(ctx context.Context, db *bun.DB) error {
|
||||
return migration.migrate(ctx, db, deploymentEnvironmentOld(), deploymentEnvironmentCurrent)
|
||||
}
|
||||
|
||||
func (migration *migrateDeploymentEnvironmentQuickFilter) Down(ctx context.Context, db *bun.DB) error {
|
||||
return migration.migrate(ctx, db, deploymentEnvironmentCurrent, deploymentEnvironmentOld())
|
||||
}
|
||||
@@ -0,0 +1,38 @@
|
||||
package sqlmigration
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"testing"
|
||||
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
func TestRewriteQuickFilterSemconv(t *testing.T) {
|
||||
oldName := deploymentEnvironmentOld()
|
||||
input := `[{"key":"service.name","dataType":"string","type":"resource"},{"key":"` + oldName + `","dataType":"string","type":"resource","custom":true}]`
|
||||
|
||||
rewritten, changed, err := rewriteQuickFilterSemconv(input, oldName, deploymentEnvironmentCurrent)
|
||||
require.NoError(t, err)
|
||||
assert.True(t, changed)
|
||||
|
||||
var filters []map[string]any
|
||||
require.NoError(t, json.Unmarshal([]byte(rewritten), &filters))
|
||||
assert.Equal(t, "service.name", filters[0]["key"])
|
||||
assert.Equal(t, deploymentEnvironmentCurrent, filters[1]["key"])
|
||||
assert.Equal(t, true, filters[1]["custom"], "unknown filter properties must be preserved")
|
||||
|
||||
restored, changed, err := rewriteQuickFilterSemconv(rewritten, deploymentEnvironmentCurrent, oldName)
|
||||
require.NoError(t, err)
|
||||
assert.True(t, changed)
|
||||
require.NoError(t, json.Unmarshal([]byte(restored), &filters))
|
||||
assert.Equal(t, oldName, filters[1]["key"])
|
||||
}
|
||||
|
||||
func TestRewriteQuickFilterSemconvNoop(t *testing.T) {
|
||||
input := `[{"key":"service.name","dataType":"string","type":"resource"}]`
|
||||
rewritten, changed, err := rewriteQuickFilterSemconv(input, deploymentEnvironmentOld(), deploymentEnvironmentCurrent)
|
||||
require.NoError(t, err)
|
||||
assert.False(t, changed)
|
||||
assert.Equal(t, input, rewritten)
|
||||
}
|
||||
@@ -10,6 +10,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/semconv"
|
||||
"github.com/SigNoz/signoz/pkg/statementbuilder"
|
||||
"github.com/SigNoz/signoz/pkg/telemetryschema/metricstelemetryschema"
|
||||
"github.com/SigNoz/signoz/pkg/types/metrictypes"
|
||||
@@ -31,6 +32,15 @@ const (
|
||||
OthersMultiTemporality = `IF(LOWER(temporality) LIKE LOWER('delta'), %s, %s) AS per_series_value`
|
||||
)
|
||||
|
||||
func metricNameValues(name string) []any {
|
||||
members := semconv.MetricNames(name)
|
||||
values := make([]any, 0, len(members))
|
||||
for _, member := range members {
|
||||
values = append(values, member)
|
||||
}
|
||||
return values
|
||||
}
|
||||
|
||||
type StatementBuilder struct {
|
||||
logger *slog.Logger
|
||||
metadataStore telemetrytypes.MetadataStore
|
||||
@@ -341,7 +351,7 @@ func (b *StatementBuilder) buildReducedTimeSeriesCTE(
|
||||
sb.SelectMore(col)
|
||||
}
|
||||
sb.Where(
|
||||
sb.In("metric_name", query.Aggregations[0].MetricName),
|
||||
sb.In("metric_name", metricNameValues(query.Aggregations[0].MetricName)...),
|
||||
sb.GTE("unix_milli", start),
|
||||
sb.LTE("unix_milli", end),
|
||||
)
|
||||
@@ -385,7 +395,7 @@ func (b *StatementBuilder) buildReducedSpatialAggFastPath(
|
||||
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),
|
||||
sb.In("metric_name", metricNameValues(agg.MetricName)...),
|
||||
sb.GTE("unix_milli", start),
|
||||
sb.LT("unix_milli", end),
|
||||
)
|
||||
@@ -427,7 +437,7 @@ func (b *StatementBuilder) buildReducedTemporalAggregationCTE(
|
||||
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),
|
||||
sb.In("metric_name", metricNameValues(agg.MetricName)...),
|
||||
sb.GTE("unix_milli", start),
|
||||
sb.LT("unix_milli", end),
|
||||
)
|
||||
@@ -505,7 +515,7 @@ func (b *StatementBuilder) buildTemporalAggDeltaFastPath(
|
||||
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),
|
||||
sb.In("metric_name", metricNameValues(query.Aggregations[0].MetricName)...),
|
||||
sb.GTE("unix_milli", start),
|
||||
sb.LT("unix_milli", end),
|
||||
)
|
||||
@@ -560,7 +570,7 @@ func (b *StatementBuilder) buildTimeSeriesCTE(
|
||||
}
|
||||
|
||||
sb.Where(
|
||||
sb.In("metric_name", query.Aggregations[0].MetricName),
|
||||
sb.In("metric_name", metricNameValues(query.Aggregations[0].MetricName)...),
|
||||
sb.GTE("unix_milli", start),
|
||||
sb.LTE("unix_milli", end),
|
||||
)
|
||||
@@ -638,7 +648,7 @@ func (b *StatementBuilder) buildTemporalAggDelta(
|
||||
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),
|
||||
sb.In("metric_name", metricNameValues(query.Aggregations[0].MetricName)...),
|
||||
sb.GTE("unix_milli", start),
|
||||
sb.LT("unix_milli", end),
|
||||
)
|
||||
@@ -679,7 +689,7 @@ func (b *StatementBuilder) buildTemporalAggCumulativeOrUnspecified(
|
||||
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),
|
||||
baseSb.In("metric_name", metricNameValues(query.Aggregations[0].MetricName)...),
|
||||
baseSb.GTE("unix_milli", start),
|
||||
baseSb.LT("unix_milli", end),
|
||||
)
|
||||
@@ -770,7 +780,7 @@ func (b *StatementBuilder) buildTemporalAggForMultipleTemporalities(
|
||||
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),
|
||||
sb.In("metric_name", metricNameValues(query.Aggregations[0].MetricName)...),
|
||||
sb.GTE("unix_milli", start),
|
||||
sb.LT("unix_milli", end),
|
||||
)
|
||||
|
||||
@@ -13,9 +13,21 @@ import (
|
||||
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
|
||||
"github.com/SigNoz/signoz/pkg/types/telemetrytypes/telemetrytypestest"
|
||||
"github.com/SigNoz/signoz/pkg/valuer"
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
func TestMetricNameValuesResolveFamily(t *testing.T) {
|
||||
assert.Equal(t,
|
||||
[]any{"k8s.pod.cpu.usage", "k8s.pod.cpu.utilization"},
|
||||
metricNameValues("k8s.pod.cpu.usage"),
|
||||
)
|
||||
assert.Equal(t,
|
||||
[]any{"container_cpu_usage", "container_cpu_utilization"},
|
||||
metricNameValues("container_cpu_utilization"),
|
||||
)
|
||||
}
|
||||
|
||||
func TestStatementBuilder(t *testing.T) {
|
||||
cases := []struct {
|
||||
name string
|
||||
|
||||
@@ -44,6 +44,73 @@ func keyIndexFilter(key *telemetrytypes.TelemetryFieldKey) any {
|
||||
return fmt.Sprintf(`%%%s%%`, key.Name)
|
||||
}
|
||||
|
||||
func memberKey(key *telemetrytypes.TelemetryFieldKey, name string) *telemetrytypes.TelemetryFieldKey {
|
||||
member := *key
|
||||
member.Name = name
|
||||
return &member
|
||||
}
|
||||
|
||||
func keyIndexCondition(sb *sqlbuilder.SelectBuilder, column string, key *telemetrytypes.TelemetryFieldKey, members []string) string {
|
||||
conditions := make([]string, 0, len(members))
|
||||
for _, member := range members {
|
||||
conditions = append(conditions, sb.Like(column, keyIndexFilter(memberKey(key, member))))
|
||||
}
|
||||
if len(conditions) == 1 {
|
||||
return conditions[0]
|
||||
}
|
||||
return sb.Or(conditions...)
|
||||
}
|
||||
|
||||
func valueIndexCondition(
|
||||
sb *sqlbuilder.SelectBuilder,
|
||||
column string,
|
||||
key *telemetrytypes.TelemetryFieldKey,
|
||||
members []string,
|
||||
op qbtypes.FilterOperator,
|
||||
value any,
|
||||
caseInsensitive bool,
|
||||
) string {
|
||||
conditions := make([]string, 0, len(members))
|
||||
for _, member := range members {
|
||||
patterns := valueForIndexFilter(op, memberKey(key, member), value)
|
||||
switch values := patterns.(type) {
|
||||
case []string:
|
||||
for _, pattern := range values {
|
||||
conditions = append(conditions, sb.Like(column, pattern))
|
||||
}
|
||||
default:
|
||||
if caseInsensitive {
|
||||
conditions = append(conditions, sb.ILike(column, values))
|
||||
} else {
|
||||
conditions = append(conditions, sb.Like(column, values))
|
||||
}
|
||||
}
|
||||
}
|
||||
if len(conditions) == 1 {
|
||||
return conditions[0]
|
||||
}
|
||||
return sb.Or(conditions...)
|
||||
}
|
||||
|
||||
func memberPresenceCondition(sb *sqlbuilder.SelectBuilder, column string, members []string, exists bool) string {
|
||||
conditions := make([]string, 0, len(members))
|
||||
for _, member := range members {
|
||||
field := fmt.Sprintf("simpleJSONHas(%s, '%s')", column, member)
|
||||
if exists {
|
||||
conditions = append(conditions, sb.E(field, true))
|
||||
} else {
|
||||
conditions = append(conditions, sb.NE(field, true))
|
||||
}
|
||||
}
|
||||
if exists {
|
||||
if len(conditions) == 1 {
|
||||
return conditions[0]
|
||||
}
|
||||
return sb.Or(conditions...)
|
||||
}
|
||||
return sb.And(conditions...)
|
||||
}
|
||||
|
||||
// SkipResourceFilter is not applicable here: the fingerprint table only stores resource attributes.
|
||||
func (b *defaultConditionBuilder) ConditionFor(
|
||||
ctx context.Context,
|
||||
@@ -115,8 +182,10 @@ func (b *defaultConditionBuilder) conditionForKey(
|
||||
// as we have not changed the resource column in the resource fingerprint table.
|
||||
column := columns[0]
|
||||
|
||||
keyIdxFilter := sb.Like(column.Name, keyIndexFilter(key))
|
||||
valueForIndexFilter := valueForIndexFilter(op, key, value)
|
||||
members := resourceSemconvMembers(key)
|
||||
isFamily := len(members) > 1
|
||||
keyIdxFilter := keyIndexCondition(sb, column.Name, key, members)
|
||||
singleValueIndexFilter := valueForIndexFilter(op, memberKey(key, members[0]), value)
|
||||
|
||||
fieldName, err := b.fm.FieldFor(ctx, valuer.UUID{}, startNs, endNs, key)
|
||||
if err != nil {
|
||||
@@ -128,12 +197,15 @@ func (b *defaultConditionBuilder) conditionForKey(
|
||||
return sb.And(
|
||||
sb.E(fieldName, formattedValue),
|
||||
keyIdxFilter,
|
||||
sb.Like(column.Name, valueForIndexFilter),
|
||||
valueIndexCondition(sb, column.Name, key, members, op, value, false),
|
||||
), nil
|
||||
case qbtypes.FilterOperatorNotEqual:
|
||||
if isFamily {
|
||||
return sb.NE(fieldName, formattedValue), nil
|
||||
}
|
||||
return sb.And(
|
||||
sb.NE(fieldName, formattedValue),
|
||||
sb.NotLike(column.Name, valueForIndexFilter),
|
||||
sb.NotLike(column.Name, singleValueIndexFilter),
|
||||
), nil
|
||||
case qbtypes.FilterOperatorGreaterThan:
|
||||
return sb.And(sb.GT(fieldName, formattedValue), keyIdxFilter), nil
|
||||
@@ -148,7 +220,7 @@ func (b *defaultConditionBuilder) conditionForKey(
|
||||
return sb.And(
|
||||
sb.ILike(fieldName, formattedValue),
|
||||
keyIdxFilter,
|
||||
sb.ILike(column.Name, valueForIndexFilter),
|
||||
valueIndexCondition(sb, column.Name, key, members, op, value, true),
|
||||
), nil
|
||||
case qbtypes.FilterOperatorNotLike, qbtypes.FilterOperatorNotILike:
|
||||
// no index filter: as cannot apply `not contains x%y` as y can be somewhere else
|
||||
@@ -185,13 +257,11 @@ func (b *defaultConditionBuilder) conditionForKey(
|
||||
inConditions = append(inConditions, sb.E(fieldName, querybuilder.FormatValueForContains(v)))
|
||||
}
|
||||
mainCondition := sb.Or(inConditions...)
|
||||
valConditions := make([]string, 0, len(values))
|
||||
if valuesForIndexFilter, ok := valueForIndexFilter.([]string); ok {
|
||||
for _, v := range valuesForIndexFilter {
|
||||
valConditions = append(valConditions, sb.Like(column.Name, v))
|
||||
}
|
||||
}
|
||||
mainCondition = sb.And(mainCondition, keyIdxFilter, sb.Or(valConditions...))
|
||||
mainCondition = sb.And(
|
||||
mainCondition,
|
||||
keyIdxFilter,
|
||||
valueIndexCondition(sb, column.Name, key, members, op, value, false),
|
||||
)
|
||||
|
||||
return mainCondition, nil
|
||||
case qbtypes.FilterOperatorNotIn:
|
||||
@@ -204,8 +274,11 @@ func (b *defaultConditionBuilder) conditionForKey(
|
||||
notInConditions = append(notInConditions, sb.NE(fieldName, querybuilder.FormatValueForContains(v)))
|
||||
}
|
||||
mainCondition := sb.And(notInConditions...)
|
||||
if isFamily {
|
||||
return mainCondition, nil
|
||||
}
|
||||
valConditions := make([]string, 0, len(values))
|
||||
if valuesForIndexFilter, ok := valueForIndexFilter.([]string); ok {
|
||||
if valuesForIndexFilter, ok := singleValueIndexFilter.([]string); ok {
|
||||
for _, v := range valuesForIndexFilter {
|
||||
valConditions = append(valConditions, sb.NotLike(column.Name, v))
|
||||
}
|
||||
@@ -215,13 +288,11 @@ func (b *defaultConditionBuilder) conditionForKey(
|
||||
|
||||
case qbtypes.FilterOperatorExists:
|
||||
return sb.And(
|
||||
sb.E(fmt.Sprintf("simpleJSONHas(%s, '%s')", column.Name, key.Name), true),
|
||||
memberPresenceCondition(sb, column.Name, members, true),
|
||||
keyIdxFilter,
|
||||
), nil
|
||||
case qbtypes.FilterOperatorNotExists:
|
||||
return sb.And(
|
||||
sb.NE(fmt.Sprintf("simpleJSONHas(%s, '%s')", column.Name, key.Name), true),
|
||||
), nil
|
||||
return memberPresenceCondition(sb, column.Name, members, false), nil
|
||||
|
||||
case qbtypes.FilterOperatorRegexp:
|
||||
return sb.And(
|
||||
@@ -237,7 +308,7 @@ func (b *defaultConditionBuilder) conditionForKey(
|
||||
return sb.And(
|
||||
sb.ILike(fieldName, fmt.Sprintf(`%%%s%%`, formattedValue)),
|
||||
keyIdxFilter,
|
||||
sb.ILike(column.Name, valueForIndexFilter),
|
||||
valueIndexCondition(sb, column.Name, key, members, op, value, true),
|
||||
), nil
|
||||
case qbtypes.FilterOperatorNotContains:
|
||||
// no index filter: as cannot apply `not contains x%y` as y can be somewhere else
|
||||
|
||||
@@ -220,3 +220,230 @@ func TestConditionBuilder(t *testing.T) {
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestFamilyPositiveFilterExcludesKeylessRows(t *testing.T) {
|
||||
key := &telemetrytypes.TelemetryFieldKey{
|
||||
Name: "deployment.environment.name",
|
||||
Signal: telemetrytypes.SignalTraces,
|
||||
FieldContext: telemetrytypes.FieldContextResource,
|
||||
FieldDataType: telemetrytypes.FieldDataTypeString,
|
||||
SemconvMembers: []string{"deployment.environment.name", "deployment.environment"},
|
||||
}
|
||||
sb := sqlbuilder.NewSelectBuilder()
|
||||
|
||||
conditions, _, err := NewConditionBuilder(NewFieldMapper()).ConditionFor(
|
||||
context.Background(), valuer.UUID{}, 0, 0, key,
|
||||
map[string][]*telemetrytypes.TelemetryFieldKey{key.Name: {key}},
|
||||
qbtypes.ConditionBuilderOptions{}, qbtypes.FilterOperatorEqual, "production", sb,
|
||||
)
|
||||
require.NoError(t, err)
|
||||
sb.Where(conditions...)
|
||||
sql, args := sb.BuildWithFlavor(sqlbuilder.ClickHouse)
|
||||
|
||||
assert.Contains(t, sql, "COALESCE(NULLIF(simpleJSONExtractString(labels, 'deployment.environment.name'), ''), NULLIF(simpleJSONExtractString(labels, 'deployment.environment'), ''), '') = ? AND (labels LIKE ? OR labels LIKE ?) AND (labels LIKE ? OR labels LIKE ?)")
|
||||
assert.Equal(t, []any{
|
||||
"production",
|
||||
"%deployment.environment.name%",
|
||||
"%deployment.environment%",
|
||||
`%deployment.environment.name":"production%`,
|
||||
`%deployment.environment":"production%`,
|
||||
}, args)
|
||||
}
|
||||
|
||||
func TestFamilyNotEqualIncludesKeylessRows(t *testing.T) {
|
||||
key := &telemetrytypes.TelemetryFieldKey{
|
||||
Name: "deployment.environment.name",
|
||||
Signal: telemetrytypes.SignalTraces,
|
||||
FieldContext: telemetrytypes.FieldContextResource,
|
||||
FieldDataType: telemetrytypes.FieldDataTypeString,
|
||||
SemconvMembers: []string{"deployment.environment.name", "deployment.environment"},
|
||||
}
|
||||
sb := sqlbuilder.NewSelectBuilder()
|
||||
|
||||
conditions, _, err := NewConditionBuilder(NewFieldMapper()).ConditionFor(
|
||||
context.Background(), valuer.UUID{}, 0, 0, key,
|
||||
map[string][]*telemetrytypes.TelemetryFieldKey{key.Name: {key}},
|
||||
qbtypes.ConditionBuilderOptions{}, qbtypes.FilterOperatorNotEqual, "staging", sb,
|
||||
)
|
||||
require.NoError(t, err)
|
||||
sb.Where(conditions...)
|
||||
sql, args := sb.BuildWithFlavor(sqlbuilder.ClickHouse)
|
||||
|
||||
assert.Contains(t, sql, "COALESCE(NULLIF(simpleJSONExtractString(labels, 'deployment.environment.name'), ''), NULLIF(simpleJSONExtractString(labels, 'deployment.environment'), ''), '') <> ?")
|
||||
assert.Equal(t, []any{"staging"}, args)
|
||||
}
|
||||
|
||||
func TestFamilyNotInIncludesKeylessRows(t *testing.T) {
|
||||
key := &telemetrytypes.TelemetryFieldKey{
|
||||
Name: "deployment.environment.name",
|
||||
Signal: telemetrytypes.SignalTraces,
|
||||
FieldContext: telemetrytypes.FieldContextResource,
|
||||
FieldDataType: telemetrytypes.FieldDataTypeString,
|
||||
SemconvMembers: []string{"deployment.environment.name", "deployment.environment"},
|
||||
}
|
||||
sb := sqlbuilder.NewSelectBuilder()
|
||||
|
||||
conditions, _, err := NewConditionBuilder(NewFieldMapper()).ConditionFor(
|
||||
context.Background(), valuer.UUID{}, 0, 0, key,
|
||||
map[string][]*telemetrytypes.TelemetryFieldKey{key.Name: {key}},
|
||||
qbtypes.ConditionBuilderOptions{}, qbtypes.FilterOperatorNotIn, []any{"staging", "dev"}, sb,
|
||||
)
|
||||
require.NoError(t, err)
|
||||
sb.Where(conditions...)
|
||||
sql, args := sb.BuildWithFlavor(sqlbuilder.ClickHouse)
|
||||
|
||||
assert.Contains(t, sql, "(COALESCE(NULLIF(simpleJSONExtractString(labels, 'deployment.environment.name'), ''), NULLIF(simpleJSONExtractString(labels, 'deployment.environment'), ''), '') <> ? AND COALESCE(NULLIF(simpleJSONExtractString(labels, 'deployment.environment.name'), ''), NULLIF(simpleJSONExtractString(labels, 'deployment.environment'), ''), '') <> ?)")
|
||||
assert.Equal(t, []any{"staging", "dev"}, args)
|
||||
}
|
||||
|
||||
func TestFamilyNotLikeIncludesKeylessRows(t *testing.T) {
|
||||
key := &telemetrytypes.TelemetryFieldKey{
|
||||
Name: "deployment.environment.name",
|
||||
Signal: telemetrytypes.SignalTraces,
|
||||
FieldContext: telemetrytypes.FieldContextResource,
|
||||
FieldDataType: telemetrytypes.FieldDataTypeString,
|
||||
SemconvMembers: []string{"deployment.environment.name", "deployment.environment"},
|
||||
}
|
||||
sb := sqlbuilder.NewSelectBuilder()
|
||||
|
||||
conditions, _, err := NewConditionBuilder(NewFieldMapper()).ConditionFor(
|
||||
context.Background(), valuer.UUID{}, 0, 0, key,
|
||||
map[string][]*telemetrytypes.TelemetryFieldKey{key.Name: {key}},
|
||||
qbtypes.ConditionBuilderOptions{}, qbtypes.FilterOperatorNotLike, "%stag%", sb,
|
||||
)
|
||||
require.NoError(t, err)
|
||||
sb.Where(conditions...)
|
||||
sql, args := sb.BuildWithFlavor(sqlbuilder.ClickHouse)
|
||||
|
||||
assert.Contains(t, sql, "LOWER(COALESCE(NULLIF(simpleJSONExtractString(labels, 'deployment.environment.name'), ''), NULLIF(simpleJSONExtractString(labels, 'deployment.environment'), ''), '')) NOT LIKE LOWER(?)")
|
||||
assert.Equal(t, []any{"%stag%"}, args)
|
||||
}
|
||||
|
||||
func TestFamilyNotContainsIncludesKeylessRows(t *testing.T) {
|
||||
key := &telemetrytypes.TelemetryFieldKey{
|
||||
Name: "deployment.environment.name",
|
||||
Signal: telemetrytypes.SignalTraces,
|
||||
FieldContext: telemetrytypes.FieldContextResource,
|
||||
FieldDataType: telemetrytypes.FieldDataTypeString,
|
||||
SemconvMembers: []string{"deployment.environment.name", "deployment.environment"},
|
||||
}
|
||||
sb := sqlbuilder.NewSelectBuilder()
|
||||
|
||||
conditions, _, err := NewConditionBuilder(NewFieldMapper()).ConditionFor(
|
||||
context.Background(), valuer.UUID{}, 0, 0, key,
|
||||
map[string][]*telemetrytypes.TelemetryFieldKey{key.Name: {key}},
|
||||
qbtypes.ConditionBuilderOptions{}, qbtypes.FilterOperatorNotContains, "stag", sb,
|
||||
)
|
||||
require.NoError(t, err)
|
||||
sb.Where(conditions...)
|
||||
sql, args := sb.BuildWithFlavor(sqlbuilder.ClickHouse)
|
||||
|
||||
assert.Contains(t, sql, "LOWER(COALESCE(NULLIF(simpleJSONExtractString(labels, 'deployment.environment.name'), ''), NULLIF(simpleJSONExtractString(labels, 'deployment.environment'), ''), '')) NOT LIKE LOWER(?)")
|
||||
assert.Equal(t, []any{"%stag%"}, args)
|
||||
}
|
||||
|
||||
func TestFamilyNotRegexpIncludesKeylessRows(t *testing.T) {
|
||||
key := &telemetrytypes.TelemetryFieldKey{
|
||||
Name: "deployment.environment.name",
|
||||
Signal: telemetrytypes.SignalTraces,
|
||||
FieldContext: telemetrytypes.FieldContextResource,
|
||||
FieldDataType: telemetrytypes.FieldDataTypeString,
|
||||
SemconvMembers: []string{"deployment.environment.name", "deployment.environment"},
|
||||
}
|
||||
sb := sqlbuilder.NewSelectBuilder()
|
||||
|
||||
conditions, _, err := NewConditionBuilder(NewFieldMapper()).ConditionFor(
|
||||
context.Background(), valuer.UUID{}, 0, 0, key,
|
||||
map[string][]*telemetrytypes.TelemetryFieldKey{key.Name: {key}},
|
||||
qbtypes.ConditionBuilderOptions{}, qbtypes.FilterOperatorNotRegexp, "stag.*", sb,
|
||||
)
|
||||
require.NoError(t, err)
|
||||
sb.Where(conditions...)
|
||||
sql, args := sb.BuildWithFlavor(sqlbuilder.ClickHouse)
|
||||
|
||||
assert.Contains(t, sql, "NOT match(COALESCE(NULLIF(simpleJSONExtractString(labels, 'deployment.environment.name'), ''), NULLIF(simpleJSONExtractString(labels, 'deployment.environment'), ''), ''), ?)")
|
||||
assert.Equal(t, []any{"stag.*"}, args)
|
||||
}
|
||||
|
||||
func TestFamilyExistsChecksEveryMember(t *testing.T) {
|
||||
key := &telemetrytypes.TelemetryFieldKey{
|
||||
Name: "deployment.environment.name",
|
||||
Signal: telemetrytypes.SignalTraces,
|
||||
FieldContext: telemetrytypes.FieldContextResource,
|
||||
FieldDataType: telemetrytypes.FieldDataTypeString,
|
||||
SemconvMembers: []string{"deployment.environment.name", "deployment.environment"},
|
||||
}
|
||||
sb := sqlbuilder.NewSelectBuilder()
|
||||
|
||||
conditions, _, err := NewConditionBuilder(NewFieldMapper()).ConditionFor(
|
||||
context.Background(), valuer.UUID{}, 0, 0, key,
|
||||
map[string][]*telemetrytypes.TelemetryFieldKey{key.Name: {key}},
|
||||
qbtypes.ConditionBuilderOptions{}, qbtypes.FilterOperatorExists, nil, sb,
|
||||
)
|
||||
require.NoError(t, err)
|
||||
sb.Where(conditions...)
|
||||
sql, args := sb.BuildWithFlavor(sqlbuilder.ClickHouse)
|
||||
|
||||
assert.Contains(t, sql, "(simpleJSONHas(labels, 'deployment.environment.name') = ? OR simpleJSONHas(labels, 'deployment.environment') = ?) AND (labels LIKE ? OR labels LIKE ?)")
|
||||
assert.Equal(t, []any{true, true, "%deployment.environment.name%", "%deployment.environment%"}, args)
|
||||
}
|
||||
|
||||
func TestFamilyNotExistsChecksEveryMember(t *testing.T) {
|
||||
key := &telemetrytypes.TelemetryFieldKey{
|
||||
Name: "deployment.environment.name",
|
||||
Signal: telemetrytypes.SignalTraces,
|
||||
FieldContext: telemetrytypes.FieldContextResource,
|
||||
FieldDataType: telemetrytypes.FieldDataTypeString,
|
||||
SemconvMembers: []string{"deployment.environment.name", "deployment.environment"},
|
||||
}
|
||||
sb := sqlbuilder.NewSelectBuilder()
|
||||
|
||||
conditions, _, err := NewConditionBuilder(NewFieldMapper()).ConditionFor(
|
||||
context.Background(), valuer.UUID{}, 0, 0, key,
|
||||
map[string][]*telemetrytypes.TelemetryFieldKey{key.Name: {key}},
|
||||
qbtypes.ConditionBuilderOptions{}, qbtypes.FilterOperatorNotExists, nil, sb,
|
||||
)
|
||||
require.NoError(t, err)
|
||||
sb.Where(conditions...)
|
||||
sql, args := sb.BuildWithFlavor(sqlbuilder.ClickHouse)
|
||||
|
||||
assert.Contains(t, sql, "simpleJSONHas(labels, 'deployment.environment.name') <> ? AND simpleJSONHas(labels, 'deployment.environment') <> ?")
|
||||
assert.Equal(t, []any{true, true}, args)
|
||||
}
|
||||
|
||||
func TestLogSemconvFamilyResolvesEveryStoredMember(t *testing.T) {
|
||||
current := &telemetrytypes.TelemetryFieldKey{
|
||||
Name: "deployment.environment.name",
|
||||
Signal: telemetrytypes.SignalLogs,
|
||||
FieldContext: telemetrytypes.FieldContextResource,
|
||||
FieldDataType: telemetrytypes.FieldDataTypeString,
|
||||
}
|
||||
old := &telemetrytypes.TelemetryFieldKey{
|
||||
Name: "deployment.environment",
|
||||
Signal: telemetrytypes.SignalLogs,
|
||||
FieldContext: telemetrytypes.FieldContextResource,
|
||||
FieldDataType: telemetrytypes.FieldDataTypeString,
|
||||
}
|
||||
sb := sqlbuilder.NewSelectBuilder()
|
||||
|
||||
conditions, _, err := NewConditionBuilder(NewFieldMapper()).ConditionFor(
|
||||
context.Background(), valuer.UUID{}, 0, 0, current,
|
||||
map[string][]*telemetrytypes.TelemetryFieldKey{
|
||||
current.Name: {current},
|
||||
old.Name: {old},
|
||||
},
|
||||
qbtypes.ConditionBuilderOptions{}, qbtypes.FilterOperatorEqual, "production", sb,
|
||||
)
|
||||
require.NoError(t, err)
|
||||
sb.Where(conditions...)
|
||||
sql, args := sb.BuildWithFlavor(sqlbuilder.ClickHouse)
|
||||
|
||||
assert.Contains(t, sql, "COALESCE(NULLIF(simpleJSONExtractString(labels, 'deployment.environment.name'), ''), NULLIF(simpleJSONExtractString(labels, 'deployment.environment'), ''), '') = ? AND (labels LIKE ? OR labels LIKE ?) AND (labels LIKE ? OR labels LIKE ?)")
|
||||
assert.Equal(t, []any{
|
||||
"production",
|
||||
"%deployment.environment.name%",
|
||||
"%deployment.environment%",
|
||||
`%deployment.environment.name":"production%`,
|
||||
`%deployment.environment":"production%`,
|
||||
}, args)
|
||||
}
|
||||
|
||||
@@ -3,8 +3,10 @@ package resourcefilter
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"strings"
|
||||
|
||||
schema "github.com/SigNoz/signoz-otel-collector/cmd/signozschemamigrator/schema_migrator"
|
||||
"github.com/SigNoz/signoz/pkg/semconv"
|
||||
qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
|
||||
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
|
||||
"github.com/SigNoz/signoz/pkg/valuer"
|
||||
@@ -32,6 +34,21 @@ func NewFieldMapper() *defaultFieldMapper {
|
||||
return &defaultFieldMapper{}
|
||||
}
|
||||
|
||||
func resourceSemconvMembers(key *telemetrytypes.TelemetryFieldKey) []string {
|
||||
if (key.Signal != telemetrytypes.SignalTraces && key.Signal != telemetrytypes.SignalLogs) ||
|
||||
key.FieldContext != telemetrytypes.FieldContextResource {
|
||||
return []string{key.Name}
|
||||
}
|
||||
if len(key.SemconvMembers) > 0 {
|
||||
return key.SemconvMembers
|
||||
}
|
||||
return semconv.Members(semconv.KindAttribute, telemetrytypes.FieldKeySelector{
|
||||
Name: key.Name,
|
||||
Signal: telemetrytypes.SignalTraces,
|
||||
FieldContext: telemetrytypes.FieldContextResource,
|
||||
})
|
||||
}
|
||||
|
||||
func (m *defaultFieldMapper) getColumn(
|
||||
_ context.Context,
|
||||
_, _ uint64,
|
||||
@@ -66,7 +83,15 @@ func (m *defaultFieldMapper) FieldFor(
|
||||
return "", err
|
||||
}
|
||||
if key.FieldContext == telemetrytypes.FieldContextResource {
|
||||
return fmt.Sprintf("simpleJSONExtractString(%s, '%s')", columns[0].Name, key.Name), nil
|
||||
members := resourceSemconvMembers(key)
|
||||
if len(members) > 1 {
|
||||
values := make([]string, 0, len(members))
|
||||
for _, member := range members {
|
||||
values = append(values, fmt.Sprintf("NULLIF(simpleJSONExtractString(%s, '%s'), '')", columns[0].Name, member))
|
||||
}
|
||||
return "COALESCE(" + strings.Join(values, ", ") + ", '')", nil
|
||||
}
|
||||
return fmt.Sprintf("simpleJSONExtractString(%s, '%s')", columns[0].Name, members[0]), nil
|
||||
}
|
||||
return columns[0].Name, nil
|
||||
}
|
||||
|
||||
@@ -14,6 +14,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/semconv"
|
||||
"github.com/SigNoz/signoz/pkg/telemetryschema/audittelemetryschema"
|
||||
"github.com/SigNoz/signoz/pkg/telemetryschema/logstelemetryschema"
|
||||
"github.com/SigNoz/signoz/pkg/telemetryschema/metertelemetryschema"
|
||||
@@ -151,6 +152,26 @@ func (t *telemetryMetaStore) tracesTblStatementToFieldKeys(ctx context.Context)
|
||||
return materialisedKeys, nil
|
||||
}
|
||||
|
||||
func attributeSemconvMembers(name string, signal telemetrytypes.Signal, fieldContext telemetrytypes.FieldContext) []string {
|
||||
return semconv.AttributeMembers(telemetrytypes.FieldKeySelector{
|
||||
Name: name,
|
||||
Signal: signal,
|
||||
FieldContext: fieldContext,
|
||||
})
|
||||
}
|
||||
|
||||
func traceSemconvMembers(name string, fieldContext telemetrytypes.FieldContext) []string {
|
||||
return attributeSemconvMembers(name, telemetrytypes.SignalTraces, fieldContext)
|
||||
}
|
||||
|
||||
func inStrings(sb *sqlbuilder.SelectBuilder, column string, values []string) string {
|
||||
args := make([]any, 0, len(values))
|
||||
for _, value := range values {
|
||||
args = append(args, value)
|
||||
}
|
||||
return sb.In(column, args...)
|
||||
}
|
||||
|
||||
// getTracesKeys returns the keys from the spans that match the field selection criteria.
|
||||
func (t *telemetryMetaStore) getTracesKeys(ctx context.Context, fieldKeySelectors []*telemetrytypes.FieldKeySelector) ([]*telemetrytypes.TelemetryFieldKey, bool, error) {
|
||||
ctx = ctxtypes.NewContextWithCommentVals(ctx, map[string]string{
|
||||
@@ -1312,6 +1333,37 @@ func (t *telemetryMetaStore) GetKeysMulti(ctx context.Context, orgID valuer.UUID
|
||||
}
|
||||
}
|
||||
|
||||
// GetKeys backs key suggestions and remains literal. Internal multi-key
|
||||
// lookups expand selectors so query builders can inspect stored family members.
|
||||
expandSelectors := func(selectors []*telemetrytypes.FieldKeySelector, signal telemetrytypes.Signal) []*telemetrytypes.FieldKeySelector {
|
||||
expanded := make([]*telemetrytypes.FieldKeySelector, 0, len(selectors))
|
||||
for _, selector := range selectors {
|
||||
metricNames := []string{""}
|
||||
if signal == telemetrytypes.SignalMetrics && selector.MetricContext != nil && selector.MetricContext.MetricName != "" {
|
||||
metricNames = semconv.MetricNames(selector.MetricContext.MetricName)
|
||||
}
|
||||
for _, member := range attributeSemconvMembers(selector.Name, signal, selector.FieldContext) {
|
||||
for _, metricName := range metricNames {
|
||||
memberSelector := *selector
|
||||
memberSelector.Name = member
|
||||
if selector.MetricContext != nil {
|
||||
metricContext := *selector.MetricContext
|
||||
if metricName != "" {
|
||||
metricContext.MetricName = metricName
|
||||
}
|
||||
memberSelector.MetricContext = &metricContext
|
||||
}
|
||||
expanded = append(expanded, &memberSelector)
|
||||
}
|
||||
}
|
||||
}
|
||||
return expanded
|
||||
}
|
||||
logsSelectors = expandSelectors(logsSelectors, telemetrytypes.SignalLogs)
|
||||
tracesSelectors = expandSelectors(tracesSelectors, telemetrytypes.SignalTraces)
|
||||
metricsSelectors = expandSelectors(metricsSelectors, telemetrytypes.SignalMetrics)
|
||||
meterSourceMetricsSelectors = expandSelectors(meterSourceMetricsSelectors, telemetrytypes.SignalMetrics)
|
||||
|
||||
logsKeys, logsComplete, err := t.getLogsKeys(ctx, orgID, logsSelectors)
|
||||
if err != nil {
|
||||
return nil, false, err
|
||||
@@ -1542,7 +1594,16 @@ func (t *telemetryMetaStore) getSpanFieldValues(ctx context.Context, fieldValueS
|
||||
sb := sqlbuilder.Select("DISTINCT string_value, number_value").From(t.tracesDBName + "." + t.tracesFieldsTblName)
|
||||
|
||||
if fieldValueSelector.Name != "" {
|
||||
sb.Where(sb.E("tag_key", fieldValueSelector.Name))
|
||||
members := traceSemconvMembers(fieldValueSelector.Name, fieldValueSelector.FieldContext)
|
||||
if len(members) == 1 {
|
||||
sb.Where(sb.E("tag_key", members[0]))
|
||||
} else {
|
||||
memberValues := make([]any, 0, len(members))
|
||||
for _, member := range members {
|
||||
memberValues = append(memberValues, member)
|
||||
}
|
||||
sb.Where(sb.In("tag_key", memberValues...))
|
||||
}
|
||||
}
|
||||
|
||||
// now look at the field context
|
||||
@@ -1632,7 +1693,12 @@ func (t *telemetryMetaStore) getLogFieldValues(ctx context.Context, fieldValueSe
|
||||
sb := sqlbuilder.Select("DISTINCT string_value, number_value").From(t.logsDBName + "." + t.logsFieldsTblName)
|
||||
|
||||
if fieldValueSelector.Name != "" {
|
||||
sb.Where(sb.E("tag_key", fieldValueSelector.Name))
|
||||
members := attributeSemconvMembers(fieldValueSelector.Name, telemetrytypes.SignalLogs, fieldValueSelector.FieldContext)
|
||||
memberValues := make([]any, 0, len(members))
|
||||
for _, member := range members {
|
||||
memberValues = append(memberValues, member)
|
||||
}
|
||||
sb.Where(sb.In("tag_key", memberValues...))
|
||||
}
|
||||
|
||||
if fieldValueSelector.FieldContext != telemetrytypes.FieldContextUnspecified {
|
||||
@@ -1830,7 +1896,9 @@ func (t *telemetryMetaStore) getMetricFieldValues(ctx context.Context, orgID val
|
||||
From(t.metricsDBName + "." + t.metricsFieldsTblName)
|
||||
|
||||
if fieldValueSelector.Name != "" {
|
||||
sb.Where(sb.E("attr_name", fieldValueSelector.Name))
|
||||
sb.Where(inStrings(sb, "attr_name", attributeSemconvMembers(
|
||||
fieldValueSelector.Name, telemetrytypes.SignalMetrics, fieldValueSelector.FieldContext,
|
||||
)))
|
||||
}
|
||||
|
||||
if fieldValueSelector.FieldContext != telemetrytypes.FieldContextUnspecified {
|
||||
@@ -1842,7 +1910,7 @@ func (t *telemetryMetaStore) getMetricFieldValues(ctx context.Context, orgID val
|
||||
}
|
||||
|
||||
if fieldValueSelector.MetricContext != nil && fieldValueSelector.MetricContext.MetricName != "" {
|
||||
sb.Where(sb.E("metric_name", fieldValueSelector.MetricContext.MetricName))
|
||||
sb.Where(inStrings(sb, "metric_name", semconv.MetricNames(fieldValueSelector.MetricContext.MetricName)))
|
||||
}
|
||||
if fieldValueSelector.MetricContext != nil && fieldValueSelector.MetricContext.MetricNamespace != "" {
|
||||
sb.Where(sb.Like("metric_name", escapeForLike(fieldValueSelector.MetricContext.MetricNamespace)+"%"))
|
||||
@@ -1976,7 +2044,7 @@ func (t *telemetryMetaStore) getIntrinsicMetricFieldValuesForTable(ctx context.C
|
||||
From(t.metricsDBName + "." + tableName)
|
||||
|
||||
if fieldValueSelector.MetricContext != nil && fieldValueSelector.MetricContext.MetricName != "" {
|
||||
sb.Where(sb.E("metric_name", fieldValueSelector.MetricContext.MetricName))
|
||||
sb.Where(inStrings(sb, "metric_name", semconv.MetricNames(fieldValueSelector.MetricContext.MetricName)))
|
||||
}
|
||||
if fieldValueSelector.MetricContext != nil && fieldValueSelector.MetricContext.MetricNamespace != "" {
|
||||
sb.Where(sb.Like("metric_name", escapeForLike(fieldValueSelector.MetricContext.MetricNamespace)+"%"))
|
||||
@@ -2038,13 +2106,18 @@ func (t *telemetryMetaStore) getMeterSourceMetricFieldValues(ctx context.Context
|
||||
sb := sqlbuilder.Select("DISTINCT arrayJoin(JSONExtractKeysAndValues(labels, 'String')) AS attr").
|
||||
From(t.meterDBName + "." + t.meterFieldsTblName)
|
||||
|
||||
duplicateFactor := 1
|
||||
if fieldValueSelector.Name != "" {
|
||||
sb.Where(sb.E("attr.1", fieldValueSelector.Name))
|
||||
members := attributeSemconvMembers(fieldValueSelector.Name, telemetrytypes.SignalMetrics, fieldValueSelector.FieldContext)
|
||||
duplicateFactor *= len(members)
|
||||
sb.Where(inStrings(sb, "attr.1", members))
|
||||
}
|
||||
sb.Where(sb.NotLike("attr.1", "\\_\\_%"))
|
||||
|
||||
if fieldValueSelector.MetricContext != nil && fieldValueSelector.MetricContext.MetricName != "" {
|
||||
sb.Where(sb.E("metric_name", fieldValueSelector.MetricContext.MetricName))
|
||||
metricNames := semconv.MetricNames(fieldValueSelector.MetricContext.MetricName)
|
||||
duplicateFactor *= len(metricNames)
|
||||
sb.Where(inStrings(sb, "metric_name", metricNames))
|
||||
}
|
||||
if fieldValueSelector.MetricContext != nil && fieldValueSelector.MetricContext.MetricNamespace != "" {
|
||||
sb.Where(sb.Like("metric_name", escapeForLike(fieldValueSelector.MetricContext.MetricNamespace)+"%"))
|
||||
@@ -2063,8 +2136,10 @@ func (t *telemetryMetaStore) getMeterSourceMetricFieldValues(ctx context.Context
|
||||
if limit == 0 {
|
||||
limit = 50
|
||||
}
|
||||
// query one extra to check if we hit the limit
|
||||
sb.Limit(limit + 1)
|
||||
// A value can be present under several physical spellings; over-fetch and
|
||||
// de-duplicate after scanning so one family cannot consume the result limit.
|
||||
dbLimit := limit * duplicateFactor
|
||||
sb.Limit(dbLimit + 1)
|
||||
|
||||
query, args := sb.BuildWithFlavor(sqlbuilder.ClickHouse)
|
||||
rows, err := t.telemetrystore.ClickhouseDB().Query(ctx, query, args...)
|
||||
@@ -2074,11 +2149,13 @@ func (t *telemetryMetaStore) getMeterSourceMetricFieldValues(ctx context.Context
|
||||
defer rows.Close()
|
||||
|
||||
values := &telemetrytypes.TelemetryFieldValues{}
|
||||
seen := make(map[string]bool)
|
||||
rowCount := 0
|
||||
uniqueCount := 0
|
||||
for rows.Next() {
|
||||
rowCount++
|
||||
// reached the limit, we know there are more results
|
||||
if rowCount > limit {
|
||||
if rowCount > dbLimit {
|
||||
break
|
||||
}
|
||||
|
||||
@@ -2087,12 +2164,16 @@ func (t *telemetryMetaStore) getMeterSourceMetricFieldValues(ctx context.Context
|
||||
return nil, false, errors.Wrap(err, errors.TypeInternal, errors.CodeInternal, ErrFailedToGetMeterValues.Error())
|
||||
}
|
||||
if len(attribute) > 1 {
|
||||
values.StringValues = append(values.StringValues, attribute[1])
|
||||
if !seen[attribute[1]] && uniqueCount < limit {
|
||||
values.StringValues = append(values.StringValues, attribute[1])
|
||||
seen[attribute[1]] = true
|
||||
uniqueCount++
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// hit the limit?
|
||||
complete := rowCount <= limit
|
||||
complete := rowCount <= dbLimit && uniqueCount < limit
|
||||
return values, complete, nil
|
||||
}
|
||||
|
||||
|
||||
@@ -12,6 +12,7 @@ import (
|
||||
"github.com/SigNoz/signoz/pkg/telemetrystore"
|
||||
"github.com/SigNoz/signoz/pkg/telemetrystore/telemetrystoretest"
|
||||
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
|
||||
"github.com/SigNoz/signoz/pkg/valuer"
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
@@ -83,3 +84,38 @@ func TestGetFirstSeenFromMetricMetadata(t *testing.T) {
|
||||
t.Errorf("there were unfulfilled expectations: %s", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestGetAllValuesReturnsValuesFromEveryTraceSemconvFamilyMember(t *testing.T) {
|
||||
mockTelemetryStore := telemetrystoretest.New(telemetrystore.Config{}, ®exMatcher{})
|
||||
mock := mockTelemetryStore.Mock()
|
||||
|
||||
metadata := NewTelemetryMetaStore(
|
||||
instrumentationtest.New().ToProviderSettings(),
|
||||
mockTelemetryStore,
|
||||
flaggertest.New(t),
|
||||
)
|
||||
|
||||
mock.ExpectQuery(`SELECT DISTINCT string_value, number_value FROM signoz_traces\.distributed_tag_attributes_v2 WHERE tag_key IN \(\?, \?\) AND tag_type = \? AND tag_data_type = \? LIMIT \?`).
|
||||
WithArgs("deployment.environment.name", "deployment.environment", "resource", "string", 51).
|
||||
WillReturnRows(cmock.NewRows([]cmock.ColumnType{
|
||||
{Name: "string_value", Type: "String"},
|
||||
{Name: "number_value", Type: "Float64"},
|
||||
}, [][]any{
|
||||
{"production", float64(0)},
|
||||
{"staging", float64(0)},
|
||||
{"production", float64(0)},
|
||||
}))
|
||||
|
||||
values, complete, err := metadata.GetAllValues(context.Background(), valuer.UUID{}, &telemetrytypes.FieldValueSelector{
|
||||
FieldKeySelector: &telemetrytypes.FieldKeySelector{
|
||||
Signal: telemetrytypes.SignalTraces,
|
||||
FieldContext: telemetrytypes.FieldContextResource,
|
||||
FieldDataType: telemetrytypes.FieldDataTypeString,
|
||||
Name: "deployment.environment",
|
||||
},
|
||||
})
|
||||
require.NoError(t, err)
|
||||
assert.True(t, complete)
|
||||
assert.Equal(t, []string{"production", "staging"}, values.StringValues)
|
||||
assert.NoError(t, mock.ExpectationsWereMet(), "all expected metadata queries should be executed")
|
||||
}
|
||||
|
||||
@@ -89,10 +89,12 @@ func (v *TelemetryFieldVisitor) VisitColumnDef(expr *parser.ColumnDef) error {
|
||||
|
||||
// Create and store the TelemetryFieldKey
|
||||
field := &telemetrytypes.TelemetryFieldKey{
|
||||
Name: fieldName,
|
||||
FieldContext: fieldContext,
|
||||
FieldDataType: fieldDataType,
|
||||
Materialized: true,
|
||||
Name: fieldName,
|
||||
FieldContext: fieldContext,
|
||||
FieldDataType: fieldDataType,
|
||||
Materialized: true,
|
||||
MaterializedColumnName: columnName,
|
||||
MaterializedSemconv: strings.Count(defaultExprStr, "['") > 1,
|
||||
}
|
||||
|
||||
v.Fields = append(v.Fields, field)
|
||||
|
||||
@@ -5,8 +5,25 @@ import (
|
||||
"testing"
|
||||
|
||||
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
func TestExtractFieldKeyPreservesHistoricalMaterializedColumnName(t *testing.T) {
|
||||
statement := `CREATE TABLE signoz_traces.signoz_index_v3
|
||||
(
|
||||
attributes_string Map(LowCardinality(String), String),
|
||||
` + "`attribute_string_db$$system`" + ` LowCardinality(String)
|
||||
DEFAULT if(mapContains(attributes_string, 'db.system.name'), attributes_string['db.system.name'], attributes_string['db.system'])
|
||||
) ENGINE = MergeTree ORDER BY tuple()`
|
||||
keys, err := ExtractFieldKeysFromTblStatement(statement)
|
||||
require.NoError(t, err, "table statement should parse")
|
||||
require.Len(t, keys, 1, "table statement should contain one materialized key")
|
||||
assert.Equal(t, "db.system.name", keys[0].Name)
|
||||
assert.Equal(t, "attribute_string_db$$system", keys[0].MaterializedColumnName)
|
||||
assert.True(t, keys[0].MaterializedSemconv)
|
||||
}
|
||||
|
||||
func TestExtractFieldKeysFromTblStatement(t *testing.T) {
|
||||
|
||||
var statement = `CREATE TABLE signoz_logs.logs_v2
|
||||
|
||||
@@ -453,6 +453,9 @@ func (c *conditionBuilder) ConditionFor(
|
||||
sb *sqlbuilder.SelectBuilder,
|
||||
) ([]string, []string, error) {
|
||||
matches := querybuilder.MatchingFieldKeys(key, fieldKeys)
|
||||
if options.ExactSemconv {
|
||||
matches = querybuilder.MatchingFieldKeysExact(key, fieldKeys)
|
||||
}
|
||||
skipResourceFilter := options.SkipResourceFilter
|
||||
|
||||
// search() resolves its own (optional) scope; handle it before key resolution.
|
||||
@@ -499,6 +502,9 @@ func (c *conditionBuilder) ConditionFor(
|
||||
warnings = append(warnings, querybuilder.NewKeyNotFoundWarning(key.Name))
|
||||
}
|
||||
}
|
||||
if options.ExactSemconv {
|
||||
keys = querybuilder.ExactSemconvKeys(keys)
|
||||
}
|
||||
|
||||
if skipResourceFilter && !synthesized {
|
||||
filtered := make([]*telemetrytypes.TelemetryFieldKey, 0, len(keys))
|
||||
|
||||
@@ -10,6 +10,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/semconv"
|
||||
"github.com/SigNoz/signoz/pkg/types/featuretypes"
|
||||
qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
|
||||
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
|
||||
@@ -67,6 +68,20 @@ type fieldMapper struct {
|
||||
fl flagger.Flagger
|
||||
}
|
||||
|
||||
func logSemconvMembers(key *telemetrytypes.TelemetryFieldKey) []string {
|
||||
if key.FieldContext != telemetrytypes.FieldContextResource && key.FieldContext != telemetrytypes.FieldContextAttribute {
|
||||
return []string{key.Name}
|
||||
}
|
||||
if len(key.SemconvMembers) > 0 {
|
||||
return key.SemconvMembers
|
||||
}
|
||||
return semconv.AttributeMembers(telemetrytypes.FieldKeySelector{
|
||||
Name: key.Name,
|
||||
Signal: telemetrytypes.SignalLogs,
|
||||
FieldContext: key.FieldContext,
|
||||
})
|
||||
}
|
||||
|
||||
func NewFieldMapper(fl flagger.Flagger) qbtypes.FieldMapper {
|
||||
return &fieldMapper{fl: fl}
|
||||
}
|
||||
@@ -141,8 +156,22 @@ func (m *fieldMapper) FieldFor(ctx context.Context, orgID valuer.UUID, tsStart,
|
||||
case schema.ColumnTypeEnumJSON:
|
||||
switch key.FieldContext {
|
||||
case telemetrytypes.FieldContextResource:
|
||||
exprs = append(exprs, fmt.Sprintf("%s.`%s`::String", columnName, key.Name))
|
||||
existExpr = append(existExpr, fmt.Sprintf("%s.`%s` IS NOT NULL", columnName, key.Name))
|
||||
members := logSemconvMembers(key)
|
||||
if len(members) > 1 {
|
||||
values := make([]string, 0, len(members))
|
||||
guards := make([]string, 0, len(members))
|
||||
for _, member := range members {
|
||||
values = append(values, fmt.Sprintf("NULLIF(%s.`%s`::String, '')", columnName, member))
|
||||
guards = append(guards, fmt.Sprintf("%s.`%s` IS NOT NULL", columnName, member))
|
||||
}
|
||||
// Missing Dynamic paths are NULL, so this family expression must
|
||||
// retain the same NULL result as a single JSON-path lookup.
|
||||
exprs = append(exprs, "COALESCE("+strings.Join(values, ", ")+")")
|
||||
existExpr = append(existExpr, "("+strings.Join(guards, " OR ")+")")
|
||||
} else {
|
||||
exprs = append(exprs, fmt.Sprintf("%s.`%s`::String", columnName, members[0]))
|
||||
existExpr = append(existExpr, fmt.Sprintf("%s.`%s` IS NOT NULL", columnName, members[0]))
|
||||
}
|
||||
case telemetrytypes.FieldContextBody:
|
||||
if key.Name == messageSubField {
|
||||
exprs = append(exprs, messageSubColumn)
|
||||
@@ -181,13 +210,34 @@ func (m *fieldMapper) FieldFor(ctx context.Context, orgID valuer.UUID, tsStart,
|
||||
|
||||
switch valueType := column.Type.(schema.MapColumnType).ValueType; valueType.GetType() {
|
||||
case schema.ColumnTypeEnumString, schema.ColumnTypeEnumBool, schema.ColumnTypeEnumFloat64:
|
||||
// a key could have been materialized, if so return the materialized column name
|
||||
if key.Materialized {
|
||||
members := logSemconvMembers(key)
|
||||
if key.Materialized && (len(members) == 1 || key.MaterializedSemconv) {
|
||||
exprs = append(exprs, telemetrytypes.FieldKeyToMaterializedColumnName(key))
|
||||
existExpr = append(existExpr, telemetrytypes.FieldKeyToMaterializedColumnNameForExists(key))
|
||||
} else if len(members) > 1 {
|
||||
guards := make([]string, 0, len(members))
|
||||
for _, member := range members {
|
||||
guards = append(guards, fmt.Sprintf("mapContains(%s, '%s')", columnName, member))
|
||||
}
|
||||
if valueType.GetType() == schema.ColumnTypeEnumString {
|
||||
values := make([]string, 0, len(members))
|
||||
for _, member := range members {
|
||||
values = append(values, fmt.Sprintf("NULLIF(%s['%s'], '')", columnName, member))
|
||||
}
|
||||
exprs = append(exprs, "COALESCE("+strings.Join(values, ", ")+", '')")
|
||||
} else {
|
||||
branches := make([]string, 0, len(members)*2+1)
|
||||
for i, member := range members {
|
||||
branches = append(branches, guards[i], fmt.Sprintf("%s['%s']", columnName, member))
|
||||
}
|
||||
// Numeric and boolean maps return zero for an absent key. If a
|
||||
// family of either type is enabled, this tail must become zero too.
|
||||
exprs = append(exprs, "multiIf("+strings.Join(branches, ", ")+", NULL)")
|
||||
}
|
||||
existExpr = append(existExpr, "("+strings.Join(guards, " OR ")+")")
|
||||
} else {
|
||||
exprs = append(exprs, fmt.Sprintf("%s['%s']", columnName, key.Name))
|
||||
existExpr = append(existExpr, fmt.Sprintf("mapContains(%s, '%s')", columnName, key.Name))
|
||||
exprs = append(exprs, fmt.Sprintf("%s['%s']", columnName, members[0]))
|
||||
existExpr = append(existExpr, fmt.Sprintf("mapContains(%s, '%s')", columnName, members[0]))
|
||||
}
|
||||
default:
|
||||
return "", errors.NewInvalidInputf(errors.CodeInvalidInput, "exists operator is not supported for map column type %s", valueType)
|
||||
|
||||
@@ -2,6 +2,7 @@ package logstelemetryschema
|
||||
|
||||
import (
|
||||
"context"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
@@ -14,6 +15,32 @@ import (
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
func TestFieldForSemconvFamily(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
fm := NewFieldMapper(flaggertest.New(t)).(*fieldMapper)
|
||||
key := &telemetrytypes.TelemetryFieldKey{
|
||||
Name: "db.system.name",
|
||||
Signal: telemetrytypes.SignalLogs,
|
||||
FieldContext: telemetrytypes.FieldContextAttribute,
|
||||
FieldDataType: telemetrytypes.FieldDataTypeString,
|
||||
}
|
||||
|
||||
expression, err := fm.FieldFor(ctx, valuer.UUID{}, 0, 0, key)
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, "COALESCE(NULLIF(attributes_string['db.system.name'], ''), NULLIF(attributes_string['db.system'], ''), '')", expression)
|
||||
assert.Less(t, strings.Index(expression, "db.system.name"), strings.Index(expression, "db.system']"), "current spelling must win")
|
||||
|
||||
exists, err := fm.existsExpressionFor(ctx, valuer.UUID{}, 0, 0, key, true)
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, "(mapContains(attributes_string, 'db.system.name') OR mapContains(attributes_string, 'db.system'))", exists)
|
||||
|
||||
exact := *key
|
||||
exact.SemconvMembers = []string{"db.system"}
|
||||
expression, err = fm.FieldFor(ctx, valuer.UUID{}, 0, 0, &exact)
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, "attributes_string['db.system']", expression)
|
||||
}
|
||||
|
||||
func TestGetColumn(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
|
||||
|
||||
@@ -4,6 +4,7 @@ import (
|
||||
"context"
|
||||
"fmt"
|
||||
"slices"
|
||||
"strings"
|
||||
|
||||
"github.com/SigNoz/signoz/pkg/errors"
|
||||
"github.com/SigNoz/signoz/pkg/querybuilder"
|
||||
@@ -135,10 +136,22 @@ func (c *conditionBuilder) conditionFor(
|
||||
return "true", nil
|
||||
}
|
||||
|
||||
if operator == qbtypes.FilterOperatorExists {
|
||||
return fmt.Sprintf("has(JSONExtractKeys(labels), '%s')", key.Name), nil
|
||||
members := metricAttributeMembers(key)
|
||||
guards := make([]string, 0, len(members))
|
||||
for _, member := range members {
|
||||
guards = append(guards, fmt.Sprintf("has(JSONExtractKeys(labels), '%s')", member))
|
||||
}
|
||||
return fmt.Sprintf("not has(JSONExtractKeys(labels), '%s')", key.Name), nil
|
||||
guard := strings.Join(guards, " OR ")
|
||||
if len(guards) > 1 {
|
||||
guard = "(" + guard + ")"
|
||||
}
|
||||
if operator == qbtypes.FilterOperatorExists {
|
||||
return guard, nil
|
||||
}
|
||||
if len(guards) == 1 {
|
||||
return "not " + guard, nil
|
||||
}
|
||||
return "NOT " + guard, nil
|
||||
}
|
||||
return "", errors.NewInvalidInputf(errors.CodeInvalidInput, "unsupported operator: %v", operator)
|
||||
}
|
||||
@@ -151,7 +164,7 @@ func (c *conditionBuilder) ConditionFor(
|
||||
endNs uint64,
|
||||
key *telemetrytypes.TelemetryFieldKey,
|
||||
fieldKeys map[string][]*telemetrytypes.TelemetryFieldKey,
|
||||
_ qbtypes.ConditionBuilderOptions,
|
||||
options qbtypes.ConditionBuilderOptions,
|
||||
operator qbtypes.FilterOperator,
|
||||
value any,
|
||||
sb *sqlbuilder.SelectBuilder,
|
||||
@@ -162,7 +175,15 @@ func (c *conditionBuilder) ConditionFor(
|
||||
return nil, nil, err
|
||||
}
|
||||
|
||||
requestedKey := *key
|
||||
if requestedKey.Signal == telemetrytypes.SignalUnspecified {
|
||||
requestedKey.Signal = telemetrytypes.SignalMetrics
|
||||
}
|
||||
key = &requestedKey
|
||||
keys := querybuilder.MatchingFieldKeys(key, fieldKeys)
|
||||
if options.ExactSemconv {
|
||||
keys = querybuilder.MatchingFieldKeysExact(key, fieldKeys)
|
||||
}
|
||||
var warnings []string
|
||||
if len(keys) == 0 {
|
||||
if _, isColumn := timeSeriesV4Columns[key.Name]; isColumn {
|
||||
@@ -180,6 +201,9 @@ func (c *conditionBuilder) ConditionFor(
|
||||
}
|
||||
}
|
||||
}
|
||||
if options.ExactSemconv {
|
||||
keys = querybuilder.ExactSemconvKeys(keys)
|
||||
}
|
||||
|
||||
conds := make([]string, 0, len(keys))
|
||||
for _, k := range keys {
|
||||
|
||||
@@ -390,3 +390,45 @@ func TestConditionForKeyNotInMetadata(t *testing.T) {
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestConditionForSemconvMetricLabels(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
fm := NewFieldMapper()
|
||||
conditionBuilder := NewConditionBuilder(fm)
|
||||
requested := telemetrytypes.TelemetryFieldKey{
|
||||
Name: "db.system.name",
|
||||
Signal: telemetrytypes.SignalMetrics,
|
||||
FieldContext: telemetrytypes.FieldContextResource,
|
||||
FieldDataType: telemetrytypes.FieldDataTypeString,
|
||||
}
|
||||
current := requested
|
||||
legacyNormalized := requested
|
||||
legacyNormalized.Name = "resource_db_system"
|
||||
fieldKeys := map[string][]*telemetrytypes.TelemetryFieldKey{
|
||||
current.Name: {¤t},
|
||||
legacyNormalized.Name: {&legacyNormalized},
|
||||
}
|
||||
|
||||
sb := sqlbuilder.NewSelectBuilder()
|
||||
conditions, warnings, err := conditionBuilder.ConditionFor(
|
||||
ctx, valuer.UUID{}, 0, 0, &requested, fieldKeys, qbtypes.ConditionBuilderOptions{},
|
||||
qbtypes.FilterOperatorEqual, "postgresql", sb,
|
||||
)
|
||||
require.NoError(t, err)
|
||||
assert.Empty(t, warnings)
|
||||
sb.Where(conditions...)
|
||||
query, args := sb.BuildWithFlavor(sqlbuilder.ClickHouse)
|
||||
assert.Contains(t, query, "COALESCE(NULLIF(JSONExtractString(labels, 'db.system.name'), ''), NULLIF(JSONExtractString(labels, 'resource_db_system'), ''), '') = ?")
|
||||
assert.Equal(t, []any{"postgresql"}, args)
|
||||
|
||||
sb = sqlbuilder.NewSelectBuilder()
|
||||
conditions, _, err = conditionBuilder.ConditionFor(
|
||||
ctx, valuer.UUID{}, 0, 0, &requested, fieldKeys, qbtypes.ConditionBuilderOptions{ExactSemconv: true},
|
||||
qbtypes.FilterOperatorExists, nil, sb,
|
||||
)
|
||||
require.NoError(t, err)
|
||||
sb.Where(conditions...)
|
||||
query, _ = sb.BuildWithFlavor(sqlbuilder.ClickHouse)
|
||||
assert.Contains(t, query, "has(JSONExtractKeys(labels), 'db.system.name')")
|
||||
assert.NotContains(t, query, "resource_db_system")
|
||||
}
|
||||
|
||||
@@ -4,8 +4,10 @@ import (
|
||||
"context"
|
||||
"fmt"
|
||||
"slices"
|
||||
"strings"
|
||||
|
||||
schema "github.com/SigNoz/signoz-otel-collector/cmd/signozschemamigrator/schema_migrator"
|
||||
"github.com/SigNoz/signoz/pkg/semconv"
|
||||
qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
|
||||
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
|
||||
"github.com/SigNoz/signoz/pkg/valuer"
|
||||
@@ -38,6 +40,23 @@ var (
|
||||
|
||||
type fieldMapper struct{}
|
||||
|
||||
func metricAttributeMembers(key *telemetrytypes.TelemetryFieldKey) []string {
|
||||
if key.FieldContext != telemetrytypes.FieldContextResource &&
|
||||
key.FieldContext != telemetrytypes.FieldContextScope &&
|
||||
key.FieldContext != telemetrytypes.FieldContextAttribute &&
|
||||
key.FieldContext != telemetrytypes.FieldContextUnspecified {
|
||||
return []string{key.Name}
|
||||
}
|
||||
if len(key.SemconvMembers) > 0 {
|
||||
return key.SemconvMembers
|
||||
}
|
||||
return semconv.AttributeMembers(telemetrytypes.FieldKeySelector{
|
||||
Name: key.Name,
|
||||
Signal: telemetrytypes.SignalMetrics,
|
||||
FieldContext: key.FieldContext,
|
||||
})
|
||||
}
|
||||
|
||||
// CandidateKeys returns nil: metrics has no attribute-map fallback, so a context-missing
|
||||
// key stays unresolved and the caller errors.
|
||||
func (m *fieldMapper) CandidateKeys(_ context.Context, _ valuer.UUID, _ *telemetrytypes.TelemetryFieldKey, _ any, _ map[string][]*telemetrytypes.TelemetryFieldKey) []*telemetrytypes.TelemetryFieldKey {
|
||||
@@ -80,19 +99,30 @@ func (m *fieldMapper) FieldFor(ctx context.Context, _ valuer.UUID, startNs, endN
|
||||
|
||||
switch key.FieldContext {
|
||||
case telemetrytypes.FieldContextResource, telemetrytypes.FieldContextScope, telemetrytypes.FieldContextAttribute:
|
||||
return fmt.Sprintf("JSONExtractString(%s, '%s')", columns[0].Name, key.Name), nil
|
||||
return metricLabelExpression(columns[0].Name, metricAttributeMembers(key)), nil
|
||||
case telemetrytypes.FieldContextMetric:
|
||||
return columns[0].Name, nil
|
||||
case telemetrytypes.FieldContextUnspecified:
|
||||
if slices.Contains(IntrinsicFields, key.Name) {
|
||||
return columns[0].Name, nil
|
||||
}
|
||||
return fmt.Sprintf("JSONExtractString(%s, '%s')", columns[0].Name, key.Name), nil
|
||||
return metricLabelExpression(columns[0].Name, metricAttributeMembers(key)), nil
|
||||
}
|
||||
|
||||
return columns[0].Name, nil
|
||||
}
|
||||
|
||||
func metricLabelExpression(columnName string, members []string) string {
|
||||
if len(members) == 1 {
|
||||
return fmt.Sprintf("JSONExtractString(%s, '%s')", columnName, members[0])
|
||||
}
|
||||
values := make([]string, 0, len(members))
|
||||
for _, member := range members {
|
||||
values = append(values, fmt.Sprintf("NULLIF(JSONExtractString(%s, '%s'), '')", columnName, member))
|
||||
}
|
||||
return "COALESCE(" + strings.Join(values, ", ") + ", '')"
|
||||
}
|
||||
|
||||
func (m *fieldMapper) ColumnFor(ctx context.Context, _ valuer.UUID, tsStart, tsEnd uint64, key *telemetrytypes.TelemetryFieldKey) ([]*schema.Column, error) {
|
||||
return m.getColumn(ctx, tsStart, tsEnd, key)
|
||||
}
|
||||
|
||||
@@ -2,6 +2,7 @@ package metricstelemetryschema
|
||||
|
||||
import (
|
||||
"context"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
schema "github.com/SigNoz/signoz-otel-collector/cmd/signozschemamigrator/schema_migrator"
|
||||
@@ -12,6 +13,32 @@ import (
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
func TestMetricLabelSemconvSpellings(t *testing.T) {
|
||||
fm := NewFieldMapper()
|
||||
key := &telemetrytypes.TelemetryFieldKey{
|
||||
Name: "db.system.name",
|
||||
Signal: telemetrytypes.SignalMetrics,
|
||||
FieldContext: telemetrytypes.FieldContextResource,
|
||||
FieldDataType: telemetrytypes.FieldDataTypeString,
|
||||
}
|
||||
|
||||
expression, err := fm.FieldFor(context.Background(), valuer.UUID{}, 0, 0, key)
|
||||
require.NoError(t, err)
|
||||
for _, member := range []string{
|
||||
"resource_db.system.name", "resource_db_system_name", "db.system.name", "db_system_name",
|
||||
"resource_db.system", "resource_db_system", "db.system", "db_system",
|
||||
} {
|
||||
assert.Contains(t, expression, "'"+member+"'")
|
||||
}
|
||||
assert.Less(t, strings.Index(expression, "resource_db.system.name"), strings.Index(expression, "resource_db.system'"))
|
||||
|
||||
exact := *key
|
||||
exact.SemconvMembers = []string{"resource_db_system"}
|
||||
expression, err = fm.FieldFor(context.Background(), valuer.UUID{}, 0, 0, &exact)
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, "JSONExtractString(labels, 'resource_db_system')", expression)
|
||||
}
|
||||
|
||||
func TestGetColumn(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
|
||||
|
||||
@@ -18,12 +18,12 @@ import (
|
||||
)
|
||||
|
||||
type conditionBuilder struct {
|
||||
fm qbtypes.FieldMapper
|
||||
fm *fieldMapper
|
||||
}
|
||||
|
||||
var _ qbtypes.ConditionBuilder = (*conditionBuilder)(nil)
|
||||
|
||||
func NewConditionBuilder(fm qbtypes.FieldMapper) *conditionBuilder {
|
||||
func NewConditionBuilder(fm *fieldMapper) *conditionBuilder {
|
||||
return &conditionBuilder{fm: fm}
|
||||
}
|
||||
|
||||
@@ -154,6 +154,17 @@ func (c *conditionBuilder) conditionFor(
|
||||
// in the query builder, `exists` and `not exists` are used for
|
||||
// key membership checks, so depending on the column type, the condition changes
|
||||
case qbtypes.FilterOperatorExists, qbtypes.FilterOperatorNotExists:
|
||||
// A semantic-convention family is represented by one current-first value
|
||||
// expression, but presence still has to inspect every physical member. In
|
||||
// particular, using ExistsExpression below with the requested key would add
|
||||
// a mapContains check for only that spelling and reject fallback-only rows.
|
||||
if isTraceSemconvFamily(key) {
|
||||
pred, err := c.fm.existsExpressionFor(ctx, orgID, startNs, endNs, key, operator == qbtypes.FilterOperatorExists)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
return sqlbuilder.Escape(pred), nil
|
||||
}
|
||||
columns, err := c.fm.ColumnFor(ctx, orgID, startNs, endNs, key)
|
||||
if err != nil {
|
||||
return "", err
|
||||
@@ -211,6 +222,9 @@ func (c *conditionBuilder) ConditionFor(
|
||||
}
|
||||
|
||||
matches := querybuilder.MatchingFieldKeys(key, fieldKeys)
|
||||
if options.ExactSemconv {
|
||||
matches = querybuilder.MatchingFieldKeysExact(key, fieldKeys)
|
||||
}
|
||||
skipResourceFilter := options.SkipResourceFilter
|
||||
|
||||
keys, warning := querybuilder.ResolveKeys(key, matches)
|
||||
@@ -256,6 +270,9 @@ func (c *conditionBuilder) ConditionFor(
|
||||
synthesized = true
|
||||
warnings = append(warnings, querybuilder.NewKeyNotFoundWarning(key.Name))
|
||||
}
|
||||
if options.ExactSemconv {
|
||||
keys = querybuilder.ExactSemconvKeys(keys)
|
||||
}
|
||||
|
||||
// When a resource sub-query already covers the term, drop resource keys from the main
|
||||
// query. Synthesized keys are exempt: the sub-query skips keys absent from metadata.
|
||||
|
||||
@@ -308,6 +308,82 @@ func TestConditionFor(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestConditionForSemconvFamilyPositiveFilterChecksPresence(t *testing.T) {
|
||||
key := &telemetrytypes.TelemetryFieldKey{
|
||||
Name: "deployment.environment.name",
|
||||
Signal: telemetrytypes.SignalTraces,
|
||||
FieldContext: telemetrytypes.FieldContextAttribute,
|
||||
FieldDataType: telemetrytypes.FieldDataTypeString,
|
||||
SemconvMembers: []string{"deployment.environment.name", "deployment.environment"},
|
||||
}
|
||||
sb := sqlbuilder.NewSelectBuilder()
|
||||
|
||||
conditions, warnings, err := NewConditionBuilder(NewFieldMapper()).ConditionFor(
|
||||
context.Background(), valuer.UUID{}, 0, 0, key,
|
||||
map[string][]*telemetrytypes.TelemetryFieldKey{key.Name: {key}},
|
||||
qbtypes.ConditionBuilderOptions{}, qbtypes.FilterOperatorEqual, "production", sb,
|
||||
)
|
||||
require.NoError(t, err)
|
||||
sb.Where(conditions...)
|
||||
sql, args := sb.BuildWithFlavor(sqlbuilder.ClickHouse)
|
||||
|
||||
assert.Empty(t, warnings)
|
||||
assert.Contains(t, sql, "(COALESCE(NULLIF(attributes_string['deployment.environment.name'], ''), NULLIF(attributes_string['deployment.environment'], ''), '') = ? AND ((mapContains(attributes_string, 'deployment.environment.name') OR mapContains(attributes_string, 'deployment.environment'))))")
|
||||
assert.Equal(t, []any{"production"}, args)
|
||||
}
|
||||
|
||||
func TestConditionForSemconvFamilyPreservesMaterializedMemberExistsColumn(t *testing.T) {
|
||||
key := &telemetrytypes.TelemetryFieldKey{
|
||||
Name: "deployment.environment.name",
|
||||
Signal: telemetrytypes.SignalTraces,
|
||||
FieldContext: telemetrytypes.FieldContextAttribute,
|
||||
FieldDataType: telemetrytypes.FieldDataTypeString,
|
||||
SemconvMembers: []string{"deployment.environment.name", "deployment.environment"},
|
||||
SemconvMaterializedColumns: map[string]string{
|
||||
"deployment.environment": "attribute_string_deployment$$environment",
|
||||
},
|
||||
}
|
||||
sb := sqlbuilder.NewSelectBuilder()
|
||||
|
||||
conditions, warnings, err := NewConditionBuilder(NewFieldMapper()).ConditionFor(
|
||||
context.Background(), valuer.UUID{}, 0, 0, key,
|
||||
map[string][]*telemetrytypes.TelemetryFieldKey{key.Name: {key}},
|
||||
qbtypes.ConditionBuilderOptions{}, qbtypes.FilterOperatorEqual, "production", sb,
|
||||
)
|
||||
require.NoError(t, err)
|
||||
sb.Where(conditions...)
|
||||
sql, args := sb.BuildWithFlavor(sqlbuilder.ClickHouse)
|
||||
|
||||
assert.Empty(t, warnings)
|
||||
assert.Contains(t, sql, "`attribute_string_deployment$$environment_exists`")
|
||||
assert.NotContains(t, sql, "`attribute_string_deployment$environment_exists`")
|
||||
assert.Equal(t, []any{"production"}, args)
|
||||
}
|
||||
|
||||
func TestConditionForSemconvFamilyNotExistsChecksEveryMember(t *testing.T) {
|
||||
key := &telemetrytypes.TelemetryFieldKey{
|
||||
Name: "deployment.environment.name",
|
||||
Signal: telemetrytypes.SignalTraces,
|
||||
FieldContext: telemetrytypes.FieldContextAttribute,
|
||||
FieldDataType: telemetrytypes.FieldDataTypeString,
|
||||
SemconvMembers: []string{"deployment.environment.name", "deployment.environment"},
|
||||
}
|
||||
sb := sqlbuilder.NewSelectBuilder()
|
||||
|
||||
conditions, warnings, err := NewConditionBuilder(NewFieldMapper()).ConditionFor(
|
||||
context.Background(), valuer.UUID{}, 0, 0, key,
|
||||
map[string][]*telemetrytypes.TelemetryFieldKey{key.Name: {key}},
|
||||
qbtypes.ConditionBuilderOptions{}, qbtypes.FilterOperatorNotExists, nil, sb,
|
||||
)
|
||||
require.NoError(t, err)
|
||||
sb.Where(conditions...)
|
||||
sql, args := sb.BuildWithFlavor(sqlbuilder.ClickHouse)
|
||||
|
||||
assert.Empty(t, warnings)
|
||||
assert.Contains(t, sql, "NOT (((mapContains(attributes_string, 'deployment.environment.name') OR mapContains(attributes_string, 'deployment.environment'))))")
|
||||
assert.Empty(t, args)
|
||||
}
|
||||
|
||||
func TestConditionForResourceWithEvolution(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
releaseTime := time.Date(2025, 1, 1, 0, 0, 0, 0, time.UTC)
|
||||
|
||||
@@ -8,6 +8,7 @@ import (
|
||||
schema "github.com/SigNoz/signoz-otel-collector/cmd/signozschemamigrator/schema_migrator"
|
||||
"github.com/SigNoz/signoz/pkg/errors"
|
||||
"github.com/SigNoz/signoz/pkg/querybuilder"
|
||||
"github.com/SigNoz/signoz/pkg/semconv"
|
||||
qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
|
||||
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
|
||||
"github.com/SigNoz/signoz/pkg/valuer"
|
||||
@@ -167,6 +168,44 @@ func NewFieldMapper() *fieldMapper {
|
||||
return &fieldMapper{}
|
||||
}
|
||||
|
||||
func traceSemconvMembers(key *telemetrytypes.TelemetryFieldKey) []string {
|
||||
if key.FieldContext != telemetrytypes.FieldContextResource && key.FieldContext != telemetrytypes.FieldContextAttribute {
|
||||
return []string{key.Name}
|
||||
}
|
||||
if len(key.SemconvMembers) > 0 {
|
||||
return key.SemconvMembers
|
||||
}
|
||||
return semconv.Members(semconv.KindAttribute, telemetrytypes.FieldKeySelector{
|
||||
Name: key.Name,
|
||||
Signal: telemetrytypes.SignalTraces,
|
||||
FieldContext: key.FieldContext,
|
||||
})
|
||||
}
|
||||
|
||||
func isTraceSemconvFamily(key *telemetrytypes.TelemetryFieldKey) bool {
|
||||
if key.FieldContext != telemetrytypes.FieldContextResource && key.FieldContext != telemetrytypes.FieldContextAttribute {
|
||||
return false
|
||||
}
|
||||
_, ok := semconv.Lookup(semconv.KindAttribute, telemetrytypes.FieldKeySelector{
|
||||
Name: key.Name,
|
||||
Signal: telemetrytypes.SignalTraces,
|
||||
FieldContext: key.FieldContext,
|
||||
})
|
||||
return ok
|
||||
}
|
||||
|
||||
func traceSemconvMapMemberExpressions(columnName string, key *telemetrytypes.TelemetryFieldKey, member string) (string, string) {
|
||||
if materializedColumn, ok := key.SemconvMaterializedColumns[member]; ok {
|
||||
return fmt.Sprintf("`%s`", materializedColumn), fmt.Sprintf("`%s_exists`", materializedColumn)
|
||||
}
|
||||
if key.Materialized && key.Name == member {
|
||||
physicalKey := *key
|
||||
physicalKey.Name = member
|
||||
return telemetrytypes.FieldKeyToMaterializedColumnName(&physicalKey), telemetrytypes.FieldKeyToMaterializedColumnNameForExists(&physicalKey)
|
||||
}
|
||||
return fmt.Sprintf("%s['%s']", columnName, member), fmt.Sprintf("mapContains(%s, '%s')", columnName, member)
|
||||
}
|
||||
|
||||
func (m *fieldMapper) getColumn(
|
||||
_ context.Context,
|
||||
_, _ uint64,
|
||||
@@ -291,10 +330,27 @@ func (m *fieldMapper) resolveColumnExprs(
|
||||
if key.FieldContext != telemetrytypes.FieldContextResource {
|
||||
return nil, nil, nil, errors.Newf(errors.TypeInvalidInput, errors.CodeInvalidInput, "only resource context fields are supported for json columns, got %s", key.FieldContext.String)
|
||||
}
|
||||
// have to add ::string as clickHouse throws an error :- data types Variant/Dynamic are not allowed in GROUP BY
|
||||
// once clickHouse dependency is updated, we need to check if we can remove it.
|
||||
exprs = append(exprs, fmt.Sprintf("%s.`%s`::String", columnName, key.Name))
|
||||
existExprs = append(existExprs, fmt.Sprintf("%s.`%s` IS NOT NULL", columnName, key.Name))
|
||||
members := traceSemconvMembers(key)
|
||||
if len(members) > 1 {
|
||||
values := make([]string, 0, len(members))
|
||||
guards := make([]string, 0, len(members))
|
||||
for _, member := range members {
|
||||
// The String cast is required because ClickHouse does not allow
|
||||
// Variant/Dynamic values in GROUP BY.
|
||||
value := fmt.Sprintf("%s.`%s`::String", columnName, member)
|
||||
values = append(values, fmt.Sprintf("NULLIF(%s, '')", value))
|
||||
guards = append(guards, fmt.Sprintf("%s.`%s` IS NOT NULL", columnName, member))
|
||||
}
|
||||
// Missing Dynamic paths are NULL, so this family expression must
|
||||
// retain the same NULL result as a single JSON-path lookup.
|
||||
exprs = append(exprs, "COALESCE("+strings.Join(values, ", ")+")")
|
||||
existExprs = append(existExprs, "("+strings.Join(guards, " OR ")+")")
|
||||
} else {
|
||||
// have to add ::string as clickHouse throws an error :- data types Variant/Dynamic are not allowed in GROUP BY
|
||||
// once ClickHouse is updated, check whether this cast can be removed.
|
||||
exprs = append(exprs, fmt.Sprintf("%s.`%s`::String", columnName, members[0]))
|
||||
existExprs = append(existExprs, fmt.Sprintf("%s.`%s` IS NOT NULL", columnName, members[0]))
|
||||
}
|
||||
case schema.ColumnTypeEnumString,
|
||||
schema.ColumnTypeEnumUInt64,
|
||||
schema.ColumnTypeEnumUInt32,
|
||||
@@ -319,13 +375,37 @@ func (m *fieldMapper) resolveColumnExprs(
|
||||
|
||||
switch valueType := column.Type.(schema.MapColumnType).ValueType; valueType.GetType() {
|
||||
case schema.ColumnTypeEnumString, schema.ColumnTypeEnumFloat64, schema.ColumnTypeEnumBool:
|
||||
// a key could have been materialized, if so return the materialized column name
|
||||
if key.Materialized {
|
||||
members := traceSemconvMembers(key)
|
||||
if key.Materialized && (len(members) == 1 || key.MaterializedSemconv) {
|
||||
exprs = append(exprs, telemetrytypes.FieldKeyToMaterializedColumnName(key))
|
||||
existExprs = append(existExprs, telemetrytypes.FieldKeyToMaterializedColumnNameForExists(key))
|
||||
} else if len(members) > 1 {
|
||||
guards := make([]string, 0, len(members))
|
||||
memberValues := make([]string, 0, len(members))
|
||||
for _, member := range members {
|
||||
valueExpression, existsExpression := traceSemconvMapMemberExpressions(columnName, key, member)
|
||||
memberValues = append(memberValues, valueExpression)
|
||||
guards = append(guards, existsExpression)
|
||||
}
|
||||
if valueType.GetType() == schema.ColumnTypeEnumString {
|
||||
values := make([]string, 0, len(members))
|
||||
for _, memberValue := range memberValues {
|
||||
values = append(values, fmt.Sprintf("NULLIF(%s, '')", memberValue))
|
||||
}
|
||||
exprs = append(exprs, "COALESCE("+strings.Join(values, ", ")+", '')")
|
||||
} else {
|
||||
branches := make([]string, 0, len(members)*2)
|
||||
for i, memberValue := range memberValues {
|
||||
branches = append(branches, guards[i], memberValue)
|
||||
}
|
||||
// Numeric and boolean maps return zero for an absent key. If a
|
||||
// family of either type is enabled, this tail must become zero too.
|
||||
exprs = append(exprs, "multiIf("+strings.Join(branches, ", ")+", NULL)")
|
||||
}
|
||||
existExprs = append(existExprs, "("+strings.Join(guards, " OR ")+")")
|
||||
} else {
|
||||
exprs = append(exprs, fmt.Sprintf("%s['%s']", columnName, key.Name))
|
||||
existExprs = append(existExprs, fmt.Sprintf("mapContains(%s, '%s')", columnName, key.Name))
|
||||
exprs = append(exprs, fmt.Sprintf("%s['%s']", columnName, members[0]))
|
||||
existExprs = append(existExprs, fmt.Sprintf("mapContains(%s, '%s')", columnName, members[0]))
|
||||
}
|
||||
default:
|
||||
return nil, nil, nil, errors.NewInvalidInputf(errors.CodeInvalidInput, "value type %s is not supported for map column type %s", valueType, column.Type)
|
||||
@@ -529,6 +609,25 @@ func (m *fieldMapper) existsExpressionFor(
|
||||
key *telemetrytypes.TelemetryFieldKey,
|
||||
exists bool,
|
||||
) (string, error) {
|
||||
if isTraceSemconvFamily(key) {
|
||||
_, existExprs, _, err := m.resolveColumnExprs(ctx, tsStart, tsEnd, key)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
if len(existExprs) == 0 {
|
||||
return "", errors.NewInvalidInputf(errors.CodeInvalidInput, "no existence expression found for field %s", key.Name)
|
||||
}
|
||||
parts := make([]string, 0, len(existExprs))
|
||||
for _, expression := range existExprs {
|
||||
parts = append(parts, "("+expression+")")
|
||||
}
|
||||
combined := strings.Join(parts, " OR ")
|
||||
if exists {
|
||||
return combined, nil
|
||||
}
|
||||
return "NOT (" + combined + ")", nil
|
||||
}
|
||||
|
||||
columns, err := m.getColumn(ctx, tsStart, tsEnd, key)
|
||||
if err != nil {
|
||||
return "", err
|
||||
|
||||
@@ -5,6 +5,8 @@ import (
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
schema "github.com/SigNoz/signoz-otel-collector/cmd/signozschemamigrator/schema_migrator"
|
||||
"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/valuer"
|
||||
@@ -80,7 +82,7 @@ func TestGetFieldKeyName(t *testing.T) {
|
||||
Materialized: true,
|
||||
Evolutions: mockEvolution,
|
||||
},
|
||||
expectedResult: "multiIf(resource.`deployment.environment` IS NOT NULL, resource.`deployment.environment`::String, `resource_string_deployment$$environment_exists`, `resource_string_deployment$$environment`, NULL)",
|
||||
expectedResult: "multiIf((resource.`deployment.environment.name` IS NOT NULL OR resource.`deployment.environment` IS NOT NULL), COALESCE(NULLIF(resource.`deployment.environment.name`::String, ''), NULLIF(resource.`deployment.environment`::String, '')), (mapContains(resources_string, 'deployment.environment.name') OR `resource_string_deployment$$environment_exists`), COALESCE(NULLIF(resources_string['deployment.environment.name'], ''), NULLIF(`resource_string_deployment$$environment`, ''), ''), NULL)",
|
||||
expectedError: nil,
|
||||
},
|
||||
{
|
||||
@@ -120,6 +122,63 @@ func TestGetFieldKeyName(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestFieldForResolvesCurrentTraceSemconvAttributeName(t *testing.T) {
|
||||
key := telemetrytypes.TelemetryFieldKey{
|
||||
Name: "deployment.environment.name",
|
||||
FieldContext: telemetrytypes.FieldContextAttribute,
|
||||
FieldDataType: telemetrytypes.FieldDataTypeString,
|
||||
}
|
||||
|
||||
expression, err := NewFieldMapper().FieldFor(context.Background(), valuer.UUID{}, 0, 0, &key)
|
||||
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, "COALESCE(NULLIF(attributes_string['deployment.environment.name'], ''), NULLIF(attributes_string['deployment.environment'], ''), '')", expression)
|
||||
}
|
||||
|
||||
func TestFieldForResolvesOldTraceSemconvAttributeName(t *testing.T) {
|
||||
key := telemetrytypes.TelemetryFieldKey{
|
||||
Name: "deployment.environment",
|
||||
FieldContext: telemetrytypes.FieldContextAttribute,
|
||||
FieldDataType: telemetrytypes.FieldDataTypeString,
|
||||
}
|
||||
|
||||
expression, err := NewFieldMapper().FieldFor(context.Background(), valuer.UUID{}, 0, 0, &key)
|
||||
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, "COALESCE(NULLIF(attributes_string['deployment.environment.name'], ''), NULLIF(attributes_string['deployment.environment'], ''), '')", expression)
|
||||
}
|
||||
|
||||
func TestFieldForPreservesResourceStorageDefaultsForSemconvFamily(t *testing.T) {
|
||||
key := telemetrytypes.TelemetryFieldKey{
|
||||
Name: "deployment.environment.name",
|
||||
FieldContext: telemetrytypes.FieldContextResource,
|
||||
FieldDataType: telemetrytypes.FieldDataTypeString,
|
||||
Materialized: true,
|
||||
Evolutions: MockEvolutionData(time.Date(2024, 6, 2, 0, 0, 0, 0, time.UTC)),
|
||||
}
|
||||
start := uint64(time.Date(2024, 6, 1, 0, 0, 0, 0, time.UTC).UnixNano())
|
||||
end := uint64(time.Date(2024, 6, 5, 0, 0, 0, 0, time.UTC).UnixNano())
|
||||
|
||||
expression, err := NewFieldMapper().FieldFor(context.Background(), valuer.UUID{}, start, end, &key)
|
||||
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, "multiIf((resource.`deployment.environment.name` IS NOT NULL OR resource.`deployment.environment` IS NOT NULL), COALESCE(NULLIF(resource.`deployment.environment.name`::String, ''), NULLIF(resource.`deployment.environment`::String, '')), (`resource_string_deployment$$environment$$name_exists` OR mapContains(resources_string, 'deployment.environment')), COALESCE(NULLIF(`resource_string_deployment$$environment$$name`, ''), NULLIF(resources_string['deployment.environment'], ''), ''), NULL)", expression)
|
||||
}
|
||||
|
||||
func TestFieldForUsesAvailableTraceSemconvMember(t *testing.T) {
|
||||
key := telemetrytypes.TelemetryFieldKey{
|
||||
Name: "deployment.environment",
|
||||
FieldContext: telemetrytypes.FieldContextAttribute,
|
||||
FieldDataType: telemetrytypes.FieldDataTypeString,
|
||||
SemconvMembers: []string{"deployment.environment.name"},
|
||||
}
|
||||
|
||||
expression, err := NewFieldMapper().FieldFor(context.Background(), valuer.UUID{}, 0, 0, &key)
|
||||
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, "attributes_string['deployment.environment.name']", expression)
|
||||
}
|
||||
|
||||
func TestFieldForResourceWithEvolution(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
releaseTime := time.Date(2025, 1, 1, 0, 0, 0, 0, time.UTC)
|
||||
@@ -176,7 +235,7 @@ func TestFieldForResourceWithEvolution(t *testing.T) {
|
||||
},
|
||||
tsStart: uint64(time.Date(2025, 6, 1, 0, 0, 0, 0, time.UTC).UnixNano()),
|
||||
tsEnd: uint64(time.Date(2025, 7, 1, 0, 0, 0, 0, time.UTC).UnixNano()),
|
||||
expectedResult: "resource.`deployment.environment`::String",
|
||||
expectedResult: "COALESCE(NULLIF(resource.`deployment.environment.name`::String, ''), NULLIF(resource.`deployment.environment`::String, ''))",
|
||||
},
|
||||
{
|
||||
name: "Window straddles release - materialized resource",
|
||||
@@ -189,7 +248,7 @@ func TestFieldForResourceWithEvolution(t *testing.T) {
|
||||
},
|
||||
tsStart: uint64(time.Date(2024, 6, 1, 0, 0, 0, 0, time.UTC).UnixNano()),
|
||||
tsEnd: uint64(time.Date(2025, 6, 1, 0, 0, 0, 0, time.UTC).UnixNano()),
|
||||
expectedResult: "multiIf(resource.`deployment.environment` IS NOT NULL, resource.`deployment.environment`::String, `resource_string_deployment$$environment_exists`, `resource_string_deployment$$environment`, NULL)",
|
||||
expectedResult: "multiIf((resource.`deployment.environment.name` IS NOT NULL OR resource.`deployment.environment` IS NOT NULL), COALESCE(NULLIF(resource.`deployment.environment.name`::String, ''), NULLIF(resource.`deployment.environment`::String, '')), (mapContains(resources_string, 'deployment.environment.name') OR `resource_string_deployment$$environment_exists`), COALESCE(NULLIF(resources_string['deployment.environment.name'], ''), NULLIF(`resource_string_deployment$$environment`, ''), ''), NULL)",
|
||||
},
|
||||
}
|
||||
|
||||
@@ -303,3 +362,27 @@ func TestColumnExpressionForTimestampAttributeCollision(t *testing.T) {
|
||||
assert.Contains(t, result, "attributes_number['timestamp']")
|
||||
})
|
||||
}
|
||||
|
||||
func TestDBSystemFamilyUsesSemconvAwareMaterializedColumn(t *testing.T) {
|
||||
fm := NewFieldMapper()
|
||||
key := &telemetrytypes.TelemetryFieldKey{
|
||||
Name: "db.system.name",
|
||||
Signal: telemetrytypes.SignalTraces,
|
||||
FieldContext: telemetrytypes.FieldContextAttribute,
|
||||
FieldDataType: telemetrytypes.FieldDataTypeString,
|
||||
Materialized: true,
|
||||
MaterializedColumnName: "attribute_string_db$$system",
|
||||
MaterializedSemconv: true,
|
||||
SemconvMembers: []string{"db.system.name", "db.system"},
|
||||
}
|
||||
|
||||
expression, err := fm.FieldFor(context.Background(), valuer.UUID{}, 0, 0, key)
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, "`attribute_string_db$$system`", expression)
|
||||
|
||||
exists, err := querybuilder.ExistsExpression(
|
||||
[]*schema.Column{indexV3Columns["attributes_string"]}, key, 0, 0, expression, true,
|
||||
)
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, "`attribute_string_db$$system_exists`", exists)
|
||||
}
|
||||
|
||||
@@ -155,6 +155,8 @@ var operatorInverseMapping = map[FilterOperator]FilterOperator{
|
||||
// doesn't have value "redis"
|
||||
// Since we don't know the intent, we don't add the exists filter. They are expected
|
||||
// to add exists filter themselves if exclusion is desired.
|
||||
// Negative predicates therefore include rows where the key is absent; value
|
||||
// expressions must preserve the storage column's absent-key default.
|
||||
//
|
||||
// For the positive predicates, the key existence is implied.
|
||||
func (f FilterOperator) AddDefaultExistsFilter() bool {
|
||||
|
||||
@@ -45,6 +45,10 @@ type ConditionBuilder interface {
|
||||
type ConditionBuilderOptions struct {
|
||||
// SkipResourceFilter drops the resource context from the candidate set.
|
||||
SkipResourceFilter bool
|
||||
// ExactSemconv disables semantic-convention family expansion. It is an
|
||||
// internal escape hatch for diagnostics and migrations that must address one
|
||||
// physical spelling only; public query APIs continue to resolve families.
|
||||
ExactSemconv bool
|
||||
}
|
||||
type AggExprRewriter interface {
|
||||
// Rewrite rewrites the aggregation expression to be used in the query.
|
||||
|
||||
@@ -141,7 +141,7 @@ func NewSignalFilterFromStorableQuickFilter(storableQuickFilter *StorableQuickFi
|
||||
func NewDefaultQuickFilter(orgID valuer.UUID) ([]*StorableQuickFilter, error) {
|
||||
tracesFilters := []map[string]interface{}{
|
||||
{"key": "duration_nano", "dataType": "float64", "type": "tag"},
|
||||
{"key": "deployment.environment", "dataType": "string", "type": "resource"},
|
||||
{"key": "deployment.environment.name", "dataType": "string", "type": "resource"},
|
||||
{"key": "hasError", "dataType": "bool", "type": "tag"},
|
||||
{"key": "service.name", "dataType": "string", "type": "resource"},
|
||||
{"key": "name", "dataType": "string", "type": "tag"},
|
||||
@@ -166,13 +166,13 @@ func NewDefaultQuickFilter(orgID valuer.UUID) ([]*StorableQuickFilter, error) {
|
||||
}
|
||||
|
||||
apiMonitoringFilters := []map[string]interface{}{
|
||||
{"key": "deployment.environment", "dataType": "string", "type": "resource"},
|
||||
{"key": "deployment.environment.name", "dataType": "string", "type": "resource"},
|
||||
{"key": "service.name", "dataType": "string", "type": "resource"},
|
||||
{"key": "rpc.method", "dataType": "string", "type": "tag"},
|
||||
}
|
||||
|
||||
exceptionsFilters := []map[string]interface{}{
|
||||
{"key": "deployment.environment", "dataType": "string", "type": "resource"},
|
||||
{"key": "deployment.environment.name", "dataType": "string", "type": "resource"},
|
||||
{"key": "service.name", "dataType": "string", "type": "resource"},
|
||||
{"key": "host.name", "dataType": "string", "type": "resource"},
|
||||
{"key": "k8s.cluster.name", "dataType": "string", "type": "resource"},
|
||||
|
||||
37
pkg/types/quickfiltertypes/filter_test.go
Normal file
37
pkg/types/quickfiltertypes/filter_test.go
Normal file
@@ -0,0 +1,37 @@
|
||||
package quickfiltertypes
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"testing"
|
||||
|
||||
v3 "github.com/SigNoz/signoz/pkg/query-service/model/v3"
|
||||
"github.com/SigNoz/signoz/pkg/valuer"
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
func TestDefaultTraceQuickFiltersUseCurrentEnvironmentName(t *testing.T) {
|
||||
filters, err := NewDefaultQuickFilter(valuer.GenerateUUID())
|
||||
require.NoError(t, err)
|
||||
|
||||
traceSignals := map[string]bool{
|
||||
SignalTraces.StringValue(): true,
|
||||
SignalApiMonitoring.StringValue(): true,
|
||||
SignalExceptions.StringValue(): true,
|
||||
}
|
||||
for _, filter := range filters {
|
||||
if !traceSignals[filter.Signal.StringValue()] {
|
||||
continue
|
||||
}
|
||||
var keys []v3.AttributeKey
|
||||
require.NoError(t, json.Unmarshal([]byte(filter.Filter), &keys))
|
||||
found := false
|
||||
for _, key := range keys {
|
||||
if key.Key == "deployment.environment.name" {
|
||||
found = true
|
||||
}
|
||||
assert.NotEqual(t, "deployment.environment", key.Key)
|
||||
}
|
||||
assert.True(t, found, "missing environment quick filter for %s", filter.Signal.StringValue())
|
||||
}
|
||||
}
|
||||
@@ -46,8 +46,19 @@ type TelemetryFieldKey struct {
|
||||
JSONPlan JSONAccessPlan `json:"-"`
|
||||
Indexes []TelemetryFieldKeySkipIndex `json:"-"`
|
||||
Materialized bool `json:"-"` // refers to promoted in case of body.... fields
|
||||
// MaterializedColumnName preserves the physical column when its DEFAULT
|
||||
// expression resolves a newer semantic-convention key than the historical
|
||||
// column identifier (for example db.system.name in db$$system).
|
||||
MaterializedColumnName string `json:"-"`
|
||||
// MaterializedSemconv is true when the column's DEFAULT expression already
|
||||
// coalesces every enabled family member and is therefore safe for a family query.
|
||||
MaterializedSemconv bool `json:"-"`
|
||||
|
||||
Evolutions []*EvolutionEntry `json:"-"`
|
||||
Evolutions []*EvolutionEntry `json:"-"`
|
||||
SemconvMembers []string `json:"-"`
|
||||
// SemconvMaterializedColumns maps a physical family spelling to its
|
||||
// materialized column name. It is populated only on resolved query keys.
|
||||
SemconvMaterializedColumns map[string]string `json:"-"`
|
||||
}
|
||||
|
||||
func (f *TelemetryFieldKey) KeyNameContainsArray() bool {
|
||||
@@ -126,8 +137,12 @@ func (f *TelemetryFieldKey) OverrideMetadataFrom(src *TelemetryFieldKey) {
|
||||
f.FieldDataType = src.FieldDataType
|
||||
f.Indexes = src.Indexes
|
||||
f.Materialized = src.Materialized
|
||||
f.MaterializedColumnName = src.MaterializedColumnName
|
||||
f.MaterializedSemconv = src.MaterializedSemconv
|
||||
f.JSONPlan = src.JSONPlan
|
||||
f.Evolutions = src.Evolutions
|
||||
f.SemconvMembers = src.SemconvMembers
|
||||
f.SemconvMaterializedColumns = src.SemconvMaterializedColumns
|
||||
}
|
||||
|
||||
func (f *TelemetryFieldKey) Equal(key *TelemetryFieldKey) bool {
|
||||
@@ -202,6 +217,9 @@ func TelemetryFieldKeyToText(key *TelemetryFieldKey) string {
|
||||
}
|
||||
|
||||
func FieldKeyToMaterializedColumnName(key *TelemetryFieldKey) string {
|
||||
if key.MaterializedColumnName != "" {
|
||||
return fmt.Sprintf("`%s`", key.MaterializedColumnName)
|
||||
}
|
||||
return fmt.Sprintf("`%s_%s_%s`",
|
||||
key.FieldContext.String,
|
||||
fieldDataTypes[key.FieldDataType.StringValue()].StringValue(),
|
||||
@@ -210,6 +228,9 @@ func FieldKeyToMaterializedColumnName(key *TelemetryFieldKey) string {
|
||||
}
|
||||
|
||||
func FieldKeyToMaterializedColumnNameForExists(key *TelemetryFieldKey) string {
|
||||
if key.MaterializedColumnName != "" {
|
||||
return fmt.Sprintf("`%s_exists`", key.MaterializedColumnName)
|
||||
}
|
||||
return fmt.Sprintf("`%s_%s_%s_exists`",
|
||||
key.FieldContext.String,
|
||||
fieldDataTypes[key.FieldDataType.StringValue()].StringValue(),
|
||||
|
||||
@@ -7,5 +7,31 @@ default_enabled: false
|
||||
families:
|
||||
deployment.environment.name:
|
||||
enabled: true
|
||||
contexts: [resource, attribute]
|
||||
signals: [traces, logs, metrics]
|
||||
db.system.name:
|
||||
enabled: true
|
||||
contexts: [resource, attribute]
|
||||
signals: [traces, logs, metrics]
|
||||
|
||||
# These metric renames predate the schema history vendored above. Keep them
|
||||
# in the same generated registry so every v5 metric query uses one source of
|
||||
# truth instead of the legacy hand-written transition table.
|
||||
k8s.pod.cpu.usage:
|
||||
enabled: true
|
||||
kind: metric
|
||||
old: [k8s.pod.cpu.utilization]
|
||||
contexts: [metric]
|
||||
signals: [metrics]
|
||||
k8s.node.cpu.usage:
|
||||
enabled: true
|
||||
kind: metric
|
||||
old: [k8s.node.cpu.utilization]
|
||||
contexts: [metric]
|
||||
signals: [metrics]
|
||||
container.cpu.usage:
|
||||
enabled: true
|
||||
kind: metric
|
||||
old: [container.cpu.utilization]
|
||||
contexts: [metric]
|
||||
signals: [metrics]
|
||||
|
||||
@@ -34,6 +34,7 @@ pytest_plugins = [
|
||||
"fixtures.role",
|
||||
"fixtures.savedview",
|
||||
"fixtures.seed_golden_dataset",
|
||||
"fixtures.semconv",
|
||||
]
|
||||
|
||||
|
||||
|
||||
66
tests/fixtures/semconv.py
vendored
Normal file
66
tests/fixtures/semconv.py
vendored
Normal file
@@ -0,0 +1,66 @@
|
||||
from collections.abc import Callable, Generator
|
||||
from datetime import UTC, datetime, timedelta
|
||||
|
||||
import pytest
|
||||
|
||||
from fixtures import types
|
||||
from fixtures.traces import TraceIdGenerator, Traces, TracesKind, TracesStatusCode
|
||||
|
||||
SEMCONV_PHASE1_CURRENT = "deployment.environment.name"
|
||||
SEMCONV_PHASE1_OLD = "deployment.environment"
|
||||
SEMCONV_PHASE1_PREFIX = "semconv-phase1"
|
||||
|
||||
|
||||
@pytest.fixture(name="semconv_phase1_data")
|
||||
def semconv_phase1_data(
|
||||
insert_traces: Callable[[list[Traces]], None],
|
||||
clickhouse: types.TestContainerClickhouse,
|
||||
) -> Generator[datetime]:
|
||||
now = datetime.now(tz=UTC).replace(microsecond=0) - timedelta(minutes=2)
|
||||
records = [
|
||||
(now - timedelta(seconds=5), "old", {SEMCONV_PHASE1_OLD: "production"}),
|
||||
(now - timedelta(seconds=4), "current", {SEMCONV_PHASE1_CURRENT: "production"}),
|
||||
(now - timedelta(seconds=3), "both", {SEMCONV_PHASE1_OLD: "production", SEMCONV_PHASE1_CURRENT: "production"}),
|
||||
(now - timedelta(seconds=2), "conflict", {SEMCONV_PHASE1_OLD: "staging", SEMCONV_PHASE1_CURRENT: "production"}),
|
||||
(now - timedelta(seconds=1), "staging", {SEMCONV_PHASE1_OLD: "staging"}),
|
||||
(now, "missing", {}),
|
||||
]
|
||||
traces = []
|
||||
for timestamp, suffix, environment in records:
|
||||
service = f"{SEMCONV_PHASE1_PREFIX}-{suffix}"
|
||||
traces.append(
|
||||
Traces(
|
||||
timestamp=timestamp,
|
||||
duration=timedelta(milliseconds=10),
|
||||
trace_id=TraceIdGenerator.trace_id(),
|
||||
span_id=TraceIdGenerator.span_id(),
|
||||
name=service,
|
||||
kind=TracesKind.SPAN_KIND_SERVER,
|
||||
status_code=TracesStatusCode.STATUS_CODE_OK,
|
||||
resources={"service.name": service, **environment},
|
||||
attributes=dict(environment),
|
||||
)
|
||||
)
|
||||
insert_traces(traces)
|
||||
|
||||
# Service-map rows are derived by the collector in production. Seed the
|
||||
# derived table directly so this test isolates the backend alias allowlist;
|
||||
# the collector repository owns its write-path integration test.
|
||||
for environment, suffix in (("production", "production"), ("staging", "staging")):
|
||||
clickhouse.conn.command(
|
||||
f"""
|
||||
INSERT INTO signoz_traces.distributed_dependency_graph_minutes_v2
|
||||
(src, dest, duration_quantiles_state, error_count, total_count, timestamp,
|
||||
deployment_environment, k8s_cluster_name, k8s_namespace_name)
|
||||
SELECT
|
||||
'{SEMCONV_PHASE1_PREFIX}-map-{suffix}', '{SEMCONV_PHASE1_PREFIX}-map-child',
|
||||
quantilesState(0.5, 0.75, 0.9, 0.95, 0.99)(toFloat64(1000000)),
|
||||
toUInt64(0), toUInt64(1), toDateTime({int(now.timestamp())}),
|
||||
'{environment}', '', ''
|
||||
"""
|
||||
)
|
||||
|
||||
yield now
|
||||
|
||||
cluster = clickhouse.env["SIGNOZ_TELEMETRYSTORE_CLICKHOUSE_CLUSTER"]
|
||||
clickhouse.conn.command(f"ALTER TABLE signoz_traces.dependency_graph_minutes_v2 ON CLUSTER '{cluster}' DELETE WHERE startsWith(src, '{SEMCONV_PHASE1_PREFIX}-map-') SETTINGS mutations_sync = 1")
|
||||
154
tests/integration/tests/queriersemconv/02_cross_signal.py
Normal file
154
tests/integration/tests/queriersemconv/02_cross_signal.py
Normal file
@@ -0,0 +1,154 @@
|
||||
"""Phase 2 semantic-convention checks across logs and metrics."""
|
||||
|
||||
from collections.abc import Callable
|
||||
from datetime import UTC, datetime, timedelta
|
||||
from http import HTTPStatus
|
||||
|
||||
import requests
|
||||
|
||||
from fixtures import querier, types
|
||||
from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
|
||||
from fixtures.logs import Logs
|
||||
from fixtures.metrics import Metrics
|
||||
|
||||
DB_CURRENT = "db.system.name"
|
||||
DB_OLD = "db.system"
|
||||
METRIC_CURRENT = "container.cpu.usage"
|
||||
METRIC_OLD = "container.cpu.utilization"
|
||||
PREFIX = "semconv-phase2"
|
||||
|
||||
|
||||
def test_logs_resolve_db_system_family(
|
||||
signoz: types.SigNoz,
|
||||
create_user_admin: None, # pylint: disable=unused-argument
|
||||
get_token: Callable[[str, str], str],
|
||||
insert_logs: Callable[[list[Logs]], None],
|
||||
) -> None:
|
||||
now = datetime.now(tz=UTC).replace(microsecond=0) - timedelta(minutes=1)
|
||||
rows = [
|
||||
("old", {DB_OLD: "postgresql"}),
|
||||
("current", {DB_CURRENT: "postgresql"}),
|
||||
("conflict", {DB_OLD: "mysql", DB_CURRENT: "postgresql"}),
|
||||
("missing", {}),
|
||||
]
|
||||
insert_logs(
|
||||
[
|
||||
Logs(
|
||||
timestamp=now + timedelta(seconds=index),
|
||||
resources={"service.name": PREFIX, **attributes},
|
||||
attributes=attributes,
|
||||
body=f"{PREFIX}-{suffix}",
|
||||
)
|
||||
for index, (suffix, attributes) in enumerate(rows)
|
||||
]
|
||||
)
|
||||
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
|
||||
start_ms = int((now - timedelta(minutes=2)).timestamp() * 1000)
|
||||
end_ms = int((now + timedelta(minutes=1)).timestamp() * 1000)
|
||||
present = {f"{PREFIX}-old", f"{PREFIX}-current", f"{PREFIX}-conflict"}
|
||||
|
||||
for context in ("attribute", "resource"):
|
||||
for requested in (DB_CURRENT, DB_OLD):
|
||||
field = f"{context}.{requested}"
|
||||
query_cases = {
|
||||
"equal": (f'{field} = "postgresql"', present),
|
||||
"exists": (f"{field} EXISTS", present),
|
||||
"not_exists": (f"{field} NOT EXISTS", {f"{PREFIX}-missing"}),
|
||||
}
|
||||
response = querier.make_query_request(
|
||||
signoz,
|
||||
token,
|
||||
start_ms=start_ms,
|
||||
end_ms=end_ms,
|
||||
request_type=querier.RequestType.RAW,
|
||||
queries=[querier.build_raw_query(name, "logs", limit=100, filter_expression=expression) for name, (expression, _) in query_cases.items()],
|
||||
)
|
||||
assert response.status_code == HTTPStatus.OK, response.text
|
||||
results = response.json()["data"]["data"]["results"]
|
||||
for name, (_, expected_bodies) in query_cases.items():
|
||||
result = querier.find_named_result(results, name)
|
||||
assert result is not None, name
|
||||
assert {row["data"]["body"] for row in (result.get("rows") or [])} == expected_bodies
|
||||
|
||||
values_response = requests.get(
|
||||
signoz.self.host_configs["8080"].get("/api/v1/fields/values"),
|
||||
timeout=5,
|
||||
headers={"authorization": f"Bearer {token}"},
|
||||
params={
|
||||
"signal": "logs",
|
||||
"name": requested,
|
||||
"fieldContext": context,
|
||||
"fieldDataType": "string",
|
||||
},
|
||||
)
|
||||
assert values_response.status_code == HTTPStatus.OK, values_response.text
|
||||
assert set(values_response.json()["data"]["values"].get("stringValues") or []) == {"postgresql", "mysql"}
|
||||
|
||||
|
||||
def test_metrics_resolve_label_and_metric_name_families(
|
||||
signoz: types.SigNoz,
|
||||
create_user_admin: None, # pylint: disable=unused-argument
|
||||
get_token: Callable[[str, str], str],
|
||||
insert_metrics: Callable[[list[Metrics]], None],
|
||||
) -> None:
|
||||
now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
|
||||
insert_metrics(
|
||||
[
|
||||
Metrics(
|
||||
metric_name=METRIC_CURRENT,
|
||||
labels={DB_CURRENT: "postgresql"},
|
||||
timestamp=now - timedelta(seconds=3),
|
||||
temporality="Unspecified",
|
||||
type_="Gauge",
|
||||
is_monotonic=False,
|
||||
value=10,
|
||||
),
|
||||
Metrics(
|
||||
metric_name=METRIC_OLD,
|
||||
labels={"db_system": "mysql"},
|
||||
timestamp=now - timedelta(seconds=2),
|
||||
temporality="Unspecified",
|
||||
type_="Gauge",
|
||||
is_monotonic=False,
|
||||
value=20,
|
||||
),
|
||||
Metrics(
|
||||
metric_name=METRIC_OLD,
|
||||
labels={DB_OLD: "mysql", DB_CURRENT: "postgresql", "series": "conflict"},
|
||||
timestamp=now - timedelta(seconds=1),
|
||||
temporality="Unspecified",
|
||||
type_="Gauge",
|
||||
is_monotonic=False,
|
||||
value=30,
|
||||
),
|
||||
]
|
||||
)
|
||||
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
|
||||
|
||||
for metric_name in (METRIC_CURRENT, METRIC_OLD):
|
||||
for requested_label in (DB_CURRENT, DB_OLD):
|
||||
response = querier.make_scalar_query_request(
|
||||
signoz,
|
||||
token,
|
||||
now,
|
||||
[
|
||||
querier.build_scalar_query(
|
||||
name="A",
|
||||
signal="metrics",
|
||||
aggregations=[
|
||||
querier.build_metrics_aggregation(
|
||||
metric_name,
|
||||
"latest",
|
||||
"sum",
|
||||
"unspecified",
|
||||
reduce_to="last",
|
||||
)
|
||||
],
|
||||
group_by=[querier.build_group_by_field(requested_label, "string", "attribute")],
|
||||
filter_expression=f"attribute.{requested_label} EXISTS",
|
||||
)
|
||||
],
|
||||
)
|
||||
assert response.status_code == HTTPStatus.OK, response.text
|
||||
data = {row[0]: row[-1] for row in querier.get_scalar_table_data(response.json())}
|
||||
assert data == {"postgresql": 40.0, "mysql": 20.0}, (metric_name, requested_label, data)
|
||||
166
tests/integration/tests/queriertraces/13_semconv_evolution.py
Normal file
166
tests/integration/tests/queriertraces/13_semconv_evolution.py
Normal file
@@ -0,0 +1,166 @@
|
||||
"""Phase 1 end-to-end checks for semantic-convention name evolution."""
|
||||
|
||||
from collections.abc import Callable
|
||||
from datetime import datetime, timedelta
|
||||
from http import HTTPStatus
|
||||
|
||||
import requests
|
||||
|
||||
from fixtures import querier, types
|
||||
from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
|
||||
from fixtures.semconv import SEMCONV_PHASE1_CURRENT as CURRENT
|
||||
from fixtures.semconv import SEMCONV_PHASE1_OLD as OLD
|
||||
from fixtures.semconv import SEMCONV_PHASE1_PREFIX as PREFIX
|
||||
|
||||
PRODUCTION_SPANS = {
|
||||
f"{PREFIX}-old",
|
||||
f"{PREFIX}-current",
|
||||
f"{PREFIX}-both",
|
||||
f"{PREFIX}-conflict",
|
||||
}
|
||||
STAGING_SPANS = {f"{PREFIX}-staging"}
|
||||
MISSING_SPANS = {f"{PREFIX}-missing"}
|
||||
|
||||
|
||||
def test_semconv_phase1_mixed_sdk_generations( # pylint: disable=too-many-statements
|
||||
signoz: types.SigNoz,
|
||||
create_user_admin: None, # pylint: disable=unused-argument
|
||||
get_token: Callable[[str, str], str],
|
||||
semconv_phase1_data: datetime,
|
||||
) -> None:
|
||||
now = semconv_phase1_data
|
||||
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
|
||||
start_ms = int((now - timedelta(minutes=2)).timestamp() * 1000)
|
||||
end_ms = int((now + timedelta(minutes=1)).timestamp() * 1000)
|
||||
|
||||
# Resource and span-attribute paths share the same matrix. Run every
|
||||
# operator with both the saved-query (old) and current request spellings.
|
||||
for context in ("resource", "attribute"):
|
||||
for requested in (CURRENT, OLD):
|
||||
field = f"{context}.{requested}"
|
||||
query_cases = {
|
||||
"production": (f"{field} = 'production'", PRODUCTION_SPANS),
|
||||
"staging": (f"{field} = 'staging'", STAGING_SPANS),
|
||||
# Negative operators intentionally include rows where no family
|
||||
# member exists; explicit EXISTS is the opt-in presence filter.
|
||||
"negative": (f"{field} != 'production'", STAGING_SPANS | MISSING_SPANS),
|
||||
"exists": (f"{field} EXISTS", PRODUCTION_SPANS | STAGING_SPANS),
|
||||
"not_exists": (f"{field} NOT EXISTS", MISSING_SPANS),
|
||||
}
|
||||
matrix_response = querier.make_query_request(
|
||||
signoz,
|
||||
token,
|
||||
start_ms=start_ms,
|
||||
end_ms=end_ms,
|
||||
request_type=querier.RequestType.RAW,
|
||||
queries=[
|
||||
querier.BuilderQuery(
|
||||
signal="traces",
|
||||
name=name,
|
||||
limit=100,
|
||||
filter_expression=expression,
|
||||
select_fields=[querier.TelemetryFieldKey("span.name")],
|
||||
order=[querier.OrderBy(querier.TelemetryFieldKey("timestamp"), "asc")],
|
||||
).to_dict()
|
||||
for name, (expression, _) in query_cases.items()
|
||||
],
|
||||
)
|
||||
assert matrix_response.status_code == HTTPStatus.OK, matrix_response.text
|
||||
matrix_results = matrix_response.json()["data"]["data"]["results"]
|
||||
for name, (_, expected_names) in query_cases.items():
|
||||
result = querier.find_named_result(matrix_results, name)
|
||||
assert result is not None, name
|
||||
assert {row["data"]["name"] for row in (result.get("rows") or [])} == expected_names
|
||||
|
||||
grouped_response = querier.make_query_request(
|
||||
signoz,
|
||||
token,
|
||||
start_ms=start_ms,
|
||||
end_ms=end_ms,
|
||||
request_type=querier.RequestType.SCALAR,
|
||||
queries=[
|
||||
querier.BuilderQuery(
|
||||
signal="traces",
|
||||
name="A",
|
||||
filter_expression=f"{field} EXISTS",
|
||||
aggregations=[querier.Aggregation("count()")],
|
||||
group_by=[querier.TelemetryFieldKey(requested, "string", context)],
|
||||
order=[querier.OrderBy(querier.TelemetryFieldKey(requested, "string", context), "asc")],
|
||||
).to_dict()
|
||||
],
|
||||
)
|
||||
assert grouped_response.status_code == HTTPStatus.OK, grouped_response.text
|
||||
grouped_results = grouped_response.json()["data"]["data"]["results"]
|
||||
assert len(grouped_results) == 1
|
||||
assert grouped_results[0]["columns"][0]["name"] == requested, "response identity must match the request spelling"
|
||||
assert grouped_results[0]["data"] == [["production", 4], ["staging", 1]]
|
||||
|
||||
for requested in (CURRENT, OLD):
|
||||
values_response = requests.get(
|
||||
signoz.self.host_configs["8080"].get("/api/v1/fields/values"),
|
||||
timeout=5,
|
||||
headers={"authorization": f"Bearer {token}"},
|
||||
params={
|
||||
"signal": "traces",
|
||||
"name": requested,
|
||||
"fieldContext": context,
|
||||
"fieldDataType": "string",
|
||||
},
|
||||
)
|
||||
assert values_response.status_code == HTTPStatus.OK, values_response.text
|
||||
assert set(values_response.json()["data"]["values"].get("stringValues") or []) == {"production", "staging"}
|
||||
|
||||
keys_response = requests.get(
|
||||
signoz.self.host_configs["8080"].get("/api/v1/fields/keys"),
|
||||
timeout=5,
|
||||
headers={"authorization": f"Bearer {token}"},
|
||||
params={"signal": "traces", "searchText": OLD},
|
||||
)
|
||||
assert keys_response.status_code == HTTPStatus.OK, keys_response.text
|
||||
keys = keys_response.json()["data"]["keys"]
|
||||
assert CURRENT in keys
|
||||
assert OLD in keys
|
||||
|
||||
start_ns = str(int((now - timedelta(minutes=2)).timestamp() * 1_000_000_000))
|
||||
end_ns = str(int((now + timedelta(minutes=1)).timestamp() * 1_000_000_000))
|
||||
for requested in (CURRENT, OLD):
|
||||
services_response = requests.post(
|
||||
signoz.self.host_configs["8080"].get("/api/v2/services"),
|
||||
timeout=30,
|
||||
headers={"authorization": f"Bearer {token}"},
|
||||
json={
|
||||
"start": start_ns,
|
||||
"end": end_ns,
|
||||
"tags": [
|
||||
{
|
||||
"Key": requested,
|
||||
"Operator": "In",
|
||||
"StringValues": ["production"],
|
||||
"TagType": "ResourceAttribute",
|
||||
}
|
||||
],
|
||||
},
|
||||
)
|
||||
assert services_response.status_code == HTTPStatus.OK, services_response.text
|
||||
services = {item["serviceName"] for item in services_response.json()["data"]}
|
||||
assert services == PRODUCTION_SPANS
|
||||
|
||||
map_response = requests.post(
|
||||
signoz.self.host_configs["8080"].get("/api/v1/dependency_graph"),
|
||||
timeout=30,
|
||||
headers={"authorization": f"Bearer {token}"},
|
||||
json={
|
||||
"start": start_ns,
|
||||
"end": end_ns,
|
||||
"tags": [
|
||||
{
|
||||
"key": requested,
|
||||
"operator": "In",
|
||||
"stringValues": ["production"],
|
||||
"tagType": "ResourceAttribute",
|
||||
}
|
||||
],
|
||||
},
|
||||
)
|
||||
assert map_response.status_code == HTTPStatus.OK, map_response.text
|
||||
assert {edge["parent"] for edge in map_response.json()} == {f"{PREFIX}-map-production"}
|
||||
Reference in New Issue
Block a user