feat: Add celery-overview APIs

pkg/query-service/app/http_handler.go

@@ -4977,7 +4977,7 @@ func (aH *APIHandler) getQueueOverview(w http.ResponseWriter, r *http.Request) {
 
 	if err != nil {
 		zap.L().Error(err.Error())
-		RespondError(w, apiErr, nil)
+		RespondError(w, &model.ApiError{Typ: model.ErrorInternal, Err: fmt.Errorf("error building clickhouse query: %v", err)}, nil)
 		return
 	}
 
@@ -4987,7 +4987,68 @@ func (aH *APIHandler) getQueueOverview(w http.ResponseWriter, r *http.Request) {
 }
 
 func (aH *APIHandler) getCeleryOverview(w http.ResponseWriter, r *http.Request) {
-	// TODO: Implement celery overview logic for both worker and tasks types
+	messagingQueue, apiErr := ParseMessagingQueueBody(r)
+
+	if apiErr != nil {
+		zap.L().Error(apiErr.Err.Error())
+		RespondError(w, apiErr, nil)
+		return
+	}
+
+	filters, err := mq.GetCeleryFilters(messagingQueue.Variables)
+
+	if err != nil {
+		zap.L().Error(err.Error())
+		RespondError(w, &model.ApiError{Typ: model.ErrorInternal, Err: fmt.Errorf("error building filters: %v", err)}, nil)
+		return
+	}
+
+	queryRangeParams, err := mq.CeleryClickHouseQuery(messagingQueue, "celeryoverview", filters)
+
+	if err != nil {
+		zap.L().Error(err.Error())
+		RespondError(w, &model.ApiError{Typ: model.ErrorInternal, Err: fmt.Errorf("error building clickhouse query: %v", err)}, nil)
+		return
+	}
+	if err := validateQueryRangeParamsV3(queryRangeParams); err != nil {
+		zap.L().Error(err.Error())
+		RespondError(w, &model.ApiError{Typ: model.ErrorInternal, Err: fmt.Errorf("error in qyery range params: %v", err)}, nil)
+		return
+	}
+
+	resultFetchLatency, errQueriesByNameFetchLatency, err := aH.querierV2.QueryRange(r.Context(), queryRangeParams)
+
+	if err != nil {
+		apiErrObj := &model.ApiError{Typ: model.ErrorBadData, Err: err}
+		RespondError(w, apiErrObj, errQueriesByNameFetchLatency)
+		return
+	}
+
+	if filters.QueryFor[0] == "worker" {
+
+		var workerNames = make([]string, 0)
+		var workerCount int
+
+		for _, res := range resultFetchLatency {
+			for _, series := range res.Series {
+				if workerName, ok := series.Labels["worker"]; ok {
+					workerNames = append(workerNames, workerName)
+					workerCount++
+				}
+			}
+		}
+
+		aH.Respond(w, mq.WorkerResponse{
+			Count: workerCount,
+			Names: workerNames,
+		})
+		return
+	}
+
+	resp := v3.QueryRangeResponse{
+		Result: resultFetchLatency,
+	}
+	aH.Respond(w, resp)
 }
 
 func (aH *APIHandler) getCeleryTasks(w http.ResponseWriter, r *http.Request) {

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

@@ -23,6 +23,11 @@ type OnboardingResponse struct {
 	Status    string `json:"status"`
 }
 
+type WorkerResponse struct {
+	Count int      `json:"active_workers"`
+	Names []string `json:"worker_names"`
+}
+
 // QueueFilters
 // ToDo: add capability of dynamic filtering based on any of the filters
 type QueueFilters struct {
@@ -31,17 +36,23 @@ type QueueFilters struct {
 	Queue       []string
 	Destination []string
 	Kind        []string
+
+	QueryFor []string
+	Status   []string
+
+	TaskName []string
 }
 
 type CeleryTask struct {
 	kind   string
 	status string
+	name   string
 }
 
 type CeleryTasks interface {
 	GetKind() string
 	GetStatus() string
-	Set(string, string)
+	GetName() string
 }
 
 func (r *CeleryTask) GetKind() string {
@@ -52,7 +63,6 @@ func (r *CeleryTask) GetStatus() string {
 	return r.status
 }
 
-func (r *CeleryTask) Set(kind, status string) {
-	r.kind = kind
-	r.status = status
+func (r *CeleryTask) GetName() string {
+	return r.name
 }

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

@@ -2,7 +2,6 @@ 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"
@@ -21,6 +20,7 @@ func BuildQueryRangeParams(messagingQueue *MessagingQueue, queryContext string)
 	queueType := KafkaQueue
 
 	chq, err := BuildClickHouseQuery(messagingQueue, queueType, queryContext)
+
 	if err != nil {
 		return nil, err
 	}
@@ -80,10 +80,9 @@ func buildBuilderQueriesProducerBytes(
 			Type:     v3.AttributeKeyType("Gauge"),
 			IsColumn: true,
 		},
-		AggregateOperator: v3.AggregateOperatorAvg,
-		Temporality:       v3.Unspecified,
-		TimeAggregation:   v3.TimeAggregationAvg,
-		SpaceAggregation:  v3.SpaceAggregationAvg,
+		Temporality:      v3.Unspecified,
+		TimeAggregation:  v3.TimeAggregationAvg,
+		SpaceAggregation: v3.SpaceAggregationAvg,
 		Filters: &v3.FilterSet{
 			Operator: "AND",
 			Items: []v3.FilterItem{
@@ -126,6 +125,88 @@ func buildBuilderQueriesProducerBytes(
 	return bq, nil
 }
 
+func CeleryClickHouseQuery(
+	messagingQueue *MessagingQueue,
+	queryContext string, filters *QueueFilters,
+) (*v3.QueryRangeParamsV3, error) {
+
+	unixMilliStart := messagingQueue.Start / 1000000
+	unixMilliEnd := messagingQueue.End / 1000000
+
+	var cq *v3.CompositeQuery
+
+	switch queryContext {
+	case "celeryoverview":
+
+		metrics := ""
+
+		if filters.QueryFor[0] == "worker" {
+			metrics = "flower_worker_online"
+		} else if filters.QueryFor[0] == "tasks" {
+			metrics = "flower_worker_number_of_currently_executing_tasks"
+		}
+
+		query, err := buildCeleryOverviewQuery(metrics, queryContext, unixMilliStart, unixMilliEnd)
+		if err != nil {
+			return nil, err
+		}
+		cq = &v3.CompositeQuery{
+			QueryType:      v3.QueryTypeBuilder,
+			BuilderQueries: query,
+			PanelType:      v3.PanelTypeGraph,
+			FillGaps:       false,
+		}
+	}
+	queryRangeParams := &v3.QueryRangeParamsV3{
+		Start:          unixMilliStart,
+		End:            unixMilliEnd,
+		Step:           defaultStepInterval,
+		CompositeQuery: cq,
+		Version:        "v4",
+		FormatForWeb:   true,
+	}
+
+	return queryRangeParams, nil
+
+}
+
+func buildCeleryOverviewQuery(metrics string, queryContext string, unixMilliStart, unixMilliEnd int64) (map[string]*v3.BuilderQuery, error) {
+	bq := make(map[string]*v3.BuilderQuery)
+
+	chq := &v3.BuilderQuery{
+		QueryName:  queryContext,
+		DataSource: v3.DataSourceMetrics,
+
+		StepInterval:      common.MinAllowedStepInterval(unixMilliStart, unixMilliEnd),
+		AggregateOperator: v3.AggregateOperator("latest"),
+
+		AggregateAttribute: v3.AttributeKey{
+			Key:      metrics, // flower_worker_online
+			DataType: v3.AttributeKeyDataTypeFloat64,
+			Type:     v3.AttributeKeyType("Gauge"),
+			IsColumn: true,
+			IsJSON:   false,
+		},
+
+		Temporality:      v3.Unspecified,
+		TimeAggregation:  v3.TimeAggregationAnyLast,
+		SpaceAggregation: v3.SpaceAggregationAvg,
+
+		Expression: queryContext,
+		ReduceTo:   v3.ReduceToOperatorAvg,
+
+		GroupBy: []v3.AttributeKey{
+			{
+				Key:      "worker",
+				DataType: v3.AttributeKeyDataTypeString,
+				Type:     v3.AttributeKeyTypeTag,
+			},
+		},
+	}
+	bq[queryContext] = chq
+	return bq, nil
+}
+
 func buildBuilderQueriesNetwork(
 	unixMilliStart, unixMilliEnd int64,
 	attributeCache *Clients,
@@ -321,14 +402,43 @@ func BuildQRParamsWithCache(
 	return queryRangeParams, err
 }
 
+func GetCeleryFilters(variables map[string]string) (*QueueFilters, error) {
+	filters := getFilters(variables)
+	if len(filters.QueryFor) != 1 || len(filters.Status) != 1 {
+		return nil, fmt.Errorf("query_for and status not found in the request")
+	}
+	return filters, nil
+}
+
 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"]),
+	var filters QueueFilters
+
+	if val, ok := variables["service_name"]; ok && val != "" {
+		filters.ServiceName = parseFilter(val)
+	}
+	if val, ok := variables["span_name"]; ok && val != "" {
+		filters.SpanName = parseFilter(val)
+	}
+	if val, ok := variables["queue"]; ok && val != "" {
+		filters.Queue = parseFilter(val)
+	}
+	if val, ok := variables["destination"]; ok && val != "" {
+		filters.Destination = parseFilter(val)
+	}
+	if val, ok := variables["kind"]; ok && val != "" {
+		filters.Kind = parseFilter(val)
 	}
+	if val, ok := variables["query_for"]; ok && val != "" {
+		filters.QueryFor = parseFilter(val)
+	}
+	if val, ok := variables["status"]; ok && val != "" {
+		filters.Status = parseFilter(val)
+	}
+	if val, ok := variables["task_name"]; ok && val != "" {
+		filters.TaskName = parseFilter(val)
+	}
+
+	return &filters
 }
 
 // parseFilter splits a comma-separated string into a []string.