mirror of
https://github.com/SigNoz/signoz.git
synced 2026-08-06 13:10:40 +01:00
8eb3f6bc1b
- Embed statementbuilder.Config into querier.Config with mapstructure ",squash"; keys move to querier.skip_resource_fingerprint.* (env SIGNOZ_QUERIER_*). - Drop the standalone statementbuilder section and its config factory. - Pass statementbuilder.Config wholesale into NewLogQueryStatementBuilder.
1295 lines
61 KiB
Go
1295 lines
61 KiB
Go
package logsstatementbuilder
|
|
|
|
import (
|
|
"context"
|
|
"regexp"
|
|
"testing"
|
|
"time"
|
|
|
|
cmock "github.com/SigNoz/clickhouse-go-mock"
|
|
"github.com/SigNoz/signoz/pkg/errors"
|
|
"github.com/SigNoz/signoz/pkg/flagger/flaggertest"
|
|
"github.com/SigNoz/signoz/pkg/instrumentation/instrumentationtest"
|
|
"github.com/SigNoz/signoz/pkg/querybuilder"
|
|
"github.com/SigNoz/signoz/pkg/statementbuilder"
|
|
"github.com/SigNoz/signoz/pkg/telemetryschema/logstelemetryschema"
|
|
"github.com/SigNoz/signoz/pkg/telemetrystore"
|
|
"github.com/SigNoz/signoz/pkg/telemetrystore/telemetrystoretest"
|
|
qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
|
|
"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/require"
|
|
)
|
|
|
|
type regexQueryMatcher struct{}
|
|
|
|
func (m *regexQueryMatcher) Match(expectedSQL, actualSQL string) error {
|
|
re, err := regexp.Compile(expectedSQL)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if !re.MatchString(actualSQL) {
|
|
return errors.NewInvalidInputf(errors.CodeInvalidInput, "expected query to match %s, got %s", expectedSQL, actualSQL)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func TestStatementBuilderTimeSeries(t *testing.T) {
|
|
// Create a test release time
|
|
releaseTime := time.Date(2024, 1, 15, 10, 0, 0, 0, time.UTC)
|
|
releaseTimeNano := uint64(releaseTime.UnixNano())
|
|
|
|
cases := []struct {
|
|
startTs uint64
|
|
endTs uint64
|
|
name string
|
|
requestType qbtypes.RequestType
|
|
query qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]
|
|
expected qbtypes.Statement
|
|
expectedErr error
|
|
}{
|
|
{
|
|
startTs: releaseTimeNano + uint64(24*time.Hour.Nanoseconds()),
|
|
endTs: releaseTimeNano + uint64(48*time.Hour.Nanoseconds()),
|
|
name: "Time series with limit and count distinct on service.name",
|
|
requestType: qbtypes.RequestTypeTimeSeries,
|
|
query: qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]{
|
|
Signal: telemetrytypes.SignalLogs,
|
|
StepInterval: qbtypes.Step{Duration: 30 * time.Second},
|
|
Aggregations: []qbtypes.LogAggregation{
|
|
{
|
|
Expression: "count_distinct(service.name)",
|
|
},
|
|
},
|
|
Filter: &qbtypes.Filter{
|
|
Expression: "service.name = 'cartservice'",
|
|
},
|
|
Limit: 10,
|
|
GroupBy: []qbtypes.GroupByKey{
|
|
{
|
|
TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{
|
|
Name: "service.name",
|
|
},
|
|
},
|
|
},
|
|
},
|
|
expected: qbtypes.Statement{
|
|
Query: "WITH __resource_filter AS (SELECT fingerprint FROM signoz_logs.distributed_logs_v2_resource WHERE (simpleJSONExtractString(labels, 'service.name') = ? AND labels LIKE ? AND labels LIKE ?) AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ? GROUP BY fingerprint), __limit_cte AS (SELECT toString(multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, NULL)) AS `__GROUP_BY_KEY_0_service.name`, countDistinct(multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, NULL)) AS __result_0 FROM signoz_logs.distributed_logs_v2 WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter) AND timestamp >= ? AND ts_bucket_start >= ? AND timestamp < ? AND ts_bucket_start <= ? GROUP BY `__GROUP_BY_KEY_0_service.name` ORDER BY __result_0 DESC LIMIT ?) SELECT toStartOfInterval(fromUnixTimestamp64Nano(timestamp), INTERVAL 30 SECOND) AS ts, toString(multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, NULL)) AS `__GROUP_BY_KEY_0_service.name`, countDistinct(multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, NULL)) AS __result_0 FROM signoz_logs.distributed_logs_v2 WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter) AND timestamp >= ? AND ts_bucket_start >= ? AND timestamp < ? AND ts_bucket_start <= ? AND (`__GROUP_BY_KEY_0_service.name`) GLOBAL IN (SELECT `__GROUP_BY_KEY_0_service.name` FROM __limit_cte) GROUP BY ts, `__GROUP_BY_KEY_0_service.name`",
|
|
Args: []any{"cartservice", "%service.name%", "%service.name\":\"cartservice%", uint64(1705397400), uint64(1705485600), "1705399200000000000", uint64(1705397400), "1705485600000000000", uint64(1705485600), 10, "1705399200000000000", uint64(1705397400), "1705485600000000000", uint64(1705485600)},
|
|
},
|
|
expectedErr: nil,
|
|
},
|
|
{
|
|
startTs: releaseTimeNano - uint64(24*time.Hour.Nanoseconds()),
|
|
endTs: releaseTimeNano + uint64(48*time.Hour.Nanoseconds()),
|
|
name: "Time series with OR b/w resource attr and attribute filter and count distinct on service.name",
|
|
requestType: qbtypes.RequestTypeTimeSeries,
|
|
query: qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]{
|
|
Signal: telemetrytypes.SignalTraces,
|
|
StepInterval: qbtypes.Step{Duration: 30 * time.Second},
|
|
Aggregations: []qbtypes.LogAggregation{
|
|
{
|
|
Expression: "count_distinct(service.name)",
|
|
},
|
|
},
|
|
Filter: &qbtypes.Filter{
|
|
Expression: "service.name = 'redis-manual' OR http.method = 'GET'",
|
|
},
|
|
Limit: 10,
|
|
GroupBy: []qbtypes.GroupByKey{
|
|
{
|
|
TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{
|
|
Name: "service.name",
|
|
},
|
|
},
|
|
},
|
|
},
|
|
expected: qbtypes.Statement{
|
|
Query: "WITH __limit_cte AS (SELECT toString(multiIf(multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, mapContains(resources_string, 'service.name'), resources_string['service.name'], NULL) IS NOT NULL, multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, mapContains(resources_string, 'service.name'), resources_string['service.name'], NULL), NULL)) AS `__GROUP_BY_KEY_0_service.name`, countDistinct(multiIf(multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, mapContains(resources_string, 'service.name'), resources_string['service.name'], NULL) IS NOT NULL, multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, mapContains(resources_string, 'service.name'), resources_string['service.name'], NULL), NULL)) AS __result_0 FROM signoz_logs.distributed_logs_v2 WHERE ((multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, mapContains(resources_string, 'service.name'), resources_string['service.name'], NULL) = ? AND multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, mapContains(resources_string, 'service.name'), resources_string['service.name'], NULL) IS NOT NULL) OR (attributes_string['http.method'] = ? AND mapContains(attributes_string, 'http.method'))) AND timestamp >= ? AND ts_bucket_start >= ? AND timestamp < ? AND ts_bucket_start <= ? GROUP BY `__GROUP_BY_KEY_0_service.name` ORDER BY __result_0 DESC LIMIT ?) SELECT toStartOfInterval(fromUnixTimestamp64Nano(timestamp), INTERVAL 30 SECOND) AS ts, toString(multiIf(multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, mapContains(resources_string, 'service.name'), resources_string['service.name'], NULL) IS NOT NULL, multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, mapContains(resources_string, 'service.name'), resources_string['service.name'], NULL), NULL)) AS `__GROUP_BY_KEY_0_service.name`, countDistinct(multiIf(multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, mapContains(resources_string, 'service.name'), resources_string['service.name'], NULL) IS NOT NULL, multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, mapContains(resources_string, 'service.name'), resources_string['service.name'], NULL), NULL)) AS __result_0 FROM signoz_logs.distributed_logs_v2 WHERE ((multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, mapContains(resources_string, 'service.name'), resources_string['service.name'], NULL) = ? AND multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, mapContains(resources_string, 'service.name'), resources_string['service.name'], NULL) IS NOT NULL) OR (attributes_string['http.method'] = ? AND mapContains(attributes_string, 'http.method'))) AND timestamp >= ? AND ts_bucket_start >= ? AND timestamp < ? AND ts_bucket_start <= ? AND (`__GROUP_BY_KEY_0_service.name`) GLOBAL IN (SELECT `__GROUP_BY_KEY_0_service.name` FROM __limit_cte) GROUP BY ts, `__GROUP_BY_KEY_0_service.name`",
|
|
Args: []any{"redis-manual", "GET", "1705226400000000000", uint64(1705224600), "1705485600000000000", uint64(1705485600), 10, "redis-manual", "GET", "1705226400000000000", uint64(1705224600), "1705485600000000000", uint64(1705485600)},
|
|
},
|
|
expectedErr: nil,
|
|
},
|
|
{
|
|
startTs: releaseTimeNano + uint64(24*time.Hour.Nanoseconds()),
|
|
endTs: releaseTimeNano + uint64(48*time.Hour.Nanoseconds()),
|
|
name: "Time series with limit + custom order by",
|
|
requestType: qbtypes.RequestTypeTimeSeries,
|
|
query: qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]{
|
|
Signal: telemetrytypes.SignalLogs,
|
|
StepInterval: qbtypes.Step{Duration: 30 * time.Second},
|
|
Aggregations: []qbtypes.LogAggregation{
|
|
{
|
|
Expression: "count()",
|
|
},
|
|
},
|
|
Filter: &qbtypes.Filter{
|
|
Expression: "service.name = 'cartservice'",
|
|
},
|
|
Limit: 10,
|
|
GroupBy: []qbtypes.GroupByKey{
|
|
{
|
|
TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{
|
|
Name: "service.name",
|
|
},
|
|
},
|
|
},
|
|
Order: []qbtypes.OrderBy{
|
|
{
|
|
Key: qbtypes.OrderByKey{
|
|
TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{
|
|
Name: "service.name",
|
|
},
|
|
},
|
|
Direction: qbtypes.OrderDirectionDesc,
|
|
},
|
|
},
|
|
},
|
|
expected: qbtypes.Statement{
|
|
Query: "WITH __resource_filter AS (SELECT fingerprint FROM signoz_logs.distributed_logs_v2_resource WHERE (simpleJSONExtractString(labels, 'service.name') = ? AND labels LIKE ? AND labels LIKE ?) AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ? GROUP BY fingerprint), __limit_cte AS (SELECT toString(multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, NULL)) AS `__GROUP_BY_KEY_0_service.name`, count() AS __result_0 FROM signoz_logs.distributed_logs_v2 WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter) AND timestamp >= ? AND ts_bucket_start >= ? AND timestamp < ? AND ts_bucket_start <= ? GROUP BY `__GROUP_BY_KEY_0_service.name` ORDER BY `__GROUP_BY_KEY_0_service.name` desc LIMIT ?) SELECT toStartOfInterval(fromUnixTimestamp64Nano(timestamp), INTERVAL 30 SECOND) AS ts, toString(multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, NULL)) AS `__GROUP_BY_KEY_0_service.name`, count() AS __result_0 FROM signoz_logs.distributed_logs_v2 WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter) AND timestamp >= ? AND ts_bucket_start >= ? AND timestamp < ? AND ts_bucket_start <= ? AND (`__GROUP_BY_KEY_0_service.name`) GLOBAL IN (SELECT `__GROUP_BY_KEY_0_service.name` FROM __limit_cte) GROUP BY ts, `__GROUP_BY_KEY_0_service.name` ORDER BY `__GROUP_BY_KEY_0_service.name` desc, ts desc",
|
|
Args: []any{"cartservice", "%service.name%", "%service.name\":\"cartservice%", uint64(1705397400), uint64(1705485600), "1705399200000000000", uint64(1705397400), "1705485600000000000", uint64(1705485600), 10, "1705399200000000000", uint64(1705397400), "1705485600000000000", uint64(1705485600)},
|
|
},
|
|
expectedErr: nil,
|
|
},
|
|
{
|
|
startTs: releaseTimeNano + uint64(24*time.Hour.Nanoseconds()),
|
|
endTs: releaseTimeNano + uint64(48*time.Hour.Nanoseconds()),
|
|
name: "Time series with group by on materialized column",
|
|
requestType: qbtypes.RequestTypeTimeSeries,
|
|
query: qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]{
|
|
Signal: telemetrytypes.SignalLogs,
|
|
StepInterval: qbtypes.Step{Duration: 30 * time.Second},
|
|
Aggregations: []qbtypes.LogAggregation{
|
|
{
|
|
Expression: "count()",
|
|
},
|
|
},
|
|
Filter: &qbtypes.Filter{
|
|
Expression: "service.name = 'cartservice'",
|
|
},
|
|
Limit: 10,
|
|
GroupBy: []qbtypes.GroupByKey{
|
|
{
|
|
TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{
|
|
Name: "materialized.key.name",
|
|
FieldContext: telemetrytypes.FieldContextAttribute,
|
|
FieldDataType: telemetrytypes.FieldDataTypeString,
|
|
},
|
|
},
|
|
},
|
|
},
|
|
expected: qbtypes.Statement{
|
|
Query: "WITH __resource_filter AS (SELECT fingerprint FROM signoz_logs.distributed_logs_v2_resource WHERE (simpleJSONExtractString(labels, 'service.name') = ? AND labels LIKE ? AND labels LIKE ?) AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ? GROUP BY fingerprint), __limit_cte AS (SELECT toString(multiIf(`attribute_string_materialized$$key$$name_exists`, `attribute_string_materialized$$key$$name`, NULL)) AS `__GROUP_BY_KEY_0_materialized.key.name`, count() AS __result_0 FROM signoz_logs.distributed_logs_v2 WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter) AND timestamp >= ? AND ts_bucket_start >= ? AND timestamp < ? AND ts_bucket_start <= ? GROUP BY `__GROUP_BY_KEY_0_materialized.key.name` ORDER BY __result_0 DESC LIMIT ?) SELECT toStartOfInterval(fromUnixTimestamp64Nano(timestamp), INTERVAL 30 SECOND) AS ts, toString(multiIf(`attribute_string_materialized$$key$$name_exists`, `attribute_string_materialized$$key$$name`, NULL)) AS `__GROUP_BY_KEY_0_materialized.key.name`, count() AS __result_0 FROM signoz_logs.distributed_logs_v2 WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter) AND timestamp >= ? AND ts_bucket_start >= ? AND timestamp < ? AND ts_bucket_start <= ? AND (`__GROUP_BY_KEY_0_materialized.key.name`) GLOBAL IN (SELECT `__GROUP_BY_KEY_0_materialized.key.name` FROM __limit_cte) GROUP BY ts, `__GROUP_BY_KEY_0_materialized.key.name`",
|
|
Args: []any{"cartservice", "%service.name%", "%service.name\":\"cartservice%", uint64(1705397400), uint64(1705485600), "1705399200000000000", uint64(1705397400), "1705485600000000000", uint64(1705485600), 10, "1705399200000000000", uint64(1705397400), "1705485600000000000", uint64(1705485600)},
|
|
},
|
|
},
|
|
{
|
|
startTs: releaseTimeNano + uint64(24*time.Hour.Nanoseconds()),
|
|
endTs: releaseTimeNano + uint64(48*time.Hour.Nanoseconds()),
|
|
name: "Time series with materialised column using or with regex operator",
|
|
requestType: qbtypes.RequestTypeTimeSeries,
|
|
query: qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]{
|
|
Signal: telemetrytypes.SignalLogs,
|
|
StepInterval: qbtypes.Step{Duration: 30 * time.Second},
|
|
Aggregations: []qbtypes.LogAggregation{
|
|
{
|
|
Expression: "count()",
|
|
},
|
|
},
|
|
Filter: &qbtypes.Filter{
|
|
Expression: "materialized.key.name REGEXP 'redis.*' OR materialized.key.name = 'memcached'",
|
|
},
|
|
Limit: 10,
|
|
},
|
|
expected: qbtypes.Statement{
|
|
Query: "SELECT toStartOfInterval(fromUnixTimestamp64Nano(timestamp), INTERVAL 30 SECOND) AS ts, count() AS __result_0 FROM signoz_logs.distributed_logs_v2 WHERE ((match(`attribute_string_materialized$$key$$name`, ?) AND `attribute_string_materialized$$key$$name_exists`) OR (`attribute_string_materialized$$key$$name` = ? AND `attribute_string_materialized$$key$$name_exists`)) AND timestamp >= ? AND ts_bucket_start >= ? AND timestamp < ? AND ts_bucket_start <= ? GROUP BY ts",
|
|
Args: []any{"redis.*", "memcached", "1705399200000000000", uint64(1705397400), "1705485600000000000", uint64(1705485600)},
|
|
},
|
|
expectedErr: nil,
|
|
},
|
|
}
|
|
|
|
ctx := context.Background()
|
|
fl := flaggertest.New(t)
|
|
|
|
mockMetadataStore := telemetrytypestest.NewMockMetadataStore()
|
|
keysMap := logstelemetryschema.BuildCompleteFieldKeyMap(releaseTime)
|
|
|
|
mockMetadataStore.KeysMap = keysMap
|
|
|
|
fm := logstelemetryschema.NewFieldMapper(fl)
|
|
cb := logstelemetryschema.NewConditionBuilder(fm, fl)
|
|
|
|
aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, fm, cb, fl)
|
|
|
|
statementBuilder := NewLogQueryStatementBuilder(
|
|
instrumentationtest.New().ToProviderSettings(),
|
|
mockMetadataStore,
|
|
fm,
|
|
cb,
|
|
aggExprRewriter,
|
|
logstelemetryschema.DefaultFullTextColumn,
|
|
fl,
|
|
nil,
|
|
statementbuilder.Config{SkipResourceFingerprint: statementbuilder.SkipResourceFingerprint{Enabled: false, Threshold: 100000}},
|
|
)
|
|
|
|
for _, c := range cases {
|
|
t.Run(c.name, func(t *testing.T) {
|
|
|
|
q, err := statementBuilder.Build(ctx, valuer.UUID{}, c.startTs, c.endTs, c.requestType, c.query, nil)
|
|
|
|
if c.expectedErr != nil {
|
|
require.Error(t, err)
|
|
require.Contains(t, err.Error(), c.expectedErr.Error())
|
|
} else {
|
|
require.NoError(t, err)
|
|
require.Equal(t, c.expected.Query, q.Query)
|
|
require.Equal(t, c.expected.Args, q.Args)
|
|
require.Equal(t, c.expected.Warnings, q.Warnings)
|
|
}
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestStatementBuilderListQuery(t *testing.T) {
|
|
cases := []struct {
|
|
name string
|
|
requestType qbtypes.RequestType
|
|
query qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]
|
|
variables map[string]qbtypes.VariableItem
|
|
expected qbtypes.Statement
|
|
expectedErr error
|
|
expectWarn bool
|
|
}{
|
|
{
|
|
name: "default list",
|
|
requestType: qbtypes.RequestTypeRaw,
|
|
query: qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]{
|
|
Signal: telemetrytypes.SignalLogs,
|
|
Filter: &qbtypes.Filter{
|
|
Expression: "service.name = 'cartservice'",
|
|
},
|
|
Limit: 10,
|
|
},
|
|
expected: qbtypes.Statement{
|
|
Query: "WITH __resource_filter AS (SELECT fingerprint FROM signoz_logs.distributed_logs_v2_resource WHERE (simpleJSONExtractString(labels, 'service.name') = ? AND labels LIKE ? AND labels LIKE ?) AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ? GROUP BY fingerprint) SELECT timestamp, id, trace_id, span_id, trace_flags, severity_text, severity_number, scope_name, scope_version, body, attributes_string, attributes_number, attributes_bool, resources_string, scope_string FROM signoz_logs.distributed_logs_v2 WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter) AND timestamp >= ? AND ts_bucket_start >= ? AND timestamp < ? AND ts_bucket_start <= ? LIMIT ?",
|
|
Args: []any{"cartservice", "%service.name%", "%service.name\":\"cartservice%", uint64(1747945619), uint64(1747983448), "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 10},
|
|
},
|
|
expectedErr: nil,
|
|
},
|
|
{
|
|
name: "list query with mat col order by",
|
|
requestType: qbtypes.RequestTypeRaw,
|
|
query: qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]{
|
|
Signal: telemetrytypes.SignalLogs,
|
|
Filter: &qbtypes.Filter{
|
|
Expression: "service.name = 'cartservice'",
|
|
},
|
|
Limit: 10,
|
|
Order: []qbtypes.OrderBy{
|
|
{
|
|
Key: qbtypes.OrderByKey{
|
|
TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{
|
|
Name: "materialized.key.name",
|
|
FieldContext: telemetrytypes.FieldContextAttribute,
|
|
FieldDataType: telemetrytypes.FieldDataTypeString,
|
|
},
|
|
},
|
|
Direction: qbtypes.OrderDirectionDesc,
|
|
},
|
|
},
|
|
},
|
|
expected: qbtypes.Statement{
|
|
Query: "WITH __resource_filter AS (SELECT fingerprint FROM signoz_logs.distributed_logs_v2_resource WHERE (simpleJSONExtractString(labels, 'service.name') = ? AND labels LIKE ? AND labels LIKE ?) AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ? GROUP BY fingerprint) SELECT timestamp, id, trace_id, span_id, trace_flags, severity_text, severity_number, scope_name, scope_version, body, attributes_string, attributes_number, attributes_bool, resources_string, scope_string FROM signoz_logs.distributed_logs_v2 WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter) AND timestamp >= ? AND ts_bucket_start >= ? AND timestamp < ? AND ts_bucket_start <= ? ORDER BY multiIf(`attribute_string_materialized$$key$$name_exists`, `attribute_string_materialized$$key$$name`, NULL) desc LIMIT ?",
|
|
Args: []any{"cartservice", "%service.name%", "%service.name\":\"cartservice%", uint64(1747945619), uint64(1747983448), "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 10},
|
|
},
|
|
expectedErr: nil,
|
|
},
|
|
{
|
|
name: "list query with mat col using or and regex operator",
|
|
requestType: qbtypes.RequestTypeRaw,
|
|
query: qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]{
|
|
Signal: telemetrytypes.SignalLogs,
|
|
Filter: &qbtypes.Filter{
|
|
Expression: "materialized.key.name REGEXP 'redis.*' OR materialized.key.name = 'memcached'",
|
|
},
|
|
Limit: 10,
|
|
Order: []qbtypes.OrderBy{
|
|
{
|
|
Key: qbtypes.OrderByKey{
|
|
TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{
|
|
Name: "materialized.key.name",
|
|
FieldContext: telemetrytypes.FieldContextAttribute,
|
|
FieldDataType: telemetrytypes.FieldDataTypeString,
|
|
},
|
|
},
|
|
Direction: qbtypes.OrderDirectionDesc,
|
|
},
|
|
},
|
|
},
|
|
expected: qbtypes.Statement{
|
|
Query: "SELECT timestamp, id, trace_id, span_id, trace_flags, severity_text, severity_number, scope_name, scope_version, body, attributes_string, attributes_number, attributes_bool, resources_string, scope_string FROM signoz_logs.distributed_logs_v2 WHERE ((match(`attribute_string_materialized$$key$$name`, ?) AND `attribute_string_materialized$$key$$name_exists`) OR (`attribute_string_materialized$$key$$name` = ? AND `attribute_string_materialized$$key$$name_exists`)) AND timestamp >= ? AND ts_bucket_start >= ? AND timestamp < ? AND ts_bucket_start <= ? ORDER BY multiIf(`attribute_string_materialized$$key$$name_exists`, `attribute_string_materialized$$key$$name`, NULL) desc LIMIT ?",
|
|
Args: []any{"redis.*", "memcached", "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 10},
|
|
},
|
|
expectedErr: nil,
|
|
},
|
|
{
|
|
name: "filter skips entirely but emits LIKE-without-wildcards warning",
|
|
requestType: qbtypes.RequestTypeRaw,
|
|
query: qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]{
|
|
Signal: telemetrytypes.SignalLogs,
|
|
Filter: &qbtypes.Filter{
|
|
Expression: "message LIKE 'plain' OR message IN $env",
|
|
},
|
|
Limit: 10,
|
|
},
|
|
variables: map[string]qbtypes.VariableItem{
|
|
"env": {Type: qbtypes.DynamicVariableType, Value: "__all__"},
|
|
},
|
|
expectedErr: nil,
|
|
expectWarn: true,
|
|
},
|
|
}
|
|
|
|
ctx := context.Background()
|
|
fl := flaggertest.New(t)
|
|
mockMetadataStore := telemetrytypestest.NewMockMetadataStore()
|
|
fm := logstelemetryschema.NewFieldMapper(fl)
|
|
|
|
// Create a test release time
|
|
releaseTime := time.Date(2024, 1, 15, 10, 0, 0, 0, time.UTC)
|
|
mockMetadataStore.KeysMap = logstelemetryschema.BuildCompleteFieldKeyMap(releaseTime)
|
|
cb := logstelemetryschema.NewConditionBuilder(fm, fl)
|
|
|
|
aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, fm, cb, fl)
|
|
|
|
statementBuilder := NewLogQueryStatementBuilder(
|
|
instrumentationtest.New().ToProviderSettings(),
|
|
mockMetadataStore,
|
|
fm,
|
|
cb,
|
|
aggExprRewriter,
|
|
logstelemetryschema.DefaultFullTextColumn,
|
|
fl,
|
|
nil,
|
|
statementbuilder.Config{SkipResourceFingerprint: statementbuilder.SkipResourceFingerprint{Enabled: false, Threshold: 100000}},
|
|
)
|
|
|
|
for _, c := range cases {
|
|
t.Run(c.name, func(t *testing.T) {
|
|
|
|
q, err := statementBuilder.Build(ctx, valuer.UUID{}, 1747947419000, 1747983448000, c.requestType, c.query, c.variables)
|
|
|
|
if c.expectedErr != nil {
|
|
require.Error(t, err)
|
|
require.Contains(t, err.Error(), c.expectedErr.Error())
|
|
} else {
|
|
require.NoError(t, err)
|
|
if c.expectWarn {
|
|
require.NotEmpty(t, q.Warnings)
|
|
} else {
|
|
require.Equal(t, c.expected.Query, q.Query)
|
|
require.Equal(t, c.expected.Args, q.Args)
|
|
require.Equal(t, c.expected.Warnings, q.Warnings)
|
|
}
|
|
}
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestStatementBuilderListQueryResourceTests(t *testing.T) {
|
|
cases := []struct {
|
|
name string
|
|
requestType qbtypes.RequestType
|
|
query qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]
|
|
expected qbtypes.Statement
|
|
expectedErr error
|
|
}{
|
|
{
|
|
name: "List with full text search",
|
|
requestType: qbtypes.RequestTypeRaw,
|
|
query: qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]{
|
|
Signal: telemetrytypes.SignalLogs,
|
|
Filter: &qbtypes.Filter{
|
|
Expression: "hello",
|
|
},
|
|
Limit: 10,
|
|
},
|
|
expected: qbtypes.Statement{
|
|
Query: "SELECT timestamp, id, trace_id, span_id, trace_flags, severity_text, severity_number, scope_name, scope_version, body, attributes_string, attributes_number, attributes_bool, resources_string, scope_string FROM signoz_logs.distributed_logs_v2 WHERE match(LOWER(body), LOWER(?)) AND timestamp >= ? AND ts_bucket_start >= ? AND timestamp < ? AND ts_bucket_start <= ? LIMIT ?",
|
|
Args: []any{"hello", "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 10},
|
|
},
|
|
expectedErr: nil,
|
|
},
|
|
{
|
|
name: "list query with mat col order by",
|
|
requestType: qbtypes.RequestTypeRaw,
|
|
query: qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]{
|
|
Signal: telemetrytypes.SignalLogs,
|
|
Filter: &qbtypes.Filter{
|
|
Expression: "service.name = 'cartservice' hello",
|
|
},
|
|
Limit: 10,
|
|
Order: []qbtypes.OrderBy{
|
|
{
|
|
Key: qbtypes.OrderByKey{
|
|
TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{
|
|
Name: "materialized.key.name",
|
|
FieldContext: telemetrytypes.FieldContextAttribute,
|
|
FieldDataType: telemetrytypes.FieldDataTypeString,
|
|
},
|
|
},
|
|
Direction: qbtypes.OrderDirectionDesc,
|
|
},
|
|
},
|
|
},
|
|
expected: qbtypes.Statement{
|
|
Query: "WITH __resource_filter AS (SELECT fingerprint FROM signoz_logs.distributed_logs_v2_resource WHERE (simpleJSONExtractString(labels, 'service.name') = ? AND labels LIKE ? AND labels LIKE ?) AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ? GROUP BY fingerprint) SELECT timestamp, id, trace_id, span_id, trace_flags, severity_text, severity_number, scope_name, scope_version, body, attributes_string, attributes_number, attributes_bool, resources_string, scope_string FROM signoz_logs.distributed_logs_v2 WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter) AND match(LOWER(body), LOWER(?)) AND timestamp >= ? AND ts_bucket_start >= ? AND timestamp < ? AND ts_bucket_start <= ? ORDER BY multiIf(`attribute_string_materialized$$key$$name_exists`, `attribute_string_materialized$$key$$name`, NULL) desc LIMIT ?",
|
|
Args: []any{"cartservice", "%service.name%", "%service.name\":\"cartservice%", uint64(1747945619), uint64(1747983448), "hello", "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 10},
|
|
},
|
|
expectedErr: nil,
|
|
},
|
|
{
|
|
name: "List with json search",
|
|
requestType: qbtypes.RequestTypeRaw,
|
|
query: qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]{
|
|
Signal: telemetrytypes.SignalLogs,
|
|
Filter: &qbtypes.Filter{
|
|
Expression: "body.status = 'success'",
|
|
},
|
|
Limit: 10,
|
|
},
|
|
expected: qbtypes.Statement{
|
|
Query: "SELECT timestamp, id, trace_id, span_id, trace_flags, severity_text, severity_number, scope_name, scope_version, body, attributes_string, attributes_number, attributes_bool, resources_string, scope_string FROM signoz_logs.distributed_logs_v2 WHERE (JSON_VALUE(body, '$.\"status\"') = ? AND JSON_EXISTS(body, '$.\"status\"')) AND timestamp >= ? AND ts_bucket_start >= ? AND timestamp < ? AND ts_bucket_start <= ? LIMIT ?",
|
|
Args: []any{"success", "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 10},
|
|
Warnings: []string{querybuilder.NewKeyNotFoundWarning("status")},
|
|
},
|
|
expectedErr: nil,
|
|
},
|
|
{
|
|
name: "IN operator with json search",
|
|
requestType: qbtypes.RequestTypeRaw,
|
|
query: qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]{
|
|
Signal: telemetrytypes.SignalLogs,
|
|
Filter: &qbtypes.Filter{
|
|
Expression: "body.user_names[*] IN 'john_doe'",
|
|
},
|
|
Limit: 10,
|
|
},
|
|
expected: qbtypes.Statement{
|
|
Query: "SELECT timestamp, id, trace_id, span_id, trace_flags, severity_text, severity_number, scope_name, scope_version, body, attributes_string, attributes_number, attributes_bool, resources_string, scope_string FROM signoz_logs.distributed_logs_v2 WHERE ((JSONExtract(JSON_QUERY(body, '$.\"user_names\"[*]'), 'Array(String)') = ?) AND JSON_EXISTS(body, '$.\"user_names\"[*]')) AND timestamp >= ? AND ts_bucket_start >= ? AND timestamp < ? AND ts_bucket_start <= ? LIMIT ?",
|
|
Args: []any{"john_doe", "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 10},
|
|
Warnings: []string{querybuilder.NewKeyNotFoundWarning("user_names[*]")},
|
|
},
|
|
expectedErr: nil,
|
|
},
|
|
{
|
|
name: "has with json search",
|
|
requestType: qbtypes.RequestTypeRaw,
|
|
query: qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]{
|
|
Signal: telemetrytypes.SignalLogs,
|
|
Filter: &qbtypes.Filter{
|
|
Expression: "has(body.user_names[*], 'john_doe')",
|
|
},
|
|
Limit: 10,
|
|
},
|
|
expected: qbtypes.Statement{
|
|
Query: "SELECT timestamp, id, trace_id, span_id, trace_flags, severity_text, severity_number, scope_name, scope_version, body, attributes_string, attributes_number, attributes_bool, resources_string, scope_string FROM signoz_logs.distributed_logs_v2 WHERE (has(JSONExtract(JSON_QUERY(body, '$.\"user_names\"[*]'), 'Array(Nullable(String))'), ?) OR ifNull((JSON_VALUE(body, '$.\"user_names\"') = ? AND JSONType(body, 'user_names') NOT IN ('Array', 'Object', 'Null')), false)) AND timestamp >= ? AND ts_bucket_start >= ? AND timestamp < ? AND ts_bucket_start <= ? LIMIT ?",
|
|
Args: []any{"john_doe", "john_doe", "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 10},
|
|
Warnings: []string{querybuilder.NewKeyNotFoundWarning("user_names[*]")},
|
|
},
|
|
expectedErr: nil,
|
|
},
|
|
}
|
|
|
|
ctx := context.Background()
|
|
fl := flaggertest.New(t)
|
|
mockMetadataStore := telemetrytypestest.NewMockMetadataStore()
|
|
fm := logstelemetryschema.NewFieldMapper(fl)
|
|
// Create a test release time
|
|
releaseTime := time.Date(2024, 1, 15, 10, 0, 0, 0, time.UTC)
|
|
mockMetadataStore.KeysMap = logstelemetryschema.BuildCompleteFieldKeyMap(releaseTime)
|
|
cb := logstelemetryschema.NewConditionBuilder(fm, fl)
|
|
|
|
aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, fm, cb, fl)
|
|
|
|
statementBuilder := NewLogQueryStatementBuilder(
|
|
instrumentationtest.New().ToProviderSettings(),
|
|
mockMetadataStore,
|
|
fm,
|
|
cb,
|
|
aggExprRewriter,
|
|
logstelemetryschema.DefaultFullTextColumn,
|
|
fl,
|
|
nil,
|
|
statementbuilder.Config{SkipResourceFingerprint: statementbuilder.SkipResourceFingerprint{Enabled: false, Threshold: 100000}},
|
|
)
|
|
|
|
for _, c := range cases {
|
|
t.Run(c.name, func(t *testing.T) {
|
|
|
|
q, err := statementBuilder.Build(ctx, valuer.UUID{}, 1747947419000, 1747983448000, c.requestType, c.query, nil)
|
|
|
|
if c.expectedErr != nil {
|
|
require.Error(t, err)
|
|
require.Contains(t, err.Error(), c.expectedErr.Error())
|
|
} else {
|
|
require.NoError(t, err)
|
|
require.Equal(t, c.expected.Query, q.Query)
|
|
require.Equal(t, c.expected.Args, q.Args)
|
|
require.Equal(t, c.expected.Warnings, q.Warnings)
|
|
}
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestStatementBuilderTimeSeriesBodyGroupBy(t *testing.T) {
|
|
cases := []struct {
|
|
name string
|
|
requestType qbtypes.RequestType
|
|
query qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]
|
|
expected qbtypes.Statement
|
|
expectedErrContains string
|
|
}{
|
|
{
|
|
name: "Time series with limit and body group by",
|
|
requestType: qbtypes.RequestTypeTimeSeries,
|
|
query: qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]{
|
|
Signal: telemetrytypes.SignalLogs,
|
|
StepInterval: qbtypes.Step{Duration: 30 * time.Second},
|
|
Aggregations: []qbtypes.LogAggregation{
|
|
{
|
|
Expression: "count()",
|
|
},
|
|
},
|
|
Filter: &qbtypes.Filter{
|
|
Expression: "service.name = 'cartservice'",
|
|
},
|
|
Limit: 10,
|
|
GroupBy: []qbtypes.GroupByKey{
|
|
{
|
|
TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{
|
|
Name: "status",
|
|
FieldContext: telemetrytypes.FieldContextBody,
|
|
},
|
|
},
|
|
},
|
|
},
|
|
expectedErrContains: "Operation isn't available for the body column",
|
|
},
|
|
}
|
|
|
|
ctx := context.Background()
|
|
fl := flaggertest.New(t)
|
|
mockMetadataStore := telemetrytypestest.NewMockMetadataStore()
|
|
fm := logstelemetryschema.NewFieldMapper(fl)
|
|
// Create a test release time
|
|
releaseTime := time.Date(2024, 1, 15, 10, 0, 0, 0, time.UTC)
|
|
mockMetadataStore.KeysMap = logstelemetryschema.BuildCompleteFieldKeyMap(releaseTime)
|
|
cb := logstelemetryschema.NewConditionBuilder(fm, fl)
|
|
|
|
aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, fm, cb, fl)
|
|
|
|
statementBuilder := NewLogQueryStatementBuilder(
|
|
instrumentationtest.New().ToProviderSettings(),
|
|
mockMetadataStore,
|
|
fm,
|
|
cb,
|
|
aggExprRewriter,
|
|
logstelemetryschema.DefaultFullTextColumn,
|
|
fl,
|
|
nil,
|
|
statementbuilder.Config{SkipResourceFingerprint: statementbuilder.SkipResourceFingerprint{Enabled: false, Threshold: 100000}},
|
|
)
|
|
|
|
for _, c := range cases {
|
|
t.Run(c.name, func(t *testing.T) {
|
|
|
|
q, err := statementBuilder.Build(ctx, valuer.UUID{}, 1747947419000, 1747983448000, c.requestType, c.query, nil)
|
|
|
|
if c.expectedErrContains != "" {
|
|
require.Error(t, err)
|
|
require.Contains(t, err.Error(), c.expectedErrContains)
|
|
} else {
|
|
require.NoError(t, err)
|
|
require.Equal(t, c.expected.Query, q.Query)
|
|
require.Equal(t, c.expected.Args, q.Args)
|
|
require.Equal(t, c.expected.Warnings, q.Warnings)
|
|
}
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestStatementBuilderListQueryServiceCollision(t *testing.T) {
|
|
cases := []struct {
|
|
name string
|
|
requestType qbtypes.RequestType
|
|
query qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]
|
|
expected qbtypes.Statement
|
|
expectedErr error
|
|
expectWarn bool
|
|
}{
|
|
{
|
|
name: "default list",
|
|
requestType: qbtypes.RequestTypeRaw,
|
|
query: qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]{
|
|
Signal: telemetrytypes.SignalLogs,
|
|
Filter: &qbtypes.Filter{
|
|
Expression: "(service.name = 'cartservice' AND body CONTAINS 'error')",
|
|
},
|
|
Limit: 10,
|
|
},
|
|
expected: qbtypes.Statement{
|
|
Query: "WITH __resource_filter AS (SELECT fingerprint FROM signoz_logs.distributed_logs_v2_resource WHERE ((simpleJSONExtractString(labels, 'service.name') = ? AND labels LIKE ? AND labels LIKE ?)) AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ? GROUP BY fingerprint) SELECT timestamp, id, trace_id, span_id, trace_flags, severity_text, severity_number, scope_name, scope_version, body, attributes_string, attributes_number, attributes_bool, resources_string, scope_string FROM signoz_logs.distributed_logs_v2 WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter) AND (LOWER(body) LIKE LOWER(?)) AND timestamp >= ? AND ts_bucket_start >= ? AND timestamp < ? AND ts_bucket_start <= ? LIMIT ?",
|
|
Args: []any{"cartservice", "%service.name%", "%service.name\":\"cartservice%", uint64(1747945619), uint64(1747983448), "%error%", "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 10},
|
|
},
|
|
expectedErr: nil,
|
|
expectWarn: true,
|
|
},
|
|
{
|
|
name: "list query with mat col order by",
|
|
requestType: qbtypes.RequestTypeRaw,
|
|
query: qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]{
|
|
Signal: telemetrytypes.SignalLogs,
|
|
Filter: &qbtypes.Filter{
|
|
Expression: "service.name = 'cartservice' AND body CONTAINS 'error'",
|
|
},
|
|
Limit: 10,
|
|
Order: []qbtypes.OrderBy{
|
|
{
|
|
Key: qbtypes.OrderByKey{
|
|
TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{
|
|
Name: "materialized.key.name",
|
|
FieldContext: telemetrytypes.FieldContextAttribute,
|
|
FieldDataType: telemetrytypes.FieldDataTypeString,
|
|
},
|
|
},
|
|
Direction: qbtypes.OrderDirectionDesc,
|
|
},
|
|
},
|
|
},
|
|
expected: qbtypes.Statement{
|
|
Query: "WITH __resource_filter AS (SELECT fingerprint FROM signoz_logs.distributed_logs_v2_resource WHERE (simpleJSONExtractString(labels, 'service.name') = ? AND labels LIKE ? AND labels LIKE ?) AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ? GROUP BY fingerprint) SELECT timestamp, id, trace_id, span_id, trace_flags, severity_text, severity_number, scope_name, scope_version, body, attributes_string, attributes_number, attributes_bool, resources_string, scope_string FROM signoz_logs.distributed_logs_v2 WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter) AND LOWER(body) LIKE LOWER(?) AND timestamp >= ? AND ts_bucket_start >= ? AND timestamp < ? AND ts_bucket_start <= ? ORDER BY multiIf(`attribute_string_materialized$$key$$name_exists`, `attribute_string_materialized$$key$$name`, NULL) desc LIMIT ?",
|
|
Args: []any{"cartservice", "%service.name%", "%service.name\":\"cartservice%", uint64(1747945619), uint64(1747983448), "%error%", "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 10},
|
|
},
|
|
expectedErr: nil,
|
|
expectWarn: true,
|
|
},
|
|
}
|
|
|
|
ctx := context.Background()
|
|
mockMetadataStore := telemetrytypestest.NewMockMetadataStore()
|
|
fl := flaggertest.New(t)
|
|
fm := logstelemetryschema.NewFieldMapper(fl)
|
|
mockMetadataStore.KeysMap = logstelemetryschema.BuildCompleteFieldKeyMapCollision()
|
|
cb := logstelemetryschema.NewConditionBuilder(fm, fl)
|
|
|
|
aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, fm, cb, fl)
|
|
|
|
statementBuilder := NewLogQueryStatementBuilder(
|
|
instrumentationtest.New().ToProviderSettings(),
|
|
mockMetadataStore,
|
|
fm,
|
|
cb,
|
|
aggExprRewriter,
|
|
logstelemetryschema.DefaultFullTextColumn,
|
|
fl,
|
|
nil,
|
|
statementbuilder.Config{SkipResourceFingerprint: statementbuilder.SkipResourceFingerprint{Enabled: false, Threshold: 100000}},
|
|
)
|
|
|
|
for _, c := range cases {
|
|
t.Run(c.name, func(t *testing.T) {
|
|
|
|
q, err := statementBuilder.Build(ctx, valuer.UUID{}, 1747947419000, 1747983448000, c.requestType, c.query, nil)
|
|
|
|
if c.expectedErr != nil {
|
|
require.Error(t, err)
|
|
require.Contains(t, err.Error(), c.expectedErr.Error())
|
|
} else {
|
|
require.NoError(t, err)
|
|
require.Equal(t, c.expected.Query, q.Query)
|
|
require.Equal(t, c.expected.Args, q.Args)
|
|
if c.expectWarn {
|
|
require.True(t, len(q.Warnings) > 0)
|
|
}
|
|
}
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestAdjustKey(t *testing.T) {
|
|
|
|
// Create a test release time
|
|
releaseTime := time.Date(2024, 1, 15, 10, 0, 0, 0, time.UTC)
|
|
cases := []struct {
|
|
name string
|
|
inputKey telemetrytypes.TelemetryFieldKey
|
|
keysMap map[string][]*telemetrytypes.TelemetryFieldKey
|
|
expectedKey telemetrytypes.TelemetryFieldKey
|
|
}{
|
|
{
|
|
name: "intrinsic field with no other key match - use intrinsic",
|
|
inputKey: telemetrytypes.TelemetryFieldKey{
|
|
Name: "severity_text",
|
|
FieldContext: telemetrytypes.FieldContextUnspecified,
|
|
FieldDataType: telemetrytypes.FieldDataTypeUnspecified,
|
|
},
|
|
keysMap: logstelemetryschema.BuildCompleteFieldKeyMap(releaseTime),
|
|
expectedKey: logstelemetryschema.IntrinsicFields["severity_text"],
|
|
},
|
|
{
|
|
name: "intrinsic field with other key match - no override",
|
|
inputKey: telemetrytypes.TelemetryFieldKey{
|
|
Name: "body",
|
|
FieldContext: telemetrytypes.FieldContextUnspecified,
|
|
FieldDataType: telemetrytypes.FieldDataTypeUnspecified,
|
|
},
|
|
keysMap: map[string][]*telemetrytypes.TelemetryFieldKey{
|
|
"body": {
|
|
{
|
|
Name: "body",
|
|
FieldContext: telemetrytypes.FieldContextBody,
|
|
FieldDataType: telemetrytypes.FieldDataTypeUnspecified,
|
|
},
|
|
{
|
|
Name: "body",
|
|
FieldContext: telemetrytypes.FieldContextAttribute,
|
|
FieldDataType: telemetrytypes.FieldDataTypeUnspecified,
|
|
},
|
|
},
|
|
},
|
|
expectedKey: telemetrytypes.TelemetryFieldKey{
|
|
Name: "body",
|
|
FieldContext: telemetrytypes.FieldContextUnspecified,
|
|
FieldDataType: telemetrytypes.FieldDataTypeUnspecified,
|
|
},
|
|
},
|
|
{
|
|
name: "json field with no context specified",
|
|
inputKey: telemetrytypes.TelemetryFieldKey{
|
|
Name: "severity_number",
|
|
FieldContext: telemetrytypes.FieldContextBody,
|
|
FieldDataType: telemetrytypes.FieldDataTypeUnspecified,
|
|
},
|
|
keysMap: logstelemetryschema.BuildCompleteFieldKeyMap(releaseTime),
|
|
expectedKey: telemetrytypes.TelemetryFieldKey{
|
|
Name: "severity_number",
|
|
FieldContext: telemetrytypes.FieldContextBody,
|
|
FieldDataType: telemetrytypes.FieldDataTypeNumber,
|
|
},
|
|
},
|
|
{
|
|
name: "single matching key in metadata",
|
|
inputKey: telemetrytypes.TelemetryFieldKey{
|
|
Name: "service.name",
|
|
FieldContext: telemetrytypes.FieldContextUnspecified,
|
|
FieldDataType: telemetrytypes.FieldDataTypeUnspecified,
|
|
},
|
|
keysMap: logstelemetryschema.BuildCompleteFieldKeyMap(releaseTime),
|
|
expectedKey: *logstelemetryschema.BuildCompleteFieldKeyMap(releaseTime)["service.name"][0],
|
|
},
|
|
{
|
|
name: "single matching key with incorrect context specified - no override",
|
|
inputKey: telemetrytypes.TelemetryFieldKey{
|
|
Name: "service.name",
|
|
FieldContext: telemetrytypes.FieldContextAttribute,
|
|
FieldDataType: telemetrytypes.FieldDataTypeUnspecified,
|
|
},
|
|
keysMap: logstelemetryschema.BuildCompleteFieldKeyMap(releaseTime),
|
|
expectedKey: telemetrytypes.TelemetryFieldKey{
|
|
Name: "service.name",
|
|
FieldContext: telemetrytypes.FieldContextAttribute,
|
|
FieldDataType: telemetrytypes.FieldDataTypeUnspecified,
|
|
},
|
|
},
|
|
{
|
|
name: "single matching key with no context specified - override",
|
|
inputKey: telemetrytypes.TelemetryFieldKey{
|
|
Name: "service.name",
|
|
FieldContext: telemetrytypes.FieldContextUnspecified,
|
|
FieldDataType: telemetrytypes.FieldDataTypeUnspecified,
|
|
},
|
|
keysMap: logstelemetryschema.BuildCompleteFieldKeyMap(releaseTime),
|
|
expectedKey: *logstelemetryschema.BuildCompleteFieldKeyMap(releaseTime)["service.name"][0],
|
|
},
|
|
{
|
|
name: "multiple matching keys - all materialized",
|
|
inputKey: telemetrytypes.TelemetryFieldKey{
|
|
Name: "multi.mat.key",
|
|
FieldContext: telemetrytypes.FieldContextUnspecified,
|
|
FieldDataType: telemetrytypes.FieldDataTypeUnspecified,
|
|
},
|
|
keysMap: logstelemetryschema.BuildCompleteFieldKeyMap(releaseTime),
|
|
expectedKey: telemetrytypes.TelemetryFieldKey{
|
|
Name: "multi.mat.key",
|
|
FieldDataType: telemetrytypes.FieldDataTypeString,
|
|
Materialized: true,
|
|
},
|
|
},
|
|
{
|
|
name: "multiple matching keys - mixed materialization",
|
|
inputKey: telemetrytypes.TelemetryFieldKey{
|
|
Name: "mixed.materialization.key",
|
|
FieldContext: telemetrytypes.FieldContextUnspecified,
|
|
FieldDataType: telemetrytypes.FieldDataTypeUnspecified,
|
|
},
|
|
keysMap: logstelemetryschema.BuildCompleteFieldKeyMap(releaseTime),
|
|
expectedKey: telemetrytypes.TelemetryFieldKey{
|
|
Name: "mixed.materialization.key",
|
|
FieldDataType: telemetrytypes.FieldDataTypeString,
|
|
Materialized: false,
|
|
},
|
|
},
|
|
{
|
|
name: "multiple matching keys with context specified",
|
|
inputKey: telemetrytypes.TelemetryFieldKey{
|
|
Name: "mixed.materialization.key",
|
|
FieldContext: telemetrytypes.FieldContextAttribute,
|
|
FieldDataType: telemetrytypes.FieldDataTypeUnspecified,
|
|
},
|
|
keysMap: logstelemetryschema.BuildCompleteFieldKeyMap(releaseTime),
|
|
expectedKey: *logstelemetryschema.BuildCompleteFieldKeyMap(releaseTime)["mixed.materialization.key"][0],
|
|
},
|
|
{
|
|
name: "no matching keys - unknown field",
|
|
inputKey: telemetrytypes.TelemetryFieldKey{
|
|
Name: "unknown.field",
|
|
FieldContext: telemetrytypes.FieldContextUnspecified,
|
|
FieldDataType: telemetrytypes.FieldDataTypeUnspecified,
|
|
},
|
|
keysMap: logstelemetryschema.BuildCompleteFieldKeyMap(releaseTime),
|
|
expectedKey: telemetrytypes.TelemetryFieldKey{
|
|
Name: "unknown.field",
|
|
FieldContext: telemetrytypes.FieldContextUnspecified,
|
|
FieldDataType: telemetrytypes.FieldDataTypeUnspecified,
|
|
Materialized: false,
|
|
},
|
|
},
|
|
{
|
|
name: "no matching keys with context filter",
|
|
inputKey: telemetrytypes.TelemetryFieldKey{
|
|
Name: "unknown.field",
|
|
FieldContext: telemetrytypes.FieldContextAttribute,
|
|
FieldDataType: telemetrytypes.FieldDataTypeUnspecified,
|
|
},
|
|
keysMap: logstelemetryschema.BuildCompleteFieldKeyMap(releaseTime),
|
|
expectedKey: telemetrytypes.TelemetryFieldKey{
|
|
Name: "unknown.field",
|
|
FieldContext: telemetrytypes.FieldContextAttribute,
|
|
FieldDataType: telemetrytypes.FieldDataTypeUnspecified,
|
|
Materialized: false,
|
|
},
|
|
},
|
|
{
|
|
name: "materialized field",
|
|
inputKey: telemetrytypes.TelemetryFieldKey{
|
|
Name: "mat.key",
|
|
FieldContext: telemetrytypes.FieldContextUnspecified,
|
|
FieldDataType: telemetrytypes.FieldDataTypeUnspecified,
|
|
},
|
|
keysMap: logstelemetryschema.BuildCompleteFieldKeyMap(releaseTime),
|
|
expectedKey: *logstelemetryschema.BuildCompleteFieldKeyMap(releaseTime)["mat.key"][0],
|
|
},
|
|
{
|
|
name: "non-materialized field",
|
|
inputKey: telemetrytypes.TelemetryFieldKey{
|
|
Name: "user.id",
|
|
FieldContext: telemetrytypes.FieldContextUnspecified,
|
|
FieldDataType: telemetrytypes.FieldDataTypeUnspecified,
|
|
},
|
|
keysMap: logstelemetryschema.BuildCompleteFieldKeyMap(releaseTime),
|
|
expectedKey: *logstelemetryschema.BuildCompleteFieldKeyMap(releaseTime)["user.id"][0],
|
|
},
|
|
}
|
|
|
|
fl := flaggertest.New(t)
|
|
fm := logstelemetryschema.NewFieldMapper(fl)
|
|
mockMetadataStore := telemetrytypestest.NewMockMetadataStore()
|
|
mockMetadataStore.KeysMap = logstelemetryschema.BuildCompleteFieldKeyMapCollision()
|
|
cb := logstelemetryschema.NewConditionBuilder(fm, fl)
|
|
|
|
aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, fm, cb, fl)
|
|
|
|
statementBuilder := NewLogQueryStatementBuilder(
|
|
instrumentationtest.New().ToProviderSettings(),
|
|
mockMetadataStore,
|
|
fm,
|
|
cb,
|
|
aggExprRewriter,
|
|
logstelemetryschema.DefaultFullTextColumn,
|
|
fl,
|
|
nil,
|
|
statementbuilder.Config{SkipResourceFingerprint: statementbuilder.SkipResourceFingerprint{Enabled: false, Threshold: 100000}},
|
|
)
|
|
|
|
for _, c := range cases {
|
|
t.Run(c.name, func(t *testing.T) {
|
|
// Create a copy of the input key to avoid modifying the original
|
|
key := c.inputKey
|
|
|
|
// Call adjustKey
|
|
statementBuilder.adjustKey(&key, c.keysMap)
|
|
|
|
// Verify the key was adjusted as expected
|
|
require.Equal(t, c.expectedKey.Name, key.Name, "key name should match")
|
|
require.Equal(t, c.expectedKey.FieldContext, key.FieldContext, "field context should match")
|
|
require.Equal(t, c.expectedKey.FieldDataType, key.FieldDataType, "field data type should match")
|
|
require.Equal(t, c.expectedKey.Materialized, key.Materialized, "materialized should match")
|
|
require.Equal(t, c.expectedKey.Indexes, key.Indexes, "json exists should match")
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestStmtBuilderBodyField(t *testing.T) {
|
|
cases := []struct {
|
|
name string
|
|
requestType qbtypes.RequestType
|
|
query qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]
|
|
enableUseJSONBody bool
|
|
expected qbtypes.Statement
|
|
expectedErr error
|
|
}{
|
|
{
|
|
name: "body_exists",
|
|
requestType: qbtypes.RequestTypeRaw,
|
|
query: qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]{
|
|
Signal: telemetrytypes.SignalLogs,
|
|
Filter: &qbtypes.Filter{Expression: "body Exists"},
|
|
Limit: 10,
|
|
},
|
|
enableUseJSONBody: true,
|
|
expected: qbtypes.Statement{
|
|
Query: "SELECT timestamp, id, trace_id, span_id, trace_flags, severity_text, severity_number, scope_name, scope_version, body_v2 as body, attributes_string, attributes_number, attributes_bool, resources_string, scope_string FROM signoz_logs.distributed_logs_v2 WHERE body_v2.message <> '' AND timestamp >= ? AND ts_bucket_start >= ? AND timestamp < ? AND ts_bucket_start <= ? LIMIT ?",
|
|
Args: []any{"1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 10},
|
|
Warnings: []string{bodySearchDefaultWarning},
|
|
},
|
|
expectedErr: nil,
|
|
},
|
|
{
|
|
name: "body_exists_disabled",
|
|
requestType: qbtypes.RequestTypeRaw,
|
|
query: qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]{
|
|
Signal: telemetrytypes.SignalLogs,
|
|
Filter: &qbtypes.Filter{Expression: "body Exists"},
|
|
Limit: 10,
|
|
},
|
|
enableUseJSONBody: false,
|
|
expected: qbtypes.Statement{
|
|
Query: "SELECT timestamp, id, trace_id, span_id, trace_flags, severity_text, severity_number, scope_name, scope_version, body, attributes_string, attributes_number, attributes_bool, resources_string, scope_string FROM signoz_logs.distributed_logs_v2 WHERE body <> '' AND timestamp >= ? AND ts_bucket_start >= ? AND timestamp < ? AND ts_bucket_start <= ? LIMIT ?",
|
|
Args: []any{"1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 10},
|
|
},
|
|
expectedErr: nil,
|
|
},
|
|
{
|
|
name: "body_empty",
|
|
requestType: qbtypes.RequestTypeRaw,
|
|
query: qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]{
|
|
Signal: telemetrytypes.SignalLogs,
|
|
Filter: &qbtypes.Filter{Expression: "body == ''"},
|
|
Limit: 10,
|
|
},
|
|
enableUseJSONBody: true,
|
|
expected: qbtypes.Statement{
|
|
Query: "SELECT timestamp, id, trace_id, span_id, trace_flags, severity_text, severity_number, scope_name, scope_version, body_v2 as body, attributes_string, attributes_number, attributes_bool, resources_string, scope_string FROM signoz_logs.distributed_logs_v2 WHERE body_v2.message = ? AND timestamp >= ? AND ts_bucket_start >= ? AND timestamp < ? AND ts_bucket_start <= ? LIMIT ?",
|
|
Args: []any{"", "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 10},
|
|
Warnings: []string{bodySearchDefaultWarning},
|
|
},
|
|
expectedErr: nil,
|
|
},
|
|
{
|
|
name: "body_empty_disabled",
|
|
requestType: qbtypes.RequestTypeRaw,
|
|
query: qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]{
|
|
Signal: telemetrytypes.SignalLogs,
|
|
Filter: &qbtypes.Filter{Expression: "body == ''"},
|
|
Limit: 10,
|
|
},
|
|
enableUseJSONBody: false,
|
|
expected: qbtypes.Statement{
|
|
Query: "SELECT timestamp, id, trace_id, span_id, trace_flags, severity_text, severity_number, scope_name, scope_version, body, attributes_string, attributes_number, attributes_bool, resources_string, scope_string FROM signoz_logs.distributed_logs_v2 WHERE body = ? AND timestamp >= ? AND ts_bucket_start >= ? AND timestamp < ? AND ts_bucket_start <= ? LIMIT ?",
|
|
Args: []any{"", "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 10},
|
|
},
|
|
expectedErr: nil,
|
|
},
|
|
{
|
|
name: "body_contains",
|
|
requestType: qbtypes.RequestTypeRaw,
|
|
query: qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]{
|
|
Signal: telemetrytypes.SignalLogs,
|
|
Filter: &qbtypes.Filter{Expression: "body CONTAINS 'error'"},
|
|
Limit: 10,
|
|
},
|
|
enableUseJSONBody: true,
|
|
expected: qbtypes.Statement{
|
|
Query: "SELECT timestamp, id, trace_id, span_id, trace_flags, severity_text, severity_number, scope_name, scope_version, body_v2 as body, attributes_string, attributes_number, attributes_bool, resources_string, scope_string FROM signoz_logs.distributed_logs_v2 WHERE LOWER(body_v2.message) LIKE LOWER(?) AND timestamp >= ? AND ts_bucket_start >= ? AND timestamp < ? AND ts_bucket_start <= ? LIMIT ?",
|
|
Args: []any{"%error%", "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 10},
|
|
Warnings: []string{bodySearchDefaultWarning},
|
|
},
|
|
expectedErr: nil,
|
|
},
|
|
{
|
|
name: "body_contains_disabled",
|
|
requestType: qbtypes.RequestTypeRaw,
|
|
query: qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]{
|
|
Signal: telemetrytypes.SignalLogs,
|
|
Filter: &qbtypes.Filter{Expression: "body CONTAINS 'error'"},
|
|
Limit: 10,
|
|
},
|
|
enableUseJSONBody: false,
|
|
expected: qbtypes.Statement{
|
|
Query: "SELECT timestamp, id, trace_id, span_id, trace_flags, severity_text, severity_number, scope_name, scope_version, body, attributes_string, attributes_number, attributes_bool, resources_string, scope_string FROM signoz_logs.distributed_logs_v2 WHERE LOWER(body) LIKE LOWER(?) AND timestamp >= ? AND ts_bucket_start >= ? AND timestamp < ? AND ts_bucket_start <= ? LIMIT ?",
|
|
Args: []any{"%error%", "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 10},
|
|
},
|
|
expectedErr: nil,
|
|
},
|
|
}
|
|
|
|
for _, c := range cases {
|
|
t.Run(c.name, func(t *testing.T) {
|
|
fl := flaggertest.WithUseJSONBody(t, c.enableUseJSONBody)
|
|
fm := logstelemetryschema.NewFieldMapper(fl)
|
|
cb := logstelemetryschema.NewConditionBuilder(fm, fl)
|
|
// build the key map
|
|
mockMetadataStore := telemetrytypestest.NewMockMetadataStore()
|
|
for _, field := range logstelemetryschema.IntrinsicFields {
|
|
f := field
|
|
mockMetadataStore.KeysMap[field.Name] = append(mockMetadataStore.KeysMap[field.Name], &f)
|
|
}
|
|
aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, fm, cb, fl)
|
|
statementBuilder := NewLogQueryStatementBuilder(
|
|
instrumentationtest.New().ToProviderSettings(),
|
|
mockMetadataStore,
|
|
fm,
|
|
cb,
|
|
aggExprRewriter,
|
|
logstelemetryschema.DefaultFullTextColumn,
|
|
fl,
|
|
nil,
|
|
statementbuilder.Config{SkipResourceFingerprint: statementbuilder.SkipResourceFingerprint{Enabled: false, Threshold: 100000}},
|
|
)
|
|
|
|
q, err := statementBuilder.Build(context.Background(), valuer.UUID{}, 1747947419000, 1747983448000, c.requestType, c.query, nil)
|
|
if c.expectedErr != nil {
|
|
require.Error(t, err)
|
|
require.Contains(t, err.Error(), c.expectedErr.Error())
|
|
} else {
|
|
if err != nil {
|
|
_, _, _, _, _, add := errors.Unwrapb(err)
|
|
t.Logf("error additionals: %v", add)
|
|
}
|
|
require.NoError(t, err)
|
|
require.Equal(t, c.expected.Query, q.Query)
|
|
require.Equal(t, c.expected.Args, q.Args)
|
|
require.Equal(t, c.expected.Warnings, q.Warnings)
|
|
}
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestStmtBuilderBodyFullTextSearch(t *testing.T) {
|
|
cases := []struct {
|
|
name string
|
|
requestType qbtypes.RequestType
|
|
query qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]
|
|
enableUseJSONBody bool
|
|
expected qbtypes.Statement
|
|
expectedErr error
|
|
}{
|
|
{
|
|
name: "fts",
|
|
requestType: qbtypes.RequestTypeRaw,
|
|
query: qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]{
|
|
Signal: telemetrytypes.SignalLogs,
|
|
Filter: &qbtypes.Filter{Expression: "'error'"},
|
|
Limit: 10,
|
|
},
|
|
enableUseJSONBody: true,
|
|
expected: qbtypes.Statement{
|
|
Query: "SELECT timestamp, id, trace_id, span_id, trace_flags, severity_text, severity_number, scope_name, scope_version, body_v2 as body, attributes_string, attributes_number, attributes_bool, resources_string, scope_string FROM signoz_logs.distributed_logs_v2 WHERE match(LOWER(body_v2.message), LOWER(?)) AND timestamp >= ? AND ts_bucket_start >= ? AND timestamp < ? AND ts_bucket_start <= ? LIMIT ?",
|
|
Args: []any{"error", "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 10},
|
|
Warnings: []string{querybuilder.BodyFullTextSearchDefaultWarning},
|
|
},
|
|
expectedErr: nil,
|
|
},
|
|
{
|
|
name: "fts_2",
|
|
requestType: qbtypes.RequestTypeRaw,
|
|
query: qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]{
|
|
Signal: telemetrytypes.SignalLogs,
|
|
Filter: &qbtypes.Filter{Expression: "error"},
|
|
Limit: 10,
|
|
},
|
|
enableUseJSONBody: true,
|
|
expected: qbtypes.Statement{
|
|
Query: "SELECT timestamp, id, trace_id, span_id, trace_flags, severity_text, severity_number, scope_name, scope_version, body_v2 as body, attributes_string, attributes_number, attributes_bool, resources_string, scope_string FROM signoz_logs.distributed_logs_v2 WHERE match(LOWER(body_v2.message), LOWER(?)) AND timestamp >= ? AND ts_bucket_start >= ? AND timestamp < ? AND ts_bucket_start <= ? LIMIT ?",
|
|
Args: []any{"error", "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 10},
|
|
Warnings: []string{querybuilder.BodyFullTextSearchDefaultWarning},
|
|
},
|
|
expectedErr: nil,
|
|
},
|
|
{
|
|
name: "fts_disabled",
|
|
requestType: qbtypes.RequestTypeRaw,
|
|
query: qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]{
|
|
Signal: telemetrytypes.SignalLogs,
|
|
Filter: &qbtypes.Filter{Expression: "'error'"},
|
|
Limit: 10,
|
|
},
|
|
enableUseJSONBody: false,
|
|
expected: qbtypes.Statement{
|
|
Query: "SELECT timestamp, id, trace_id, span_id, trace_flags, severity_text, severity_number, scope_name, scope_version, body, attributes_string, attributes_number, attributes_bool, resources_string, scope_string FROM signoz_logs.distributed_logs_v2 WHERE match(LOWER(body), LOWER(?)) AND timestamp >= ? AND ts_bucket_start >= ? AND timestamp < ? AND ts_bucket_start <= ? LIMIT ?",
|
|
Args: []any{"error", "1747947419000000000", uint64(1747945619), "1747983448000000000", uint64(1747983448), 10},
|
|
},
|
|
expectedErr: nil,
|
|
},
|
|
}
|
|
|
|
for _, c := range cases {
|
|
t.Run(c.name, func(t *testing.T) {
|
|
fl := flaggertest.WithUseJSONBody(t, c.enableUseJSONBody)
|
|
fm := logstelemetryschema.NewFieldMapper(fl)
|
|
cb := logstelemetryschema.NewConditionBuilder(fm, fl)
|
|
// build the key map
|
|
mockMetadataStore := telemetrytypestest.NewMockMetadataStore()
|
|
for _, field := range logstelemetryschema.IntrinsicFields {
|
|
f := field
|
|
mockMetadataStore.KeysMap[field.Name] = append(mockMetadataStore.KeysMap[field.Name], &f)
|
|
}
|
|
aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, fm, cb, fl)
|
|
statementBuilder := NewLogQueryStatementBuilder(
|
|
instrumentationtest.New().ToProviderSettings(),
|
|
mockMetadataStore,
|
|
fm,
|
|
cb,
|
|
aggExprRewriter,
|
|
logstelemetryschema.DefaultFullTextColumn,
|
|
fl,
|
|
nil,
|
|
statementbuilder.Config{SkipResourceFingerprint: statementbuilder.SkipResourceFingerprint{Enabled: false, Threshold: 100000}},
|
|
)
|
|
|
|
q, err := statementBuilder.Build(context.Background(), valuer.UUID{}, 1747947419000, 1747983448000, c.requestType, c.query, nil)
|
|
if c.expectedErr != nil {
|
|
require.Error(t, err)
|
|
require.Contains(t, err.Error(), c.expectedErr.Error())
|
|
} else {
|
|
if err != nil {
|
|
_, _, _, _, _, add := errors.Unwrapb(err)
|
|
t.Logf("error additionals: %v", add)
|
|
}
|
|
require.NoError(t, err)
|
|
require.Equal(t, c.expected.Query, q.Query)
|
|
require.Equal(t, c.expected.Args, q.Args)
|
|
require.Equal(t, c.expected.Warnings, q.Warnings)
|
|
}
|
|
})
|
|
}
|
|
}
|
|
|
|
// TestSkipResourceFingerprintLogs exercises the three resolver outcomes for
|
|
// logs: use-CTE (count < threshold), fallback (count >= threshold), and the
|
|
// legacy path (feature disabled).
|
|
func TestSkipResourceFingerprintLogs(t *testing.T) {
|
|
const (
|
|
startMs = uint64(1747947419000)
|
|
endMs = uint64(1747983448000)
|
|
threshold = uint64(10)
|
|
)
|
|
|
|
query := qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]{
|
|
Signal: telemetrytypes.SignalLogs,
|
|
Filter: &qbtypes.Filter{
|
|
Expression: "service.name = 'redis-manual'",
|
|
},
|
|
Limit: 5,
|
|
}
|
|
|
|
t.Run("disabled uses the legacy CTE", func(t *testing.T) {
|
|
sb := newSkipResourceFingerprintLogsBuilder(t, nil, false, threshold)
|
|
|
|
stmt, err := sb.Build(context.Background(), valuer.UUID{}, startMs, endMs, qbtypes.RequestTypeRaw, query, nil)
|
|
require.NoError(t, err)
|
|
require.Contains(t, stmt.Query, "__resource_filter AS (SELECT fingerprint")
|
|
require.Contains(t, stmt.Query, "resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter)")
|
|
})
|
|
|
|
t.Run("CTE attached when count below threshold", func(t *testing.T) {
|
|
mockStore := telemetrystoretest.New(telemetrystore.Config{}, ®exQueryMatcher{})
|
|
mock := mockStore.Mock()
|
|
|
|
mock.ExpectQueryRow(`SELECT count\(\) FROM \(SELECT fingerprint FROM signoz_logs\.distributed_logs_v2_resource`).
|
|
WillReturnRow(cmock.NewRow([]cmock.ColumnType{
|
|
{Name: "count", Type: "UInt64"},
|
|
}, []any{uint64(2)}))
|
|
|
|
sb := newSkipResourceFingerprintLogsBuilder(t, mockStore, true, threshold)
|
|
|
|
stmt, err := sb.Build(context.Background(), valuer.UUID{}, startMs, endMs, qbtypes.RequestTypeRaw, query, nil)
|
|
require.NoError(t, err)
|
|
|
|
require.Contains(t, stmt.Query, "__resource_filter AS (SELECT fingerprint")
|
|
require.Contains(t, stmt.Query, "resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter)")
|
|
|
|
require.NoError(t, mock.ExpectationsWereMet())
|
|
})
|
|
|
|
t.Run("fallback when count at or above threshold", func(t *testing.T) {
|
|
mockStore := telemetrystoretest.New(telemetrystore.Config{}, ®exQueryMatcher{})
|
|
mock := mockStore.Mock()
|
|
|
|
mock.ExpectQueryRow(`SELECT count\(\) FROM \(SELECT fingerprint FROM signoz_logs\.distributed_logs_v2_resource`).
|
|
WillReturnRow(cmock.NewRow([]cmock.ColumnType{
|
|
{Name: "count", Type: "UInt64"},
|
|
}, []any{threshold}))
|
|
|
|
sb := newSkipResourceFingerprintLogsBuilder(t, mockStore, true, threshold)
|
|
|
|
stmt, err := sb.Build(context.Background(), valuer.UUID{}, startMs, endMs, qbtypes.RequestTypeRaw, query, nil)
|
|
require.NoError(t, err)
|
|
|
|
require.NotContains(t, stmt.Query, "__resource_filter AS")
|
|
require.NotContains(t, stmt.Query, "resource_fingerprint")
|
|
require.Contains(t, stmt.Query, "service.name")
|
|
|
|
require.NoError(t, mock.ExpectationsWereMet())
|
|
})
|
|
}
|
|
|
|
func newSkipResourceFingerprintLogsBuilder(
|
|
t *testing.T,
|
|
telemetryStore telemetrystore.TelemetryStore,
|
|
skipEnable bool,
|
|
threshold uint64,
|
|
) *logQueryStatementBuilder {
|
|
t.Helper()
|
|
|
|
fl := flaggertest.New(t)
|
|
fm := logstelemetryschema.NewFieldMapper(fl)
|
|
cb := logstelemetryschema.NewConditionBuilder(fm, fl)
|
|
mockMetadataStore := telemetrytypestest.NewMockMetadataStore()
|
|
mockMetadataStore.KeysMap = logstelemetryschema.BuildCompleteFieldKeyMap(time.Date(2024, 1, 15, 10, 0, 0, 0, time.UTC))
|
|
|
|
aggExprRewriter := querybuilder.NewAggExprRewriter(
|
|
instrumentationtest.New().ToProviderSettings(),
|
|
logstelemetryschema.DefaultFullTextColumn,
|
|
fm,
|
|
cb,
|
|
fl,
|
|
)
|
|
|
|
return NewLogQueryStatementBuilder(
|
|
instrumentationtest.New().ToProviderSettings(),
|
|
mockMetadataStore,
|
|
fm,
|
|
cb,
|
|
aggExprRewriter,
|
|
logstelemetryschema.DefaultFullTextColumn,
|
|
fl,
|
|
telemetryStore,
|
|
statementbuilder.Config{SkipResourceFingerprint: statementbuilder.SkipResourceFingerprint{Enabled: skipEnable, Threshold: threshold}},
|
|
)
|
|
}
|