GitHub-Proxy/signoz
1
0
Fork 0
mirror of https://git.mirrors.martin98.com/https://github.com/SigNoz/signoz synced 2025-06-21 15:36:34 +08:00
Code Issues Packages Projects Releases Wiki Activity

chore: move traces builder query attributes enrichment before query prep (#5917)

Browse Source
This commit is contained in:
Srikanth Chekuri 2024-09-13 16:43:56 +05:30 committed by GitHub
parent 3fdfb51e02
commit a5f3a189f8
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
13 changed files with 160 additions and 119 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

20
pkg/query-service/app/http_handler.go
View File

@ -2521,7 +2521,7 @@ func (aH *APIHandler) getNetworkData(
var result []*v3.Result var result []*v3.Result
var errQueriesByName map[string]error var errQueriesByName map[string]error
result, errQueriesByName, err = aH.querierV2.QueryRange(r.Context(), queryRangeParams, nil) result, errQueriesByName, err = aH.querierV2.QueryRange(r.Context(), queryRangeParams)
if err != nil { if err != nil {
apiErrObj := &model.ApiError{Typ: model.ErrorBadData, Err: err} apiErrObj := &model.ApiError{Typ: model.ErrorBadData, Err: err}
RespondError(w, apiErrObj, errQueriesByName) RespondError(w, apiErrObj, errQueriesByName)
@ -2556,7 +2556,7 @@ func (aH *APIHandler) getNetworkData(
return return
} }
resultFetchLatency, errQueriesByNameFetchLatency, err := aH.querierV2.QueryRange(r.Context(), queryRangeParams, nil) resultFetchLatency, errQueriesByNameFetchLatency, err := aH.querierV2.QueryRange(r.Context(), queryRangeParams)
if err != nil { if err != nil {
apiErrObj := &model.ApiError{Typ: model.ErrorBadData, Err: err} apiErrObj := &model.ApiError{Typ: model.ErrorBadData, Err: err}
RespondError(w, apiErrObj, errQueriesByNameFetchLatency) RespondError(w, apiErrObj, errQueriesByNameFetchLatency)
@ -2617,7 +2617,7 @@ func (aH *APIHandler) getProducerData(
var result []*v3.Result var result []*v3.Result
var errQuriesByName map[string]error var errQuriesByName map[string]error
result, errQuriesByName, err = aH.querierV2.QueryRange(r.Context(), queryRangeParams, nil) result, errQuriesByName, err = aH.querierV2.QueryRange(r.Context(), queryRangeParams)
if err != nil { if err != nil {
apiErrObj := &model.ApiError{Typ: model.ErrorBadData, Err: err} apiErrObj := &model.ApiError{Typ: model.ErrorBadData, Err: err}
RespondError(w, apiErrObj, errQuriesByName) RespondError(w, apiErrObj, errQuriesByName)
@ -2658,7 +2658,7 @@ func (aH *APIHandler) getConsumerData(
var result []*v3.Result var result []*v3.Result
var errQuriesByName map[string]error var errQuriesByName map[string]error
result, errQuriesByName, err = aH.querierV2.QueryRange(r.Context(), queryRangeParams, nil) result, errQuriesByName, err = aH.querierV2.QueryRange(r.Context(), queryRangeParams)
if err != nil { if err != nil {
apiErrObj := &model.ApiError{Typ: model.ErrorBadData, Err: err} apiErrObj := &model.ApiError{Typ: model.ErrorBadData, Err: err}
RespondError(w, apiErrObj, errQuriesByName) RespondError(w, apiErrObj, errQuriesByName)
@ -3020,7 +3020,7 @@ func (aH *APIHandler) calculateLogsConnectionStatus(
}, },
} }
queryRes, _, err := aH.querier.QueryRange( queryRes, _, err := aH.querier.QueryRange(
ctx, qrParams, map[string]v3.AttributeKey{}, ctx, qrParams,
) )
if err != nil { if err != nil {
return nil, model.InternalError(fmt.Errorf( return nil, model.InternalError(fmt.Errorf(
@ -3670,13 +3670,14 @@ func (aH *APIHandler) queryRangeV3(ctx context.Context, queryRangeParams *v3.Que
RespondError(w, apiErrObj, errQuriesByName) RespondError(w, apiErrObj, errQuriesByName)
return return
} }
tracesV3.Enrich(queryRangeParams, spanKeys)
} }
// WARN: Only works for AND operator in traces query // WARN: Only works for AND operator in traces query
if queryRangeParams.CompositeQuery.QueryType == v3.QueryTypeBuilder { if queryRangeParams.CompositeQuery.QueryType == v3.QueryTypeBuilder {
// check if traceID is used as filter (with equal/similar operator) in traces query if yes add timestamp filter to queryRange params // check if traceID is used as filter (with equal/similar operator) in traces query if yes add timestamp filter to queryRange params
isUsed, traceIDs := tracesV3.TraceIdFilterUsedWithEqual(queryRangeParams) isUsed, traceIDs := tracesV3.TraceIdFilterUsedWithEqual(queryRangeParams)
if isUsed == true && len(traceIDs) > 0 { if isUsed && len(traceIDs) > 0 {
zap.L().Debug("traceID used as filter in traces query") zap.L().Debug("traceID used as filter in traces query")
// query signoz_spans table with traceID to get min and max timestamp // query signoz_spans table with traceID to get min and max timestamp
min, max, err := aH.reader.GetMinAndMaxTimestampForTraceID(ctx, traceIDs) min, max, err := aH.reader.GetMinAndMaxTimestampForTraceID(ctx, traceIDs)
@ -3706,7 +3707,7 @@ func (aH *APIHandler) queryRangeV3(ctx context.Context, queryRangeParams *v3.Que
} }
} }
result, errQuriesByName, err = aH.querier.QueryRange(ctx, queryRangeParams, spanKeys) result, errQuriesByName, err = aH.querier.QueryRange(ctx, queryRangeParams)
if err != nil { if err != nil {
apiErrObj := &model.ApiError{Typ: model.ErrorBadData, Err: err} apiErrObj := &model.ApiError{Typ: model.ErrorBadData, Err: err}
@ -4044,13 +4045,14 @@ func (aH *APIHandler) queryRangeV4(ctx context.Context, queryRangeParams *v3.Que
RespondError(w, apiErrObj, errQuriesByName) RespondError(w, apiErrObj, errQuriesByName)
return return
} }
tracesV3.Enrich(queryRangeParams, spanKeys)
} }
// WARN: Only works for AND operator in traces query // WARN: Only works for AND operator in traces query
if queryRangeParams.CompositeQuery.QueryType == v3.QueryTypeBuilder { if queryRangeParams.CompositeQuery.QueryType == v3.QueryTypeBuilder {
// check if traceID is used as filter (with equal/similar operator) in traces query if yes add timestamp filter to queryRange params // check if traceID is used as filter (with equal/similar operator) in traces query if yes add timestamp filter to queryRange params
isUsed, traceIDs := tracesV3.TraceIdFilterUsedWithEqual(queryRangeParams) isUsed, traceIDs := tracesV3.TraceIdFilterUsedWithEqual(queryRangeParams)
if isUsed == true && len(traceIDs) > 0 { if isUsed && len(traceIDs) > 0 {
zap.L().Debug("traceID used as filter in traces query") zap.L().Debug("traceID used as filter in traces query")
// query signoz_spans table with traceID to get min and max timestamp // query signoz_spans table with traceID to get min and max timestamp
min, max, err := aH.reader.GetMinAndMaxTimestampForTraceID(ctx, traceIDs) min, max, err := aH.reader.GetMinAndMaxTimestampForTraceID(ctx, traceIDs)
@ -4062,7 +4064,7 @@ func (aH *APIHandler) queryRangeV4(ctx context.Context, queryRangeParams *v3.Que
} }
} }
result, errQuriesByName, err = aH.querierV2.QueryRange(ctx, queryRangeParams, spanKeys) result, errQuriesByName, err = aH.querierV2.QueryRange(ctx, queryRangeParams)
if err != nil { if err != nil {
apiErrObj := &model.ApiError{Typ: model.ErrorBadData, Err: err} apiErrObj := &model.ApiError{Typ: model.ErrorBadData, Err: err}

9
pkg/query-service/app/querier/helper.go
View File

@ -77,7 +77,6 @@ func (q *querier) runBuilderQuery(
ctx context.Context, ctx context.Context,
builderQuery *v3.BuilderQuery, builderQuery *v3.BuilderQuery,
params *v3.QueryRangeParamsV3, params *v3.QueryRangeParamsV3,
keys map[string]v3.AttributeKey,
cacheKeys map[string]string, cacheKeys map[string]string,
ch chan channelResult, ch chan channelResult,
wg *sync.WaitGroup, wg *sync.WaitGroup,
@ -196,7 +195,6 @@ func (q *querier) runBuilderQuery(
end, end,
params.CompositeQuery.PanelType, params.CompositeQuery.PanelType,
builderQuery, builderQuery,
keys,
tracesV3.Options{GraphLimitQtype: constants.FirstQueryGraphLimit, PreferRPM: preferRPM}, tracesV3.Options{GraphLimitQtype: constants.FirstQueryGraphLimit, PreferRPM: preferRPM},
) )
if err != nil { if err != nil {
@ -208,7 +206,6 @@ func (q *querier) runBuilderQuery(
end, end,
params.CompositeQuery.PanelType, params.CompositeQuery.PanelType,
builderQuery, builderQuery,
keys,
tracesV3.Options{GraphLimitQtype: constants.SecondQueryGraphLimit, PreferRPM: preferRPM}, tracesV3.Options{GraphLimitQtype: constants.SecondQueryGraphLimit, PreferRPM: preferRPM},
) )
if err != nil { if err != nil {
@ -222,7 +219,6 @@ func (q *querier) runBuilderQuery(
end, end,
params.CompositeQuery.PanelType, params.CompositeQuery.PanelType,
builderQuery, builderQuery,
keys,
tracesV3.Options{PreferRPM: preferRPM}, tracesV3.Options{PreferRPM: preferRPM},
) )
if err != nil { if err != nil {
@ -333,7 +329,6 @@ func (q *querier) runBuilderExpression(
ctx context.Context, ctx context.Context,
builderQuery *v3.BuilderQuery, builderQuery *v3.BuilderQuery,
params *v3.QueryRangeParamsV3, params *v3.QueryRangeParamsV3,
keys map[string]v3.AttributeKey,
cacheKeys map[string]string, cacheKeys map[string]string,
ch chan channelResult, ch chan channelResult,
wg *sync.WaitGroup, wg *sync.WaitGroup,
@ -342,7 +337,7 @@ func (q *querier) runBuilderExpression(
queryName := builderQuery.QueryName queryName := builderQuery.QueryName
queries, err := q.builder.PrepareQueries(params, keys) queries, err := q.builder.PrepareQueries(params)
if err != nil { if err != nil {
ch <- channelResult{Err: err, Name: queryName, Query: "", Series: nil} ch <- channelResult{Err: err, Name: queryName, Query: "", Series: nil}
return return
@ -377,7 +372,7 @@ func (q *querier) runBuilderExpression(
NoCache: params.NoCache, NoCache: params.NoCache,
CompositeQuery: params.CompositeQuery, CompositeQuery: params.CompositeQuery,
Variables: params.Variables, Variables: params.Variables,
}, keys) })
query := missQueries[queryName] query := missQueries[queryName]
series, err := q.execClickHouseQuery(ctx, query) series, err := q.execClickHouseQuery(ctx, query)
if err != nil { if err != nil {

22
pkg/query-service/app/querier/querier.go
View File

@ -297,7 +297,7 @@ func mergeSerieses(cachedSeries, missedSeries []*v3.Series) []*v3.Series {
return mergedSeries return mergedSeries
} }
func (q *querier) runBuilderQueries(ctx context.Context, params *v3.QueryRangeParamsV3, keys map[string]v3.AttributeKey) ([]*v3.Result, map[string]error, error) { func (q *querier) runBuilderQueries(ctx context.Context, params *v3.QueryRangeParamsV3) ([]*v3.Result, map[string]error, error) {
cacheKeys := q.keyGenerator.GenerateKeys(params) cacheKeys := q.keyGenerator.GenerateKeys(params)
@ -310,9 +310,9 @@ func (q *querier) runBuilderQueries(ctx context.Context, params *v3.QueryRangePa
} }
wg.Add(1) wg.Add(1)
if queryName == builderQuery.Expression { if queryName == builderQuery.Expression {
go q.runBuilderQuery(ctx, builderQuery, params, keys, cacheKeys, ch, &wg) go q.runBuilderQuery(ctx, builderQuery, params, cacheKeys, ch, &wg)
} else { } else {
go q.runBuilderExpression(ctx, builderQuery, params, keys, cacheKeys, ch, &wg) go q.runBuilderExpression(ctx, builderQuery, params, cacheKeys, ch, &wg)
} }
} }
@ -470,7 +470,7 @@ func (q *querier) runClickHouseQueries(ctx context.Context, params *v3.QueryRang
return results, errQueriesByName, err return results, errQueriesByName, err
} }
func (q *querier) runLogsListQuery(ctx context.Context, params *v3.QueryRangeParamsV3, keys map[string]v3.AttributeKey, tsRanges []utils.LogsListTsRange) ([]*v3.Result, map[string]error, error) { func (q *querier) runLogsListQuery(ctx context.Context, params *v3.QueryRangeParamsV3, tsRanges []utils.LogsListTsRange) ([]*v3.Result, map[string]error, error) {
res := make([]*v3.Result, 0) res := make([]*v3.Result, 0)
qName := "" qName := ""
pageSize := uint64(0) pageSize := uint64(0)
@ -487,7 +487,7 @@ func (q *querier) runLogsListQuery(ctx context.Context, params *v3.QueryRangePar
params.End = v.End params.End = v.End
params.CompositeQuery.BuilderQueries[qName].PageSize = pageSize - uint64(len(data)) params.CompositeQuery.BuilderQueries[qName].PageSize = pageSize - uint64(len(data))
queries, err := q.builder.PrepareQueries(params, keys) queries, err := q.builder.PrepareQueries(params)
if err != nil { if err != nil {
return nil, nil, err return nil, nil, err
} }
@ -529,7 +529,7 @@ func (q *querier) runLogsListQuery(ctx context.Context, params *v3.QueryRangePar
return res, nil, nil return res, nil, nil
} }
func (q *querier) runBuilderListQueries(ctx context.Context, params *v3.QueryRangeParamsV3, keys map[string]v3.AttributeKey) ([]*v3.Result, map[string]error, error) { func (q *querier) runBuilderListQueries(ctx context.Context, params *v3.QueryRangeParamsV3) ([]*v3.Result, map[string]error, error) {
// List query has support for only one query. // List query has support for only one query.
if q.UseLogsNewSchema && params.CompositeQuery != nil && len(params.CompositeQuery.BuilderQueries) == 1 { if q.UseLogsNewSchema && params.CompositeQuery != nil && len(params.CompositeQuery.BuilderQueries) == 1 {
for _, v := range params.CompositeQuery.BuilderQueries { for _, v := range params.CompositeQuery.BuilderQueries {
@ -537,13 +537,13 @@ func (q *querier) runBuilderListQueries(ctx context.Context, params *v3.QueryRan
if v.DataSource == v3.DataSourceLogs && len(v.OrderBy) == 1 && v.OrderBy[0].ColumnName == "timestamp" && v.OrderBy[0].Order == "desc" { if v.DataSource == v3.DataSourceLogs && len(v.OrderBy) == 1 && v.OrderBy[0].ColumnName == "timestamp" && v.OrderBy[0].Order == "desc" {
startEndArr := utils.GetLogsListTsRanges(params.Start, params.End) startEndArr := utils.GetLogsListTsRanges(params.Start, params.End)
if len(startEndArr) > 0 { if len(startEndArr) > 0 {
return q.runLogsListQuery(ctx, params, keys, startEndArr) return q.runLogsListQuery(ctx, params, startEndArr)
} }
} }
} }
} }
queries, err := q.builder.PrepareQueries(params, keys) queries, err := q.builder.PrepareQueries(params)
if err != nil { if err != nil {
return nil, nil, err return nil, nil, err
@ -590,7 +590,7 @@ func (q *querier) runBuilderListQueries(ctx context.Context, params *v3.QueryRan
return res, nil, nil return res, nil, nil
} }
func (q *querier) QueryRange(ctx context.Context, params *v3.QueryRangeParamsV3, keys map[string]v3.AttributeKey) ([]*v3.Result, map[string]error, error) { func (q *querier) QueryRange(ctx context.Context, params *v3.QueryRangeParamsV3) ([]*v3.Result, map[string]error, error) {
var results []*v3.Result var results []*v3.Result
var err error var err error
var errQueriesByName map[string]error var errQueriesByName map[string]error
@ -598,9 +598,9 @@ func (q *querier) QueryRange(ctx context.Context, params *v3.QueryRangeParamsV3,
switch params.CompositeQuery.QueryType { switch params.CompositeQuery.QueryType {
case v3.QueryTypeBuilder: case v3.QueryTypeBuilder:
if params.CompositeQuery.PanelType == v3.PanelTypeList || params.CompositeQuery.PanelType == v3.PanelTypeTrace { if params.CompositeQuery.PanelType == v3.PanelTypeList || params.CompositeQuery.PanelType == v3.PanelTypeTrace {
results, errQueriesByName, err = q.runBuilderListQueries(ctx, params, keys) results, errQueriesByName, err = q.runBuilderListQueries(ctx, params)
} else { } else {
results, errQueriesByName, err = q.runBuilderQueries(ctx, params, keys) results, errQueriesByName, err = q.runBuilderQueries(ctx, params)
} }
// in builder query, the only errors we expose are the ones that exceed the resource limits // in builder query, the only errors we expose are the ones that exceed the resource limits
// everything else is internal error as they are not actionable by the user // everything else is internal error as they are not actionable by the user

19
pkg/query-service/app/querier/querier_test.go
View File

@ -8,6 +8,7 @@ import (
"time" "time"
"go.signoz.io/signoz/pkg/query-service/app/queryBuilder" "go.signoz.io/signoz/pkg/query-service/app/queryBuilder"
tracesV3 "go.signoz.io/signoz/pkg/query-service/app/traces/v3"
"go.signoz.io/signoz/pkg/query-service/cache/inmemory" "go.signoz.io/signoz/pkg/query-service/cache/inmemory"
v3 "go.signoz.io/signoz/pkg/query-service/model/v3" v3 "go.signoz.io/signoz/pkg/query-service/model/v3"
) )
@ -584,7 +585,8 @@ func TestQueryRange(t *testing.T) {
} }
for i, param := range params { for i, param := range params {
_, errByName, err := q.QueryRange(context.Background(), param, nil) tracesV3.Enrich(param, map[string]v3.AttributeKey{})
_, errByName, err := q.QueryRange(context.Background(), param)
if err != nil { if err != nil {
t.Errorf("expected no error, got %s", err) t.Errorf("expected no error, got %s", err)
} }
@ -693,7 +695,8 @@ func TestQueryRangeValueType(t *testing.T) {
} }
for i, param := range params { for i, param := range params {
_, errByName, err := q.QueryRange(context.Background(), param, nil) tracesV3.Enrich(param, map[string]v3.AttributeKey{})
_, errByName, err := q.QueryRange(context.Background(), param)
if err != nil { if err != nil {
t.Errorf("expected no error, got %s", err) t.Errorf("expected no error, got %s", err)
} }
@ -746,7 +749,8 @@ func TestQueryRangeTimeShift(t *testing.T) {
expectedTimeRangeInQueryString := fmt.Sprintf("timestamp >= %d AND timestamp <= %d", (1675115596722-86400*1000)*1000000, ((1675115596722+120*60*1000)-86400*1000)*1000000) expectedTimeRangeInQueryString := fmt.Sprintf("timestamp >= %d AND timestamp <= %d", (1675115596722-86400*1000)*1000000, ((1675115596722+120*60*1000)-86400*1000)*1000000)
for i, param := range params { for i, param := range params {
_, errByName, err := q.QueryRange(context.Background(), param, nil) tracesV3.Enrich(param, map[string]v3.AttributeKey{})
_, errByName, err := q.QueryRange(context.Background(), param)
if err != nil { if err != nil {
t.Errorf("expected no error, got %s", err) t.Errorf("expected no error, got %s", err)
} }
@ -844,7 +848,8 @@ func TestQueryRangeTimeShiftWithCache(t *testing.T) {
} }
for i, param := range params { for i, param := range params {
_, errByName, err := q.QueryRange(context.Background(), param, nil) tracesV3.Enrich(param, map[string]v3.AttributeKey{})
_, errByName, err := q.QueryRange(context.Background(), param)
if err != nil { if err != nil {
t.Errorf("expected no error, got %s", err) t.Errorf("expected no error, got %s", err)
} }
@ -944,7 +949,8 @@ func TestQueryRangeTimeShiftWithLimitAndCache(t *testing.T) {
} }
for i, param := range params { for i, param := range params {
_, errByName, err := q.QueryRange(context.Background(), param, nil) tracesV3.Enrich(param, map[string]v3.AttributeKey{})
_, errByName, err := q.QueryRange(context.Background(), param)
if err != nil { if err != nil {
t.Errorf("expected no error, got %s", err) t.Errorf("expected no error, got %s", err)
} }
@ -1033,7 +1039,8 @@ func TestQueryRangeValueTypePromQL(t *testing.T) {
} }
for i, param := range params { for i, param := range params {
_, errByName, err := q.QueryRange(context.Background(), param, nil) tracesV3.Enrich(param, map[string]v3.AttributeKey{})
_, errByName, err := q.QueryRange(context.Background(), param)
if err != nil { if err != nil {
t.Errorf("expected no error, got %s", err) t.Errorf("expected no error, got %s", err)
} }

4
pkg/query-service/app/querier/v2/helper.go
View File

@ -77,7 +77,6 @@ func (q *querier) runBuilderQuery(
ctx context.Context, ctx context.Context,
builderQuery *v3.BuilderQuery, builderQuery *v3.BuilderQuery,
params *v3.QueryRangeParamsV3, params *v3.QueryRangeParamsV3,
keys map[string]v3.AttributeKey,
cacheKeys map[string]string, cacheKeys map[string]string,
ch chan channelResult, ch chan channelResult,
wg *sync.WaitGroup, wg *sync.WaitGroup,
@ -195,7 +194,6 @@ func (q *querier) runBuilderQuery(
end, end,
params.CompositeQuery.PanelType, params.CompositeQuery.PanelType,
builderQuery, builderQuery,
keys,
tracesV3.Options{GraphLimitQtype: constants.FirstQueryGraphLimit, PreferRPM: preferRPM}, tracesV3.Options{GraphLimitQtype: constants.FirstQueryGraphLimit, PreferRPM: preferRPM},
) )
if err != nil { if err != nil {
@ -207,7 +205,6 @@ func (q *querier) runBuilderQuery(
end, end,
params.CompositeQuery.PanelType, params.CompositeQuery.PanelType,
builderQuery, builderQuery,
keys,
tracesV3.Options{GraphLimitQtype: constants.SecondQueryGraphLimit, PreferRPM: preferRPM}, tracesV3.Options{GraphLimitQtype: constants.SecondQueryGraphLimit, PreferRPM: preferRPM},
) )
if err != nil { if err != nil {
@ -221,7 +218,6 @@ func (q *querier) runBuilderQuery(
end, end,
params.CompositeQuery.PanelType, params.CompositeQuery.PanelType,
builderQuery, builderQuery,
keys,
tracesV3.Options{PreferRPM: preferRPM}, tracesV3.Options{PreferRPM: preferRPM},
) )
if err != nil { if err != nil {

20
pkg/query-service/app/querier/v2/querier.go
View File

@ -312,7 +312,7 @@ func mergeSerieses(cachedSeries, missedSeries []*v3.Series) []*v3.Series {
return mergedSeries return mergedSeries
} }
func (q *querier) runBuilderQueries(ctx context.Context, params *v3.QueryRangeParamsV3, keys map[string]v3.AttributeKey) ([]*v3.Result, map[string]error, error) { func (q *querier) runBuilderQueries(ctx context.Context, params *v3.QueryRangeParamsV3) ([]*v3.Result, map[string]error, error) {
cacheKeys := q.keyGenerator.GenerateKeys(params) cacheKeys := q.keyGenerator.GenerateKeys(params)
@ -322,7 +322,7 @@ func (q *querier) runBuilderQueries(ctx context.Context, params *v3.QueryRangePa
for queryName, builderQuery := range params.CompositeQuery.BuilderQueries { for queryName, builderQuery := range params.CompositeQuery.BuilderQueries {
if queryName == builderQuery.Expression { if queryName == builderQuery.Expression {
wg.Add(1) wg.Add(1)
go q.runBuilderQuery(ctx, builderQuery, params, keys, cacheKeys, ch, &wg) go q.runBuilderQuery(ctx, builderQuery, params, cacheKeys, ch, &wg)
} }
} }
@ -479,7 +479,7 @@ func (q *querier) runClickHouseQueries(ctx context.Context, params *v3.QueryRang
return results, errQueriesByName, err return results, errQueriesByName, err
} }
func (q *querier) runLogsListQuery(ctx context.Context, params *v3.QueryRangeParamsV3, keys map[string]v3.AttributeKey, tsRanges []utils.LogsListTsRange) ([]*v3.Result, map[string]error, error) { func (q *querier) runLogsListQuery(ctx context.Context, params *v3.QueryRangeParamsV3, tsRanges []utils.LogsListTsRange) ([]*v3.Result, map[string]error, error) {
res := make([]*v3.Result, 0) res := make([]*v3.Result, 0)
qName := "" qName := ""
pageSize := uint64(0) pageSize := uint64(0)
@ -496,7 +496,7 @@ func (q *querier) runLogsListQuery(ctx context.Context, params *v3.QueryRangePar
params.End = v.End params.End = v.End
params.CompositeQuery.BuilderQueries[qName].PageSize = pageSize - uint64(len(data)) params.CompositeQuery.BuilderQueries[qName].PageSize = pageSize - uint64(len(data))
queries, err := q.builder.PrepareQueries(params, keys) queries, err := q.builder.PrepareQueries(params)
if err != nil { if err != nil {
return nil, nil, err return nil, nil, err
} }
@ -538,7 +538,7 @@ func (q *querier) runLogsListQuery(ctx context.Context, params *v3.QueryRangePar
return res, nil, nil return res, nil, nil
} }
func (q *querier) runBuilderListQueries(ctx context.Context, params *v3.QueryRangeParamsV3, keys map[string]v3.AttributeKey) ([]*v3.Result, map[string]error, error) { func (q *querier) runBuilderListQueries(ctx context.Context, params *v3.QueryRangeParamsV3) ([]*v3.Result, map[string]error, error) {
// List query has support for only one query. // List query has support for only one query.
if q.UseLogsNewSchema && params.CompositeQuery != nil && len(params.CompositeQuery.BuilderQueries) == 1 { if q.UseLogsNewSchema && params.CompositeQuery != nil && len(params.CompositeQuery.BuilderQueries) == 1 {
for _, v := range params.CompositeQuery.BuilderQueries { for _, v := range params.CompositeQuery.BuilderQueries {
@ -546,13 +546,13 @@ func (q *querier) runBuilderListQueries(ctx context.Context, params *v3.QueryRan
if v.DataSource == v3.DataSourceLogs && len(v.OrderBy) == 1 && v.OrderBy[0].ColumnName == "timestamp" && v.OrderBy[0].Order == "desc" { if v.DataSource == v3.DataSourceLogs && len(v.OrderBy) == 1 && v.OrderBy[0].ColumnName == "timestamp" && v.OrderBy[0].Order == "desc" {
startEndArr := utils.GetLogsListTsRanges(params.Start, params.End) startEndArr := utils.GetLogsListTsRanges(params.Start, params.End)
if len(startEndArr) > 0 { if len(startEndArr) > 0 {
return q.runLogsListQuery(ctx, params, keys, startEndArr) return q.runLogsListQuery(ctx, params, startEndArr)
} }
} }
} }
} }
queries, err := q.builder.PrepareQueries(params, keys) queries, err := q.builder.PrepareQueries(params)
if err != nil { if err != nil {
return nil, nil, err return nil, nil, err
@ -601,7 +601,7 @@ func (q *querier) runBuilderListQueries(ctx context.Context, params *v3.QueryRan
// QueryRange is the main function that runs the queries // QueryRange is the main function that runs the queries
// and returns the results // and returns the results
func (q *querier) QueryRange(ctx context.Context, params *v3.QueryRangeParamsV3, keys map[string]v3.AttributeKey) ([]*v3.Result, map[string]error, error) { func (q *querier) QueryRange(ctx context.Context, params *v3.QueryRangeParamsV3) ([]*v3.Result, map[string]error, error) {
var results []*v3.Result var results []*v3.Result
var err error var err error
var errQueriesByName map[string]error var errQueriesByName map[string]error
@ -609,9 +609,9 @@ func (q *querier) QueryRange(ctx context.Context, params *v3.QueryRangeParamsV3,
switch params.CompositeQuery.QueryType { switch params.CompositeQuery.QueryType {
case v3.QueryTypeBuilder: case v3.QueryTypeBuilder:
if params.CompositeQuery.PanelType == v3.PanelTypeList || params.CompositeQuery.PanelType == v3.PanelTypeTrace { if params.CompositeQuery.PanelType == v3.PanelTypeList || params.CompositeQuery.PanelType == v3.PanelTypeTrace {
results, errQueriesByName, err = q.runBuilderListQueries(ctx, params, keys) results, errQueriesByName, err = q.runBuilderListQueries(ctx, params)
} else { } else {
results, errQueriesByName, err = q.runBuilderQueries(ctx, params, keys) results, errQueriesByName, err = q.runBuilderQueries(ctx, params)
} }
// in builder query, the only errors we expose are the ones that exceed the resource limits // in builder query, the only errors we expose are the ones that exceed the resource limits
// everything else is internal error as they are not actionable by the user // everything else is internal error as they are not actionable by the user

19
pkg/query-service/app/querier/v2/querier_test.go
View File

@ -8,6 +8,7 @@ import (
"time" "time"
"go.signoz.io/signoz/pkg/query-service/app/queryBuilder" "go.signoz.io/signoz/pkg/query-service/app/queryBuilder"
tracesV3 "go.signoz.io/signoz/pkg/query-service/app/traces/v3"
"go.signoz.io/signoz/pkg/query-service/cache/inmemory" "go.signoz.io/signoz/pkg/query-service/cache/inmemory"
v3 "go.signoz.io/signoz/pkg/query-service/model/v3" v3 "go.signoz.io/signoz/pkg/query-service/model/v3"
) )
@ -593,7 +594,8 @@ func TestV2QueryRangePanelGraph(t *testing.T) {
} }
for i, param := range params { for i, param := range params {
_, errByName, err := q.QueryRange(context.Background(), param, nil) tracesV3.Enrich(param, map[string]v3.AttributeKey{})
_, errByName, err := q.QueryRange(context.Background(), param)
if err != nil { if err != nil {
t.Errorf("expected no error, got %s", err) t.Errorf("expected no error, got %s", err)
} }
@ -738,7 +740,8 @@ func TestV2QueryRangeValueType(t *testing.T) {
} }
for i, param := range params { for i, param := range params {
_, errByName, err := q.QueryRange(context.Background(), param, nil) tracesV3.Enrich(param, map[string]v3.AttributeKey{})
_, errByName, err := q.QueryRange(context.Background(), param)
if err != nil { if err != nil {
t.Errorf("expected no error, got %s", err) t.Errorf("expected no error, got %s", err)
} }
@ -792,7 +795,8 @@ func TestV2QueryRangeTimeShift(t *testing.T) {
expectedTimeRangeInQueryString := fmt.Sprintf("timestamp >= %d AND timestamp <= %d", (1675115596722-86400*1000)*1000000, ((1675115596722+120*60*1000)-86400*1000)*1000000) expectedTimeRangeInQueryString := fmt.Sprintf("timestamp >= %d AND timestamp <= %d", (1675115596722-86400*1000)*1000000, ((1675115596722+120*60*1000)-86400*1000)*1000000)
for i, param := range params { for i, param := range params {
_, errByName, err := q.QueryRange(context.Background(), param, nil) tracesV3.Enrich(param, map[string]v3.AttributeKey{})
_, errByName, err := q.QueryRange(context.Background(), param)
if err != nil { if err != nil {
t.Errorf("expected no error, got %s", err) t.Errorf("expected no error, got %s", err)
} }
@ -892,7 +896,8 @@ func TestV2QueryRangeTimeShiftWithCache(t *testing.T) {
} }
for i, param := range params { for i, param := range params {
_, errByName, err := q.QueryRange(context.Background(), param, nil) tracesV3.Enrich(param, map[string]v3.AttributeKey{})
_, errByName, err := q.QueryRange(context.Background(), param)
if err != nil { if err != nil {
t.Errorf("expected no error, got %s", err) t.Errorf("expected no error, got %s", err)
} }
@ -994,7 +999,8 @@ func TestV2QueryRangeTimeShiftWithLimitAndCache(t *testing.T) {
} }
for i, param := range params { for i, param := range params {
_, errByName, err := q.QueryRange(context.Background(), param, nil) tracesV3.Enrich(param, map[string]v3.AttributeKey{})
_, errByName, err := q.QueryRange(context.Background(), param)
if err != nil { if err != nil {
t.Errorf("expected no error, got %s", err) t.Errorf("expected no error, got %s", err)
} }
@ -1085,7 +1091,8 @@ func TestV2QueryRangeValueTypePromQL(t *testing.T) {
} }
for i, param := range params { for i, param := range params {
_, errByName, err := q.QueryRange(context.Background(), param, nil) tracesV3.Enrich(param, map[string]v3.AttributeKey{})
_, errByName, err := q.QueryRange(context.Background(), param)
if err != nil { if err != nil {
t.Errorf("expected no error, got %s", err) t.Errorf("expected no error, got %s", err)
} }

14
pkg/query-service/app/queryBuilder/query_builder.go
View File

@ -43,7 +43,7 @@ var SupportedFunctions = []string{
var EvalFuncs = map[string]govaluate.ExpressionFunction{} var EvalFuncs = map[string]govaluate.ExpressionFunction{}
type prepareTracesQueryFunc func(start, end int64, panelType v3.PanelType, bq *v3.BuilderQuery, keys map[string]v3.AttributeKey, options tracesV3.Options) (string, error) type prepareTracesQueryFunc func(start, end int64, panelType v3.PanelType, bq *v3.BuilderQuery, options tracesV3.Options) (string, error)
type prepareLogsQueryFunc func(start, end int64, queryType v3.QueryType, panelType v3.PanelType, bq *v3.BuilderQuery, options logsV3.Options) (string, error) type prepareLogsQueryFunc func(start, end int64, queryType v3.QueryType, panelType v3.PanelType, bq *v3.BuilderQuery, options logsV3.Options) (string, error)
type prepareMetricQueryFunc func(start, end int64, queryType v3.QueryType, panelType v3.PanelType, bq *v3.BuilderQuery, options metricsV3.Options) (string, error) type prepareMetricQueryFunc func(start, end int64, queryType v3.QueryType, panelType v3.PanelType, bq *v3.BuilderQuery, options metricsV3.Options) (string, error)
@ -173,7 +173,7 @@ func (qb *QueryBuilder) PrepareLiveTailQuery(params *v3.QueryRangeParamsV3) (str
return queryStr, nil return queryStr, nil
} }
func (qb *QueryBuilder) PrepareQueries(params *v3.QueryRangeParamsV3, args ...interface{}) (map[string]string, error) { func (qb *QueryBuilder) PrepareQueries(params *v3.QueryRangeParamsV3) (map[string]string, error) {
queries := make(map[string]string) queries := make(map[string]string)
compositeQuery := params.CompositeQuery compositeQuery := params.CompositeQuery
@ -193,19 +193,15 @@ func (qb *QueryBuilder) PrepareQueries(params *v3.QueryRangeParamsV3, args ...in
if query.Expression == queryName { if query.Expression == queryName {
switch query.DataSource { switch query.DataSource {
case v3.DataSourceTraces: case v3.DataSourceTraces:
keys := map[string]v3.AttributeKey{}
if len(args) > 0 {
keys = args[0].(map[string]v3.AttributeKey)
}
// for ts query with group by and limit form two queries // for ts query with group by and limit form two queries
if compositeQuery.PanelType == v3.PanelTypeGraph && query.Limit > 0 && len(query.GroupBy) > 0 { if compositeQuery.PanelType == v3.PanelTypeGraph && query.Limit > 0 && len(query.GroupBy) > 0 {
limitQuery, err := qb.options.BuildTraceQuery(start, end, compositeQuery.PanelType, query, limitQuery, err := qb.options.BuildTraceQuery(start, end, compositeQuery.PanelType, query,
keys, tracesV3.Options{GraphLimitQtype: constants.FirstQueryGraphLimit, PreferRPM: PreferRPMFeatureEnabled}) tracesV3.Options{GraphLimitQtype: constants.FirstQueryGraphLimit, PreferRPM: PreferRPMFeatureEnabled})
if err != nil { if err != nil {
return nil, err return nil, err
} }
placeholderQuery, err := qb.options.BuildTraceQuery(start, end, compositeQuery.PanelType, placeholderQuery, err := qb.options.BuildTraceQuery(start, end, compositeQuery.PanelType,
query, keys, tracesV3.Options{GraphLimitQtype: constants.SecondQueryGraphLimit, PreferRPM: PreferRPMFeatureEnabled}) query, tracesV3.Options{GraphLimitQtype: constants.SecondQueryGraphLimit, PreferRPM: PreferRPMFeatureEnabled})
if err != nil { if err != nil {
return nil, err return nil, err
} }
@ -213,7 +209,7 @@ func (qb *QueryBuilder) PrepareQueries(params *v3.QueryRangeParamsV3, args ...in
queries[queryName] = query queries[queryName] = query
} else { } else {
queryString, err := qb.options.BuildTraceQuery(start, end, compositeQuery.PanelType, queryString, err := qb.options.BuildTraceQuery(start, end, compositeQuery.PanelType,
query, keys, tracesV3.Options{PreferRPM: PreferRPMFeatureEnabled, GraphLimitQtype: ""}) query, tracesV3.Options{PreferRPM: PreferRPMFeatureEnabled, GraphLimitQtype: ""})
if err != nil { if err != nil {
return nil, err return nil, err
} }

2
pkg/query-service/app/server.go
View File

@ -710,7 +710,7 @@ func (s *Server) Stop() error {
} }
func makeRulesManager( func makeRulesManager(
promConfigPath, _,
alertManagerURL string, alertManagerURL string,
ruleRepoURL string, ruleRepoURL string,
db *sqlx.DB, db *sqlx.DB,

100
pkg/query-service/app/traces/v3/query_builder.go
View File

@ -58,8 +58,7 @@ var tracesOperatorMappingV3 = map[v3.FilterOperator]string{
v3.FilterOperatorNotExists: "NOT has(%s%s, '%s')", v3.FilterOperatorNotExists: "NOT has(%s%s, '%s')",
} }
func getColumnName(key v3.AttributeKey, keys map[string]v3.AttributeKey) string { func getColumnName(key v3.AttributeKey) string {
key = enrichKeyWithMetadata(key, keys)
if key.IsColumn { if key.IsColumn {
return key.Key return key.Key
} }
@ -102,13 +101,13 @@ func enrichKeyWithMetadata(key v3.AttributeKey, keys map[string]v3.AttributeKey)
} }
// getSelectLabels returns the select labels for the query based on groupBy and aggregateOperator // getSelectLabels returns the select labels for the query based on groupBy and aggregateOperator
func getSelectLabels(aggregatorOperator v3.AggregateOperator, groupBy []v3.AttributeKey, keys map[string]v3.AttributeKey) string { func getSelectLabels(aggregatorOperator v3.AggregateOperator, groupBy []v3.AttributeKey) string {
var selectLabels string var selectLabels string
if aggregatorOperator == v3.AggregateOperatorNoOp { if aggregatorOperator == v3.AggregateOperatorNoOp {
selectLabels = "" selectLabels = ""
} else { } else {
for _, tag := range groupBy { for _, tag := range groupBy {
filterName := getColumnName(tag, keys) filterName := getColumnName(tag)
selectLabels += fmt.Sprintf(" %s as `%s`,", filterName, tag.Key) selectLabels += fmt.Sprintf(" %s as `%s`,", filterName, tag.Key)
} }
} }
@ -127,10 +126,10 @@ func getSelectKeys(aggregatorOperator v3.AggregateOperator, groupBy []v3.Attribu
return strings.Join(selectLabels, ",") return strings.Join(selectLabels, ",")
} }
func getSelectColumns(sc []v3.AttributeKey, keys map[string]v3.AttributeKey) string { func getSelectColumns(sc []v3.AttributeKey) string {
var columns []string var columns []string
for _, tag := range sc { for _, tag := range sc {
columnName := getColumnName(tag, keys) columnName := getColumnName(tag)
columns = append(columns, fmt.Sprintf("%s as `%s` ", columnName, tag.Key)) columns = append(columns, fmt.Sprintf("%s as `%s` ", columnName, tag.Key))
} }
return strings.Join(columns, ",") return strings.Join(columns, ",")
@ -150,20 +149,19 @@ func getZerosForEpochNano(epoch int64) int64 {
return int64(math.Pow(10, float64(19-count))) return int64(math.Pow(10, float64(19-count)))
} }
func buildTracesFilterQuery(fs *v3.FilterSet, keys map[string]v3.AttributeKey) (string, error) { func buildTracesFilterQuery(fs *v3.FilterSet) (string, error) {
var conditions []string var conditions []string
if fs != nil && len(fs.Items) != 0 { if fs != nil && len(fs.Items) != 0 {
for _, item := range fs.Items { for _, item := range fs.Items {
val := item.Value val := item.Value
// generate the key // generate the key
columnName := getColumnName(item.Key, keys) columnName := getColumnName(item.Key)
var fmtVal string var fmtVal string
key := enrichKeyWithMetadata(item.Key, keys)
item.Operator = v3.FilterOperator(strings.ToLower(strings.TrimSpace(string(item.Operator)))) item.Operator = v3.FilterOperator(strings.ToLower(strings.TrimSpace(string(item.Operator))))
if item.Operator != v3.FilterOperatorExists && item.Operator != v3.FilterOperatorNotExists { if item.Operator != v3.FilterOperatorExists && item.Operator != v3.FilterOperatorNotExists {
var err error var err error
val, err = utils.ValidateAndCastValue(val, key.DataType) val, err = utils.ValidateAndCastValue(val, item.Key.DataType)
if err != nil { if err != nil {
return "", fmt.Errorf("invalid value for key %s: %v", item.Key.Key, err) return "", fmt.Errorf("invalid value for key %s: %v", item.Key.Key, err)
} }
@ -179,15 +177,15 @@ func buildTracesFilterQuery(fs *v3.FilterSet, keys map[string]v3.AttributeKey) (
case v3.FilterOperatorRegex, v3.FilterOperatorNotRegex: case v3.FilterOperatorRegex, v3.FilterOperatorNotRegex:
conditions = append(conditions, fmt.Sprintf(operator, columnName, fmtVal)) conditions = append(conditions, fmt.Sprintf(operator, columnName, fmtVal))
case v3.FilterOperatorExists, v3.FilterOperatorNotExists: case v3.FilterOperatorExists, v3.FilterOperatorNotExists:
if key.IsColumn { if item.Key.IsColumn {
subQuery, err := existsSubQueryForFixedColumn(key, item.Operator) subQuery, err := existsSubQueryForFixedColumn(item.Key, item.Operator)
if err != nil { if err != nil {
return "", err return "", err
} }
conditions = append(conditions, subQuery) conditions = append(conditions, subQuery)
} else { } else {
columnType, columnDataType := getClickhouseTracesColumnDataTypeAndType(key) columnType, columnDataType := getClickhouseTracesColumnDataTypeAndType(item.Key)
conditions = append(conditions, fmt.Sprintf(operator, columnDataType, columnType, key.Key)) conditions = append(conditions, fmt.Sprintf(operator, columnDataType, columnType, item.Key.Key))
} }
default: default:
@ -218,12 +216,11 @@ func existsSubQueryForFixedColumn(key v3.AttributeKey, op v3.FilterOperator) (st
} }
} }
func handleEmptyValuesInGroupBy(keys map[string]v3.AttributeKey, groupBy []v3.AttributeKey) (string, error) { func handleEmptyValuesInGroupBy(groupBy []v3.AttributeKey) (string, error) {
filterItems := []v3.FilterItem{} filterItems := []v3.FilterItem{}
if len(groupBy) != 0 { if len(groupBy) != 0 {
for _, item := range groupBy { for _, item := range groupBy {
key := enrichKeyWithMetadata(item, keys) if !item.IsColumn {
if !key.IsColumn {
filterItems = append(filterItems, v3.FilterItem{ filterItems = append(filterItems, v3.FilterItem{
Key: item, Key: item,
Operator: v3.FilterOperatorExists, Operator: v3.FilterOperatorExists,
@ -236,21 +233,21 @@ func handleEmptyValuesInGroupBy(keys map[string]v3.AttributeKey, groupBy []v3.At
Operator: "AND", Operator: "AND",
Items: filterItems, Items: filterItems,
} }
return buildTracesFilterQuery(&filterSet, keys) return buildTracesFilterQuery(&filterSet)
} }
return "", nil return "", nil
} }
func buildTracesQuery(start, end, step int64, mq *v3.BuilderQuery, tableName string, keys map[string]v3.AttributeKey, panelType v3.PanelType, options Options) (string, error) { func buildTracesQuery(start, end, step int64, mq *v3.BuilderQuery, _ string, panelType v3.PanelType, options Options) (string, error) {
filterSubQuery, err := buildTracesFilterQuery(mq.Filters, keys) filterSubQuery, err := buildTracesFilterQuery(mq.Filters)
if err != nil { if err != nil {
return "", err return "", err
} }
// timerange will be sent in epoch millisecond // timerange will be sent in epoch millisecond
spanIndexTableTimeFilter := fmt.Sprintf("(timestamp >= '%d' AND timestamp <= '%d')", start*getZerosForEpochNano(start), end*getZerosForEpochNano(end)) spanIndexTableTimeFilter := fmt.Sprintf("(timestamp >= '%d' AND timestamp <= '%d')", start*getZerosForEpochNano(start), end*getZerosForEpochNano(end))
selectLabels := getSelectLabels(mq.AggregateOperator, mq.GroupBy, keys) selectLabels := getSelectLabels(mq.AggregateOperator, mq.GroupBy)
having := having(mq.Having) having := having(mq.Having)
if having != "" { if having != "" {
@ -283,7 +280,7 @@ func buildTracesQuery(start, end, step int64, mq *v3.BuilderQuery, tableName str
queryTmpl = "SELECT " + getSelectKeys(mq.AggregateOperator, mq.GroupBy) + " from (" + queryTmpl + ")" queryTmpl = "SELECT " + getSelectKeys(mq.AggregateOperator, mq.GroupBy) + " from (" + queryTmpl + ")"
} }
emptyValuesInGroupByFilter, err := handleEmptyValuesInGroupBy(keys, mq.GroupBy) emptyValuesInGroupByFilter, err := handleEmptyValuesInGroupBy(mq.GroupBy)
if err != nil { if err != nil {
return "", err return "", err
} }
@ -293,8 +290,7 @@ func buildTracesQuery(start, end, step int64, mq *v3.BuilderQuery, tableName str
if groupBy != "" { if groupBy != "" {
groupBy = " group by " + groupBy groupBy = " group by " + groupBy
} }
enrichedOrderBy := enrichOrderBy(mq.OrderBy, keys) orderBy := orderByAttributeKeyTags(panelType, mq.OrderBy, mq.GroupBy)
orderBy := orderByAttributeKeyTags(panelType, enrichedOrderBy, mq.GroupBy, keys)
if orderBy != "" { if orderBy != "" {
orderBy = " order by " + orderBy orderBy = " order by " + orderBy
} }
@ -305,7 +301,7 @@ func buildTracesQuery(start, end, step int64, mq *v3.BuilderQuery, tableName str
aggregationKey := "" aggregationKey := ""
if mq.AggregateAttribute.Key != "" { if mq.AggregateAttribute.Key != "" {
aggregationKey = getColumnName(mq.AggregateAttribute, keys) aggregationKey = getColumnName(mq.AggregateAttribute)
} }
switch mq.AggregateOperator { switch mq.AggregateOperator {
@ -342,14 +338,13 @@ func buildTracesQuery(start, end, step int64, mq *v3.BuilderQuery, tableName str
return query, nil return query, nil
case v3.AggregateOperatorCount: case v3.AggregateOperatorCount:
if mq.AggregateAttribute.Key != "" { if mq.AggregateAttribute.Key != "" {
key := enrichKeyWithMetadata(mq.AggregateAttribute, keys) if mq.AggregateAttribute.IsColumn {
if key.IsColumn { subQuery, err := existsSubQueryForFixedColumn(mq.AggregateAttribute, v3.FilterOperatorExists)
subQuery, err := existsSubQueryForFixedColumn(key, v3.FilterOperatorExists)
if err == nil { if err == nil {
filterSubQuery = fmt.Sprintf("%s AND %s", filterSubQuery, subQuery) filterSubQuery = fmt.Sprintf("%s AND %s", filterSubQuery, subQuery)
} }
} else { } else {
columnType, columnDataType := getClickhouseTracesColumnDataTypeAndType(key) columnType, columnDataType := getClickhouseTracesColumnDataTypeAndType(mq.AggregateAttribute)
filterSubQuery = fmt.Sprintf("%s AND has(%s%s, '%s')", filterSubQuery, columnDataType, columnType, mq.AggregateAttribute.Key) filterSubQuery = fmt.Sprintf("%s AND has(%s%s, '%s')", filterSubQuery, columnDataType, columnType, mq.AggregateAttribute.Key)
} }
} }
@ -373,7 +368,7 @@ func buildTracesQuery(start, end, step int64, mq *v3.BuilderQuery, tableName str
if len(mq.SelectColumns) == 0 { if len(mq.SelectColumns) == 0 {
return "", fmt.Errorf("select columns cannot be empty for panelType %s", panelType) return "", fmt.Errorf("select columns cannot be empty for panelType %s", panelType)
} }
selectColumns := getSelectColumns(mq.SelectColumns, keys) selectColumns := getSelectColumns(mq.SelectColumns)
queryNoOpTmpl := fmt.Sprintf("SELECT timestamp as timestamp_datetime, spanID, traceID, "+"%s ", selectColumns) + "from " + constants.SIGNOZ_TRACE_DBNAME + "." + constants.SIGNOZ_SPAN_INDEX_TABLENAME + " where %s %s" + "%s" queryNoOpTmpl := fmt.Sprintf("SELECT timestamp as timestamp_datetime, spanID, traceID, "+"%s ", selectColumns) + "from " + constants.SIGNOZ_TRACE_DBNAME + "." + constants.SIGNOZ_SPAN_INDEX_TABLENAME + " where %s %s" + "%s"
query = fmt.Sprintf(queryNoOpTmpl, spanIndexTableTimeFilter, filterSubQuery, orderBy) query = fmt.Sprintf(queryNoOpTmpl, spanIndexTableTimeFilter, filterSubQuery, orderBy)
} else { } else {
@ -423,7 +418,7 @@ func groupByAttributeKeyTags(panelType v3.PanelType, graphLimitQtype string, tag
// orderBy returns a string of comma separated tags for order by clause // orderBy returns a string of comma separated tags for order by clause
// if there are remaining items which are not present in tags they are also added // if there are remaining items which are not present in tags they are also added
// if the order is not specified, it defaults to ASC // if the order is not specified, it defaults to ASC
func orderBy(panelType v3.PanelType, items []v3.OrderBy, tagLookup map[string]struct{}, keys map[string]v3.AttributeKey) []string { func orderBy(panelType v3.PanelType, items []v3.OrderBy, tagLookup map[string]struct{}) []string {
var orderBy []string var orderBy []string
for _, item := range items { for _, item := range items {
@ -433,7 +428,7 @@ func orderBy(panelType v3.PanelType, items []v3.OrderBy, tagLookup map[string]st
orderBy = append(orderBy, fmt.Sprintf("`%s` %s", item.ColumnName, item.Order)) orderBy = append(orderBy, fmt.Sprintf("`%s` %s", item.ColumnName, item.Order))
} else if panelType == v3.PanelTypeList { } else if panelType == v3.PanelTypeList {
attr := v3.AttributeKey{Key: item.ColumnName, DataType: item.DataType, Type: item.Type, IsColumn: item.IsColumn} attr := v3.AttributeKey{Key: item.ColumnName, DataType: item.DataType, Type: item.Type, IsColumn: item.IsColumn}
name := getColumnName(attr, keys) name := getColumnName(attr)
if item.IsColumn { if item.IsColumn {
orderBy = append(orderBy, fmt.Sprintf("`%s` %s", name, item.Order)) orderBy = append(orderBy, fmt.Sprintf("`%s` %s", name, item.Order))
} else { } else {
@ -445,13 +440,13 @@ func orderBy(panelType v3.PanelType, items []v3.OrderBy, tagLookup map[string]st
return orderBy return orderBy
} }
func orderByAttributeKeyTags(panelType v3.PanelType, items []v3.OrderBy, tags []v3.AttributeKey, keys map[string]v3.AttributeKey) string { func orderByAttributeKeyTags(panelType v3.PanelType, items []v3.OrderBy, tags []v3.AttributeKey) string {
tagLookup := map[string]struct{}{} tagLookup := map[string]struct{}{}
for _, v := range tags { for _, v := range tags {
tagLookup[v.Key] = struct{}{} tagLookup[v.Key] = struct{}{}
} }
orderByArray := orderBy(panelType, items, tagLookup, keys) orderByArray := orderBy(panelType, items, tagLookup)
if len(orderByArray) == 0 { if len(orderByArray) == 0 {
if panelType == v3.PanelTypeList { if panelType == v3.PanelTypeList {
@ -474,7 +469,7 @@ func having(items []v3.Having) string {
return strings.Join(having, " AND ") return strings.Join(having, " AND ")
} }
func reduceToQuery(query string, reduceTo v3.ReduceToOperator, aggregateOperator v3.AggregateOperator) (string, error) { func reduceToQuery(query string, reduceTo v3.ReduceToOperator, _ v3.AggregateOperator) (string, error) {
var groupBy string var groupBy string
switch reduceTo { switch reduceTo {
@ -508,13 +503,13 @@ func addOffsetToQuery(query string, offset uint64) string {
// PrepareTracesQuery returns the query string for traces // PrepareTracesQuery returns the query string for traces
// start and end are in epoch millisecond // start and end are in epoch millisecond
// step is in seconds // step is in seconds
func PrepareTracesQuery(start, end int64, panelType v3.PanelType, mq *v3.BuilderQuery, keys map[string]v3.AttributeKey, options Options) (string, error) { func PrepareTracesQuery(start, end int64, panelType v3.PanelType, mq *v3.BuilderQuery, options Options) (string, error) {
// adjust the start and end time to the step interval // adjust the start and end time to the step interval
start = start - (start % (mq.StepInterval * 1000)) start = start - (start % (mq.StepInterval * 1000))
end = end - (end % (mq.StepInterval * 1000)) end = end - (end % (mq.StepInterval * 1000))
if options.GraphLimitQtype == constants.FirstQueryGraphLimit { if options.GraphLimitQtype == constants.FirstQueryGraphLimit {
// give me just the group by names // give me just the group by names
query, err := buildTracesQuery(start, end, mq.StepInterval, mq, constants.SIGNOZ_SPAN_INDEX_TABLENAME, keys, panelType, options) query, err := buildTracesQuery(start, end, mq.StepInterval, mq, constants.SIGNOZ_SPAN_INDEX_TABLENAME, panelType, options)
if err != nil { if err != nil {
return "", err return "", err
} }
@ -522,14 +517,14 @@ func PrepareTracesQuery(start, end int64, panelType v3.PanelType, mq *v3.Builder
return query, nil return query, nil
} else if options.GraphLimitQtype == constants.SecondQueryGraphLimit { } else if options.GraphLimitQtype == constants.SecondQueryGraphLimit {
query, err := buildTracesQuery(start, end, mq.StepInterval, mq, constants.SIGNOZ_SPAN_INDEX_TABLENAME, keys, panelType, options) query, err := buildTracesQuery(start, end, mq.StepInterval, mq, constants.SIGNOZ_SPAN_INDEX_TABLENAME, panelType, options)
if err != nil { if err != nil {
return "", err return "", err
} }
return query, nil return query, nil
} }
query, err := buildTracesQuery(start, end, mq.StepInterval, mq, constants.SIGNOZ_SPAN_INDEX_TABLENAME, keys, panelType, options) query, err := buildTracesQuery(start, end, mq.StepInterval, mq, constants.SIGNOZ_SPAN_INDEX_TABLENAME, panelType, options)
if err != nil { if err != nil {
return "", err return "", err
} }
@ -545,3 +540,30 @@ func PrepareTracesQuery(start, end int64, panelType v3.PanelType, mq *v3.Builder
} }
return query, err return query, err
} }
func Enrich(params *v3.QueryRangeParamsV3, keys map[string]v3.AttributeKey) {
if params.CompositeQuery.QueryType == v3.QueryTypeBuilder {
for _, query := range params.CompositeQuery.BuilderQueries {
if query.DataSource == v3.DataSourceTraces {
// enrich aggregate attribute
query.AggregateAttribute = enrichKeyWithMetadata(query.AggregateAttribute, keys)
// enrich filter items
if query.Filters != nil && len(query.Filters.Items) > 0 {
for idx, filter := range query.Filters.Items {
query.Filters.Items[idx].Key = enrichKeyWithMetadata(filter.Key, keys)
}
}
// enrich group by
for idx, groupBy := range query.GroupBy {
query.GroupBy[idx] = enrichKeyWithMetadata(groupBy, keys)
}
// enrich order by
query.OrderBy = enrichOrderBy(query.OrderBy, keys)
// enrich select columns
for idx, selectColumn := range query.SelectColumns {
query.SelectColumns[idx] = enrichKeyWithMetadata(selectColumn, keys)
}
}
}
}
}

44
pkg/query-service/app/traces/v3/query_builder_test.go
View File

@ -133,7 +133,7 @@ var buildFilterQueryData = []struct {
func TestBuildTracesFilterQuery(t *testing.T) { func TestBuildTracesFilterQuery(t *testing.T) {
for _, tt := range buildFilterQueryData { for _, tt := range buildFilterQueryData {
Convey("TestBuildTracesFilterQuery", t, func() { Convey("TestBuildTracesFilterQuery", t, func() {
query, err := buildTracesFilterQuery(tt.FilterSet, map[string]v3.AttributeKey{}) query, err := buildTracesFilterQuery(tt.FilterSet)
So(err, ShouldBeNil) So(err, ShouldBeNil)
So(query, ShouldEqual, tt.ExpectedFilter) So(query, ShouldEqual, tt.ExpectedFilter)
}) })
@ -169,7 +169,7 @@ var handleEmptyValuesInGroupByData = []struct {
func TestBuildTracesHandleEmptyValuesInGroupBy(t *testing.T) { func TestBuildTracesHandleEmptyValuesInGroupBy(t *testing.T) {
for _, tt := range handleEmptyValuesInGroupByData { for _, tt := range handleEmptyValuesInGroupByData {
Convey("TestBuildTracesHandleEmptyValuesInGroupBy", t, func() { Convey("TestBuildTracesHandleEmptyValuesInGroupBy", t, func() {
query, err := handleEmptyValuesInGroupBy(map[string]v3.AttributeKey{}, tt.GroupBy) query, err := handleEmptyValuesInGroupBy(tt.GroupBy)
So(err, ShouldBeNil) So(err, ShouldBeNil)
So(query, ShouldEqual, tt.ExpectedFilter) So(query, ShouldEqual, tt.ExpectedFilter)
}) })
@ -220,8 +220,9 @@ var testColumnName = []struct {
func TestColumnName(t *testing.T) { func TestColumnName(t *testing.T) {
for _, tt := range testColumnName { for _, tt := range testColumnName {
tt.AttributeKey = enrichKeyWithMetadata(tt.AttributeKey, map[string]v3.AttributeKey{})
Convey("testColumnName", t, func() { Convey("testColumnName", t, func() {
Column := getColumnName(tt.AttributeKey, map[string]v3.AttributeKey{}) Column := getColumnName(tt.AttributeKey)
So(Column, ShouldEqual, tt.ExpectedColumn) So(Column, ShouldEqual, tt.ExpectedColumn)
}) })
} }
@ -265,7 +266,7 @@ var testGetSelectLabelsData = []struct {
func TestGetSelectLabels(t *testing.T) { func TestGetSelectLabels(t *testing.T) {
for _, tt := range testGetSelectLabelsData { for _, tt := range testGetSelectLabelsData {
Convey("testGetSelectLabelsData", t, func() { Convey("testGetSelectLabelsData", t, func() {
selectLabels := getSelectLabels(tt.AggregateOperator, tt.GroupByTags, map[string]v3.AttributeKey{}) selectLabels := getSelectLabels(tt.AggregateOperator, tt.GroupByTags)
So(selectLabels, ShouldEqual, tt.SelectLabels) So(selectLabels, ShouldEqual, tt.SelectLabels)
}) })
} }
@ -304,7 +305,7 @@ var testGetSelectColumnsData = []struct {
func TestGetSelectColumns(t *testing.T) { func TestGetSelectColumns(t *testing.T) {
for _, tt := range testGetSelectColumnsData { for _, tt := range testGetSelectColumnsData {
Convey("testGetSelectColumnsData", t, func() { Convey("testGetSelectColumnsData", t, func() {
selectColumns := getSelectColumns(tt.sc, map[string]v3.AttributeKey{}) selectColumns := getSelectColumns(tt.sc)
So(selectColumns, ShouldEqual, tt.SelectColumns) So(selectColumns, ShouldEqual, tt.SelectColumns)
}) })
} }
@ -464,13 +465,15 @@ var testOrderBy = []struct {
} }
func TestOrderBy(t *testing.T) { func TestOrderBy(t *testing.T) {
for _, tt := range testOrderBy { keys := map[string]v3.AttributeKey{
Convey("testOrderBy", t, func() {
res := orderByAttributeKeyTags(tt.PanelType, tt.Items, tt.Tags, map[string]v3.AttributeKey{
"name": {Key: "name", DataType: v3.AttributeKeyDataTypeString, Type: v3.AttributeKeyTypeTag, IsColumn: true}, "name": {Key: "name", DataType: v3.AttributeKeyDataTypeString, Type: v3.AttributeKeyTypeTag, IsColumn: true},
"bytes": {Key: "bytes", DataType: v3.AttributeKeyDataTypeString, Type: v3.AttributeKeyTypeTag, IsColumn: true}, "bytes": {Key: "bytes", DataType: v3.AttributeKeyDataTypeString, Type: v3.AttributeKeyTypeTag, IsColumn: true},
"response_time": {Key: "response_time", DataType: v3.AttributeKeyDataTypeString, Type: v3.AttributeKeyTypeTag, IsColumn: false}, "response_time": {Key: "response_time", DataType: v3.AttributeKeyDataTypeString, Type: v3.AttributeKeyTypeTag, IsColumn: false},
}) }
for _, tt := range testOrderBy {
Convey("testOrderBy", t, func() {
tt.Items = enrichOrderBy(tt.Items, keys)
res := orderByAttributeKeyTags(tt.PanelType, tt.Items, tt.Tags)
So(res, ShouldResemble, tt.Result) So(res, ShouldResemble, tt.Result)
}) })
} }
@ -1171,11 +1174,24 @@ var testBuildTracesQueryData = []struct {
} }
func TestBuildTracesQuery(t *testing.T) { func TestBuildTracesQuery(t *testing.T) {
for _, tt := range testBuildTracesQueryData { keys := map[string]v3.AttributeKey{
Convey("TestBuildTracesQuery", t, func() {
query, err := buildTracesQuery(tt.Start, tt.End, tt.BuilderQuery.StepInterval, tt.BuilderQuery, tt.TableName, map[string]v3.AttributeKey{
"name": {Key: "name", DataType: v3.AttributeKeyDataTypeString, Type: v3.AttributeKeyTypeTag, IsColumn: true}, "name": {Key: "name", DataType: v3.AttributeKeyDataTypeString, Type: v3.AttributeKeyTypeTag, IsColumn: true},
}, tt.PanelType, tt.Options) }
for _, tt := range testBuildTracesQueryData {
tt.BuilderQuery.DataSource = v3.DataSourceTraces
params := &v3.QueryRangeParamsV3{
Version: "v4",
CompositeQuery: &v3.CompositeQuery{
QueryType: v3.QueryTypeBuilder,
BuilderQueries: map[string]*v3.BuilderQuery{
"A": tt.BuilderQuery,
},
},
}
Enrich(params, keys)
Convey("TestBuildTracesQuery", t, func() {
query, err := buildTracesQuery(tt.Start, tt.End, tt.BuilderQuery.StepInterval, tt.BuilderQuery, tt.TableName, tt.PanelType, tt.Options)
So(err, ShouldBeNil) So(err, ShouldBeNil)
So(query, ShouldEqual, tt.ExpectedQuery) So(query, ShouldEqual, tt.ExpectedQuery)
}) })
@ -1400,7 +1416,7 @@ var testPrepTracesQueryData = []struct {
func TestPrepareTracesQuery(t *testing.T) { func TestPrepareTracesQuery(t *testing.T) {
for _, tt := range testPrepTracesQueryData { for _, tt := range testPrepTracesQueryData {
Convey("TestPrepareTracesQuery", t, func() { Convey("TestPrepareTracesQuery", t, func() {
query, err := PrepareTracesQuery(tt.Start, tt.End, tt.PanelType, tt.BuilderQuery, tt.Keys, tt.Options) query, err := PrepareTracesQuery(tt.Start, tt.End, tt.PanelType, tt.BuilderQuery, tt.Options)
So(err, ShouldBeNil) So(err, ShouldBeNil)
So(query, ShouldEqual, tt.ExpectedQuery) So(query, ShouldEqual, tt.ExpectedQuery)
}) })

2
pkg/query-service/interfaces/interface.go
View File

@ -126,7 +126,7 @@ type Reader interface {
} }
type Querier interface { type Querier interface {
QueryRange(context.Context, *v3.QueryRangeParamsV3, map[string]v3.AttributeKey) ([]*v3.Result, map[string]error, error) QueryRange(context.Context, *v3.QueryRangeParamsV3) ([]*v3.Result, map[string]error, error)
// test helpers // test helpers
QueriesExecuted() []string QueriesExecuted() []string

4
pkg/query-service/rules/threshold_rule.go
View File

@ -504,9 +504,9 @@ func (r *ThresholdRule) buildAndRunQuery(ctx context.Context, ts time.Time) (Vec
var queryErrors map[string]error var queryErrors map[string]error
if r.version == "v4" { if r.version == "v4" {
results, queryErrors, err = r.querierV2.QueryRange(ctx, params, map[string]v3.AttributeKey{}) results, queryErrors, err = r.querierV2.QueryRange(ctx, params)
} else { } else {
results, queryErrors, err = r.querier.QueryRange(ctx, params, map[string]v3.AttributeKey{}) results, queryErrors, err = r.querier.QueryRange(ctx, params)
} }
if err != nil { if err != nil {