mirror of
https://github.com/SigNoz/signoz.git
synced 2026-09-21 02:40:40 +01:00
9c886be120
The semantic convention family as first call citizen revealed that the
current state of the query builder needs a bit refactoring for long term
maintenance.
The `FieldMapper` and `ConditionBuilder` are now one abstraction
`Storage`.
A storage now answers
- what the compiler cannot know i.e one read per field key (the bare
SQL, the membership present or absent, what an absent row reads, and
whether the read keeps its type or filters only).
- the fallback for a key metadata does not report
- its traits
- and one Condition compilation part.
And we introduce a new type to use in the system, `Resolved`
```
// Resolved is what resolution produces for one key: its meanings, and how
// they came to be. It is the only thing the compilers receive. Compile it
// with the operator and value it was resolved with.
type Resolved struct {
Key *telemetrytypes.TelemetryFieldKey
Fields []*telemetrytypes.LogicalField
// FromFallback: the fields came from the storage's fallback, not from
// metadata matches.
FromFallback bool
// Ambiguous: the matches held several interpretations.
Ambiguous bool
// Skipped: the storage contributes nothing for this key.
Skipped bool
Warnings []string
}
```
The prepared SQL has no changes, where it changed, it specifically made
the expression better by removing the redundant part.
- The prepared SQL remains identical with this refactoring
- No changes to integration tests
Assisted-by: Claude Fable 5.1
1274 lines
60 KiB
Go
1274 lines
60 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(resource.`service.name` IS NOT NULL, resource.`service.name`::String, mapContains(resources_string, 'service.name'), resources_string['service.name'], NULL)) AS `__GROUP_BY_KEY_0_service.name`, countDistinct(multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, mapContains(resources_string, 'service.name'), resources_string['service.name'], 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) = ? 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(resource.`service.name` IS NOT NULL, resource.`service.name`::String, mapContains(resources_string, 'service.name'), resources_string['service.name'], NULL)) AS `__GROUP_BY_KEY_0_service.name`, countDistinct(multiIf(resource.`service.name` IS NOT NULL, resource.`service.name`::String, mapContains(resources_string, 'service.name'), resources_string['service.name'], 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) = ? 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
|
|
|
|
storage := logstelemetryschema.NewStorage()
|
|
|
|
aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, storage, fl, telemetrytypes.SignalLogs)
|
|
|
|
statementBuilder := NewLogQueryStatementBuilder(
|
|
instrumentationtest.New().ToProviderSettings(),
|
|
mockMetadataStore,
|
|
storage,
|
|
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()
|
|
storage := logstelemetryschema.NewStorage()
|
|
|
|
// Create a test release time
|
|
releaseTime := time.Date(2024, 1, 15, 10, 0, 0, 0, time.UTC)
|
|
mockMetadataStore.KeysMap = logstelemetryschema.BuildCompleteFieldKeyMap(releaseTime)
|
|
|
|
aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, storage, fl, telemetrytypes.SignalLogs)
|
|
|
|
statementBuilder := NewLogQueryStatementBuilder(
|
|
instrumentationtest.New().ToProviderSettings(),
|
|
mockMetadataStore,
|
|
storage,
|
|
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,
|
|
},
|
|
{
|
|
// The `[*]` path is extracted per value, not as an Array(String) compared to a
|
|
// scalar — ClickHouse rejects that outright (code 130).
|
|
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 ((JSON_VALUE(body, '$.\"user_names\"[*]') = ?) 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()
|
|
storage := logstelemetryschema.NewStorage()
|
|
// Create a test release time
|
|
releaseTime := time.Date(2024, 1, 15, 10, 0, 0, 0, time.UTC)
|
|
mockMetadataStore.KeysMap = logstelemetryschema.BuildCompleteFieldKeyMap(releaseTime)
|
|
|
|
aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, storage, fl, telemetrytypes.SignalLogs)
|
|
|
|
statementBuilder := NewLogQueryStatementBuilder(
|
|
instrumentationtest.New().ToProviderSettings(),
|
|
mockMetadataStore,
|
|
storage,
|
|
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: "can be filtered but not selected or grouped",
|
|
},
|
|
}
|
|
|
|
ctx := context.Background()
|
|
fl := flaggertest.New(t)
|
|
mockMetadataStore := telemetrytypestest.NewMockMetadataStore()
|
|
storage := logstelemetryschema.NewStorage()
|
|
// Create a test release time
|
|
releaseTime := time.Date(2024, 1, 15, 10, 0, 0, 0, time.UTC)
|
|
mockMetadataStore.KeysMap = logstelemetryschema.BuildCompleteFieldKeyMap(releaseTime)
|
|
|
|
aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, storage, fl, telemetrytypes.SignalLogs)
|
|
|
|
statementBuilder := NewLogQueryStatementBuilder(
|
|
instrumentationtest.New().ToProviderSettings(),
|
|
mockMetadataStore,
|
|
storage,
|
|
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)
|
|
_, _, message, _, _, _ := errors.Unwrapb(err)
|
|
require.Contains(t, message, 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)
|
|
storage := logstelemetryschema.NewStorage()
|
|
mockMetadataStore.KeysMap = logstelemetryschema.BuildCompleteFieldKeyMapCollision()
|
|
|
|
aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, storage, fl, telemetrytypes.SignalLogs)
|
|
|
|
statementBuilder := NewLogQueryStatementBuilder(
|
|
instrumentationtest.New().ToProviderSettings(),
|
|
mockMetadataStore,
|
|
storage,
|
|
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)
|
|
storage := logstelemetryschema.NewStorage()
|
|
mockMetadataStore := telemetrytypestest.NewMockMetadataStore()
|
|
mockMetadataStore.KeysMap = logstelemetryschema.BuildCompleteFieldKeyMapCollision()
|
|
|
|
aggExprRewriter := querybuilder.NewAggExprRewriter(instrumentationtest.New().ToProviderSettings(), nil, storage, fl, telemetrytypes.SignalLogs)
|
|
|
|
statementBuilder := NewLogQueryStatementBuilder(
|
|
instrumentationtest.New().ToProviderSettings(),
|
|
mockMetadataStore,
|
|
storage,
|
|
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)
|
|
storage := logstelemetryschema.NewStorage()
|
|
// 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, storage, fl, telemetrytypes.SignalLogs)
|
|
statementBuilder := NewLogQueryStatementBuilder(
|
|
instrumentationtest.New().ToProviderSettings(),
|
|
mockMetadataStore,
|
|
storage,
|
|
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)
|
|
storage := logstelemetryschema.NewStorage()
|
|
// 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, storage, fl, telemetrytypes.SignalLogs)
|
|
statementBuilder := NewLogQueryStatementBuilder(
|
|
instrumentationtest.New().ToProviderSettings(),
|
|
mockMetadataStore,
|
|
storage,
|
|
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)
|
|
storage := logstelemetryschema.NewStorage()
|
|
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, storage, fl, telemetrytypes.SignalLogs)
|
|
|
|
return NewLogQueryStatementBuilder(
|
|
instrumentationtest.New().ToProviderSettings(),
|
|
mockMetadataStore,
|
|
storage,
|
|
aggExprRewriter,
|
|
logstelemetryschema.DefaultFullTextColumn,
|
|
fl,
|
|
telemetryStore,
|
|
statementbuilder.Config{SkipResourceFingerprint: statementbuilder.SkipResourceFingerprint{Enabled: skipEnable, Threshold: threshold}},
|
|
)
|
|
}
|