mirror of
https://github.com/SigNoz/signoz.git
synced 2026-08-18 10:40:46 +01:00
Compare commits
1 Commits
issue-2988
...
feat/semco
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
e165bff99b |
@@ -336,7 +336,7 @@ func (r *ClickHouseReader) GetTopLevelOperations(ctx context.Context, start, end
|
||||
return &operations, nil
|
||||
}
|
||||
|
||||
func (r *ClickHouseReader) buildResourceSubQuery(tags []model.TagQueryParam, svc string, start, end time.Time) (string, error) {
|
||||
func (r *ClickHouseReader) buildResourceSubQuery(ctx context.Context, orgID valuer.UUID, tags []model.TagQueryParam, svc string, start, end time.Time) (string, error) {
|
||||
// assuming all will be resource attributes.
|
||||
// and resource attributes are string for traces
|
||||
filterSet := v3.FilterSet{}
|
||||
@@ -387,7 +387,8 @@ func (r *ClickHouseReader) buildResourceSubQuery(tags []model.TagQueryParam, svc
|
||||
&filterSet,
|
||||
[]v3.AttributeKey{},
|
||||
v3.AttributeKey{},
|
||||
false)
|
||||
false,
|
||||
r.fl.BooleanOrEmpty(ctx, flagger.FeatureResolveSemconvFamilies, featuretypes.NewFlaggerEvaluationContext(orgID)))
|
||||
if err != nil {
|
||||
r.logger.Error("Error in processing sql query", errorsV2.Attr(err))
|
||||
return "", err
|
||||
@@ -395,7 +396,7 @@ func (r *ClickHouseReader) buildResourceSubQuery(tags []model.TagQueryParam, svc
|
||||
return resourceSubQuery, nil
|
||||
}
|
||||
|
||||
func (r *ClickHouseReader) GetServices(ctx context.Context, queryParams *model.GetServicesParams) (*[]model.ServiceItem, *model.ApiError) {
|
||||
func (r *ClickHouseReader) GetServices(ctx context.Context, orgID valuer.UUID, queryParams *model.GetServicesParams) (*[]model.ServiceItem, *model.ApiError) {
|
||||
|
||||
ctx = ctxtypes.NewContextWithCommentVals(ctx, map[string]string{
|
||||
instrumentationtypes.TelemetrySignal: telemetrytypes.SignalTraces.StringValue(),
|
||||
@@ -467,7 +468,7 @@ func (r *ClickHouseReader) GetServices(ctx context.Context, queryParams *model.G
|
||||
clickhouse.Named("names", ops),
|
||||
)
|
||||
|
||||
resourceSubQuery, err := r.buildResourceSubQuery(queryParams.Tags, svc, *queryParams.Start, *queryParams.End)
|
||||
resourceSubQuery, err := r.buildResourceSubQuery(ctx, orgID, queryParams.Tags, svc, *queryParams.Start, *queryParams.End)
|
||||
if err != nil {
|
||||
r.logger.Error("Error in processing sql query", errorsV2.Attr(err))
|
||||
return
|
||||
@@ -703,9 +704,9 @@ func addExistsOperator(item model.TagQuery, tagMapType string, not bool) (string
|
||||
return fmt.Sprintf(" AND %s (%s)", notStr, strings.Join(tagOperatorPair, " OR ")), args
|
||||
}
|
||||
|
||||
func (r *ClickHouseReader) GetEntryPointOperations(ctx context.Context, queryParams *model.GetTopOperationsParams) (*[]model.TopOperationsItem, error) {
|
||||
func (r *ClickHouseReader) GetEntryPointOperations(ctx context.Context, orgID valuer.UUID, queryParams *model.GetTopOperationsParams) (*[]model.TopOperationsItem, error) {
|
||||
// Step 1: Get top operations for the given service
|
||||
topOps, err := r.GetTopOperations(ctx, queryParams)
|
||||
topOps, err := r.GetTopOperations(ctx, orgID, queryParams)
|
||||
if err != nil {
|
||||
return nil, errorsV2.Wrapf(err, errorsV2.TypeInternal, errorsV2.CodeInternal, "Error in getting Top Operations")
|
||||
}
|
||||
@@ -757,7 +758,7 @@ func (r *ClickHouseReader) GetEntryPointOperations(ctx context.Context, queryPar
|
||||
return &filtered, nil
|
||||
}
|
||||
|
||||
func (r *ClickHouseReader) GetTopOperations(ctx context.Context, queryParams *model.GetTopOperationsParams) (*[]model.TopOperationsItem, *model.ApiError) {
|
||||
func (r *ClickHouseReader) GetTopOperations(ctx context.Context, orgID valuer.UUID, queryParams *model.GetTopOperationsParams) (*[]model.TopOperationsItem, *model.ApiError) {
|
||||
|
||||
ctx = ctxtypes.NewContextWithCommentVals(ctx, map[string]string{
|
||||
instrumentationtypes.TelemetrySignal: telemetrytypes.SignalTraces.StringValue(),
|
||||
@@ -787,7 +788,7 @@ func (r *ClickHouseReader) GetTopOperations(ctx context.Context, queryParams *mo
|
||||
r.TraceDB, r.traceTableName,
|
||||
)
|
||||
|
||||
resourceSubQuery, err := r.buildResourceSubQuery(queryParams.Tags, queryParams.ServiceName, *queryParams.Start, *queryParams.End)
|
||||
resourceSubQuery, err := r.buildResourceSubQuery(ctx, orgID, queryParams.Tags, queryParams.ServiceName, *queryParams.Start, *queryParams.End)
|
||||
if err != nil {
|
||||
r.logger.Error("Error in processing sql query", errorsV2.Attr(err))
|
||||
return nil, &model.ApiError{Typ: model.ErrorExec, Err: fmt.Errorf("error in processing sql query")}
|
||||
@@ -858,7 +859,7 @@ func (r *ClickHouseReader) GetUsage(ctx context.Context, queryParams *model.GetU
|
||||
return &usageItems, nil
|
||||
}
|
||||
|
||||
func (r *ClickHouseReader) GetDependencyGraph(ctx context.Context, queryParams *model.GetServicesParams) (*[]model.ServiceMapDependencyResponseItem, error) {
|
||||
func (r *ClickHouseReader) GetDependencyGraph(ctx context.Context, orgID valuer.UUID, queryParams *model.GetServicesParams) (*[]model.ServiceMapDependencyResponseItem, error) {
|
||||
|
||||
ctx = ctxtypes.NewContextWithCommentVals(ctx, map[string]string{
|
||||
instrumentationtypes.TelemetrySignal: telemetrytypes.SignalTraces.StringValue(),
|
||||
@@ -895,7 +896,7 @@ func (r *ClickHouseReader) GetDependencyGraph(ctx context.Context, queryParams *
|
||||
)
|
||||
|
||||
tags := createTagQueryFromTagQueryParams(queryParams.Tags)
|
||||
filterQuery, filterArgs := services.BuildServiceMapQuery(tags)
|
||||
filterQuery, filterArgs := services.BuildServiceMapQuery(tags, r.fl.BooleanOrEmpty(ctx, flagger.FeatureResolveSemconvFamilies, featuretypes.NewFlaggerEvaluationContext(orgID)))
|
||||
query += filterQuery + " GROUP BY src, dest;"
|
||||
args = append(args, filterArgs...)
|
||||
|
||||
|
||||
@@ -1128,13 +1128,19 @@ func (aH *APIHandler) registerEvent(w http.ResponseWriter, r *http.Request) {
|
||||
}
|
||||
|
||||
func (aH *APIHandler) getTopOperations(w http.ResponseWriter, r *http.Request) {
|
||||
claims, err := authtypes.ClaimsFromContext(r.Context())
|
||||
if err != nil {
|
||||
render.Error(w, err)
|
||||
return
|
||||
}
|
||||
orgID := valuer.MustNewUUID(claims.OrgID)
|
||||
|
||||
query, err := parseGetTopOperationsRequest(r)
|
||||
if aH.HandleError(w, err, http.StatusBadRequest) {
|
||||
return
|
||||
}
|
||||
|
||||
result, apiErr := aH.reader.GetTopOperations(r.Context(), query)
|
||||
result, apiErr := aH.reader.GetTopOperations(r.Context(), orgID, query)
|
||||
|
||||
if apiErr != nil && aH.HandleError(w, apiErr.Err, http.StatusInternalServerError) {
|
||||
return
|
||||
@@ -1145,13 +1151,20 @@ func (aH *APIHandler) getTopOperations(w http.ResponseWriter, r *http.Request) {
|
||||
}
|
||||
|
||||
func (aH *APIHandler) getEntryPointOps(w http.ResponseWriter, r *http.Request) {
|
||||
claims, err := authtypes.ClaimsFromContext(r.Context())
|
||||
if err != nil {
|
||||
render.Error(w, err)
|
||||
return
|
||||
}
|
||||
orgID := valuer.MustNewUUID(claims.OrgID)
|
||||
|
||||
query, err := parseGetTopOperationsRequest(r)
|
||||
if err != nil {
|
||||
render.Error(w, err)
|
||||
return
|
||||
}
|
||||
|
||||
result, apiErr := aH.reader.GetEntryPointOperations(r.Context(), query)
|
||||
result, apiErr := aH.reader.GetEntryPointOperations(r.Context(), orgID, query)
|
||||
if apiErr != nil {
|
||||
render.Error(w, apiErr)
|
||||
return
|
||||
@@ -1226,12 +1239,19 @@ func (aH *APIHandler) getServicesTopLevelOps(w http.ResponseWriter, r *http.Requ
|
||||
}
|
||||
|
||||
func (aH *APIHandler) getServices(w http.ResponseWriter, r *http.Request) {
|
||||
claims, err := authtypes.ClaimsFromContext(r.Context())
|
||||
if err != nil {
|
||||
render.Error(w, err)
|
||||
return
|
||||
}
|
||||
orgID := valuer.MustNewUUID(claims.OrgID)
|
||||
|
||||
query, err := parseGetServicesRequest(r)
|
||||
if aH.HandleError(w, err, http.StatusBadRequest) {
|
||||
return
|
||||
}
|
||||
|
||||
result, apiErr := aH.reader.GetServices(r.Context(), query)
|
||||
result, apiErr := aH.reader.GetServices(r.Context(), orgID, query)
|
||||
if apiErr != nil && aH.HandleError(w, apiErr.Err, http.StatusInternalServerError) {
|
||||
return
|
||||
}
|
||||
@@ -1240,13 +1260,19 @@ func (aH *APIHandler) getServices(w http.ResponseWriter, r *http.Request) {
|
||||
}
|
||||
|
||||
func (aH *APIHandler) dependencyGraph(w http.ResponseWriter, r *http.Request) {
|
||||
claims, err := authtypes.ClaimsFromContext(r.Context())
|
||||
if err != nil {
|
||||
render.Error(w, err)
|
||||
return
|
||||
}
|
||||
orgID := valuer.MustNewUUID(claims.OrgID)
|
||||
|
||||
query, err := parseGetServicesRequest(r)
|
||||
if aH.HandleError(w, err, http.StatusBadRequest) {
|
||||
return
|
||||
}
|
||||
|
||||
result, err := aH.reader.GetDependencyGraph(r.Context(), query)
|
||||
result, err := aH.reader.GetDependencyGraph(r.Context(), orgID, query)
|
||||
if aH.HandleError(w, err, http.StatusBadRequest) {
|
||||
return
|
||||
}
|
||||
|
||||
@@ -383,7 +383,7 @@ func buildLogsQuery(panelType v3.PanelType, start, end, step int64, mq *v3.Build
|
||||
}
|
||||
|
||||
// build the where clause for resource table
|
||||
resourceSubQuery, err := resource.BuildResourceSubQuery(DB_NAME, DISTRIBUTED_LOGS_V2_RESOURCE, bucketStart, bucketEnd, mq.Filters, mq.GroupBy, mq.AggregateAttribute, false)
|
||||
resourceSubQuery, err := resource.BuildResourceSubQuery(DB_NAME, DISTRIBUTED_LOGS_V2_RESOURCE, bucketStart, bucketEnd, mq.Filters, mq.GroupBy, mq.AggregateAttribute, false, false)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
@@ -475,7 +475,7 @@ func buildLogsLiveTailQuery(mq *v3.BuilderQuery) (string, error) {
|
||||
}
|
||||
|
||||
// no values for bucket start and end
|
||||
resourceSubQuery, err := resource.BuildResourceSubQuery(DB_NAME, DISTRIBUTED_LOGS_V2_RESOURCE, 0, 0, mq.Filters, mq.GroupBy, mq.AggregateAttribute, true)
|
||||
resourceSubQuery, err := resource.BuildResourceSubQuery(DB_NAME, DISTRIBUTED_LOGS_V2_RESOURCE, 0, 0, mq.Filters, mq.GroupBy, mq.AggregateAttribute, true, false)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
|
||||
@@ -6,6 +6,9 @@ import (
|
||||
|
||||
v3 "github.com/SigNoz/signoz/pkg/query-service/model/v3"
|
||||
"github.com/SigNoz/signoz/pkg/query-service/utils"
|
||||
"github.com/SigNoz/signoz/pkg/querybuilder"
|
||||
"github.com/SigNoz/signoz/pkg/semconv"
|
||||
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
|
||||
)
|
||||
|
||||
var resourceLogOperators = map[v3.FilterOperator]string{
|
||||
@@ -30,22 +33,49 @@ var resourceLogOperators = map[v3.FilterOperator]string{
|
||||
}
|
||||
|
||||
// buildResourceFilter builds a clickhouse filter string for resource labels
|
||||
func buildResourceFilter(logsOp string, key string, op v3.FilterOperator, value interface{}) string {
|
||||
func buildResourceFilter(logsOp string, key string, op v3.FilterOperator, value interface{}, members []string) string {
|
||||
// for all operators except contains and like
|
||||
searchKey := fmt.Sprintf("simpleJSONExtractString(labels, '%s')", key)
|
||||
searchKey := fmt.Sprintf("simpleJSONExtractString(labels, %s)", querybuilder.ClickHouseStringLiteral(key))
|
||||
if len(members) > 1 {
|
||||
values := make([]string, 0, len(members))
|
||||
for _, member := range members {
|
||||
values = append(values, fmt.Sprintf("NULLIF(simpleJSONExtractString(labels, %s), '')", querybuilder.ClickHouseStringLiteral(member)))
|
||||
}
|
||||
searchKey = "COALESCE(" + strings.Join(values, ", ") + ", '')"
|
||||
}
|
||||
|
||||
// for contains and like it will be case insensitive
|
||||
lowerSearchKey := fmt.Sprintf("simpleJSONExtractString(lower(labels), '%s')", key)
|
||||
lowerSearchKey := fmt.Sprintf("simpleJSONExtractString(lower(labels), %s)", querybuilder.ClickHouseStringLiteral(key))
|
||||
if len(members) > 1 {
|
||||
lowerSearchKey = "lower(" + searchKey + ")"
|
||||
}
|
||||
|
||||
chFmtVal := utils.ClickHouseFormattedValue(value)
|
||||
|
||||
lowerValue := strings.ToLower(fmt.Sprintf("%s", value))
|
||||
|
||||
switch op {
|
||||
case v3.FilterOperatorExists:
|
||||
return fmt.Sprintf("simpleJSONHas(labels, '%s')", key)
|
||||
case v3.FilterOperatorNotExists:
|
||||
return fmt.Sprintf("not simpleJSONHas(labels, '%s')", key)
|
||||
case v3.FilterOperatorExists, v3.FilterOperatorNotExists:
|
||||
exists := op == v3.FilterOperatorExists
|
||||
if len(members) == 1 {
|
||||
if exists {
|
||||
return fmt.Sprintf("simpleJSONHas(labels, %s)", querybuilder.ClickHouseStringLiteral(key))
|
||||
}
|
||||
return fmt.Sprintf("not simpleJSONHas(labels, %s)", querybuilder.ClickHouseStringLiteral(key))
|
||||
}
|
||||
presence := make([]string, 0, len(members))
|
||||
for _, member := range members {
|
||||
if exists {
|
||||
presence = append(presence, fmt.Sprintf("simpleJSONHas(labels, %s)", querybuilder.ClickHouseStringLiteral(member)))
|
||||
} else {
|
||||
presence = append(presence, fmt.Sprintf("not simpleJSONHas(labels, %s)", querybuilder.ClickHouseStringLiteral(member)))
|
||||
}
|
||||
}
|
||||
separator := " OR "
|
||||
if !exists {
|
||||
separator = " AND "
|
||||
}
|
||||
return "(" + strings.Join(presence, separator) + ")"
|
||||
case v3.FilterOperatorRegex, v3.FilterOperatorNotRegex:
|
||||
return fmt.Sprintf(logsOp, searchKey, chFmtVal)
|
||||
case v3.FilterOperatorContains, v3.FilterOperatorNotContains:
|
||||
@@ -93,9 +123,10 @@ func buildIndexFilterForInOperator(key string, op v3.FilterOperator, value inter
|
||||
|
||||
// if there are no values to filter on, return an empty string
|
||||
if len(values) > 0 {
|
||||
escapedKey := utils.QuoteEscapedStringForContains(key, true)
|
||||
for _, v := range values {
|
||||
value := utils.QuoteEscapedStringForContains(v, true)
|
||||
conditions = append(conditions, fmt.Sprintf("labels %s '%%\"%s\":\"%s\"%%'", sqlOp, key, value))
|
||||
conditions = append(conditions, fmt.Sprintf("labels %s '%%\"%s\":\"%s\"%%'", sqlOp, escapedKey, value))
|
||||
}
|
||||
return "(" + strings.Join(conditions, separator) + ")"
|
||||
}
|
||||
@@ -109,8 +140,34 @@ func buildIndexFilterForInOperator(key string, op v3.FilterOperator, value inter
|
||||
// for like/contains we will use lower index
|
||||
// we can use lower index for =, in etc but it's difficult to do it for !=, NIN etc
|
||||
// if as x != "ABC" we cannot predict something like "not lower(labels) like '%%x%%abc%%'". It has it be "not lower(labels) like '%%x%%ABC%%'"
|
||||
func buildResourceIndexFilter(key string, op v3.FilterOperator, value interface{}) string {
|
||||
func buildResourceIndexFilter(key string, op v3.FilterOperator, value interface{}, members []string) string {
|
||||
if len(members) > 1 {
|
||||
// A negated hint would drop rows where another member holds the value.
|
||||
switch op {
|
||||
case v3.FilterOperatorNotEqual,
|
||||
v3.FilterOperatorNotLike,
|
||||
v3.FilterOperatorNotILike,
|
||||
v3.FilterOperatorNotContains,
|
||||
v3.FilterOperatorNotExists,
|
||||
v3.FilterOperatorNotRegex,
|
||||
v3.FilterOperatorNotIn:
|
||||
return ""
|
||||
}
|
||||
|
||||
conditions := make([]string, 0, len(members))
|
||||
for _, member := range members {
|
||||
if condition := buildResourceIndexFilter(member, op, value, []string{member}); condition != "" {
|
||||
conditions = append(conditions, condition)
|
||||
}
|
||||
}
|
||||
if len(conditions) == 0 {
|
||||
return ""
|
||||
}
|
||||
return "(" + strings.Join(conditions, " OR ") + ")"
|
||||
}
|
||||
|
||||
// not using clickhouseFormattedValue as we don't wan't the quotes
|
||||
escapedKey := utils.QuoteEscapedStringForContains(key, true)
|
||||
strVal := fmt.Sprintf("%s", value)
|
||||
fmtValEscapedForContains := utils.QuoteEscapedStringForContains(strVal, true)
|
||||
fmtValEscapedForContainsLower := strings.ToLower(fmtValEscapedForContains)
|
||||
@@ -119,36 +176,36 @@ func buildResourceIndexFilter(key string, op v3.FilterOperator, value interface{
|
||||
// add index filters
|
||||
switch op {
|
||||
case v3.FilterOperatorEqual:
|
||||
return fmt.Sprintf("labels like '%%%s\":\"%s%%'", key, fmtValEscapedForContains)
|
||||
return fmt.Sprintf("labels like '%%%s\":\"%s%%'", escapedKey, fmtValEscapedForContains)
|
||||
case v3.FilterOperatorNotEqual:
|
||||
return fmt.Sprintf("labels not like '%%%s\":\"%s%%'", key, fmtValEscapedForContains)
|
||||
return fmt.Sprintf("labels not like '%%%s\":\"%s%%'", escapedKey, fmtValEscapedForContains)
|
||||
case v3.FilterOperatorLike, v3.FilterOperatorILike:
|
||||
return fmt.Sprintf("lower(labels) like '%%%s%%%s%%'", key, fmtValEscapedLower)
|
||||
return fmt.Sprintf("lower(labels) like '%%%s%%%s%%'", escapedKey, fmtValEscapedLower)
|
||||
case v3.FilterOperatorNotLike, v3.FilterOperatorNotILike:
|
||||
// cannot apply not contains x%y as y can be somewhere else
|
||||
return ""
|
||||
case v3.FilterOperatorContains:
|
||||
return fmt.Sprintf("lower(labels) like '%%%s%%%s%%'", key, fmtValEscapedForContainsLower)
|
||||
return fmt.Sprintf("lower(labels) like '%%%s%%%s%%'", escapedKey, fmtValEscapedForContainsLower)
|
||||
case v3.FilterOperatorNotContains:
|
||||
// cannot apply not contains x%y as y can be somewhere else
|
||||
return ""
|
||||
case v3.FilterOperatorExists:
|
||||
return fmt.Sprintf("lower(labels) like '%%%s%%'", key)
|
||||
return fmt.Sprintf("lower(labels) like '%%%s%%'", escapedKey)
|
||||
case v3.FilterOperatorNotExists:
|
||||
return fmt.Sprintf("lower(labels) not like '%%%s%%'", key)
|
||||
return fmt.Sprintf("lower(labels) not like '%%%s%%'", escapedKey)
|
||||
case v3.FilterOperatorRegex, v3.FilterOperatorNotRegex:
|
||||
// don't try to do anything for regex.
|
||||
return ""
|
||||
case v3.FilterOperatorIn, v3.FilterOperatorNotIn:
|
||||
return buildIndexFilterForInOperator(key, op, value)
|
||||
default:
|
||||
return fmt.Sprintf("labels like '%%%s%%'", key)
|
||||
return fmt.Sprintf("labels like '%%%s%%'", escapedKey)
|
||||
}
|
||||
}
|
||||
|
||||
// buildResourceFiltersFromFilterItems builds a list of clickhouse filter strings for resource labels from a FilterSet.
|
||||
// It skips any filter items that are not resource attributes and checks that the operator is supported and the data type is correct.
|
||||
func buildResourceFiltersFromFilterItems(fs *v3.FilterSet) ([]string, error) {
|
||||
func buildResourceFiltersFromFilterItems(fs *v3.FilterSet, resolveSemconvFamilies bool) ([]string, error) {
|
||||
var conditions []string
|
||||
if fs == nil || len(fs.Items) == 0 {
|
||||
return nil, nil
|
||||
@@ -182,12 +239,20 @@ func buildResourceFiltersFromFilterItems(fs *v3.FilterSet) ([]string, error) {
|
||||
}
|
||||
|
||||
if logsOp, ok := resourceLogOperators[op]; ok {
|
||||
members := []string{keyName}
|
||||
if resolveSemconvFamilies {
|
||||
members = semconv.Members(semconv.KindAttribute, telemetrytypes.FieldKeySelector{
|
||||
Name: keyName,
|
||||
Signal: telemetrytypes.SignalTraces,
|
||||
FieldContext: telemetrytypes.FieldContextResource,
|
||||
})
|
||||
}
|
||||
// the filter
|
||||
if resourceFilter := buildResourceFilter(logsOp, keyName, op, value); resourceFilter != "" {
|
||||
if resourceFilter := buildResourceFilter(logsOp, keyName, op, value, members); resourceFilter != "" {
|
||||
conditions = append(conditions, resourceFilter)
|
||||
}
|
||||
// the additional filter for better usage of the index
|
||||
if resourceIndexFilter := buildResourceIndexFilter(keyName, op, value); resourceIndexFilter != "" {
|
||||
if resourceIndexFilter := buildResourceIndexFilter(keyName, op, value, members); resourceIndexFilter != "" {
|
||||
conditions = append(conditions, resourceIndexFilter)
|
||||
}
|
||||
} else {
|
||||
@@ -219,12 +284,12 @@ func buildResourceFiltersFromAggregateAttribute(aggregateAttribute v3.AttributeK
|
||||
return ""
|
||||
}
|
||||
|
||||
func BuildResourceSubQuery(dbName, tableName string, bucketStart, bucketEnd int64, fs *v3.FilterSet, groupBy []v3.AttributeKey, aggregateAttribute v3.AttributeKey, isLiveTail bool) (string, error) {
|
||||
func BuildResourceSubQuery(dbName, tableName string, bucketStart, bucketEnd int64, fs *v3.FilterSet, groupBy []v3.AttributeKey, aggregateAttribute v3.AttributeKey, isLiveTail bool, resolveSemconvFamilies bool) (string, error) {
|
||||
|
||||
// BUILD THE WHERE CLAUSE
|
||||
var conditions []string
|
||||
// only add the resource attributes to the filters here
|
||||
rs, err := buildResourceFiltersFromFilterItems(fs)
|
||||
rs, err := buildResourceFiltersFromFilterItems(fs, resolveSemconvFamilies)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
|
||||
@@ -5,6 +5,7 @@ import (
|
||||
"testing"
|
||||
|
||||
v3 "github.com/SigNoz/signoz/pkg/query-service/model/v3"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
func Test_buildResourceFilter(t *testing.T) {
|
||||
@@ -88,7 +89,7 @@ func Test_buildResourceFilter(t *testing.T) {
|
||||
}
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
if got := buildResourceFilter(tt.args.logsOp, tt.args.key, tt.args.op, tt.args.value); got != tt.want {
|
||||
if got := buildResourceFilter(tt.args.logsOp, tt.args.key, tt.args.op, tt.args.value, []string{tt.args.key}); got != tt.want {
|
||||
t.Errorf("buildResourceFilter() = %v, want %v", got, tt.want)
|
||||
}
|
||||
})
|
||||
@@ -282,7 +283,7 @@ func Test_buildResourceIndexFilter(t *testing.T) {
|
||||
}
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
if got := buildResourceIndexFilter(tt.args.key, tt.args.op, tt.args.value); got != tt.want {
|
||||
if got := buildResourceIndexFilter(tt.args.key, tt.args.op, tt.args.value, []string{tt.args.key}); got != tt.want {
|
||||
t.Errorf("buildResourceIndexFilter() = %v, want %v", got, tt.want)
|
||||
}
|
||||
})
|
||||
@@ -379,7 +380,7 @@ func Test_buildResourceFiltersFromFilterItems(t *testing.T) {
|
||||
}
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
got, err := buildResourceFiltersFromFilterItems(tt.args.fs)
|
||||
got, err := buildResourceFiltersFromFilterItems(tt.args.fs, false)
|
||||
if (err != nil) != tt.wantErr {
|
||||
t.Errorf("buildResourceFiltersFromFilterItems() error = %v, wantErr %v", err, tt.wantErr)
|
||||
return
|
||||
@@ -541,7 +542,7 @@ func Test_buildResourceSubQuery(t *testing.T) {
|
||||
}
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
got, err := BuildResourceSubQuery("signoz_logs", "distributed_logs_v2_resource", tt.args.bucketStart, tt.args.bucketEnd, tt.args.fs, tt.args.groupBy, tt.args.aggregateAttribute, false)
|
||||
got, err := BuildResourceSubQuery("signoz_logs", "distributed_logs_v2_resource", tt.args.bucketStart, tt.args.bucketEnd, tt.args.fs, tt.args.groupBy, tt.args.aggregateAttribute, false, false)
|
||||
if (err != nil) != tt.wantErr {
|
||||
t.Errorf("buildResourceSubQuery() error = %v, wantErr %v", err, tt.wantErr)
|
||||
return
|
||||
@@ -552,3 +553,58 @@ func Test_buildResourceSubQuery(t *testing.T) {
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func Test_buildResourceFilterFamily(t *testing.T) {
|
||||
members := []string{"deployment.environment.name", "deployment.environment"}
|
||||
|
||||
require.Equal(t,
|
||||
"COALESCE(NULLIF(simpleJSONExtractString(labels, 'deployment.environment.name'), ''), NULLIF(simpleJSONExtractString(labels, 'deployment.environment'), ''), '') = 'production'",
|
||||
buildResourceFilter("=", "deployment.environment.name", v3.FilterOperatorEqual, "production", members))
|
||||
|
||||
require.Equal(t,
|
||||
"COALESCE(NULLIF(simpleJSONExtractString(labels, 'deployment.environment.name'), ''), NULLIF(simpleJSONExtractString(labels, 'deployment.environment'), ''), '') != 'production'",
|
||||
buildResourceFilter("!=", "deployment.environment.name", v3.FilterOperatorNotEqual, "production", members))
|
||||
|
||||
require.Equal(t,
|
||||
"(simpleJSONHas(labels, 'deployment.environment.name') OR simpleJSONHas(labels, 'deployment.environment'))",
|
||||
buildResourceFilter("", "deployment.environment.name", v3.FilterOperatorExists, nil, members))
|
||||
|
||||
require.Equal(t,
|
||||
"(not simpleJSONHas(labels, 'deployment.environment.name') AND not simpleJSONHas(labels, 'deployment.environment'))",
|
||||
buildResourceFilter("", "deployment.environment.name", v3.FilterOperatorNotExists, nil, members))
|
||||
}
|
||||
|
||||
func Test_buildResourceIndexFilterFamily(t *testing.T) {
|
||||
members := []string{"deployment.environment.name", "deployment.environment"}
|
||||
|
||||
require.Equal(t,
|
||||
`(labels like '%deployment.environment.name":"production%' OR labels like '%deployment.environment":"production%')`,
|
||||
buildResourceIndexFilter("deployment.environment.name", v3.FilterOperatorEqual, "production", members))
|
||||
|
||||
require.Equal(t, "",
|
||||
buildResourceIndexFilter("deployment.environment.name", v3.FilterOperatorNotEqual, "production", members))
|
||||
require.Equal(t, "",
|
||||
buildResourceIndexFilter("deployment.environment.name", v3.FilterOperatorNotIn, []interface{}{"production"}, members))
|
||||
}
|
||||
|
||||
func TestBuildResourceSubQueryFamily(t *testing.T) {
|
||||
fs := &v3.FilterSet{Items: []v3.FilterItem{{
|
||||
Key: v3.AttributeKey{
|
||||
Key: "deployment.environment.name",
|
||||
DataType: v3.AttributeKeyDataTypeString,
|
||||
Type: v3.AttributeKeyTypeResource,
|
||||
},
|
||||
Operator: v3.FilterOperatorEqual,
|
||||
Value: "production",
|
||||
}}}
|
||||
|
||||
familyOn, err := BuildResourceSubQuery("signoz_traces", "distributed_traces_v3_resource", 1, 2, fs, nil, v3.AttributeKey{}, false, true)
|
||||
require.NoError(t, err)
|
||||
require.Contains(t, familyOn, "COALESCE(NULLIF(simpleJSONExtractString(labels, 'deployment.environment.name'), ''), NULLIF(simpleJSONExtractString(labels, 'deployment.environment'), ''), '') = 'production'")
|
||||
require.Contains(t, familyOn, `(labels like '%deployment.environment.name":"production%' OR labels like '%deployment.environment":"production%')`)
|
||||
|
||||
familyOff, err := BuildResourceSubQuery("signoz_traces", "distributed_traces_v3_resource", 1, 2, fs, nil, v3.AttributeKey{}, false, false)
|
||||
require.NoError(t, err)
|
||||
require.Contains(t, familyOff, "simpleJSONExtractString(labels, 'deployment.environment.name') = 'production'")
|
||||
require.NotContains(t, familyOff, "COALESCE")
|
||||
}
|
||||
|
||||
@@ -6,17 +6,25 @@ import (
|
||||
|
||||
"github.com/ClickHouse/clickhouse-go/v2"
|
||||
"github.com/SigNoz/signoz/pkg/query-service/model"
|
||||
"github.com/SigNoz/signoz/pkg/semconv"
|
||||
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
|
||||
)
|
||||
|
||||
var (
|
||||
columns = map[string]struct{}{
|
||||
"deployment_environment": {},
|
||||
"k8s_cluster_name": {},
|
||||
"k8s_namespace_name": {},
|
||||
func BuildServiceMapQuery(tags []model.TagQuery, resolveSemconvFamilies bool) (string, []interface{}) {
|
||||
columns := map[string]string{
|
||||
"deployment_environment": "deployment_environment",
|
||||
"k8s_cluster_name": "k8s_cluster_name",
|
||||
"k8s_namespace_name": "k8s_namespace_name",
|
||||
}
|
||||
if resolveSemconvFamilies {
|
||||
for _, member := range semconv.Members(semconv.KindAttribute, telemetrytypes.FieldKeySelector{
|
||||
Name: "deployment.environment.name",
|
||||
Signal: telemetrytypes.SignalTraces,
|
||||
FieldContext: telemetrytypes.FieldContextResource,
|
||||
}) {
|
||||
columns[strings.ReplaceAll(member, ".", "_")] = "deployment_environment"
|
||||
}
|
||||
}
|
||||
)
|
||||
|
||||
func BuildServiceMapQuery(tags []model.TagQuery) (string, []interface{}) {
|
||||
var filterQuery string
|
||||
var namedArgs []interface{}
|
||||
for _, tag := range tags {
|
||||
@@ -24,39 +32,40 @@ func BuildServiceMapQuery(tags []model.TagQuery) (string, []interface{}) {
|
||||
operator := tag.GetOperator()
|
||||
value := tag.GetValues()
|
||||
|
||||
if _, ok := columns[key]; !ok {
|
||||
column, ok := columns[key]
|
||||
if !ok {
|
||||
continue
|
||||
}
|
||||
|
||||
switch operator {
|
||||
case model.InOperator:
|
||||
filterQuery += fmt.Sprintf(" AND %s IN @%s", key, key)
|
||||
filterQuery += fmt.Sprintf(" AND %s IN @%s", column, key)
|
||||
namedArgs = append(namedArgs, clickhouse.Named(key, value))
|
||||
case model.NotInOperator:
|
||||
filterQuery += fmt.Sprintf(" AND %s NOT IN @%s", key, key)
|
||||
filterQuery += fmt.Sprintf(" AND %s NOT IN @%s", column, key)
|
||||
namedArgs = append(namedArgs, clickhouse.Named(key, value))
|
||||
case model.EqualOperator:
|
||||
filterQuery += fmt.Sprintf(" AND %s = @%s", key, key)
|
||||
filterQuery += fmt.Sprintf(" AND %s = @%s", column, key)
|
||||
namedArgs = append(namedArgs, clickhouse.Named(key, value))
|
||||
case model.NotEqualOperator:
|
||||
filterQuery += fmt.Sprintf(" AND %s != @%s", key, key)
|
||||
filterQuery += fmt.Sprintf(" AND %s != @%s", column, key)
|
||||
namedArgs = append(namedArgs, clickhouse.Named(key, value))
|
||||
case model.ContainsOperator:
|
||||
filterQuery += fmt.Sprintf(" AND %s LIKE @%s", key, key)
|
||||
filterQuery += fmt.Sprintf(" AND %s LIKE @%s", column, key)
|
||||
namedArgs = append(namedArgs, clickhouse.Named(key, fmt.Sprintf("%%%s%%", value)))
|
||||
case model.NotContainsOperator:
|
||||
filterQuery += fmt.Sprintf(" AND %s NOT LIKE @%s", key, key)
|
||||
filterQuery += fmt.Sprintf(" AND %s NOT LIKE @%s", column, key)
|
||||
namedArgs = append(namedArgs, clickhouse.Named(key, fmt.Sprintf("%%%s%%", value)))
|
||||
case model.StartsWithOperator:
|
||||
filterQuery += fmt.Sprintf(" AND %s LIKE @%s", key, key)
|
||||
filterQuery += fmt.Sprintf(" AND %s LIKE @%s", column, key)
|
||||
namedArgs = append(namedArgs, clickhouse.Named(key, fmt.Sprintf("%s%%", value)))
|
||||
case model.NotStartsWithOperator:
|
||||
filterQuery += fmt.Sprintf(" AND %s NOT LIKE @%s", key, key)
|
||||
filterQuery += fmt.Sprintf(" AND %s NOT LIKE @%s", column, key)
|
||||
namedArgs = append(namedArgs, clickhouse.Named(key, fmt.Sprintf("%s%%", value)))
|
||||
case model.ExistsOperator:
|
||||
filterQuery += fmt.Sprintf(" AND %s IS NOT NULL", key)
|
||||
filterQuery += fmt.Sprintf(" AND %s IS NOT NULL", column)
|
||||
case model.NotExistsOperator:
|
||||
filterQuery += fmt.Sprintf(" AND %s IS NULL", key)
|
||||
filterQuery += fmt.Sprintf(" AND %s IS NULL", column)
|
||||
}
|
||||
}
|
||||
return filterQuery, namedArgs
|
||||
|
||||
37
pkg/query-service/app/services/map_test.go
Normal file
37
pkg/query-service/app/services/map_test.go
Normal file
@@ -0,0 +1,37 @@
|
||||
package services
|
||||
|
||||
import (
|
||||
"testing"
|
||||
|
||||
"github.com/SigNoz/signoz/pkg/query-service/model"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
func TestBuildServiceMapQueryFamily(t *testing.T) {
|
||||
newSpelling := []model.TagQuery{model.NewTagQueryString(model.TagQueryParam{
|
||||
Key: "deployment.environment.name",
|
||||
StringValues: []string{"production"},
|
||||
Operator: model.EqualOperator,
|
||||
})}
|
||||
oldSpelling := []model.TagQuery{model.NewTagQueryString(model.TagQueryParam{
|
||||
Key: "deployment.environment",
|
||||
StringValues: []string{"production"},
|
||||
Operator: model.EqualOperator,
|
||||
})}
|
||||
|
||||
query, args := BuildServiceMapQuery(newSpelling, true)
|
||||
require.Equal(t, " AND deployment_environment = @deployment_environment_name", query)
|
||||
require.Len(t, args, 1)
|
||||
|
||||
query, args = BuildServiceMapQuery(oldSpelling, true)
|
||||
require.Equal(t, " AND deployment_environment = @deployment_environment", query)
|
||||
require.Len(t, args, 1)
|
||||
|
||||
query, args = BuildServiceMapQuery(newSpelling, false)
|
||||
require.Equal(t, "", query)
|
||||
require.Empty(t, args)
|
||||
|
||||
query, args = BuildServiceMapQuery(oldSpelling, false)
|
||||
require.Equal(t, " AND deployment_environment = @deployment_environment", query)
|
||||
require.Len(t, args, 1)
|
||||
}
|
||||
@@ -282,7 +282,7 @@ func buildTracesQuery(start, end, step int64, mq *v3.BuilderQuery, panelType v3.
|
||||
filterSubQuery = filterSubQuery + " AND " + emptyValuesInGroupByFilter
|
||||
}
|
||||
|
||||
resourceSubQuery, err := resource.BuildResourceSubQuery("signoz_traces", "distributed_traces_v3_resource", bucketStart, bucketEnd, mq.Filters, mq.GroupBy, mq.AggregateAttribute, false)
|
||||
resourceSubQuery, err := resource.BuildResourceSubQuery("signoz_traces", "distributed_traces_v3_resource", bucketStart, bucketEnd, mq.Filters, mq.GroupBy, mq.AggregateAttribute, false, false)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
|
||||
@@ -17,12 +17,12 @@ type Reader interface {
|
||||
GetInstantQueryMetricsResult(ctx context.Context, query *model.InstantQueryMetricsParams) (*promql.Result, *stats.QueryStats, *model.ApiError)
|
||||
GetQueryRangeResult(ctx context.Context, query *model.QueryRangeParams) (*promql.Result, *stats.QueryStats, *model.ApiError)
|
||||
GetTopLevelOperations(ctx context.Context, start, end time.Time, services []string) (*map[string][]string, *model.ApiError)
|
||||
GetEntryPointOperations(ctx context.Context, query *model.GetTopOperationsParams) (*[]model.TopOperationsItem, error)
|
||||
GetServices(ctx context.Context, query *model.GetServicesParams) (*[]model.ServiceItem, *model.ApiError)
|
||||
GetTopOperations(ctx context.Context, query *model.GetTopOperationsParams) (*[]model.TopOperationsItem, *model.ApiError)
|
||||
GetEntryPointOperations(ctx context.Context, orgID valuer.UUID, query *model.GetTopOperationsParams) (*[]model.TopOperationsItem, error)
|
||||
GetServices(ctx context.Context, orgID valuer.UUID, query *model.GetServicesParams) (*[]model.ServiceItem, *model.ApiError)
|
||||
GetTopOperations(ctx context.Context, orgID valuer.UUID, query *model.GetTopOperationsParams) (*[]model.TopOperationsItem, *model.ApiError)
|
||||
GetUsage(ctx context.Context, query *model.GetUsageParams) (*[]model.UsageItem, error)
|
||||
GetServicesList(ctx context.Context) (*[]string, error)
|
||||
GetDependencyGraph(ctx context.Context, query *model.GetServicesParams) (*[]model.ServiceMapDependencyResponseItem, error)
|
||||
GetDependencyGraph(ctx context.Context, orgID valuer.UUID, query *model.GetServicesParams) (*[]model.ServiceMapDependencyResponseItem, error)
|
||||
|
||||
GetTTL(ctx context.Context, orgID string, ttlParams *retentiontypes.GetTTLParams) (*retentiontypes.GetTTLResponseItem, *model.ApiError)
|
||||
GetCustomRetentionTTL(ctx context.Context, orgID string) (*retentiontypes.GetCustomRetentionTTLResponse, error)
|
||||
|
||||
Reference in New Issue
Block a user