package controllers import ( "apiserver/backend" "apiserver/common" "apiserver/models" "apiserver/services" "apiserver/store" "apiserver/utils" logger "apiserver/utils" "errors" "strings" "github.com/gin-gonic/gin" "go.uber.org/zap" ) type QueueController struct{} func (q QueueController) GetQueuesInfo(ctx *gin.Context) { // Get all queues logger.Debug("Enter GetQueuesInfo() function from ", zap.String("client_ip", ctx.ClientIP())) // get logon user token := ctx.GetHeader("Authorization") logonUser, err := GetUsernameFromToken(token) if err != nil { ReturnError(ctx, common.ERROR_CODE, err.Error(), nil) return } var fields []string filters := make([]utils.FilterCondition, 0) sortOrders := make([]utils.SortOrder, 0) // ---- check parameters ---- for param := range ctx.Request.URL.Query() { if !utils.ContainsField(utils.SupportedQueryParams, param) { err := errors.New("invalid queues info query parameters of : " + param) logger.Error(err.Error()) ReturnError(ctx, common.INVALID_PARAMS_CODE, common.INVALID_PARAMS_MSG+": "+err.Error(), nil) return } } // Get query parameters filterStr := ctx.Query("filter") fieldsStr := ctx.Query("fields") sortStr := ctx.Query("sort") // sort=queue_name:asc,submit_time:desc pagetokenStr := ctx.Query("page_token") // page_token=Aafjakjf0023A limitStr := ctx.Query("limit") // eg: limit=100 logger.Debug("Queues query values are:", zap.String("fieldsStr", fieldsStr), zap.String("filterStr", filterStr), zap.String("sortStr", sortStr), zap.String("limitStr", limitStr), zap.String("pagetokenStr", pagetokenStr)) // ---- check limit value ---- limit, err := utils.ParseLimit(limitStr) if err != nil { logger.Error(err.Error()) ReturnError(ctx, common.INVALID_PARAMS_CODE, err.Error(), nil) return } // ---- check filter ---- if filterStr != "" { filters = utils.ParseFilters(filterStr) if filters == nil { err := errors.New("filter " + filterStr + " has syntax error") logger.Error(err.Error()) ReturnError(ctx, common.INVALID_PARAMS_CODE, common.INVALID_PARAMS_MSG+": "+err.Error(), nil) return } } //filtercds := make([]utils.Condition, len(filters)) for _, filter := range filters { // check the filter is supported. if !utils.ContainsField(models.QUEUE_SUPPORT_FILTERS, filter.Field) { err := errors.New("unsupported filter: " + filter.Field) logger.Error(err.Error()) ReturnError(ctx, common.INVALID_PARAMS_CODE, common.INVALID_PARAMS_MSG+": "+err.Error(), nil) return } } var queues models.QueueInfo filtercds, err := utils.ConvertToNewFilter(filters, &queues) if err != nil { logger.Error(err.Error()) ReturnError(ctx, common.INVALID_PARAMS_CODE, common.INVALID_PARAMS_MSG+": "+err.Error(), nil) return } // ---- check fields ---- // get fields and do check, if not set, use default display columns if fieldsStr != "" { fields = utils.ParseFields(fieldsStr) found := false // verify the fields is in supported list supportedFields := utils.GetSupportedFields(models.QueueInfo{}) logger.Debug("queue supportedFields: %v", zap.Any("supportedFields", supportedFields)) for _, field := range fields { if !utils.ContainsField(supportedFields, field) { err := errors.New("unsupported query field: " + field) logger.Error(err.Error()) ReturnError(ctx, common.INVALID_PARAMS_CODE, common.INVALID_PARAMS_MSG+": "+err.Error(), nil) return } // if user not specify queue_name in fields, add queue_name as default field. if field == "queue_name" { found = true } } if !found { logger.Debug("user doesn't specify queue name in query fields, add it.") fields = append(fields, "queue_name") } } else { fields = models.QUEUE_DEFAULT_COLUMN } // ---- check sort ---- // handle sort, when no sortStr specified, need add the primary key of queue_info // table into sortStr as default value if sortStr == "" { logger.Debug("sort is not specified, set sort=queue_name:asc as default") sortStr = "queue_name:asc" } // parse sortStr sorts, err := utils.ParseSortParam(sortStr, models.QUEUE_SUPPORT_SORT) if err != nil { logger.Error(err.Error()) ReturnError(ctx, common.INVALID_PARAMS_CODE, common.INVALID_PARAMS_MSG+": "+err.Error(), nil) return } // ---- check paging --- if pagetokenStr != "" { pagetokenStr, err = utils.ParsePageToken(pagetokenStr) if err != nil { logger.Error(err.Error()) ReturnError(ctx, common.INVALID_PARAMS_CODE, common.INVALID_PARAMS_MSG+": "+err.Error(), nil) return } } //the primary key of queue_info table queue_name should always be added into sortStr if user doesn't specify. found := false for _, sortent := range sorts { if sortent.Column == "queue_name" { found = true break } } // user not specify the queue_name, only append queue_name at the end. if !found { sorts = append(sorts, utils.SortOrder{ Column: "queue_name", Ascending: true, }) } //check all sort column is in fields err = utils.CheckAllMapKeyInList(sorts, fields) if err != nil { logger.Error(err.Error()) ReturnError(ctx, common.INVALID_PARAMS_CODE, common.INVALID_PARAMS_MSG+": "+err.Error(), nil) return } sortOrders = sorts /* //fill QueryOptions qqueryNew := utils.QueryOptions{ Conditions: filtercds, Orders: sortOrders, Limits: limit, Bookmark: pagetokenStr, } */ queryOpts, err := utils.BuildQueryOptions(filtercds, nil, sortOrders, limit, pagetokenStr) if err != nil { logger.Error(err.Error()) ReturnError(ctx, common.ERROR_CODE, common.ERROR_QUERY_QUEUEINFO_MSG+": "+err.Error(), nil) return } qqueryReply, pagetoken, err := services.QueryData(services.TableQueueInfo, queryOpts, fields) if err != nil { logger.Error("Failed to query queue info.", zap.Error(err)) ReturnError(ctx, common.ERROR_CODE, common.ERROR_QUERY_QUEUEINFO_MSG+": "+err.Error(), nil) return } // check rbac accessRow := make([]utils.OrderMap, 0) for _, row := range qqueryReply { if queue_name, exist := row.Data["queue_name"]; exist { queueStr, ok := queue_name.(string) if !ok { logger.Error("queue_name is not a string", zap.Any("queue_name", queue_name)) ReturnError(ctx, common.ERROR_CODE, "queue_name is not a string", nil) return } queuePermission, err := GetRequireResourcePermission(logonUser, "queue:view") if err != nil { ReturnError(ctx, common.ERROR_CODE, common.ERROR_ACCESS_FORBIDDEN_MSG, nil) return } if queuePermission.DomainType == models.DomainTypeQueue { queueDomainValues := queuePermission.DomainValues if utils.ContainsField(queueDomainValues, queueStr) { accessRow = append(accessRow, row) } } else if queuePermission.DomainType == models.DomainTypeAll { accessRow = append(accessRow, row) } } } logger.Debug("After call GetRequireResourcePermission for all queues info", zap.Error(err), zap.Any("accessRow", accessRow)) if err != nil && len(accessRow) == 0 { ReturnError(ctx, common.ERROR_CODE, common.ERROR_ACCESS_FORBIDDEN_MSG, nil) return } logger.Debug("the queue info query reply:", zap.Any("qqueryReply", qqueryReply), zap.String("pagetoken", pagetoken)) //convert to readable string. queryResult := processQueueData(accessRow, pagetoken) ReturnSuccess(ctx, common.SUCCESS_CODE, common.SUCCESS_MSG, queryResult) } func processQueueData(rows []utils.OrderMap, page_token string) utils.QueryResult { for _, row := range rows { if v, exist := row.Data["queue_status"]; exist && v != nil { stateVal, ok := utils.IntOrInt64(v) if !ok { stateVal = -1 } if stateVal >= 0 { row.Data["queue_status"] = formatQueueStatus(int64(stateVal)) } else { row.Data["queue_status"] = "UNKNOWN" } } // Process JSONB fields based on predefined list for _, jsonbField := range models.JSONBQFields { if v, exist := row.Data[jsonbField]; exist { if strValue, ok := v.(string); ok { if jsonbData, err := utils.ParseJSONB(strValue); err == nil { row.Data[jsonbField] = jsonbData } else { logger.Error("Failed to parse queue info JSONB field.", zap.Error(err)) } } } } //utils.FormatRowsOutput(row.Data) } return utils.QueryResult{ Rows: rows, Count: len(rows), Pagetoken: page_token, } } func (q QueueController) GetQueueByName(ctx *gin.Context) { logger.Debug("Enter GetQueueByName() function from ", zap.String("client_ip", ctx.ClientIP())) // get logon user token := ctx.GetHeader("Authorization") logonUser, err := GetUsernameFromToken(token) if err != nil { ReturnError(ctx, common.ERROR_CODE, err.Error(), nil) return } // ---- check parameters ---- for param := range ctx.Request.URL.Query() { if !utils.ContainsField(utils.SupportedQueryParams, param) { err := errors.New("invalid specific queue info query parameters of : " + param) logger.Error(err.Error()) ReturnError(ctx, common.INVALID_PARAMS_CODE, common.INVALID_PARAMS_MSG+": "+err.Error(), nil) return } } queueNamesStr := ctx.Param("queuename") fieldsStr := ctx.Query("fields") logger.Debug("origin query string queuename, fieldsStr values are:", zap.String("queuename", queueNamesStr), zap.String("fields", fieldsStr)) //add jobid list as filter. queueNames := strings.Split(queueNamesStr, ",") // queue_name: q1,q2,q3 mergeQqueryReply := make([]utils.OrderMap, 0) var page_token = "" for _, queuename := range queueNames { queuename = strings.TrimSpace(queuename) // check rbac queuePermission, rbacErr := GetRequireResourcePermission(logonUser, "queue:view") if rbacErr != nil { ReturnError(ctx, common.ERROR_AUTH_CODE, common.ERROR_ACCESS_FORBIDDEN_MSG, nil) return } if queuePermission.DomainType == models.DomainTypeQueue { queueDomainValues := queuePermission.DomainValues if !utils.ContainsField(queueDomainValues, queuename) { err := errors.New("access forbidden for queue: " + queuename) logger.Error(err.Error()) ReturnError(ctx, common.ERROR_AUTH_CODE, common.ERROR_ACCESS_FORBIDDEN_MSG, nil) return } } filters := make([]utils.FilterCondition, 0) sortOrders := make([]utils.SortOrder, 0) filters = append(filters, utils.FilterCondition{ Field: "queue_name", Operator: "eq", Value: queuename, }) //filtercds := make([]utils.Condition, len(filters)) for _, filter := range filters { // check the filter is supported. if !utils.ContainsField(models.QUEUE_SUPPORT_FILTERS, filter.Field) { err := errors.New("unsupported filter: " + filter.Field) logger.Error(err.Error()) ReturnError(ctx, common.INVALID_PARAMS_CODE, common.INVALID_PARAMS_MSG+": "+err.Error(), nil) return } } var queue models.QueueInfo var err error filtercds, err := utils.ConvertToNewFilter(filters, &queue) if err != nil { logger.Error(err.Error()) ReturnError(ctx, common.INVALID_PARAMS_CODE, common.INVALID_PARAMS_MSG+": "+err.Error(), nil) return } var fields []string if fieldsStr != "" { fields = utils.ParseFields(fieldsStr) found := false // verify the fields is in supported list supportedFields := utils.GetSupportedFields(models.QueueInfo{}) logger.Debug("queue supportedFields: %v", zap.Any("supportedFields", supportedFields)) for _, field := range fields { if !utils.ContainsField(supportedFields, field) { err := errors.New("unsupported query field: " + field) logger.Error(err.Error()) ReturnError(ctx, common.INVALID_PARAMS_CODE, common.INVALID_PARAMS_MSG+": "+err.Error(), nil) return } // if user not specify queue_name in fields, add queue_name as default field. if field == "queue_name" { found = true } } if !found { logger.Debug("user doesn't specify queue_name in query fields, add it.") fields = append(fields, "queue_name") } } else { // default display columns fields = models.QUEUE_DEFAULT_COLUMN } // add a default order: sort=queue_name:asc sortOrders = append(sortOrders, utils.SortOrder{ Column: "queue_name", Ascending: true, }) /* qqueryNew := utils.QueryOptions{ Conditions: filtercds, Orders: sortOrders, Limits: utils.PER_PAGE_LIMIT, Bookmark: "", // for one queue, no need page_token } */ queryOpts, err := utils.BuildQueryOptions(filtercds, nil, sortOrders, utils.PER_PAGE_LIMIT, "") if err != nil { logger.Error(err.Error()) ReturnError(ctx, common.ERROR_CODE, common.ERROR_QUERY_JOBINFO_MSG+": "+err.Error(), nil) return } qqueryReply, page_token, err := services.QueryData(services.TableQueueInfo, queryOpts, fields) if err != nil { logger.Error("Failed to query the specific queue info.", zap.Error(err)) ReturnError(ctx, common.ERROR_CODE, common.ERROR_QUERY_JOBINFO_MSG+": "+err.Error(), nil) return } logger.Debug("the specific queue query reply:", zap.Any("qqueryReply", qqueryReply), zap.String("pagetoken", page_token)) mergeQqueryReply = append(mergeQqueryReply, qqueryReply...) } // end for queue_name logger.Debug("the specific queue query reply:", zap.Any("qqueryReply", mergeQqueryReply)) //convert job_status to readable string. queryResult := processQueueData(mergeQqueryReply, page_token) ReturnSuccess(ctx, common.SUCCESS_CODE, common.SUCCESS_MSG, queryResult) } func (q *QueueController) ControlQueue(ctx *gin.Context) { logger.Debug("Enter ControlQueue() function") // get the logon user token := ctx.GetHeader("Authorization") submitUser, err := GetUsernameFromToken(token) if err != nil { ReturnError(ctx, common.ERROR_CODE, err.Error(), nil) return } var req models.QueueControlRequest // use strict json format check if err := utils.BindJSONStrict(ctx, &req); err != nil { logger.Error("Invalid request parameters", zap.Error(err)) ReturnError(ctx, common.ERROR_CODE, err.Error(), nil) return } allFlag, err := preCheckQueueControlRequest(&req) if err != nil { logger.Error("Failed to send queue control request", zap.Error(err)) ReturnError(ctx, common.ERROR_CODE, err.Error(), nil) return } baseAction := req.ControlAction logger.Debug("ControlQueue baseAction is ", zap.Any("baseAction", baseAction)) var actionType int actionType, exists := models.ActionMap[baseAction] if !exists { logger.Error("Unsupported action type", zap.String("baseAction", baseAction)) ReturnError(ctx, common.INVALID_PARAMS_CODE, "Unsupported action type: "+baseAction, nil) return } logger.Debug("ControlQueue actionType is ", zap.Any("actionType", actionType)) successQueues := []map[string]interface{}{} failedQueues := []map[string]interface{}{} validatedSuccessQueues, validatedFailedQueues := validateQueue(actionType, req.QueueName) if allFlag { queueNames := store.GetAllQueueNames() validatedSuccessQueues, validatedFailedQueues = validateQueue(actionType, queueNames) } logger.Debug("validatedSuccessQueues:", zap.Any("validatedSuccessQueues", validatedSuccessQueues)) for _, queuename := range validatedSuccessQueues { // check rbac queuePermission, err := GetRequireResourcePermission(submitUser, "queue:control") if err != nil { logger.Error(err.Error()) ReturnError(ctx, common.ERROR_AUTH_CODE, common.ERROR_ACCESS_FORBIDDEN_MSG, nil) return } if queuePermission.DomainType == models.DomainTypeQueue { queueDomainValues := queuePermission.DomainValues if !utils.ContainsField(queueDomainValues, queuename) { failedQueues = append(failedQueues, map[string]interface{}{ "queue_name": queuename, "msg": "queue is not allowed to be controlled", }) continue } } successQueues = append(successQueues, map[string]interface{}{ "queue_name": queuename, "msg": "queue is being controlled", }) } for queuename, reason := range validatedFailedQueues { failedQueues = append(failedQueues, map[string]interface{}{ "queue_name": queuename, "msg": reason, }) } for _, queue := range successQueues { queuename := queue["queue_name"].(string) err = backend.ControlQueue(queuename, actionType) if err != nil { logger.Error("Failed to send queue control", zap.Error(err)) failedQueues = append(failedQueues, map[string]interface{}{ "queue_name": queuename, "msg": "failed to control queue: " + err.Error(), }) } } response, successMsg := BuildBatchOperationResponse(baseAction, "Queue", successQueues, failedQueues) overallStatus := response["status"].(string) logger.Info("Finished processing queue control request", zap.String("action", baseAction), zap.String("overallStatus", overallStatus), zap.Int("succeededCount", len(successQueues)), zap.Int("failedCount", len(failedQueues)), zap.String("client_ip", ctx.ClientIP())) ReturnSuccess(ctx, common.ACCEPTED, successMsg, response) return } func preCheckQueueControlRequest(req *models.QueueControlRequest) (bool, error) { logger.Debug("Enter preCheckQueueControlRequest") if err := utils.ValidateControlAction(req.ControlAction, models.SupportedQueueActions); err != nil { return false, err } req.ControlAction = strings.ToLower(strings.TrimSpace(req.ControlAction)) isAll := req.QueueName != nil && len(req.QueueName) == 1 && req.QueueName[0] == "all" if req.Comment != "" { if len(req.Comment) > 512 { return false, errors.New("comment length exceeds the limit of 512 characters") } } return isAll, nil } func validateQueue(control int, queuenames []string) (successQueues []string, failedQueues map[string]string) { logger.Debug("Validating queuenames: %v", zap.Any("queuenames", queuenames)) successQueues = []string{} failedQueues = make(map[string]string) if len(queuenames) == 0 { logger.Error("queuenames cannot be empty") return } for _, queuename := range queuenames { queuename = strings.TrimSpace(queuename) if queuename == "" { logger.Error("queuenames cannot be empty") continue } if store.Queues.Has(queuename) { logger.Debug("Found queuename in store", zap.String("queuename", queuename)) successQueues = append(successQueues, queuename) } else { logger.Error("Invalid queuename", zap.String("queuename", queuename)) failedQueues[queuename] = common.ERROR_QUEUE_NOT_FOUND_MSG } } return } func isQueueControlAllowed(controlType int, queueStateStr string) bool { parts := strings.Split(queueStateStr, ":") mainState := parts[0] secondState := parts[1] switch controlType { case models.QUEUE_CLOSE: return mainState == "open" case models.QUEUE_OPEN: return mainState == "closed" case models.QUEUE_ACT: return secondState == "inactive_adm" case models.QUEUE_INACT: return secondState != "inactive_adm" default: logger.Error("Unsupported control type", zap.Int("controlType", controlType)) return false } } func formatQueueStatus(state int64) string { var status string if state&(int64(models.QUEUESTAT["CLOSED"])) != 0 { // CLOSED status = "closed" } else { status = "open" } if state&int64(models.QUEUESTAT["INACTIVE_ADM"]) != 0 { // INACT_ADMIN status += ":inactive_adm" } else if state&int64(models.QUEUESTAT["INACTIVE_WIN"]) != 0 { // INACT_WINDOW status += ":inactive_win" } else { status += ":active" } return status }