fix: Modifies messaging queue paylod (#6783)

* fix: use filterset

Signed-off-by: Shivanshu Raj Shrivastava <[email protected]>

pkg/query-service/app/integrations/messagingQueues/celery/translator.go

@@ -0,0 +1 @@
+package celery

pkg/query-service/app/integrations/messagingQueues/kafka/model.go

@@ -22,37 +22,3 @@ type OnboardingResponse struct {
 	Message   string `json:"error_message"`
 	Status    string `json:"status"`
 }
-
-// QueueFilters
-// ToDo: add capability of dynamic filtering based on any of the filters
-type QueueFilters struct {
-	ServiceName []string
-	SpanName    []string
-	Queue       []string
-	Destination []string
-	Kind        []string
-}
-
-type CeleryTask struct {
-	kind   string
-	status string
-}
-
-type CeleryTasks interface {
-	GetKind() string
-	GetStatus() string
-	Set(string, string)
-}
-
-func (r *CeleryTask) GetKind() string {
-	return r.kind
-}
-
-func (r *CeleryTask) GetStatus() string {
-	return r.status
-}
-
-func (r *CeleryTask) Set(kind, status string) {
-	r.kind = kind
-	r.status = status
-}

pkg/query-service/app/integrations/messagingQueues/kafka/sql.go

@@ -2,7 +2,6 @@ package kafka
 
 import (
 	"fmt"
-	"strings"
 )
 
 func generateConsumerSQL(start, end int64, topic, partition, consumerGroup, queueType string) string {
@@ -319,139 +318,6 @@ GROUP BY
 	return query
 }
 
-// generateOverviewSQL builds the ClickHouse SQL query with optional filters.
-// If a filter slice is empty, the query does not constrain on that field.
-func generateOverviewSQL(start, end int64, filters *QueueFilters) string {
-	// Convert from nanoseconds to float seconds in Go to avoid decimal overflow in ClickHouse
-	startSeconds := float64(start) / 1e9
-	endSeconds := float64(end) / 1e9
-
-	// Compute time range difference in Go
-	timeRangeSecs := endSeconds - startSeconds
-
-	// Example ts_bucket boundaries (could be your own logic)
-	tsBucketStart := startSeconds - 1800
-	tsBucketEnd := endSeconds
-
-	// Build WHERE clauses for optional filters
-	// We always require messaging_system IN ('kafka', 'celery'), but
-	// we add additional AND conditions only if the slices are non-empty.
-	var whereClauses []string
-
-	// Mandatory base filter: show only kafka/celery
-	whereClauses = append(whereClauses, "messaging_system IN ('kafka', 'celery')")
-
-	if len(filters.ServiceName) > 0 {
-		whereClauses = append(whereClauses, inClause("service_name", filters.ServiceName))
-	}
-	if len(filters.SpanName) > 0 {
-		whereClauses = append(whereClauses, inClause("span_name", filters.SpanName))
-	}
-	if len(filters.Queue) > 0 {
-		// "queue" in the struct refers to the messaging_system in the DB
-		whereClauses = append(whereClauses, inClause("messaging_system", filters.Queue))
-	}
-	if len(filters.Destination) > 0 {
-		whereClauses = append(whereClauses, inClause("destination", filters.Destination))
-	}
-	if len(filters.Kind) > 0 {
-		whereClauses = append(whereClauses, inClause("kind_string", filters.Kind))
-	}
-
-	// Combine all WHERE clauses with AND
-	whereSQL := strings.Join(whereClauses, "\n    AND ")
-
-	if len(whereSQL) > 0 {
-		whereSQL = fmt.Sprintf("AND %s", whereSQL)
-	}
-
-	// Final query string
-	// Note the use of %f for float64 values in fmt.Sprintf
-	query := fmt.Sprintf(`
-WITH
-    processed_traces AS (
-        SELECT
-            resource_string_service$$name AS service_name,
-            name AS span_name,
-            CASE
-                WHEN attribute_string_messaging$$system != '' THEN attribute_string_messaging$$system
-                WHEN (has(attributes_string, 'celery.action') OR has(attributes_string, 'celery.task_name')) THEN 'celery'
-                ELSE 'undefined'
-            END AS messaging_system,
-            kind_string,
-            COALESCE(
-                NULLIF(attributes_string['messaging.destination.name'], ''),
-                NULLIF(attributes_string['messaging.destination'], '')
-            ) AS destination,
-            durationNano,
-            status_code
-        FROM signoz_traces.distributed_signoz_index_v3
-        WHERE
-            timestamp >= toDateTime64(%f, 9)
-            AND timestamp <= toDateTime64(%f, 9)
-            AND ts_bucket_start >= toDateTime64(%f, 9)
-            AND ts_bucket_start <= toDateTime64(%f, 9)
-            AND (
-                attribute_string_messaging$$system = 'kafka'
-                OR has(attributes_string, 'celery.action')
-                OR has(attributes_string, 'celery.task_name')
-            )
-		    %s
-    ),
-    aggregated_metrics AS (
-        SELECT
-            service_name,
-            span_name,
-            messaging_system,
-            destination,
-            kind_string,
-            count(*) AS total_count,
-            sumIf(1, status_code = 2) AS error_count,
-            quantile(0.95)(durationNano) / 1000000 AS p95_latency -- Convert to ms
-        FROM
-            processed_traces
-        GROUP BY
-            service_name,
-            span_name,
-            messaging_system,
-            destination,
-            kind_string
-    )
-SELECT
-    aggregated_metrics.service_name,
-    aggregated_metrics.span_name,
-    aggregated_metrics.messaging_system,
-    aggregated_metrics.destination,
-    aggregated_metrics.kind_string,
-    COALESCE(aggregated_metrics.total_count / %f, 0) AS throughput,
-    COALESCE((aggregated_metrics.error_count * 100.0) / aggregated_metrics.total_count, 0) AS error_percentage,
-    aggregated_metrics.p95_latency
-FROM
-    aggregated_metrics
-ORDER BY
-    aggregated_metrics.service_name,
-    aggregated_metrics.span_name;
-`,
-		startSeconds, endSeconds,
-		tsBucketStart, tsBucketEnd,
-		whereSQL, timeRangeSecs,
-	)
-
-	return query
-}
-
-// inClause returns SQL like "fieldName IN ('val1','val2','val3')"
-func inClause(fieldName string, values []string) string {
-	// Quote and escape each value for safety
-	var quoted []string
-	for _, v := range values {
-		// Simple escape: replace any single quotes in v
-		safeVal := strings.ReplaceAll(v, "'", "''")
-		quoted = append(quoted, fmt.Sprintf("'%s'", safeVal))
-	}
-	return fmt.Sprintf("%s IN (%s)", fieldName, strings.Join(quoted, ","))
-}
-
 func generateProducerSQL(start, end int64, topic, partition, queueType string) string {
 	timeRange := (end - start) / 1000000000
 	tsBucketStart := (start / 1000000000) - 1800

pkg/query-service/app/integrations/messagingQueues/kafka/translator.go

@@ -2,11 +2,9 @@ package kafka
 
 import (
 	"fmt"
-
 	"go.signoz.io/signoz/pkg/query-service/common"
 	"go.signoz.io/signoz/pkg/query-service/constants"
 	v3 "go.signoz.io/signoz/pkg/query-service/model/v3"
-	"strings"
 )
 
 var defaultStepInterval int64 = 60
@@ -21,6 +19,7 @@ func BuildQueryRangeParams(messagingQueue *MessagingQueue, queryContext string)
 	queueType := KafkaQueue
 
 	chq, err := BuildClickHouseQuery(messagingQueue, queueType, queryContext)
+
 	if err != nil {
 		return nil, err
 	}
@@ -321,34 +320,6 @@ func BuildQRParamsWithCache(
 	return queryRangeParams, err
 }
 
-func getFilters(variables map[string]string) *QueueFilters {
-	return &QueueFilters{
-		ServiceName: parseFilter(variables["service_name"]),
-		SpanName:    parseFilter(variables["span_name"]),
-		Queue:       parseFilter(variables["queue"]),
-		Destination: parseFilter(variables["destination"]),
-		Kind:        parseFilter(variables["kind"]),
-	}
-}
-
-// parseFilter splits a comma-separated string into a []string.
-// Returns an empty slice if the input is blank.
-func parseFilter(val string) []string {
-	if val == "" {
-		return []string{}
-	}
-	// Split on commas, trim whitespace around each part
-	parts := strings.Split(val, ",")
-	var out []string
-	for _, p := range parts {
-		trimmed := strings.TrimSpace(p)
-		if trimmed != "" {
-			out = append(out, trimmed)
-		}
-	}
-	return out
-}
-
 func BuildClickHouseQuery(
 	messagingQueue *MessagingQueue,
 	queueType string,
@@ -385,8 +356,6 @@ func BuildClickHouseQuery(
 	var query string
 
 	switch queryContext {
-	case "overview":
-		query = generateOverviewSQL(start, end, getFilters(messagingQueue.Variables))
 	case "producer":
 		query = generateProducerSQL(start, end, topic, partition, queueType)
 	case "consumer":

pkg/query-service/app/integrations/messagingQueues/queues/model.go

@@ -0,0 +1,27 @@
+package queues
+
+import (
+	"fmt"
+
+	v3 "go.signoz.io/signoz/pkg/query-service/model/v3"
+)
+
+type QueueListRequest struct {
+	Start   int64         `json:"start"` // unix nano
+	End     int64         `json:"end"`   // unix nano
+	Filters *v3.FilterSet `json:"filters"`
+	Limit   int           `json:"limit"`
+}
+
+func (qr *QueueListRequest) Validate() error {
+
+	err := qr.Filters.Validate()
+	if err != nil {
+		return err
+	}
+
+	if qr.Start < 0 || qr.End < 0 {
+		return fmt.Errorf("start and end must be unixnano time")
+	}
+	return nil
+}

pkg/query-service/app/integrations/messagingQueues/queues/queueOverview.go

@@ -0,0 +1,19 @@
+package queues
+
+import (
+	v3 "go.signoz.io/signoz/pkg/query-service/model/v3"
+)
+
+func BuildOverviewQuery(queueList *QueueListRequest) (*v3.ClickHouseQuery, error) {
+
+	err := queueList.Validate()
+	if err != nil {
+		return nil, err
+	}
+
+	query := generateOverviewSQL(queueList.Start, queueList.End, queueList.Filters.Items)
+
+	return &v3.ClickHouseQuery{
+		Query: query,
+	}, nil
+}