feat: Adds celery performance APIs

pkg/query-service/app/http_handler.go

@@ -4987,13 +4987,114 @@ 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, apiErr, nil)
+		return
+	}
+
+	queryRangeParams, err := mq.CeleryClickHouseQuery(messagingQueue, "celeryoverview", filters)
+
+	if err != nil {
+		zap.L().Error(err.Error())
+		RespondError(w, apiErr, nil)
+		return
+	}
+	if err := validateQueryRangeParamsV3(queryRangeParams); err != nil {
+		zap.L().Error(err.Error())
+		RespondError(w, apiErr, 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) {
-	// TODO: Implement celery tasks logic for both state and list types
+
 }
 
 func (aH *APIHandler) getCeleryPerformance(w http.ResponseWriter, r *http.Request) {
-	// TODO: Implement celery performance logic for error, rate, and latency 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, apiErr, nil)
+		return
+	}
+
+	queryRangeParams, err := mq.CeleryClickHouseQuery(messagingQueue, "celeryperformance", filters)
+
+	if err != nil {
+		zap.L().Error(err.Error())
+		RespondError(w, apiErr, nil)
+		return
+	}
+	if err := validateQueryRangeParamsV3(queryRangeParams); err != nil {
+		zap.L().Error(err.Error())
+		RespondError(w, apiErr, 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
+	}
+
+	resp := v3.QueryRangeResponse{
+		Result: resultFetchLatency,
+	}
+	aH.Respond(w, resp)
 }

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,25 @@ type QueueFilters struct {
 	Queue       []string
 	Destination []string
 	Kind        []string
+
+	QueryFor []string
+	Status   []string
+
+	TaskName []string
+
+	LatencyType []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 +65,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
 }