feat: support environment filtering in service map (#2481)

pkg/query-service/app/clickhouseReader/options.go

@@ -26,7 +26,7 @@ const (
 	defaultDurationTable           string        = "distributed_durationSort"
 	defaultUsageExplorerTable      string        = "distributed_usage_explorer"
 	defaultSpansTable              string        = "distributed_signoz_spans"
-	defaultDependencyGraphTable    string        = "distributed_dependency_graph_minutes"
+	defaultDependencyGraphTable    string        = "distributed_dependency_graph_minutes_v2"
 	defaultTopLevelOperationsTable string        = "distributed_top_level_operations"
 	defaultLogsDB                  string        = "signoz_logs"
 	defaultLogsTable               string        = "distributed_logs"

pkg/query-service/app/clickhouseReader/reader.go

@@ -41,6 +41,7 @@ import (
 
 	promModel "github.com/prometheus/common/model"
 	"go.signoz.io/signoz/pkg/query-service/app/logs"
+	"go.signoz.io/signoz/pkg/query-service/app/services"
 	"go.signoz.io/signoz/pkg/query-service/constants"
 	am "go.signoz.io/signoz/pkg/query-service/integrations/alertManager"
 	"go.signoz.io/signoz/pkg/query-service/interfaces"
@@ -1996,20 +1997,22 @@ func (r *ClickHouseReader) GetDependencyGraph(ctx context.Context, queryParams *
 			sum(total_count)/ @duration AS callRate,
 			sum(error_count)/sum(total_count) * 100 as errorRate
 		FROM %s.%s
-		WHERE toUInt64(toDateTime(timestamp)) >= @start AND toUInt64(toDateTime(timestamp)) <= @end
+		WHERE toUInt64(toDateTime(timestamp)) >= @start AND toUInt64(toDateTime(timestamp)) <= @end`,
-		GROUP BY
-			src,
-			dest`,
 		r.TraceDB, r.dependencyGraphTable,
 	)
 
+	tags := createTagQueryFromTagQueryParams(queryParams.Tags)
+	filterQuery, filterArgs := services.BuildServiceMapQuery(tags)
+	query += filterQuery + " GROUP BY src, dest;"
+	args = append(args, filterArgs...)
+
 	zap.S().Debug(query, args)
 
 	err := r.db.Select(ctx, &response, query, args...)
 
 	if err != nil {
 		zap.S().Error("Error in processing sql query: ", err)
-		return nil, fmt.Errorf("Error in processing sql query")
+		return nil, fmt.Errorf("error in processing sql query %w", err)
 	}
 
 	return &response, nil

pkg/query-service/app/services/map.go

@@ -0,0 +1,63 @@
+package services
+
+import (
+	"fmt"
+	"strings"
+
+	"github.com/ClickHouse/clickhouse-go/v2"
+	"go.signoz.io/signoz/pkg/query-service/model"
+)
+
+var (
+	columns = map[string]struct{}{
+		"deployment_environment": {},
+		"k8s_cluster_name":       {},
+		"k8s_namespace_name":     {},
+	}
+)
+
+func BuildServiceMapQuery(tags []model.TagQuery) (string, []interface{}) {
+	var filterQuery string
+	var namedArgs []interface{}
+	for _, tag := range tags {
+		key := strings.ReplaceAll(tag.GetKey(), ".", "_")
+		operator := tag.GetOperator()
+		value := tag.GetValues()
+
+		if _, ok := columns[key]; !ok {
+			continue
+		}
+
+		switch operator {
+		case model.InOperator:
+			filterQuery += fmt.Sprintf(" AND %s IN @%s", key, key)
+			namedArgs = append(namedArgs, clickhouse.Named(key, value))
+		case model.NotInOperator:
+			filterQuery += fmt.Sprintf(" AND %s NOT IN @%s", key, key)
+			namedArgs = append(namedArgs, clickhouse.Named(key, value))
+		case model.EqualOperator:
+			filterQuery += fmt.Sprintf(" AND %s = @%s", key, key)
+			namedArgs = append(namedArgs, clickhouse.Named(key, value))
+		case model.NotEqualOperator:
+			filterQuery += fmt.Sprintf(" AND %s != @%s", key, key)
+			namedArgs = append(namedArgs, clickhouse.Named(key, value))
+		case model.ContainsOperator:
+			filterQuery += fmt.Sprintf(" AND %s LIKE @%s", key, key)
+			namedArgs = append(namedArgs, clickhouse.Named(key, fmt.Sprintf("%%%s%%", value)))
+		case model.NotContainsOperator:
+			filterQuery += fmt.Sprintf(" AND %s NOT LIKE @%s", key, key)
+			namedArgs = append(namedArgs, clickhouse.Named(key, fmt.Sprintf("%%%s%%", value)))
+		case model.StartsWithOperator:
+			filterQuery += fmt.Sprintf(" AND %s LIKE @%s", key, key)
+			namedArgs = append(namedArgs, clickhouse.Named(key, fmt.Sprintf("%s%%", value)))
+		case model.NotStartsWithOperator:
+			filterQuery += fmt.Sprintf(" AND %s NOT LIKE @%s", key, key)
+			namedArgs = append(namedArgs, clickhouse.Named(key, fmt.Sprintf("%s%%", value)))
+		case model.ExistsOperator:
+			filterQuery += fmt.Sprintf(" AND %s IS NOT NULL", key)
+		case model.NotExistsOperator:
+			filterQuery += fmt.Sprintf(" AND %s IS NULL", key)
+		}
+	}
+	return filterQuery, namedArgs
+}